diff --git a/PROJECT_STATUS.md b/PROJECT_STATUS.md index f8ce9fa..7163e76 100644 --- a/PROJECT_STATUS.md +++ b/PROJECT_STATUS.md @@ -1,7 +1,7 @@ # PROJECT STATUS — Audit Hardening Tool -> Cập nhật lần cuối: 2026-09-04 09:12 (bởi: Hermes) -> Evaluator hiện tại: grok-build (grok-4.6) +> Cập nhật lần cuối: 2026-09-04 (bởi: Hermes + Codex gpt-5.6-sol) +> Evaluator hiện tại: Hermes verification; Kimi kiến trúc đã hoàn thành, Grok CLI chưa đăng nhập ## Phase hiện tại - [x] Phase 1 — Plan (`.hermes/plans/2026-09-04_091241-upload-checklist-config.md`) @@ -11,21 +11,27 @@ - [x] Phase 5 — Tracking ## Tổng quan -Bổ sung upload/thay thế checklist XLSX và rule JSON trên trang quản lý công cụ. Chức năng giới hạn admin, sử dụng mapping OS cố định và backup file cũ. +Đã chuyển checklist/rule sang registry dữ liệu phân cấp Category → Component → Profile → ProfileVersion, có checksum, artifact bất biến, active version và rollback; vẫn tương thích `os_type` cũ. ## Việc đã xong - [2026-09-04] Khảo sát `OS_CONFIG`, trang `admin_tools`, quyền admin/analyzer và bind mount `config`. - [2026-09-04] Chốt thiết kế upload an toàn, không nhận path/tên đích từ client. - [2026-09-04] Hoàn thành endpoint admin, UI upload checklist/rule, backup atomic và Docker RW mount; 8 regression tests pass. +- [2026-09-04] Thêm `RegistryService`, manifest schema v2 và atomic manifest/artifact writes; seed 5 OS cùng legacy aliases vào `config/registry`. +- [2026-09-04] Thêm read API `/api/v2/registry`, `/api/v2/profiles/active` và admin API/UI tạo category/component/profile, upload/activate/rollback version. +- [2026-09-04] Upload Web/async/API giữ `os_type` bắt buộc để tương thích và nhận thêm `profile_id` tùy chọn; audit UI cho phép override active profile mà không đổi auto OS mặc định. +- [2026-09-04] Xác nhận quyền: checklist/profile mutation chỉ admin; analyzer chỉ quản lý tool; `dev_guest` chỉ có luồng audit upload. ## Việc đang làm / Bước tiếp theo -- Commit/push lên Gitea, deploy NAS và kiểm thử upload thực tế qua Web. +- Push lên Gitea, deploy NAS và kiểm thử Web với một profile OS cũ cùng một profile component mới. ## Quyết định đã chốt - Chỉ admin được upload checklist/rule config; analyzer chỉ upload tool. - Checklist chỉ nhận `.xlsx`, rule config chỉ nhận JSON hợp lệ. - Tên file đích lấy từ `OS_CONFIG`; không cho path traversal. - Backup file hiện tại trước khi thay thế. +- Registry nằm dưới persistent mount riêng `/app/registry`; legacy files được giữ nguyên và compatibility upload tạo version mới thay vì ghi đè artifact active. +- `profile_id` override `os_type`; nếu bỏ trống, alias `os_type` resolve tới active profile version. ## Đánh giá các phase | Phase | Ngày | Evaluator | Kết quả | Ghi chú | @@ -34,6 +40,7 @@ Bổ sung upload/thay thế checklist XLSX và rule JSON trên trang quản lý | Direction | 2026-09-04 | grok-build | BLOCKED | Grok CLI chưa đăng nhập | | Knowledge | 2026-09-04 | Hermes | PASS | Dựa trên code/mount thực tế | | Execution | 2026-09-04 | Hermes | PASS | 8 tests, compile, Jinja, Compose, diff check đều pass; Grok review bị chặn auth | +| Registry architecture | 2026-09-04 | Codex + Hermes | PASS | 23 tests gọi trực tiếp pass; py_compile, Jinja parse, Compose config và diff check pass. | ## Kiến thức thu thập - `OS_CONFIG` định nghĩa đúng cặp checklist/config cho Windows, CentOS, RHEL, Ubuntu, Oracle Linux. @@ -42,6 +49,7 @@ Bổ sung upload/thay thế checklist XLSX và rule JSON trên trang quản lý ## Rủi ro / Vướng mắc - File checklist/rule sai nội dung có thể làm xử lý audit lỗi; giảm thiểu bằng validation JSON, extension, backup và atomic replace. +- `pytest` package chưa có trong runtime local; test suite đã chạy bằng deterministic lightweight runner, gồm fixture `tmp_path`, tổng cộng 23 tests pass. ## Lệnh thường dùng - `python -m py_compile app/web_app_fastapi.py` diff --git a/app/registry_service.py b/app/registry_service.py new file mode 100644 index 0000000..0aec7f8 --- /dev/null +++ b/app/registry_service.py @@ -0,0 +1,402 @@ +"""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 diff --git a/app/templates/admin_tools.html b/app/templates/admin_tools.html index b0739dc..d1a090e 100644 --- a/app/templates/admin_tools.html +++ b/app/templates/admin_tools.html @@ -428,6 +428,54 @@ {% if user_info and user_info.role == 'admin' %} +
+

