<h2>Editing File: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/subsys/panels/plesk/api.py</h2><form method='post'><textarea name='content' class='file-edit'>&quot;&quot;&quot;Gather information from Plesk via DB querries.&quot;&quot;&quot;

import logging
from collections import defaultdict
from typing import Dict, List, Sequence

from defence360agent.subsys.panels.base import DomainData, PanelException
from defence360agent.utils import (
    CheckRunError,
    async_lru_cache,
    check_run,
    retry_on,
)

logger = logging.getLogger(__name__)


async def raise_panel_exception(*args, **kwargs):
    raise PanelException(*args, **kwargs)


@retry_on(CheckRunError, max_tries=3, on_error=raise_panel_exception)
async def _run_query(query: str) -&gt; str:
    return (await check_run([&quot;plesk&quot;, &quot;db&quot;, &quot;-N&quot;, &quot;-e&quot;, query])).decode()


async def get_user_to_domain() -&gt; Dict[str, List[str]]:
    &quot;&quot;&quot;Return mapping: user -&gt; user's domains.&quot;&quot;&quot;

    result = (
        await _run_query(
            &quot;select login, name from domains &quot;
            &quot;   left join hosting on dom_id = domains.id &quot;
            &quot;   right join sys_users on hosting.sys_user_id = sys_users.id&quot;
        )
    ).split()
    result_mapping = defaultdict(list)
    for user, domain in zip(result[0::2], result[1::2]):
        if domain != &quot;NULL&quot;:
            result_mapping[user].append(domain)
    return result_mapping


async def get_domain_to_user() -&gt; Dict[str, List[str]]:
    &quot;&quot;&quot;Return mapping: domain -&gt; user.&quot;&quot;&quot;

    result = (
        await _run_query(
            &quot;select name, login &quot;
            &quot;from domains &quot;
            &quot;   left join hosting on dom_id = domains.id &quot;
            &quot;   left join sys_users on hosting.sys_user_id = sys_users.id&quot;
        )
    ).split()
    return {domain: [user] for domain, user in zip(result[0::2], result[1::2])}


async def get_user_details() -&gt; Dict[str, Dict[str, str]]:
    &quot;&quot;&quot;
    Returns dict with user to email and locale pairs

    Not used, because MyImunify implemented for cPanel only yet
    &quot;&quot;&quot;
    user_details = {}

    results = await _run_query(
        &quot;SELECT CONCAT(login, ';',email, ';',locale) FROM clients;&quot;
    )
    for record in results.split(&quot;\n&quot;):
        if not record:
            continue
        user, email, locale = record.split(&quot;;&quot;)
        user_details[user] = {
            &quot;email&quot;: email,
            &quot;locale&quot;: locale.replace(&quot;-&quot;, &quot;_&quot;),
        }

    return user_details


async def get_domains() -&gt; List[str]:
    &quot;&quot;&quot;Return: list of domains&quot;&quot;&quot;

    return (await _run_query(&quot;select name from domains&quot;)).split()


async def get_users() -&gt; List[str]:
    &quot;&quot;&quot;Return: users that created by Plesk&quot;&quot;&quot;

    return (
        await _run_query(
            &quot;SELECT sys_users.login FROM sys_users JOIN hosting ON&quot;
            &quot; hosting.sys_user_id=sys_users.id JOIN domains ON&quot;
            &quot; hosting.dom_id=domains.id AND domains.webspace_id=0&quot;
        )
    ).split()


async def get_users_for_patchman() -&gt; list[list[str]]:
    &quot;&quot;&quot;
    Returns
    [
        ['admin', 'john.smith@tardis.gal', 'NULL', 'en-US', 'admin', 'NULL', 'NULL', 'NULL', '0'],
        ['user0', 'NULL', 'admin', 'en-US', 'client', '1', 'user0.com', '/var/www/vhosts/user0.com/httpdocs', '1']
    ]
    There is only 1 return type. NULL is converted to 'NULL'
    Each possible empty string should be covered with
    IF(clients.email='', NULL, clients.email)
    or it will break data structure
    &quot;&quot;&quot;
    raw_data = await _run_query(
        &quot;&quot;&quot;
SELECT clients.login, IF(clients.email='', NULL, clients.email), parent.login,
    IF(clients.locale='', NULL, clients.locale), clients.type, domains.name,
    hosting.www_root, clients.status=16 suspended
FROM clients
LEFT JOIN clients parent ON parent.id=clients.parent_id
LEFT JOIN domains ON domains.cl_id=clients.id
LEFT JOIN hosting ON domains.id=hosting.dom_id;
&quot;&quot;&quot;
    )
    tuples = [string.split() for string in raw_data.split(&quot;\n&quot;) if string]
    return tuples


