ACIL FM

Pilih semua

Nama	Ukuran	Aksi
fs_scanner	-	Open
rpc	-	Open
__pycache__	-	Open
config.py	1.86 MB	View DL Edit
constants.py	908 B	View DL Edit
custom_integration.py	5.18 MB	View DL Edit
exceptions.py	826 B	View DL Edit
feature.py	3.42 MB	View DL Edit
license.py	5.29 MB	View DL Edit
model.py	2.01 MB	View DL Edit
state.py	921 B	View DL Edit
utils.py	1.41 MB	View DL Edit
__init__.py	0 B	View DL Edit

Edit file: /opt/imunify360/venv/lib/python3.11/site-packages/imav/patchman/custom_integration.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/>.

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

Batal

Isi Zip:

Create