🧩 Registry Category → Component → Profile → Version

+

Các phiên bản là bất biến; chỉ phiên bản active được dùng khi audit. Mọi thao tác bên dưới chỉ dành cho admin.

+
+
+
+
+ +
+
+
+
+
+ +
+
+
+
+
+
+ +
+
+ {% for category in registry.categories %} +

{{ category.name }} {{ category.id }}

+ {% for component in category.components %} +

{{ component.name }} {{ component.id }}

+ {% for profile in component.profiles %} +
+ {{ profile.name }} {{ profile.id }} +
+
+
+
+
+ +
+ +
+ {% for version in profile.versions %} + {{ version.version }}{% if version.active %} ✅ active{% else %} +
{% endif %}
+ {% endfor %} +
+
+ {% endfor %}{% endfor %}{% endfor %} +
+
@@ -681,14 +729,28 @@ e.stopPropagation(); } - ['dragenter', 'dragover'].forEach(eventName => { - uploadForm.addEventListener(eventName, () => uploadForm.classList.add('dragover'), false); - }); + if (uploadForm) { + ['dragenter', 'dragover'].forEach(eventName => uploadForm.addEventListener(eventName, () => uploadForm.classList.add('dragover'), false)); + ['dragleave', 'drop'].forEach(eventName => uploadForm.addEventListener(eventName, () => uploadForm.classList.remove('dragover'), false)); + } - ['dragleave', 'drop'].forEach(eventName => { - uploadForm.addEventListener(eventName, () => uploadForm.classList.remove('dragover'), false); - }); + document.querySelectorAll('.registry-json-form').forEach(form => form.addEventListener('submit', async event => { + event.preventDefault(); + const body = Object.fromEntries(new FormData(form).entries()); + if (body.aliases !== undefined) body.aliases = body.aliases.split(',').map(x => x.trim()).filter(Boolean); + const response = await fetch(form.dataset.endpoint, {method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify(body)}); + const result = await response.json(); + if (!response.ok) return alert(result.detail || 'Registry update failed'); + location.reload(); + })); + document.querySelectorAll('.registry-action-form').forEach(form => form.addEventListener('submit', async event => { + event.preventDefault(); + const response = await fetch(form.action, {method: 'POST', body: new FormData(form)}); + const result = await response.json(); + if (!response.ok) return alert(result.detail || 'Registry action failed'); + location.reload(); + })); - \ No newline at end of file + diff --git a/app/templates/index.html b/app/templates/index.html index ee2351f..cff94b5 100644 --- a/app/templates/index.html +++ b/app/templates/index.html @@ -1309,6 +1309,29 @@

+
+ + +
+
+ + +
+
+ + +

Chọn profile cho OS, Database, Web Server, Network Device...; để trống vẫn giữ auto-detect OS tương thích cũ.