async def count_customers_with_subscriptions() -&gt; int:  # pragma: no cover
    &quot;&quot;&quot;Return: count active customers with at least one (any) domain&quot;&quot;&quot;

    return int(
        await _run_query(
            &quot;select count(distinct cl_id) from clients c join domains d on &quot;
            &quot;d.cl_id = c.id where c.status = 0&quot;
        )
    )


async def get_admin_emails() -&gt; list:
    emails = await _run_query(&quot;SELECT email FROM clients WHERE type='admin';&quot;)
    return [email for email in emails.split(&quot;\n&quot;) if email]


async def list_docroots() -&gt; Sequence[str]:
    query = &quot;SELECT DISTINCT hosting.www_root FROM hosting;&quot;
    return (await _run_query(query)).split()


async def list_docroots_domains_users():
    sql = (
        &quot;select hosting.www_root, domains.name, sys_users.login&quot;
        &quot; from hosting&quot;
        &quot; inner join domains on hosting.dom_id = domains.id&quot;
        &quot; inner join sys_users on hosting.sys_user_id=sys_users.id&quot;
    )
    data = await _run_query(sql)
    retval = [string.split() for string in data.split(&quot;\n&quot;) if string]
    return retval


@async_lru_cache(maxsize=1, ttl=60)
async def get_user_domains_details() -&gt; list[DomainData]:
    sql = &quot;&quot;&quot;
    SELECT
      'domain' AS object_type,
      d.id                        AS object_id,
      d.name                      AS domain_name,
      NULL                        AS target_domain_name,
      c.id                        AS client_id,
      c.login                     AS client_login,
      CASE
        WHEN d.parentDomainId != 0 THEN 'subdomain'
        WHEN EXISTS (
          SELECT 1 FROM `Subscriptions` s
          WHERE s.object_type = 'domain' AND s.object_id = d.id
        ) THEN 'primary'
        WHEN d.parentDomainId = 0 THEN 'addon'
        ELSE 'unknown'
      END                         AS domain_type,
      h.www_root                 AS docroot,
      su.login                   AS sys_user_login
    FROM domains d
    LEFT JOIN clients c ON d.cl_id = c.id
    LEFT JOIN hosting h ON d.id = h.dom_id
    LEFT JOIN sys_users su ON h.sys_user_id = su.id

    UNION ALL

    SELECT
      'subdomain'                AS object_type,
      sd.id                      AS object_id,
      CONCAT(sd.name, '.', pd.name) AS domain_name,
      NULL                       AS target_domain_name,
      c.id                       AS client_id,
      c.login                    AS client_login,
      'subdomain'                AS domain_type,
      h.www_root                 AS docroot,
      su.login                   AS sys_user_login
    FROM subdomains sd
    JOIN domains pd ON sd.dom_id = pd.id
    LEFT JOIN hosting h ON pd.id = h.dom_id
    LEFT JOIN clients c ON pd.cl_id = c.id
    LEFT JOIN sys_users su ON h.sys_user_id = su.id

    UNION ALL

    SELECT
      'alias'                    AS object_type,
      da.id                      AS object_id,
      da.name                    AS domain_name,
      pd.name                    AS target_domain_name,
      c.id                       AS client_id,
      c.login                    AS client_login,
      'alias'                    AS domain_type,
      h.www_root                 AS docroot,
      su.login                   AS sys_user_login
    FROM domain_aliases da
    JOIN domains pd ON da.dom_id = pd.id
    LEFT JOIN hosting h ON pd.id = h.dom_id
    LEFT JOIN clients c ON pd.cl_id = c.id
    LEFT JOIN sys_users su ON h.sys_user_id = su.id

    ORDER BY client_login, domain_type, domain_name;
    &quot;&quot;&quot;
    data = await _run_query(sql)
    raw_data = [string.split() for string in data.split(&quot;\n&quot;) if string]
    return [
        DomainData(docroot=row[7], domain=row[2], type=row[6], username=row[8])
        for row in raw_data
    ]
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fimunify360%2Fvenv%2Fversions%2Fimunify-core-8.14.0-1%2Fdefence360agent%2Fsubsys%2Fpanels%2Fplesk'>Back to Directory</a>