refactor: consolidate checklist management into registry

.hermes/plans/2026-09-04_091241-upload-checklist-config.md

@@ -0,0 +1,38 @@
+# Upload Checklist Configuration Implementation Plan
+
+> **For Hermes:** Implement task-by-task with TDD and verify the Docker deployment contract.
+
+**Goal:** Bổ sung tại trang quản lý công cụ chức năng admin upload/thay thế cặp file checklist `.xlsx` và rule config `.json` cho từng OS.
+
+**Architecture:** Dùng mapping `OS_CONFIG` hiện có làm allowlist tên đích; không cho client tự chọn đường dẫn/tên file. Backend validate role admin, extension, JSON, tạo `.bak`, rồi ghi file vào `config/`. Đổi bind mount `config` sang read-write để chức năng hoạt động trong Docker.
+
+**Tech Stack:** FastAPI, Jinja2, Docker Compose, Python stdlib JSON/file I/O.
+
+---
+
+### Task 1: Add regression tests
+- Create `tests/test_checklist_config_upload.py`.
+- Assert route exists, admin authorization is required, safe OS mapping is used, JSON is validated, and Compose config mount is writable.
+- Run test and confirm RED.
+
+### Task 2: Implement secure upload endpoint
+- Modify `app/web_app_fastapi.py`.
+- Add `POST /admin/checklists/upload`.
+- Require admin session.
+- Accept `os_type`, optional checklist `.xlsx`, optional config `.json`; require at least one.
+- Resolve destination only from `OS_CONFIG`.
+- Validate JSON before write; write temporary file then atomic replace; back up existing files.
+- Redirect with PRG status message.
+
+### Task 3: Add admin UI
+- Modify `app/templates/admin_tools.html`.
+- Add admin-only card showing current file names/status and multipart upload form.
+- Analyzer remains limited to tool upload and cannot change checklist/rules.
+
+### Task 4: Enable persistence in Docker
+- Modify `docker-compose.yml` from `./config:/app/config:ro` to `./config:/app/config`.
+- Keep `config/` as the only writable configuration directory required by this feature.
+
+### Task 5: Verify
+- Run regression tests, Python compile/AST, Jinja parse, JSON checks, `docker compose config`, and `git diff --check`.
+- Grok execution review; fix any blocking findings.

PROJECT_STATUS.md

@@ -0,0 +1,57 @@
+# PROJECT STATUS — Audit Hardening Tool
+
+> 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`)
+- [x] Phase 2 — Xác định hướng xây dựng
+- [x] Phase 3 — Thu thập kiến thức
+- [x] Phase 4 — Thực hiện
+- [x] Phase 5 — Tracking
+
+## Tổng quan
+Đã 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
+- 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ú |
+|-------|------|-----------|---------|---------|
+| Plan | 2026-09-04 | grok-build | BLOCKED | Grok CLI chưa đăng nhập |
+| 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.
+- `config/` hiện bind mount read-only nên phải đổi sang RW để upload persistent.
+- `require_admin_session` là guard phù hợp; analyzer không nên sửa rules/checklist.
+
+## 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`
+- `docker compose config`
+- `git diff --check`

app/config_upload.py

