C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// 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, valu