TripleDNN

Current Directory: /opt/imunify360/venv/lib/python3.11/site-packages/imav/patchman

File Name	Size	Actions
📁 __pycache__	Folder	Edit \| Delete \| Rename \| Download
📁 fs_scanner	Folder	Edit \| Delete \| Rename \| Download
📁 rpc	Folder	Edit \| Delete \| Rename \| Download
__init__.py	0 bytes	Edit \| Delete \| Rename \| Download
config.py	1908 bytes	Edit \| Delete \| Rename \| Download
constants.py	908 bytes	Edit \| Delete \| Rename \| Download
custom_integration.py	5309 bytes	Edit \| Delete \| Rename \| Download
exceptions.py	826 bytes	Edit \| Delete \| Rename \| Download
feature.py	3498 bytes	Edit \| Delete \| Rename \| Download
license.py	5309 bytes	Edit \| Delete \| Rename \| Download
model.py	2055 bytes	Edit \| Delete \| Rename \| Download
state.py	921 bytes	Edit \| Delete \| Rename \| Download
utils.py	1443 bytes	Edit \| Delete \| Rename \| Download

"""
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/>.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import json
import logging
from enum import StrEnum
from pathlib import Path
from typing import Any, Dict, List

import cerberus

from defence360agent.utils import check_run, nice_iterator

logger = logging.getLogger(__name__)

class PatchmanCustomIntegrationType(StrEnum):
    FILE = "file"
    SCRIPT = "script"

class PatchmanCustomIntegration:
    """
    Patchman custom integration uses pre-defined user files that can be
    either files with info about users (file integration) or scripts (script
    integration). This class is designed to retrieve the users and provide them
    to the DB the same way panels do.

More info:
    https://docs.imunify360.com/Patchman_custom_integration.pdf
    """

USER_DATA_SCHEMA = {
        "homedir": {"type": "string", "required": True},
        "email": {"type": "string", "required": True},
        "parent": {"type": "string", "required": True, "nullable": True},
        "language": {"type": "string", "required": True},
        "suspended": {"type": "boolean", "required": True},
        "level": {"type": "integer", "allowed": [1, 2, 3], "required": True},
        "domains": {"type": "dict", "required": True},
    }

def __init__(self, integration_type: str, metadata_path: str):
        self._integration_type = integration_type
        self._metadata_path = metadata_path

@property
    def integration_type(self):
        return PatchmanCustomIntegrationType(self._integration_type)

@property
    def metadata_path(self):
        return Path(self._metadata_path)

def validate(self, users_data: dict) -> dict:
        """
        Return only valid *users_data*
        """
        validator = cerberus.Validator(self.USER_DATA_SCHEMA)
        result = {}
        for username, data in users_data.items():
            if not validator.validate(data):
                logger.warning(
                    "Skip not valid data for '%s': %s",
                    username,
                    validator.errors,
                )
                continue
            result[username] = data
        return result

async def _get_users_data(self) -> dict:
        users_data = {}
        if self.integration_type == PatchmanCustomIntegrationType.SCRIPT:
            if not self.metadata_path.is_file():
                raise PatchmanCustomIntegrationError(
                    "Provided metadata path is not a file"
                )
            result = await check_run([str(self.metadata_path)])
            users_data = json.loads(result.decode())
        else:  # file-based
            # *metadata_path* should be a dir that contains JSON files;
            # there should be one file per existing user
            if not self.metadata_path.is_dir():
                raise PatchmanCustomIntegrationError(
                    "Provided metadata path is not a directory"
                )
            # don't block the main loop for too long
            async for user_data_file in nice_iterator(
                self.metadata_path.glob("*.json"), chunk_size=50
            ):
                try:
                    with user_data_file.open(encoding="utf-8") as f:
                        data = json.load(f)
                except json.JSONDecodeError:
                    logger.warning(
                        "Skip invalid JSON file: %s", str(user_data_file)
                    )
                    continue
                if len(data) != 1:
                    logger.warning(
                        "Skip %s: expect one JSON object per file",
                        str(user_data_file),
                    )
                    continue
                users_data.update(data)
        return self.validate(users_data)

async def patchman_users(self) -> List[Dict[str, Any]]:
        try:
            users = []
            metadata_dict = await self._get_users_data()
            for username, details in metadata_dict.items():
                orig_domains = details.pop("domains", {})
                domains = [
                    {"domain": domain_name, "paths": paths}
                    for domain_name, paths in orig_domains.items()
                ]
                users.append(
                    {"username": username, "domains": domains, **details}
                )

return users
        except PatchmanCustomIntegrationError:
            raise
        except Exception as e:
            raise PatchmanCustomIntegrationError from e

class PatchmanCustomIntegrationError(Exception):
    pass