@@ -0,0 +1,49 @@
+import json
+import os
+import shutil
+import zipfile
+from io import BytesIO
+from pathlib import Path
+from typing import Iterable
+
+
+def validate_xlsx(content: bytes) -> bytes:
+    """Validate that content is an XLSX-compatible ZIP containing workbook metadata."""
+    try:
+        with zipfile.ZipFile(BytesIO(content)) as archive:
+            names = set(archive.namelist())
+    except (zipfile.BadZipFile, OSError) as exc:
+        raise ValueError("Checklist không phải file XLSX hợp lệ") from exc
+    if "[Content_Types].xml" not in names or "xl/workbook.xml" not in names:
+        raise ValueError("Checklist không phải file XLSX hợp lệ")
+    return content
+
+
+def normalize_json(content: bytes) -> bytes:
+    """Parse and normalize an object/array JSON configuration."""
+    try:
+        parsed = json.loads(content.decode("utf-8-sig"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError(f"Rule config JSON không hợp lệ: {exc}") from exc
+    if not isinstance(parsed, (dict, list)):
+        raise ValueError("Rule config JSON phải là object hoặc array")
+    return json.dumps(parsed, ensure_ascii=False, indent=2).encode("utf-8") + b"\n"
+
+
+def install_config_files(config_root: str | Path, files: Iterable[tuple[str, bytes]]) -> None:
+    """Atomically replace allowlisted config files and keep one .bak copy."""
+    root = Path(config_root)
+    root.mkdir(parents=True, exist_ok=True)
+    for target_name, content in files:
+        if Path(target_name).name != target_name:
+            raise ValueError("Tên file đích không hợp lệ")
+        target = root / target_name
+        temporary = root / f"{target_name}.tmp"
+        backup = root / f"{target_name}.bak"
+        with temporary.open("wb") as output:
+            output.write(content)
+            output.flush()
+            os.fsync(output.fileno())
+        if target.exists():
+            shutil.copy2(target, backup)
+        os.replace(temporary, target)

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

app/templates/admin_tools.html

@@ -428,6 +428,54 @@
         </div>
 
         {% if user_info and user_info.role == 'admin' %}
+        <div class="card" id="profile-registry">
+            <div class="card-header"><h2 class="card-title">🧩 Registry Category → Component → Profile → Version</h2></div>
+            <p style="color: var(--text-muted);">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.</p>
+            <div class="form-row">
+                <form class="registry-json-form" data-endpoint="{{ url_for('registry_create_category_v2') }}">
+                    <div class="form-group"><label>Category ID</label><input name="id" placeholder="database-platforms" required></div>
+                    <div class="form-group"><label>Tên category</label><input name="name" required></div>
+                    <button class="btn btn-primary" type="submit">Thêm category</button>
+                </form>
+                <form class="registry-json-form" data-endpoint="{{ url_for('registry_create_component_v2') }}">
+                    <div class="form-group"><label>Category ID</label><input name="category_id" required></div>
+                    <div class="form-group"><label>Component ID</label><input name="id" required></div>
+                    <div class="form-group"><label>Tên component</label><input name="name" required></div>
+                    <button class="btn btn-primary" type="submit">Thêm component</button>
+                </form>
+                <form class="registry-json-form" data-endpoint="{{ url_for('registry_create_profile_v2') }}">
+                    <div class="form-group"><label>Component ID</label><input name="component_id" required></div>
+                    <div class="form-group"><label>Profile ID</label><input name="id" required></div>
+                    <div class="form-group"><label>Tên profile</label><input name="name" required></div>
+                    <div class="form-group"><label>Aliases (phân cách dấu phẩy)</label><input name="aliases"></div>
+                    <button class="btn btn-primary" type="submit">Thêm profile</button>
+                </form>
+            </div>
+            {% for category in registry.categories %}
+            <h3>{{ category.name }} <code>{{ category.id }}</code></h3>
+            {% for component in category.components %}
+            <h4>{{ component.name }} <code>{{ component.id }}</code></h4>
+            {% for profile in component.profiles %}
+            <div style="border:1px solid var(--border-color);padding:14px;margin:10px 0;border-radius:8px;">
+                <strong>{{ profile.name }}</strong> <code>{{ profile.id }}</code>
+                <form action="{{ url_for('registry_upload_version_v2', profile_id=profile.id) }}" method="post" enctype="multipart/form-data" class="upload-form registry-action-form">
+                    <div class="form-row">
+                        <div class="form-group"><label>Version</label><input name="version" required></div>
+                        <div class="form-group"><label>Checklist XLSX</label><input type="file" name="checklist_file" accept=".xlsx" required></div>
+                        <div class="form-group"><label>Rule JSON</label><input type="file" name="config_file" accept=".json" required></div>
+                        <label><input type="checkbox" name="activate" value="true"> Kích hoạt ngay</label>
+                    </div>
+                    <button class="btn btn-primary" type="submit">Upload version</button>
+                </form>
+                {% for version in profile.versions %}
+                <span style="display:inline-block;margin:5px;"><code>{{ version.version }}</code>{% if version.active %} ✅ active{% else %}
+                    <form class="registry-action-form" style="display:inline" action="{{ url_for('registry_activate_v2', profile_id=profile.id, version=version.version) }}" method="post"><button class="btn" type="submit">Activate</button></form>{% endif %}</span>
+                {% endfor %}
+                <form class="registry-action-form" style="display:inline" action="{{ url_for('registry_rollback_v2', profile_id=profile.id) }}" method="post"><button class="btn" type="submit">Rollback</button></form>
+            </div>
+            {% endfor %}{% endfor %}{% endfor %}
+        </div>
+
         <!-- Upload Documentation Form -->
         <div class="card">
             <div class="card-header">
@@ -649,13 +697,27 @@
             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();
+        }));
     </script>
 </body>
 

app/templates/index.html

@@ -1309,6 +1309,29 @@
                 </p>
             </div>
 
+            <div class="form-group">
+                <label for="profile_category">2. Loại thành phần đánh giá</label>
+                <select id="profile_category">
+                    <option value="">Tất cả loại thành phần</option>
+                </select>
+            </div>
+            <div class="form-group">
+                <label for="profile_component">3. Thành phần</label>
+                <select id="profile_component">
+                    <option value="">Tất cả thành phần</option>
+                </select>
+            </div>
+            <div class="form-group">
+                <label for="profile_id">4. Assessment Profile (tùy chọn)</label>
+                <select id="profile_id" name="profile_id">
+                    <option value="">Dùng profile đang active theo OS tự động</option>
+                    {% for profile in active_profiles %}
+                    <option value="{{ profile.id }}" data-category="{{ profile.category_id }}" data-component="{{ profile.component_id }}">{{ profile.category }} → {{ profile.component }} → {{ profile.name }} ({{ profile.version }})</option>
+                    {% endfor %}
+                </select>
+                <p class="hint">Chọn profile cho OS, Database, Web Server, Network Device...; để trống vẫn giữ auto-detect OS tương thích cũ.</p>
+            </div>
+
             <button type="button" class="btn-submit" id="submitBtn" onclick="handleUploadClick()">
                 ⚡ Xử lý File
             </button>
