modules/integrity_module.py - master - gabelchik/dpcs-diagnostics

modules/integrity_module.py
86 строк3 KB
Leonid Kalintsev
messages
21 июл 2026, 15:09
21 июл 2026, 15:0955529c2
from typing import Dict, Any
from core.interfaces import DiagnosticModule


class IntegrityModule(DiagnosticModule):
    def __init__(self, name: str, mod_cfg: Dict[str, Any]):
        super().__init__(name)
        self._integrity_registry = mod_cfg.get("registry") or {}

    def execute(self, actual_state: dict = None) -> Dict[str, Any]:
        components = {}
        mismatched_permissions = []
        problems = []

        if not self._integrity_registry:
            return {
                "status": "FAILURE",
                "message": "Эталонные данные отсуствуют",
                "components": {},
                "mismatched_permissions": []
            }

        if not actual_state:
            return {
                "status": "FAILURE",
                "message": "Данные от узла отсутствуют",
                "components": {},
                "mismatched_permissions": []
            }

        actual_integrity = actual_state.get(self.name) or {}
        actual_registry = actual_integrity.get("registry") or {}

        for component_name, info in self._integrity_registry.items():
            file_path = info.get("path")
            expected_hash = info.get("sha256")
            expected_perms = info.get("permissions")

            component_report = {
                "path": file_path,
                "exists": False,
                "permissions_ok": False,
                "checksum_ok": False,
                "status": "FAILED",
            }

            actual_component = actual_registry.get(component_name)

            if not actual_component or actual_component.get("path") != file_path:
                components[component_name] = component_report
                problems.append(f"Файл '{component_name}' ({file_path}) не найден на узле")
                continue

            component_report["exists"] = True

            perms_ok =  (expected_perms is None) or (actual_component.get("permissions") == expected_perms)
            hash_ok = (expected_hash is None) or (actual_component.get("sha256") == expected_hash)

            component_report["checksum_ok"] = hash_ok
            component_report["permissions_ok"] = perms_ok

            if hash_ok and perms_ok:
                component_report["status"] = "OK"
            else:
                if not hash_ok:
                    problems.append(f"У компонента '{component_name}' не совпал SHA256")
                if not perms_ok:
                    problems.append(
                        f"У компонента '{component_name}' не совпали права доступа ({actual_component.get('permissions')} != {expected_perms})")
                    mismatched_permissions.append({
                        "process": component_name,
                        "path": actual_component.get("path"),
                        "expected_permissions": expected_perms
                    })

            components[component_name] = component_report

        has_errors = len(problems) > 0
        message = "; ".join(problems) if has_errors else "Целостность всех файлов и права доступа подтверждены"

        return {
            "status": "FAILURE" if has_errors else "SUCCESS",
            "message": message,
            "components": components,
            "mismatched_permissions": mismatched_permissions
        }