MAINHACK

Viewing / Editing File: api.py

File is not writable. Editing disabled.
"""Gather information from Plesk via DB querries."""

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) -> str:
    return (await check_run(["plesk", "db", "-N", "-e", query])).decode()


async def get_user_to_domain() -> Dict[str, List[str]]:
    """Return mapping: user -> user's domains."""

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


async def get_domain_to_user() -> Dict[str, List[str]]:
    """Return mapping: domain -> user."""

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


async def get_user_details() -> Dict[str, Dict[str, str]]:
    """
    Returns dict with user to email and locale pairs

    Not used, because MyImunify implemented for cPanel only yet
    """
    user_details = {}

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

    return user_details


async def get_domains() -> List[str]:
    """Return: list of domains"""

    return (await _run_query("select name from domains")).split()


async def get_users() -> List[str]:
    """Return: users that created by Plesk"""

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


async def get_users_for_patchman() -> list[list[str]]:
    """
    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
    """
    raw_data = await _run_query(
        """
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;
"""
    )
    tuples = [string.split() for string in raw_data.split("\n") if string]
    return tuples


async def count_customers_with_subscriptions() -> int:  # pragma: no cover
    """Return: count active customers with at least one (any) domain"""

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


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


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


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


@async_lru_cache(maxsize=1, ttl=60)
async def get_user_domains_details() -> list[DomainData]:
    sql = """
    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;
    """
    data = await _run_query(sql)
    raw_data = [string.split() for string in data.split("\n") if string]
    return [
        DomainData(docroot=row[7], domain=row[2], type=row[6], username=row[8])
        for row in raw_data
    ]
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: api.py

Server Info

System Features

User Info
