403 lines
20 KiB
Python
403 lines
20 KiB
Python
"""Persistent, data-driven audit profile registry.
|
|
|
|
The manifest is the source of truth; artifacts are immutable per profile version.
|
|
All manifest mutations are serialized and committed with ``os.replace``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
from .config_upload import normalize_json, validate_xlsx
|
|
except ImportError: # application is also launched with /app on sys.path
|
|
from config_upload import normalize_json, validate_xlsx
|
|
|
|
|
|
class RegistryError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class RegistryValidationError(RegistryError, ValueError):
|
|
pass
|
|
|
|
|
|
class RegistryNotFoundError(RegistryError, LookupError):
|
|
pass
|
|
|
|
|
|
_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
|
|
_ALIAS_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
|
|
_VERSION_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
|
|
_LEGACY = (
|
|
("windows", "Windows", "Windows_Checklist.xlsx", "windows_config.json", ("win", "winserver", "win_server")),
|
|
("centos", "CentOS", "CentOS_Checklist_NEW.xlsx", "centos_config_new.json", ("centos_linux",)),
|
|
("rhel", "RHEL (Red Hat Enterprise Linux)", "RHEL_Checklist_NEW.xlsx", "rhel_config_new.json", ("redhat", "red_hat", "red-hat")),
|
|
("ubuntu", "Ubuntu", "Ubuntu_Checklist_NEW.xlsx", "ubuntu_config_new.json", ()),
|
|
("oracle", "Oracle Linux", "Oracle_Linux_Checklist.xlsx", "oracle_linux_config.json", ("oracle_linux", "ol")),
|
|
)
|
|
|
|
|
|
class RegistryService:
|
|
def __init__(self, root: str | Path, legacy_config_root: str | Path | None = None):
|
|
self.root = Path(root)
|
|
self.legacy_config_root = Path(legacy_config_root or self.root.parent)
|
|
self.manifest_path = self.root / "registry.json"
|
|
self._lock = threading.RLock()
|
|
|
|
@staticmethod
|
|
def _id(value: str, field: str = "id") -> str:
|
|
value = (value or "").strip().lower()
|
|
if not _ID_RE.fullmatch(value):
|
|
raise RegistryValidationError(f"{field} must match {_ID_RE.pattern}")
|
|
return value
|
|
|
|
@staticmethod
|
|
def _version(value: str) -> str:
|
|
value = (value or "").strip()
|
|
if not _VERSION_RE.fullmatch(value):
|
|
raise RegistryValidationError("invalid version identifier")
|
|
return value
|
|
|
|
@staticmethod
|
|
def _alias(value: str, field: str = "alias") -> str:
|
|
value = (value or "").strip().lower()
|
|
if not _ALIAS_RE.fullmatch(value):
|
|
raise RegistryValidationError(f"{field} is invalid")
|
|
return value
|
|
|
|
@staticmethod
|
|
def _sha(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
def _empty(self) -> dict[str, Any]:
|
|
return {"schema_version": 2, "revision": 0, "categories": [], "aliases": {}}
|
|
|
|
def load(self) -> dict[str, Any]:
|
|
if not self.manifest_path.exists():
|
|
return self._empty()
|
|
try:
|
|
data = json.loads(self.manifest_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise RegistryValidationError(f"invalid registry manifest: {exc}") from exc
|
|
self._validate_manifest(data)
|
|
return data
|
|
|
|
def _validate_manifest(self, data: dict[str, Any]) -> None:
|
|
if not isinstance(data, dict) or data.get("schema_version") != 2:
|
|
raise RegistryValidationError("unsupported registry schema")
|
|
seen: dict[str, set[str]] = {"category": set(), "component": set(), "profile": set()}
|
|
for category in data.get("categories", []):
|
|
cid = self._id(category.get("id"), "category id")
|
|
if cid in seen["category"]:
|
|
raise RegistryValidationError(f"duplicate category: {cid}")
|
|
seen["category"].add(cid)
|
|
for component in category.get("components", []):
|
|
coid = self._id(component.get("id"), "component id")
|
|
if coid in seen["component"]:
|
|
raise RegistryValidationError(f"duplicate component: {coid}")
|
|
seen["component"].add(coid)
|
|
for profile in component.get("profiles", []):
|
|
pid = self._id(profile.get("id"), "profile id")
|
|
if pid in seen["profile"]:
|
|
raise RegistryValidationError(f"duplicate profile: {pid}")
|
|
seen["profile"].add(pid)
|
|
versions = profile.get("versions", [])
|
|
active = [v for v in versions if v.get("active")]
|
|
if len(active) > 1:
|
|
raise RegistryValidationError(f"multiple active versions: {pid}")
|
|
|
|
def _write(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
self._validate_manifest(data)
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
data["revision"] = int(data.get("revision", 0)) + 1
|
|
data["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
temporary = self.manifest_path.with_suffix(".json.tmp")
|
|
payload = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8") + b"\n"
|
|
with temporary.open("wb") as stream:
|
|
stream.write(payload)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, self.manifest_path)
|
|
return copy.deepcopy(data)
|
|
|
|
def _find(self, data: dict[str, Any], kind: str, object_id: str):
|
|
for category in data["categories"]:
|
|
if kind == "category" and category["id"] == object_id:
|
|
return category
|
|
for component in category.get("components", []):
|
|
if kind == "component" and component["id"] == object_id:
|
|
return component
|
|
for profile in component.get("profiles", []):
|
|
if kind == "profile" and profile["id"] == object_id:
|
|
return profile
|
|
raise RegistryNotFoundError(f"{kind} not found: {object_id}")
|
|
|
|
def create_category(self, category_id: str, name: str) -> dict[str, Any]:
|
|
category_id = self._id(category_id, "category id")
|
|
with self._lock:
|
|
data = self.load()
|
|
try:
|
|
self._find(data, "category", category_id)
|
|
except RegistryNotFoundError:
|
|
pass
|
|
else:
|
|
raise RegistryValidationError(f"duplicate category: {category_id}")
|
|
category = {"id": category_id, "name": name.strip(), "components": []}
|
|
data["categories"].append(category)
|
|
self._write(data)
|
|
return copy.deepcopy(category)
|
|
|
|
def create_component(self, category_id: str, component_id: str, name: str) -> dict[str, Any]:
|
|
component_id = self._id(component_id, "component id")
|
|
with self._lock:
|
|
data = self.load()
|
|
try:
|
|
self._find(data, "component", component_id)
|
|
except RegistryNotFoundError:
|
|
pass
|
|
else:
|
|
raise RegistryValidationError(f"duplicate component: {component_id}")
|
|
category = self._find(data, "category", category_id)
|
|
component = {"id": component_id, "name": name.strip(), "profiles": []}
|
|
category["components"].append(component)
|
|
self._write(data)
|
|
return copy.deepcopy(component)
|
|
|
|
def create_profile(self, component_id: str, profile_id: str, name: str, aliases=()) -> dict[str, Any]:
|
|
profile_id = self._id(profile_id, "profile id")
|
|
aliases = [self._alias(alias) for alias in aliases]
|
|
with self._lock:
|
|
data = self.load()
|
|
try:
|
|
self._find(data, "profile", profile_id)
|
|
except RegistryNotFoundError:
|
|
pass
|
|
else:
|
|
raise RegistryValidationError(f"duplicate profile: {profile_id}")
|
|
for alias in aliases:
|
|
if alias in data["aliases"] and data["aliases"][alias] != profile_id:
|
|
raise RegistryValidationError(f"duplicate alias: {alias}")
|
|
component = self._find(data, "component", component_id)
|
|
profile = {"id": profile_id, "name": name.strip(), "aliases": aliases,
|
|
"activation_history": [], "versions": []}
|
|
component["profiles"].append(profile)
|
|
for alias in aliases:
|
|
data["aliases"][alias] = profile_id
|
|
self._write(data)
|
|
return copy.deepcopy(profile)
|
|
|
|
def create_version(self, profile_id: str, version: str, checklist: bytes, rules: bytes,
|
|
*, activate: bool = False, created_by: str = "admin") -> dict[str, Any]:
|
|
profile_id, version = self._id(profile_id, "profile id"), self._version(version)
|
|
try:
|
|
checklist = validate_xlsx(checklist)
|
|
rules = normalize_json(rules)
|
|
except ValueError as exc:
|
|
raise RegistryValidationError(str(exc)) from exc
|
|
with self._lock:
|
|
data = self.load()
|
|
profile = self._find(data, "profile", profile_id)
|
|
if any(item["version"] == version for item in profile["versions"]):
|
|
raise RegistryValidationError(f"duplicate version: {version}")
|
|
relative = Path("artifacts") / profile_id / version
|
|
target = self.root / relative
|
|
if target.exists():
|
|
raise RegistryValidationError(f"artifact directory already exists: {version}")
|
|
temporary = self.root / "artifacts" / profile_id / f".{version}.{os.getpid()}.tmp"
|
|
temporary.mkdir(parents=True, exist_ok=False)
|
|
try:
|
|
(temporary / "checklist.xlsx").write_bytes(checklist)
|
|
(temporary / "rules.json").write_bytes(rules)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
os.replace(temporary, target)
|
|
except Exception:
|
|
shutil.rmtree(temporary, ignore_errors=True)
|
|
raise
|
|
if activate:
|
|
current = next((item["version"] for item in profile["versions"] if item.get("active")), None)
|
|
if current:
|
|
profile.setdefault("activation_history", []).append(current)
|
|
for item in profile["versions"]:
|
|
item["active"] = False
|
|
item = {
|
|
"version": version, "active": bool(activate),
|
|
"created_at": datetime.now(timezone.utc).isoformat(), "created_by": created_by,
|
|
"artifacts": {"checklist": str(relative / "checklist.xlsx").replace("\\", "/"),
|
|
"rules": str(relative / "rules.json").replace("\\", "/")},
|
|
"checksums": {"checklist": self._sha(checklist), "rules": self._sha(rules)},
|
|
}
|
|
profile["versions"].append(item)
|
|
try:
|
|
self._write(data)
|
|
except Exception:
|
|
shutil.rmtree(target, ignore_errors=True)
|
|
raise
|
|
return copy.deepcopy(item)
|
|
|
|
def update_version_artifacts(self, profile_id: str, version: str,
|
|
checklist: bytes | None = None,
|
|
rules: bytes | None = None,
|
|
*, updated_by: str = "admin") -> dict[str, Any]:
|
|
"""Replace artifacts for an inactive version while preserving a backup copy."""
|
|
if checklist is None and rules is None:
|
|
raise RegistryValidationError("checklist or rules is required")
|
|
if checklist is not None:
|
|
checklist = validate_xlsx(checklist)
|
|
if rules is not None:
|
|
rules = normalize_json(rules)
|
|
with self._lock:
|
|
data = self.load()
|
|
profile = self._find(data, "profile", self._id(profile_id, "profile id"))
|
|
target_version = next((item for item in profile["versions"] if item["version"] == version), None)
|
|
if not target_version:
|
|
raise RegistryNotFoundError(f"version not found: {version}")
|
|
if target_version.get("active"):
|
|
raise RegistryValidationError("active versions are immutable; create a new version")
|
|
updates = {"checklist": checklist, "rules": rules}
|
|
for key, content in updates.items():
|
|
if content is None:
|
|
continue
|
|
target = self.root / target_version["artifacts"][key]
|
|
temporary = target.with_suffix(target.suffix + ".tmp")
|
|
backup = target.with_suffix(target.suffix + ".bak")
|
|
temporary.write_bytes(content)
|
|
if target.exists():
|
|
shutil.copy2(target, backup)
|
|
os.replace(temporary, target)
|
|
target_version["checksums"][key] = self._sha(content)
|
|
target_version["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
target_version["updated_by"] = updated_by
|
|
self._write(data)
|
|
return copy.deepcopy(target_version)
|
|
|
|
def activate_version(self, profile_id: str, version: str) -> dict[str, Any]:
|
|
with self._lock:
|
|
data = self.load()
|
|
profile = self._find(data, "profile", profile_id)
|
|
target = next((v for v in profile["versions"] if v["version"] == version), None)
|
|
if not target:
|
|
raise RegistryNotFoundError(f"version not found: {version}")
|
|
for key, relative in target["artifacts"].items():
|
|
artifact = self.root / relative
|
|
if not artifact.is_file() or self._sha(artifact.read_bytes()) != target["checksums"][key]:
|
|
raise RegistryValidationError(f"{key} checksum mismatch for {profile_id}:{version}")
|
|
current = next((item["version"] for item in profile["versions"] if item.get("active")), None)
|
|
if current and current != version:
|
|
profile.setdefault("activation_history", []).append(current)
|
|
for item in profile["versions"]:
|
|
item["active"] = item is target
|
|
target["activated_at"] = datetime.now(timezone.utc).isoformat()
|
|
self._write(data)
|
|
return copy.deepcopy(target)
|
|
|
|
def rollback(self, profile_id: str) -> dict[str, Any]:
|
|
with self._lock:
|
|
data = self.load()
|
|
profile = self._find(data, "profile", profile_id)
|
|
history = profile.setdefault("activation_history", [])
|
|
if not history:
|
|
raise RegistryValidationError("no previous version available")
|
|
previous_version = history.pop()
|
|
previous = next((v for v in profile["versions"] if v["version"] == previous_version), None)
|
|
if not previous:
|
|
raise RegistryValidationError("rollback history references a missing version")
|
|
for key, relative in previous["artifacts"].items():
|
|
artifact = self.root / relative
|
|
if not artifact.is_file() or self._sha(artifact.read_bytes()) != previous["checksums"][key]:
|
|
raise RegistryValidationError(f"{key} checksum mismatch for {profile_id}:{previous_version}")
|
|
for item in profile["versions"]:
|
|
item["active"] = item is previous
|
|
previous["activated_at"] = datetime.now(timezone.utc).isoformat()
|
|
self._write(data)
|
|
return copy.deepcopy(previous)
|
|
|
|
def resolve(self, *, os_type: str | None = None, profile_id: str | None = None) -> dict[str, Any]:
|
|
data = self.load()
|
|
if profile_id:
|
|
resolved_id = self._id(profile_id, "profile id")
|
|
elif os_type:
|
|
alias = self._alias(os_type, "os_type")
|
|
resolved_id = data["aliases"].get(alias, alias if alias.endswith("-baseline") else "")
|
|
else:
|
|
raise RegistryValidationError("profile_id or os_type is required")
|
|
if not resolved_id:
|
|
raise RegistryNotFoundError(f"unknown os_type: {os_type}")
|
|
profile = self._find(data, "profile", resolved_id)
|
|
version = next((v for v in profile["versions"] if v.get("active")), None)
|
|
if not version:
|
|
raise RegistryNotFoundError(f"no active version for profile: {resolved_id}")
|
|
paths = {key: self.root / relative for key, relative in version["artifacts"].items()}
|
|
for key, path in paths.items():
|
|
if not path.is_file() or self._sha(path.read_bytes()) != version["checksums"][key]:
|
|
raise RegistryValidationError(f"{key} checksum mismatch for {resolved_id}:{version['version']}")
|
|
return {"profile": copy.deepcopy(profile), "version": copy.deepcopy(version),
|
|
"checklist_path": str(paths["checklist"]), "rules_path": str(paths["rules"])}
|
|
|
|
def seed_legacy_profiles(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
if self.manifest_path.exists():
|
|
return self.load()
|
|
data = self._empty()
|
|
category = {"id": "operating-systems", "name": "Operating Systems", "components": []}
|
|
data["categories"].append(category)
|
|
for os_id, name, checklist_name, rules_name, extra_aliases in _LEGACY:
|
|
profile_id = f"{os_id}-baseline"
|
|
aliases = [os_id, *extra_aliases]
|
|
checklist_path = self.legacy_config_root / checklist_name
|
|
rules_path = self.legacy_config_root / rules_name
|
|
if not checklist_path.is_file() or not rules_path.is_file():
|
|
raise RegistryValidationError(f"legacy artifacts missing for {os_id}")
|
|
checklist = validate_xlsx(checklist_path.read_bytes())
|
|
rules = normalize_json(rules_path.read_bytes())
|
|
relative = Path("artifacts") / profile_id / "legacy-1"
|
|
target = self.root / relative
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
(target / "checklist.xlsx").write_bytes(checklist)
|
|
(target / "rules.json").write_bytes(rules)
|
|
version = {
|
|
"version": "legacy-1", "active": True,
|
|
"created_at": datetime.now(timezone.utc).isoformat(), "created_by": "migration",
|
|
"artifacts": {"checklist": str(relative / "checklist.xlsx").replace("\\", "/"),
|
|
"rules": str(relative / "rules.json").replace("\\", "/")},
|
|
"checksums": {"checklist": self._sha(checklist), "rules": self._sha(rules)},
|
|
"legacy_sources": {"checklist": checklist_name, "rules": rules_name},
|
|
}
|
|
profile = {"id": profile_id, "name": f"{name} baseline", "aliases": aliases,
|
|
"activation_history": [],
|
|
"versions": [version]}
|
|
category["components"].append({"id": os_id, "name": name, "profiles": [profile]})
|
|
for alias in aliases:
|
|
data["aliases"][alias] = profile_id
|
|
return self._write(data)
|
|
|
|
def public_catalog(self, active_only: bool = False) -> dict[str, Any]:
|
|
"""Return catalog metadata without artifact paths, checksums or legacy source paths."""
|
|
data = copy.deepcopy(self.load())
|
|
data.pop("aliases", None)
|
|
for category in data["categories"]:
|
|
for component in category["components"]:
|
|
for profile in component["profiles"]:
|
|
versions = profile.get("versions", [])
|
|
if active_only:
|
|
versions = [version for version in versions if version.get("active")]
|
|
profile["versions"] = [
|
|
{
|
|
"version": version["version"],
|
|
"active": bool(version.get("active")),
|
|
"created_at": version.get("created_at"),
|
|
"created_by": version.get("created_by"),
|
|
"activated_at": version.get("activated_at"),
|
|
}
|
|
for version in versions
|
|
]
|
|
return data
|