@@ -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 = '<option value="">Tất cả thành phần</option>';
+            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 = '<option value="">Dùng profile đang active theo OS tự động</option>';
+            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;

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
@@ -26,6 +26,10 @@ import ipaddress
 import re
 import requests as http_requests
 from authenticator.radius_login import check_radius_login, check_radius_login_extended
+from config_upload import normalize_json, validate_xlsx
+from registry_service import (
+    RegistryError, RegistryNotFoundError, RegistryService, RegistryValidationError
+)
 
 # Database imports
 import models
@@ -187,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.
@@ -284,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)
 
@@ -1129,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,
@@ -1473,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)
@@ -1484,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'
@@ -1518,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:
@@ -1659,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.
@@ -1695,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:
@@ -1719,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:
@@ -1785,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
@@ -1807,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)
@@ -1829,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",
                 {
@@ -1851,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):
@@ -2577,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"""
@@ -2615,7 +2755,9 @@ async def admin_tools_page(request: Request, msg: str = None, msg_type: str = "s
             "request": request,
             "tools_list": get_available_tools(),
             "os_metadata": OS_METADATA,
+            "os_config": OS_CONFIG,
             "message": message,
+            "registry": registry_service.public_catalog(),
             "user": user,
             "user_info": user,
             "is_logged_in": True,
@@ -4008,6 +4150,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(...)
 ):
@@ -4034,7 +4177,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:
@@ -4044,16 +4187,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):
@@ -4128,6 +4275,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]
         }
 

docker-compose.yml

@@ -21,8 +21,10 @@ services:
       # Application source code (mounted from host for easy updates)
       - ./app:/app
 
-      # Config files (read-only) - mounted from host
-      - ./config:/app/config:ro
+      # 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
 

tests/test_checklist_config_upload.py

@@ -0,0 +1,26 @@
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = (ROOT / "app" / "web_app_fastapi.py").read_text(encoding="utf-8")
+TEMPLATE = (ROOT / "app" / "templates" / "admin_tools.html").read_text(encoding="utf-8")
+COMPOSE = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
+
+
+def test_legacy_checklist_upload_ui_and_route_are_removed():
+    assert "admin_checklists_upload" not in APP
+    assert "admin_checklists_upload" not in TEMPLATE
+    assert "Cập nhật Checklist & Rule Config" not in TEMPLATE
+
+
+def test_registry_is_the_single_checklist_rule_management_flow():
+    assert "registry_upload_version_v2" in APP
+    assert "registry_upload_version_v2" in TEMPLATE
+    assert 'name="checklist_file"' in TEMPLATE
+    assert 'name="config_file"' in TEMPLATE
+    assert "registry_activate_v2" in TEMPLATE
+    assert "registry_rollback_v2" in TEMPLATE
+
+
+def test_registry_mount_is_writable_and_persistent():
+    assert "./registry:/app/registry" in COMPOSE
+    assert "AUDIT_REGISTRY_ROOT=/app/registry" in COMPOSE

tests/test_config_upload_helpers.py

@@ -0,0 +1,48 @@
+import io
+import json
+import tempfile
+import zipfile
+from pathlib import Path
+
+from app.config_upload import install_config_files, normalize_json, validate_xlsx
+
+
+def _minimal_xlsx() -> bytes:
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w") as archive:
+        archive.writestr("[Content_Types].xml", "<Types/>")
+        archive.writestr("xl/workbook.xml", "<workbook/>")
+    return buffer.getvalue()
+
+
+def test_validate_xlsx_accepts_workbook_zip():
+    assert validate_xlsx(_minimal_xlsx()) == _minimal_xlsx()
+
+
+def test_validate_xlsx_rejects_arbitrary_pk_data():
+    try:
+        validate_xlsx(b"PK-not-a-real-xlsx")
+    except ValueError as exc:
+        assert "XLSX" in str(exc)
+    else:
+        raise AssertionError("invalid XLSX was accepted")
+
+
+def test_normalize_json_rejects_scalar_and_formats_object():
+    assert json.loads(normalize_json(b'{"a":1}')) == {"a": 1}
+    try:
+        normalize_json(b'"scalar"')
+    except ValueError as exc:
+        assert "object" in str(exc)
+    else:
+        raise AssertionError("scalar JSON was accepted")
+
+
+def test_install_config_files_replaces_and_backs_up_existing_file():
+    with tempfile.TemporaryDirectory() as directory:
+        root = Path(directory)
+        target = root / "rules.json"
+        target.write_bytes(b"old")
+        install_config_files(root, [("rules.json", b"new")])
+        assert target.read_bytes() == b"new"
+        assert (root / "rules.json.bak").read_bytes() == b"old"

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

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

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", "<Types/>")
+        archive.writestr("xl/workbook.xml", f"<workbook>{marker}</workbook>")
+    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")