+
+ @@ -1453,11 +1476,45 @@ const form = document.getElementById('uploadForm'); const loadingOverlay = document.getElementById('loadingOverlay'); const osSelect = document.getElementById('os_type'); + const profileSelect = document.getElementById('profile_id'); + const categorySelect = document.getElementById('profile_category'); + const componentSelect = document.getElementById('profile_component'); const fileInput = document.getElementById('encrypted_files'); const autoDetectInfo = document.getElementById('autoDetectInfo'); const autoDetectIcon = document.getElementById('autoDetectIcon'); const autoDetectMessage = document.getElementById('autoDetectMessage'); + const profileOptions = [...profileSelect.options].filter(option => option.value).map(option => ({ + value: option.value, + label: option.textContent, + category: option.dataset.category, + component: option.dataset.component + })); + const categories = [...new Set(profileOptions.map(option => option.category))]; + categories.forEach(category => categorySelect.add(new Option(category, category))); + + function refreshComponents() { + componentSelect.innerHTML = ''; + const components = [...new Set(profileOptions + .filter(option => !categorySelect.value || option.category === categorySelect.value) + .map(option => option.component))]; + components.forEach(component => componentSelect.add(new Option(component, component))); + refreshProfiles(); + } + + function refreshProfiles() { + const selected = profileSelect.value; + profileSelect.innerHTML = ''; + profileOptions.filter(option => + (!categorySelect.value || option.category === categorySelect.value) && + (!componentSelect.value || option.component === componentSelect.value) + ).forEach(option => profileSelect.add(new Option(option.label, option.value))); + if ([...profileSelect.options].some(option => option.value === selected)) profileSelect.value = selected; + } + categorySelect.addEventListener('change', refreshComponents); + componentSelect.addEventListener('change', refreshProfiles); + refreshComponents(); + // OS name display map (mirrors OS_CONFIG) const osNames = { 'centos': 'CentOS', @@ -1522,7 +1579,7 @@ } // Nếu người dùng đã chọn thủ công một OS trên form chính (không phải auto) - if (osSelect.value !== 'auto') { + if (profileSelect.value || osSelect.value !== 'auto') { // Trang trung gian 2 luôn (Xử lý ngay) startGeneratingReport(); return; @@ -1656,4 +1713,4 @@ - \ No newline at end of file + diff --git a/app/web_app_fastapi.py b/app/web_app_fastapi.py index f49a213..db5c628 100644 --- a/app/web_app_fastapi.py +++ b/app/web_app_fastapi.py @@ -14,7 +14,7 @@ import audit_decryption from colorama import Fore import uvicorn import secrets -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone import json import socket import struct @@ -27,6 +27,9 @@ import re import requests as http_requests from authenticator.radius_login import check_radius_login, check_radius_login_extended from config_upload import install_config_files, normalize_json, validate_xlsx +from registry_service import ( + RegistryError, RegistryNotFoundError, RegistryService, RegistryValidationError +) # Database imports import models @@ -188,6 +191,35 @@ OS_CONFIG = { } } +# Versioned registry. CONFIG_FOLDER is already a persistent Docker bind mount. +REGISTRY_ROOT = os.getenv("AUDIT_REGISTRY_ROOT", os.path.join(CONFIG_FOLDER, "registry")) +registry_service = RegistryService(REGISTRY_ROOT, legacy_config_root=CONFIG_FOLDER) +try: + registry_service.seed_legacy_profiles() +except RegistryError as exc: + # Keep the application bootable so an administrator can repair a bad mount. + print(Fore.RED + f"[REGISTRY] Initialization failed: {exc}" + Fore.RESET) + + +def resolve_audit_profile(os_type: Optional[str] = None, profile_id: Optional[str] = None): + """Resolve explicit profile selection or a legacy OS alias to its active version.""" + return registry_service.resolve(os_type=os_type, profile_id=profile_id) + + +def active_profile_options(): + """Flat active-profile list used by the audit form without exposing filesystem paths.""" + result = [] + for category in registry_service.public_catalog(active_only=True).get("categories", []): + for component in category.get("components", []): + for profile in component.get("profiles", []): + active = profile.get("versions", []) + if active: + result.append({"id": profile["id"], "name": profile["name"], + "category_id": category["id"], "category": category["name"], + "component_id": component["id"], "component": component["name"], + "version": active[0]["version"]}) + return result + def detect_os_from_filename(filename: str) -> Optional[str]: """ Detect OS type from filename pattern. @@ -285,6 +317,11 @@ class CompatTemplates(Jinja2Templates): raise ValueError("request is required in TemplateResponse context") if "docs_list" not in context and "get_available_docs" in globals(): context["docs_list"] = get_available_docs() + if "active_profiles" not in context and "registry_service" in globals(): + try: + context["active_profiles"] = active_profile_options() + except RegistryError: + context["active_profiles"] = [] # Call new-style API: super().TemplateResponse(request, name, context, ...) return super().TemplateResponse(request=request, name=name, context=context, **kwargs) @@ -1130,6 +1167,7 @@ async def index(request: Request): { "request": request, "os_config": OS_CONFIG, + "registry": registry_service.public_catalog(), "message": None, "is_logged_in": is_logged_in, "user_info": user_info, @@ -1474,7 +1512,8 @@ async def api_detect_os_from_content(files: List[UploadFile] = File(...)): "message": f"Error: {str(e)}" } -def process_files_background(task_id: str, session_id: str, os_type: str, enc_paths: List[str], base_url: str): +def process_files_background(task_id: str, session_id: str, os_type: str, enc_paths: List[str], base_url: str, + profile_id: Optional[str] = None): """Background task to process uploaded files""" try: tasks_status = processing_tasks.get(task_id) @@ -1485,10 +1524,10 @@ def process_files_background(task_id: str, session_id: str, os_type: str, enc_pa tasks_status['start_time'] = time.time() # Validation checks - if os_type != 'auto' and os_type not in OS_CONFIG: - tasks_status['status'] = 'error' - tasks_status['message'] = "Please select a valid OS type!" - return + if profile_id: + resolve_audit_profile(profile_id=profile_id) + elif os_type != 'auto': + resolve_audit_profile(os_type=os_type) if not enc_paths: tasks_status['status'] = 'error' @@ -1519,19 +1558,20 @@ def process_files_background(task_id: str, session_id: str, os_type: str, enc_pa # Fallback to content detection file_os_type = detect_os_from_decrypted_file(enc_path) - if not file_os_type or file_os_type not in OS_CONFIG: + if not profile_id and (not file_os_type or file_os_type not in OS_CONFIG): print(Fore.YELLOW + f"[BACKGROUND] Cannot detect OS for {enc_filename}, skipping." + Fore.RESET) continue - - os_info = OS_CONFIG[file_os_type] - checklist_path = os.path.join(CONFIG_FOLDER, os_info['checklist']) - config_path = os.path.join(CONFIG_FOLDER, os_info['config']) + + resolved = resolve_audit_profile(profile_id=profile_id) if profile_id else resolve_audit_profile(os_type=file_os_type) + checklist_path = resolved['checklist_path'] + config_path = resolved['rules_path'] + display_name = resolved['profile']['name'] if not os.path.exists(checklist_path) or not os.path.exists(config_path): print(Fore.RED + f"[BACKGROUND] Missing checklist/config for {file_os_type}, skipping." + Fore.RESET) continue - print(Fore.BLUE + f"\n=== [BACKGROUND] Processing: {enc_filename} ({os_info['name']}) ===" + Fore.RESET) + print(Fore.BLUE + f"\n=== [BACKGROUND] Processing: {enc_filename} ({display_name}) ===" + Fore.RESET) original_dir = os.getcwd() try: @@ -1660,6 +1700,7 @@ async def api_upload_async( request: Request, background_tasks: BackgroundTasks, os_type: str = Form(...), + profile_id: Optional[str] = Form(None), encrypted_files: List[UploadFile] = File(...) ): # Authenticated users and explicit dev mode may upload; admin routes stay protected. @@ -1696,7 +1737,7 @@ async def api_upload_async( try: # Detect OS dynamically from filenames (fast, no decryption needed) - if not os_type or os_type == 'auto': + if not profile_id and (not os_type or os_type == 'auto'): filenames = [f.filename for f in encrypted_files if f.filename] detected_os, all_same = detect_os_from_multiple_files(filenames) if detected_os and all_same: @@ -1720,7 +1761,7 @@ async def api_upload_async( enc_paths.append(enc_path) base_url = f"{request.url.scheme}://{request.url.netloc}" - background_tasks.add_task(process_files_background, task_id, session_id, os_type, enc_paths, base_url) + background_tasks.add_task(process_files_background, task_id, session_id, os_type, enc_paths, base_url, profile_id) return {"task_id": task_id, "status": "pending"} except Exception as e: @@ -1786,6 +1827,7 @@ async def task_result(request: Request, task_id: str): async def upload_files( request: Request, os_type: str = Form(...), + profile_id: Optional[str] = Form(None), encrypted_files: List[UploadFile] = File(...) ): # Tạo response object để set cookie @@ -1808,7 +1850,7 @@ async def upload_files( _is_logged_in_ctx = _user_info_ctx is not None # Auto-detect OS from filenames (fast, no decryption needed) - if not os_type or os_type == 'auto': + if not profile_id and (not os_type or os_type == 'auto'): print(Fore.CYAN + "[AUTO-DETECT] Mode: auto – detecting OS from filenames..." + Fore.RESET) filenames = [f.filename for f in encrypted_files if f.filename] detected_os, all_same = detect_os_from_multiple_files(filenames) @@ -1830,7 +1872,7 @@ async def upload_files( ) # Validate OS type - if os_type not in OS_CONFIG: + if not profile_id and os_type not in OS_CONFIG: return templates.TemplateResponse( "index.html", { @@ -1852,9 +1894,15 @@ async def upload_files( ) # Get checklist and config paths - os_info = OS_CONFIG[os_type] - checklist_path = os.path.join(CONFIG_FOLDER, os_info['checklist']) - config_path = os.path.join(CONFIG_FOLDER, os_info['config']) + try: + resolved_profile = resolve_audit_profile(os_type=os_type, profile_id=profile_id) + except RegistryError as exc: + return templates.TemplateResponse("index.html", {"request": request, "os_config": OS_CONFIG, + "message": {"type": "error", "text": str(exc)}}) + checklist_path = resolved_profile['checklist_path'] + config_path = resolved_profile['rules_path'] + os_info = {"name": resolved_profile['profile']['name'], + "checklist": os.path.basename(checklist_path), "config": os.path.basename(config_path)} # Verify files exist if not os.path.exists(checklist_path): @@ -2578,6 +2626,97 @@ def require_tool_manager_session(request: Request): return user return None + +def _admin_or_403(request: Request): + user = require_admin_session(request) + if not user: + raise HTTPException(status_code=403, detail="Admin role required") + return user + + +@app.get("/api/v2/registry", name="registry_catalog_v2") +async def registry_catalog_v2(active_only: bool = False): + """Read-only registry catalog. Artifact paths are intentionally not exposed.""" + return registry_service.public_catalog(active_only=active_only) + + +@app.get("/api/v2/profiles/active", name="active_profiles_v2") +async def active_profiles_v2(): + return {"profiles": active_profile_options()} + + +@app.post("/api/v2/admin/categories", name="registry_create_category_v2") +async def registry_create_category_v2(request: Request): + _admin_or_403(request) + body = await request.json() + try: + return registry_service.create_category(body.get("id", ""), body.get("name", "")) + except RegistryError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/v2/admin/components", name="registry_create_component_v2") +async def registry_create_component_v2(request: Request): + _admin_or_403(request) + body = await request.json() + try: + return registry_service.create_component(body.get("category_id", ""), body.get("id", ""), body.get("name", "")) + except RegistryNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except RegistryError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/v2/admin/profiles", name="registry_create_profile_v2") +async def registry_create_profile_v2(request: Request): + _admin_or_403(request) + body = await request.json() + try: + return registry_service.create_profile(body.get("component_id", ""), body.get("id", ""), + body.get("name", ""), body.get("aliases", [])) + except RegistryNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except RegistryError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/v2/admin/profiles/{profile_id}/versions", name="registry_upload_version_v2") +async def registry_upload_version_v2( + profile_id: str, request: Request, version: str = Form(...), activate: bool = Form(False), + checklist_file: UploadFile = File(...), config_file: UploadFile = File(...), +): + user = _admin_or_403(request) + try: + return registry_service.create_version(profile_id, version, await checklist_file.read(), + await config_file.read(), activate=activate, + created_by=user.get("username", "admin")) + except RegistryNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except RegistryError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/v2/admin/profiles/{profile_id}/versions/{version}/activate", name="registry_activate_v2") +async def registry_activate_v2(profile_id: str, version: str, request: Request): + _admin_or_403(request) + try: + return registry_service.activate_version(profile_id, version) + except RegistryNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except RegistryError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@app.post("/api/v2/admin/profiles/{profile_id}/rollback", name="registry_rollback_v2") +async def registry_rollback_v2(profile_id: str, request: Request): + _admin_or_403(request) + try: + return registry_service.rollback(profile_id) + except RegistryNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except RegistryError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + @app.get("/admin/tools", response_class=HTMLResponse, name="admin_tools") async def admin_tools_page(request: Request, msg: str = None, msg_type: str = "success"): """Admin page for managing tools - requires session authentication""" @@ -2618,6 +2757,7 @@ async def admin_tools_page(request: Request, msg: str = None, msg_type: str = "s "os_metadata": OS_METADATA, "os_config": OS_CONFIG, "message": message, + "registry": registry_service.public_catalog(), "user": user, "user_info": user, "is_logged_in": True, @@ -2971,7 +3111,22 @@ async def admin_upload_checklist_config( normalized_config = normalize_json(await config_file.read()) pending_files.append((os_info['config'], normalized_config)) - install_config_files(CONFIG_FOLDER, pending_files) + # Compatibility endpoint: translate the fixed OS form into a new immutable + # profile version. ``install_config_files`` remains available for old + # deployments, but registry artifacts are now the authoritative source. + resolved = resolve_audit_profile(os_type=os_type) + with open(resolved['checklist_path'], 'rb') as stream: + active_checklist = stream.read() + with open(resolved['rules_path'], 'rb') as stream: + active_rules = stream.read() + checklist_content = next((content for name, content in pending_files if name == os_info['checklist']), + active_checklist) + normalized_config = next((content for name, content in pending_files if name == os_info['config']), + active_rules) + version_id = datetime.now(timezone.utc).strftime("upload-%Y%m%dT%H%M%S%fZ") + registry_service.create_version(resolved['profile']['id'], version_id, checklist_content, + normalized_config, activate=True, + created_by=user.get('username', 'admin')) print(Fore.GREEN + f"[ADMIN] Checklist/config updated for {os_type} by {user.get('username')}" + Fore.RESET) changed = ', '.join(name for name, _ in pending_files) @@ -4067,6 +4222,7 @@ def get_client_ip(request: Request) -> str: async def api_upload_files( request: Request, os_type: str = Form(...), + profile_id: Optional[str] = Form(None), api_key: str = Form(...), encrypted_files: List[UploadFile] = File(...) ): @@ -4093,7 +4249,7 @@ async def api_upload_files( try: # Auto-detect OS from filename if os_type is 'auto' - if not os_type or os_type == 'auto': + if not profile_id and (not os_type or os_type == 'auto'): filenames = [f.filename for f in encrypted_files if f.filename] detected_os, _ = detect_os_from_multiple_files(filenames) if detected_os: @@ -4103,16 +4259,20 @@ async def api_upload_files( raise HTTPException(status_code=400, detail="Cannot auto-detect OS. Please specify os_type.") # Validate OS type - if os_type not in OS_CONFIG: + if not profile_id and os_type not in OS_CONFIG: raise HTTPException(status_code=400, detail=f"Invalid os_type: {os_type}") # Validate file exists if not encrypted_files: raise HTTPException(status_code=400, detail="No encrypted files uploaded") - os_info = OS_CONFIG[os_type] - checklist_path = os.path.join(CONFIG_FOLDER, os_info['checklist']) - config_path = os.path.join(CONFIG_FOLDER, os_info['config']) + try: + resolved_profile = resolve_audit_profile(os_type=os_type, profile_id=profile_id) + except RegistryError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + checklist_path = resolved_profile['checklist_path'] + config_path = resolved_profile['rules_path'] + os_info = {"checklist": os.path.basename(checklist_path), "config": os.path.basename(config_path)} # Verify exists if not os.path.exists(checklist_path): @@ -4187,6 +4347,8 @@ async def api_upload_files( "processed": len(output_files), "output_files": output_files, "os_type": os_type, + "profile_id": resolved_profile["profile"]["id"], + "profile_version": resolved_profile["version"]["version"], "download_urls": [f"/api/download/{f}" for f in output_files] } diff --git a/docker-compose.yml b/docker-compose.yml index 23a8553..78a0e14 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -23,6 +23,8 @@ services: # Checklist/rule config files (read-write for admin-managed updates) - ./config:/app/config + # Versioned multi-component registry artifacts + - ./registry:/app/registry - ./keys:/app/keys:ro - ./users_config.json:/app/users_config.json - ./radius_config.json:/app/radius_config.json:ro @@ -44,6 +46,7 @@ services: # Dev mode: unauthenticated upload; admin routes remain protected. # Set to false for production. - AUDIT_DEV_MODE=${AUDIT_DEV_MODE:-false} + - AUDIT_REGISTRY_ROOT=/app/registry restart: unless-stopped diff --git a/tests/test_multi_component_registry_integration.py b/tests/test_multi_component_registry_integration.py new file mode 100644 index 0000000..8bd5e9a --- /dev/null +++ b/tests/test_multi_component_registry_integration.py @@ -0,0 +1,47 @@ +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +APP = (ROOT / "app" / "web_app_fastapi.py").read_text(encoding="utf-8") +INDEX = (ROOT / "app" / "templates" / "index.html").read_text(encoding="utf-8") +ADMIN = (ROOT / "app" / "templates" / "admin_tools.html").read_text(encoding="utf-8") +COMPOSE = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") + + +def test_upload_endpoints_accept_profile_id_and_keep_os_type(): + assert APP.count("profile_id: Optional[str] = Form(None)") >= 3 + assert "resolve_audit_profile(os_type=os_type, profile_id=profile_id)" in APP + assert '"os_type": os_type' in APP + + +def test_v2_registry_routes_and_admin_mutations_exist(): + for route in [ + "/api/v2/registry", + "/api/v2/profiles/active", + "/api/v2/admin/categories", + "/api/v2/admin/components", + "/api/v2/admin/profiles", + "/versions/{version}/activate", + "/rollback", + ]: + assert route in APP + assert "_admin_or_403(request)" in APP + + +def test_audit_form_supports_category_component_profile_selection(): + assert 'id="profile_category"' in INDEX + assert 'id="profile_component"' in INDEX + assert 'name="profile_id"' in INDEX + assert "refreshComponents" in INDEX + assert "refreshProfiles" in INDEX + + +def test_admin_ui_supports_registry_version_lifecycle(): + assert "Registry Category → Component → Profile → Version" in ADMIN + assert "registry_upload_version_v2" in ADMIN + assert "registry_activate_v2" in ADMIN + assert "registry_rollback_v2" in ADMIN + + +def test_registry_has_dedicated_persistent_mount(): + assert "./registry:/app/registry" in COMPOSE + assert "AUDIT_REGISTRY_ROOT=/app/registry" in COMPOSE diff --git a/tests/test_registry_integration_contract.py b/tests/test_registry_integration_contract.py new file mode 100644 index 0000000..6f4ec7a --- /dev/null +++ b/tests/test_registry_integration_contract.py @@ -0,0 +1,42 @@ +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +APP = (ROOT / "app" / "web_app_fastapi.py").read_text(encoding="utf-8") +INDEX = (ROOT / "app" / "templates" / "index.html").read_text(encoding="utf-8") +ADMIN = (ROOT / "app" / "templates" / "admin_tools.html").read_text(encoding="utf-8") +COMPOSE = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") + + +def test_upload_contract_keeps_os_type_and_adds_optional_profile_id(): + for function_name in ("api_upload_async", "upload_files", "api_upload_files"): + function = APP.split(f"async def {function_name}(", 1)[1].split("):", 1)[0] + assert "os_type: str = Form(...)" in function + assert "profile_id: Optional[str] = Form(None)" in function + + +def test_registry_mutations_are_admin_guarded_and_analyzer_remains_tool_only(): + for function_name in ( + "registry_create_category_v2", "registry_create_component_v2", "registry_create_profile_v2", + "registry_upload_version_v2", "registry_activate_v2", "registry_rollback_v2", + ): + function = APP.split(f"async def {function_name}(", 1)[1].split("@app.", 1)[0] + assert "_admin_or_403(request)" in function + tool_upload = APP.split("async def admin_upload_tool(", 1)[1].split("@app.", 1)[0] + assert "require_tool_manager_session(request)" in tool_upload + + +def test_audit_and_admin_ui_expose_profile_registry_by_role(): + assert 'name="profile_id"' in INDEX + assert "active_profiles" in INDEX + admin_block = ADMIN.split("user_info.role == 'admin'", 1)[1] + assert "profile-registry" in admin_block + assert "registry_upload_version_v2" in admin_block + assert "registry_activate_v2" in admin_block + assert "registry_rollback_v2" in admin_block + + +def test_registry_uses_persistent_config_mount(): + assert "./config:/app/config" in COMPOSE + assert "./registry:/app/registry" in COMPOSE + assert "AUDIT_REGISTRY_ROOT=/app/registry" in COMPOSE diff --git a/tests/test_registry_service.py b/tests/test_registry_service.py new file mode 100644 index 0000000..574879c --- /dev/null +++ b/tests/test_registry_service.py @@ -0,0 +1,85 @@ +import hashlib +import io +import json +import zipfile +from pathlib import Path + +import pytest + +from app.registry_service import RegistryService, RegistryValidationError + + +def minimal_xlsx(marker: str = "v1") -> bytes: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr("[Content_Types].xml", "") + archive.writestr("xl/workbook.xml", f"{marker}") + return buffer.getvalue() + + +def make_legacy_config(root: Path) -> None: + names = { + "Windows_Checklist.xlsx": "windows_config.json", + "CentOS_Checklist_NEW.xlsx": "centos_config_new.json", + "RHEL_Checklist_NEW.xlsx": "rhel_config_new.json", + "Ubuntu_Checklist_NEW.xlsx": "ubuntu_config_new.json", + "Oracle_Linux_Checklist.xlsx": "oracle_linux_config.json", + } + for checklist, rules in names.items(): + (root / checklist).write_bytes(minimal_xlsx(checklist)) + (root / rules).write_text('{"rules": []}', encoding="utf-8") + + +def test_seed_migrates_five_operating_system_profiles_and_aliases(tmp_path): + make_legacy_config(tmp_path) + service = RegistryService(tmp_path / "registry", legacy_config_root=tmp_path) + + registry = service.seed_legacy_profiles() + + assert len(registry["categories"]) == 1 + assert len(registry["categories"][0]["components"]) == 5 + assert service.resolve(os_type="redhat")["profile"]["id"] == "rhel-baseline" + resolved = service.resolve(os_type="windows") + assert resolved["version"]["active"] is True + assert Path(resolved["checklist_path"]).is_file() + + +def test_create_version_validates_payloads_and_records_sha256(tmp_path): + make_legacy_config(tmp_path) + service = RegistryService(tmp_path / "registry", legacy_config_root=tmp_path) + service.seed_legacy_profiles() + checklist = minimal_xlsx("v2") + rules = b'{"rules": [{"id": "A"}]}' + + version = service.create_version("ubuntu-baseline", "2.0", checklist, rules) + + assert version["checksums"]["checklist"] == hashlib.sha256(checklist).hexdigest() + assert json.loads(Path(service.root / version["artifacts"]["rules"]).read_text())["rules"][0]["id"] == "A" + with pytest.raises(RegistryValidationError): + service.create_version("ubuntu-baseline", "bad", b"not xlsx", rules) + + +def test_activation_and_rollback_are_versioned_and_persistent(tmp_path): + make_legacy_config(tmp_path) + service = RegistryService(tmp_path / "registry", legacy_config_root=tmp_path) + service.seed_legacy_profiles() + service.create_version("centos-baseline", "2.0", minimal_xlsx("v2"), b'{"rules": []}') + + service.activate_version("centos-baseline", "2.0") + assert service.resolve(profile_id="centos-baseline")["version"]["version"] == "2.0" + service.rollback("centos-baseline") + assert service.resolve(profile_id="centos-baseline")["version"]["version"] == "legacy-1" + + +def test_rejects_bad_ids_duplicates_and_checksum_tampering(tmp_path): + make_legacy_config(tmp_path) + service = RegistryService(tmp_path / "registry", legacy_config_root=tmp_path) + service.seed_legacy_profiles() + with pytest.raises(RegistryValidationError): + service.create_category("../escape", "Escape") + with pytest.raises(RegistryValidationError): + service.create_component("operating-systems", "windows", "Duplicate") + resolved = service.resolve(os_type="ubuntu") + Path(resolved["checklist_path"]).write_bytes(b"tampered") + with pytest.raises(RegistryValidationError, match="checksum"): + service.resolve(os_type="ubuntu")