1#@!#!123s

file_get_contents 1#@!#!123s
: / opt / imunify360 / venv / lib / python3.11 / site-packages / defence360agent / rpc_tools /
Filename : validate.py
back
import datetime
import logging
import math
import os
import re
from collections import namedtuple
from functools import wraps

from cerberus.validator import Validator
from defence360agent.contracts.config import (
    ANTIVIRUS_MODE,
    BackupRestore,
    Malware,
)
from defence360agent.contracts.license import LicenseCLN
from defence360agent.subsys.backup_systems import BackupSystem, get_backend

logger = logging.getLogger(__name__)

SHA256_REGEXP = re.compile("^[A-Fa-f0-9]{64}$")


class ValidationError(Exception):
    def __init__(self, errors, extra_data=None):
        if isinstance(errors, str):
            self.errors = [errors]
        else:
            self.errors = errors
        self.extra_data = extra_data or {}


OrderByBase = namedtuple("OrderByBase", ["column_name", "desc"])


class OrderBy(OrderByBase):
    def __new__(cls, column_name, desc):
        return super().__new__(cls, column_name, desc)

    @classmethod
    def fromstring(cls, ob_string):
        """
        :param ob_string: for example: 'user+', 'id-'
        :return:
        """
        try:
            col_name, sign = re.compile("^(.+)([+|-])").split(ob_string)[1:-1]
            return cls(col_name, sign == "-")
        except ValueError as e:
            raise ValueError(
                "Incorrect order_by: ({}): {}".format(str(e), ob_string)
            )


class SchemaValidator(Validator):
    _DATE_FORMAT = "%Y-%m-%d"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.extra_data = {}

    def _normalize_coerce_order_by(self, value):
        if isinstance(value, OrderBy):
            return value
        return OrderBy.fromstring(value)

    def _normalize_coerce_sha256hash(self, value):
        return str(value).strip().lower()

    def _normalize_coerce_scan_db(self, value):
        if ANTIVIRUS_MODE:
            return False
        if value is None:
            return Malware.DATABASE_SCAN_ENABLED
        return value

    def _validate_type_order_by(self, value):
        if isinstance(value, OrderBy):
            return True
        return False

    def _validate_type_sha256hash(self, value: str):
        return SHA256_REGEXP.match(str(value).strip())

    def _validate_is_absolute_path(self, is_absolute_path, field, value):
        """{'type': 'boolean', 'empty': False}"""
        if is_absolute_path:
            if not os.path.isabs(value):
                self._error(field, "Path {} should be absolute".format(value))

    def _validate_isascii(self, isascii, field, value):
        """{'type': 'boolean'}"""
        if isascii:
            try:
                value.encode("ascii")
            except UnicodeEncodeError:
                self._error(field, "Must only contain ascii symbols")

    def _normalize_coerce_int(self, value):
        return int(value)

    def _normalize_default_setter_now(self, document) -> int:
        return math.ceil(datetime.datetime.now().timestamp())

    # for argparser support
    def _validate_cli(self, *args, **kwargs):
        """{'type': 'dict', 'empty': False, 'schema': {
        'users': {'type': 'list', 'allowed': ['non-root', 'root'],
            'empty': False},
        'require_rpc': {'type': 'string', 'empty': True, 'default': 'running',
                        'allowed': ['running', 'stopped', 'any', 'direct']}
        }}
        """

    # for argparser support
    def _validate_help(self, *args, **kwargs):
        """{'type': 'string', 'empty': False}"""

    # for argparser support
    def _validate_positional(self, *args, **kwargs):
        """{'type': 'boolean', 'empty': True, 'default': False}"""

    # metadata for response validation
    def _validate_return_type(self, *args, **kwargs):
        """{'type': 'string', 'empty': True}"""

    def _validate_cli_only(self, *args, **kwargs):
        """{'type': 'boolean', 'empty': False, 'default': False}"""

    def _validate_envvar(self, *args, **kwargs):
        """
        Parameter can be passed via the specified environment variable.
        The value specified via a CLI argument takes precedence.

        The rule's arguments are validated against this schema:
        {'type': 'string', 'empty': False}
        """

    def _validate_envvar_only(self, *args, **kwargs):
        """
        Parameter will only be accepted if provided via environment
        variable specified by `envvar`. It will be rejected if passed as
        a CLI argument.

        The rule's arguments are validated against this schema:
        {'type': 'boolean', 'default': False}
        """

    def _normalize_coerce_path(self, value: str):
        if value:
            return os.path.abspath(value)

        return value

    def _normalize_coerce_backup_system(self, value):
        if isinstance(value, BackupSystem):
            return value

        return get_backend(value)

    def _validator_backup_is_enabled(self, field, value):
        if not (BackupRestore.ENABLED and BackupRestore.backup_system()):
            self._error(field, "Backup is not enabled!")


def validate(validator, hashable, params):
    values = validator.normalized(
        {hashable: params}, always_return_document=True
    )
    if not validator.validate({hashable: values[hashable]}):
        logger.warning(
            "Validation error with command {}, params {}, errors {}".format(
                hashable, params, validator.errors
            )
        )
        raise ValidationError(validator.errors, validator.extra_data)

    return validator.document[hashable]


def validate_middleware(validator):
    def wrapped(f):
        @wraps(f)
        async def wrapper(request, *args, **kwargs):
            hashable = tuple(request["command"])
            request["params"] = validate(
                validator, hashable, request["params"]
            )
            result = await f(request, *args, **kwargs)
            return result

        return wrapper

    return wrapped


def validate_av_plus_license(func):
    """
    Decorator for CLI commands methods that ensures that the AV+ license
    is valid.

    :raises ValidationError:
    """
    exception = ValidationError("ImunifyAV+ license required")

    @wraps(func)
    async def async_wrapper(*args, **kwargs):
        if LicenseCLN.is_valid_av_plus():
            return await func(*args, **kwargs)
        raise exception

    if ANTIVIRUS_MODE:
        return async_wrapper
    return func