feat: add versioned multi-component assessment registry
This commit is contained in:
@@ -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
|
||||
@@ -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
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user