?HAMA

Edit File Size: 3.55 KB

//opt/imunify360/venv/lib64/python3.11/site-packages/defence360agent/subsys/panels/no_cp/panel.py

Text Base64

import asyncio
import functools
import json
import logging
from collections import defaultdict
from typing import Dict, List, Optional, Set

import cerberus
import yaml

from defence360agent.contracts.config import NoCP as Config
from defence360agent.utils import get_non_system_users, run

from .. import base

logger = logging.getLogger(__name__)
_SCHEMA_PATH_TMPL = (
    "/opt/imunify360/venv/share/imunify360/no_cp/schema-v{}.yaml"
)

@functools.lru_cache(maxsize=2)
def _get_validator(version: int) -> cerberus.Validator:
    """Returns a validator for given API version."""
    with open(_SCHEMA_PATH_TMPL.format(version)) as schema_file:
        schema = yaml.safe_load(schema_file)
        return cerberus.Validator(schema)

async def _get_client_data() -> Optional[dict]:
    """Runs a script, validates its JSON output and returns it."""
    try:
        result = await run([Config.CLIENT_SCRIPT, str(Config.LATEST_VERSION)])
    except FileNotFoundError:
        return None
    except asyncio.CancelledError:
        raise
    except Exception as exc:
        raise ScriptError("failed to run script") from exc
    code, stdout, _ = result
    if code != 0:
        raise ScriptError("exited with code: {}".format(code))
    try:
        data = json.loads(stdout.decode())
    except Exception as exc:
        raise ScriptError("Cannot decode output as JSON") from exc
    version = data.get("version")
    if version is None:
        raise ScriptError("output does not have`version` field")
    if not (
        isinstance(version, int)
        and version >= 1
        and version <= NoCP.LATEST_VERSION
    ):
        raise ScriptError("invalid API version: {}".format(version))
    validator = _get_validator(version)
    ok = validator.validate(data)
    if not ok:
        raise ScriptError("validation error: {}", validator.errors)
    return data

class ScriptError(base.PanelException):
    pass

class NoCP(base.AbstractPanel):
    NAME = "no panel"
    exception = ScriptError

@classmethod
    def is_installed(cls):
        return False

async def enable_imunify_plugin(self, name=None):
        pass

async def disable_imunify_plugin(self, plugin_name=None):
        pass

@classmethod
    async def version(cls):
        return "0"

async def get_user_domains(self):
        data = await _get_client_data()
        if data is None:
            return []
        return [domain["name"] for domain in data["domains"]]

async def get_users(self) -> List[str]:
        return [pw.pw_name for pw in get_non_system_users()]

async def get_domain_to_owner(self) -> Dict[str, List[str]]:
        data = await _get_client_data()
        if data is None:
            return {}
        return {
            domain["name"]: [domain["owner"]] for domain in data["domains"]
        }

async def get_domains_per_user(self) -> Dict[str, List[str]]:
        data = await _get_client_data()
        if data is None:
            return {}

user_to_domains = defaultdict(list)  # type: Dict[str, List[str]]
        for domain in data["domains"]:
            user_to_domains[domain["owner"]].append(domain["name"])

return dict(user_to_domains)

def basedirs(self) -> Set[str]:
        return set()

async def list_docroots(self) -> Dict[str, str]:  # pragma: no cover
        return dict()

async def panel_user_link(self, username) -> str:
        """
        Returns panel url
        :return: str
        """
        return ""

@classmethod
    async def get_user_domains_details(
        cls, username: str
    ) -> list[base.DomainData]:
        return []

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2026-06-08 20:24:30	Edit Download recursive recursive
panel.py	3.55 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
__init__.py	44 B	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive

Directory Contents