Na}{

File Content: `imunify_patch_id.py`

"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
See the GNU General Public License for more details.


You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.

Copyright © 2019 Cloud Linux Software Inc.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import asyncio
import json
import logging
import pwd
import uuid
from collections import defaultdict
from functools import lru_cache
from itertools import islice
from pathlib import Path
from typing import Any
from urllib.parse import urlencode

from defence360agent.internals.iaid import IndependentAgentIDAPI
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.utils import (
    async_lru_cache,
    log_error_and_ignore,
    safe_fileops,
)
from imav.contracts.config import IMUNIFY_PATCH_PURCHASE_URL
from imav.malwarelib.api.vulnerability import VulnerabilityAPI
from imav.malwarelib.config import VulnerabilityHitStatus
from imav.malwarelib.model import VulnerabilityHit

IMUNIFY_PATCH_ID_FILE = ".imunify_patch_id"
PURCHASE_URL_MAX_WEBSITES = 5  # Take only first 5 domains for better display in UI. Not used anywhere else.
MAX_SEVERITY_COUNT = 10

logger = logging.getLogger(__name__)


class ImunifyPatchIdError(Exception):
    pass


ImunifyPatchUserId = str


async def ensure_id_file(username: str) -> ImunifyPatchUserId:
    """Ensure the Imunify.Patch ID file exists for the given user.

    This function checks if the Imunify.Patch ID file exists in the user's
    home directory. If it does not exist, it generates a new ID, writes it
    to the file, and returns the ID. If the file already exists, it reads
    and returns the existing ID.

    Args:
        username (str): The username for which to ensure the ID file.

    Returns:
        ImunifyPatchUserId: The Imunify.Patch user ID.
    """
    id_file = await _get_id_file(username)
    if _id := _read_id_file(id_file):
        return _id
    _id = _generate_id()
    await _write_id_file(id_file, _id)
    return _id


@log_error_and_ignore()
async def get_imunify_patch_id(username: str) -> ImunifyPatchUserId:
    async with get_lock(username):
        return await ensure_id_file(username)


@lru_cache(maxsize=None)
def get_lock(username: str):
    return asyncio.Lock()


async def _get_id_file(username: str) -> Path:
    """Get a file with Imunify.Patch user id and create it if does not exist"""

    try:
        user_pwd = pwd.getpwnam(username)
    except KeyError as e:
        logger.error(f"No such user: {username}")
        raise ImunifyPatchIdError(f"No such user {username}") from e
    else:
        id_file = Path(user_pwd.pw_dir) / IMUNIFY_PATCH_ID_FILE
        if not id_file.exists():
            if not id_file.parent.exists():
                logger.error(f"No such user homedir: {id_file.parent}")
                raise ImunifyPatchIdError(
                    f"No such user homedir: {id_file.parent}"
                )
            try:
                await safe_fileops.touch(str(id_file))
            except (PermissionError, OSError) as e:
                logger.error(
                    "Unable to put %s in user home dir %s",
                    IMUNIFY_PATCH_ID_FILE,
                    e,
                )
                raise ImunifyPatchIdError from e
    return id_file


def _generate_id() -> ImunifyPatchUserId:
    """Generate Imunify.Patch id"""
    return uuid.uuid4().hex


def _read_id_file(id_file: Path) -> ImunifyPatchUserId | None:
    """Read Imunify.Patch id from `id_file`.
    If id is not found, return `None`.
    """

    with id_file.open("r") as f:
        for line in reversed(f.readlines()):
            if line and not line.startswith("#"):
                if imunify_patch_id := line.strip():
                    return imunify_patch_id

    logger.warning(f"Cannot parse {id_file}, file is corrupted or empty")
    return None


async def _write_id_file(id_file: Path, _id: ImunifyPatchUserId) -> None:
    """Write Imunify.Patch id to `id_file`."""
    text = (
        "# DO NOT EDIT\n"
        "# This file contains Imunify.Patch id unique to this user\n"
        "\n"
        f"{_id}\n"
    )
    try:
        await safe_fileops.write_text(str(id_file), text)
    except (OSError, PermissionError) as e:
        logger.error(
            "Unable to write %s in user home dir: %s", IMUNIFY_PATCH_ID_FILE, e
        )
        raise ImunifyPatchIdError from e


@async_lru_cache(maxsize=100, ttl=60)
async def get_imunify_patch_purchase_url(username) -> str:
    # Imunify Patch purchase URL template:
    # https://www.cloudlinux.com/purchase-imunify-patch?iaid=<iaid>
    # &imunify_patch_user_id=<imunify_patch_user_id>&server_ip=12.23.34.45
    # &username=johndoe&websites=example.com,anotherexample.com
    # defined in Jira ticket: https://cloudlinux.atlassian.net/browse/DEF-32303
    iaid = IndependentAgentIDAPI.get_iaid()
    imunify_patch_user_id = await get_imunify_patch_id(username)
    panel_manager = HostingPanel()

    server_ip = panel_manager.get_server_ip()
    user_domains = (await panel_manager.get_domains_per_user()).get(
        username, []
    )
    total_websites = len(user_domains)

    user_domain_paths = {
        domain: paths
        for domain, paths in (await panel_manager.get_domain_paths()).items()
        if domain in user_domains
    }

    hits = VulnerabilityHit.select().where(
        (VulnerabilityHit.user == username)
        & (VulnerabilityHit.status == VulnerabilityHitStatus.VULNERABLE)
    )

    vulnerable_domains = [
        domain
        for hit in hits
        for domain, doc_roots in user_domain_paths.items()
        for path in doc_roots
        if hit.orig_file.startswith(path)
    ]

    vulnerabilities = group_by_severity(
        await VulnerabilityAPI.get_details(
            VulnerabilityHit.get_vulnerabilities_ids(
                [hit.as_dict() for hit in hits]
            )
        )
    )

    url_args = urlencode(
        {
            "iaid": iaid,
            "imunify_patch_user_id": imunify_patch_user_id or "",
            "server_ip": server_ip,
            "username": username,
            "websites": ",".join(user_domains[:PURCHASE_URL_MAX_WEBSITES]),
            "total_websites": total_websites,
            "vulnerable_domains": len(vulnerable_domains),
            "vulnerabilities": json.dumps(vulnerabilities, sort_keys=True),
        }
    )
    return f"{IMUNIFY_PATCH_PURCHASE_URL}?{url_args}"


def group_by_severity(
    vulnerabilities: dict[str, Any],
    limit: int = MAX_SEVERITY_COUNT,
) -> dict[str, dict[str, int]]:
    result = defaultdict(lambda: defaultdict(int))

    for item in islice(vulnerabilities.values(), limit):
        app_name = item["app"]
        severity = item.get("severity", "UNKNOWN")

        if severity in ("HIGH", "MEDIUM", "LOW", "UNKNOWN"):
            result[app_name][severity] += 1

    return dict(result)
Name	Size	Permission
__pycache__	---	0755
__init__.py	0 bytes	0644
config.py	1866 bytes	0644
imunify_patch_id.py	7423 bytes	0644
messages.py	10683 bytes	0644
permissions.py	1079 bytes	0644
plugins.py	956 bytes	0644
File Content: imunify_patch_id.py

File Content: `imunify_patch_id.py`
