from __future__ import annotations import copy import hashlib import importlib.util import json import os from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] MODULE_PATH = ROOT / "ops" / "gitea-deploy-broker" / "registered_legacy_adoption.py" REPOSITORY = "jens/quiz-monitor" REQUEST_ID = "6f9619ff-8b86-4011-b42d-00c04fc964ff" KEY = b"k" * 32 IMAGE_ID = "sha256:" + "b" * 64 CONTAINER_ID = "c" * 64 MISMATCH_KEY = "quiz-server/KNOWLEDGE_OPTIMIZER_MODEL" STATE = b'{"schema_version":3,"live_status":"unknown"}\n' SOURCE_FP = "1" * 64 RUNTIME_FP = "2" * 64 SECRET_VALUE = "gpt-6-luna" def _load_module(): spec = importlib.util.spec_from_file_location("registered_legacy_capture_test", MODULE_PATH) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module @pytest.fixture def module(): return _load_module() @pytest.fixture def project(tmp_path): return { "slug": "quiz-monitor", "branch": "main", "compose_project": "quiz-monitor", "compose_files": ["docker-compose.yml"], "containers": ["quiz-monitor-quiz-server-1"], "source_dir": "/srv/example/appdata/quiz-monitor/source", "verify_oci_revision": True, } def _observation(): container = { "Id": CONTAINER_ID, "Image": IMAGE_ID, "Config": { "Labels": { "com.docker.compose.project": "quiz-monitor", "com.docker.compose.service": "quiz-server", }, "Env": ["KNOWLEDGE_OPTIMIZER_MODEL=" + SECRET_VALUE], }, "State": {"Running": True}, } return { "status": "observed", "sourceFingerprint": SOURCE_FP, "runtimeFingerprint": RUNTIME_FP, "imageIDs": {"quiz-monitor-quiz-server-1": IMAGE_ID}, "environmentMismatchKeys": [MISMATCH_KEY], "capturedEnvironmentKeys": [MISMATCH_KEY], "artifacts": { "containers": { "quiz-monitor-quiz-server-1": { "container_id": CONTAINER_ID, "image_id": IMAGE_ID, "image_ref": "quiz-monitor:latest", "service": "quiz-server", "running": True, }, }, }, "runtimeConfiguration": { "composeModel": { "services": { "quiz-server": { "environment": {"KNOWLEDGE_OPTIMIZER_MODEL": "gpt-5.5"}, }, }, }, "containers": {"quiz-monitor-quiz-server-1": container}, "imageEnvironments": {IMAGE_ID: {}}, }, } class _CaptureRuntime: def __init__(self, observation=None): self.observation = observation or _observation() self.strict_calls = 0 self.capture_calls = 0 def observe(self, repository, _project, *, receipt_key): self.strict_calls += 1 assert repository == REPOSITORY assert receipt_key == KEY raise RuntimeError("strict observer rejects environment mismatch") def observe_for_baseline_capture(self, repository, _project, *, receipt_key): self.capture_calls += 1 assert repository == REPOSITORY assert receipt_key == KEY return copy.deepcopy(self.observation) def _payload(): observation = _observation() return { "repository": REPOSITORY, "requestId": REQUEST_ID, "captureRuntimeEnvironment": True, "expectedSourceFingerprint": observation["sourceFingerprint"], "expectedRuntimeFingerprint": observation["runtimeFingerprint"], "expectedStateSha256": hashlib.sha256(STATE).hexdigest(), "expectedContainerIds": { "quiz-monitor-quiz-server-1": CONTAINER_ID, }, } def _read_state(_repository, _project): return {"live_status": "unknown"}, STATE def test_observe_returns_only_cas_metadata_and_environment_key_names(module, project): runtime = _CaptureRuntime() result = module.observe( {"repository": REPOSITORY}, registry={REPOSITORY: project}, read_state=_read_state, runtime=runtime, receipt_key=KEY, ) assert result == { "status": "observed", "repository": REPOSITORY, "expectedStateSha256": hashlib.sha256(STATE).hexdigest(), "sourceFingerprint": SOURCE_FP, "runtimeFingerprint": RUNTIME_FP, "containerIDs": {"quiz-monitor-quiz-server-1": CONTAINER_ID}, "imageIDs": {"quiz-monitor-quiz-server-1": IMAGE_ID}, "environmentMismatchKeys": [MISMATCH_KEY], } encoded = json.dumps(result) assert SECRET_VALUE not in encoded assert "gpt-5.5" not in encoded assert runtime.strict_calls == 0 assert runtime.capture_calls == 1 def test_explicit_capture_is_hmac_bound_private_and_idempotent(module, project, tmp_path): runtime = _CaptureRuntime() payload = _payload() first = module.baseline( payload, registry={REPOSITORY: project}, read_state=_read_state, runtime=runtime, receipt_key=KEY, state_root=tmp_path, ) second = module.baseline( payload, registry={REPOSITORY: project}, read_state=_read_state, runtime=runtime, receipt_key=KEY, state_root=tmp_path, ) assert first == second assert first["capturedEnvironmentKeys"] == [MISMATCH_KEY] assert SECRET_VALUE not in json.dumps(first) paths = module._paths(tmp_path, "quiz-monitor", REQUEST_ID) signed, _ = module._load_signed(paths["baseline"], KEY, module.BASELINE_SCHEMA) private, _ = module._load_private(paths["private"], KEY) assert signed["environmentCapture"]["containerIDs"] == payload["expectedContainerIds"] assert signed["environmentCapture"]["capturedEnvironmentKeys"] == [MISMATCH_KEY] assert private["environmentCapture"] == signed["environmentCapture"] assert SECRET_VALUE in json.dumps(private) assert module._capture_runtime_environment( private, captured_environment_keys=signed["environmentCapture"]["capturedEnvironmentKeys"], ) == {"quiz-server": {"KNOWLEDGE_OPTIMIZER_MODEL": SECRET_VALUE}} assert runtime.strict_calls == 0 @pytest.mark.skipif(os.name == "nt", reason="POSIX private-file mode enforcement") def test_candidate_compose_proof_rejects_group_readable_environment_values(module, tmp_path): paths = module._paths(tmp_path, "quiz-monitor", REQUEST_ID) module._write_signed( paths["candidate"], {"schema": module.CANDIDATE_SCHEMA, "composeModel": {"secret": SECRET_VALUE}}, KEY, ) paths["candidate"].chmod(0o640) with pytest.raises(module.RegisteredLegacyAdoptionError): module._load_signed(paths["candidate"], KEY, module.CANDIDATE_SCHEMA) @pytest.mark.parametrize( "captured_keys", [[], ["quiz-server/UNDECLARED_RUNTIME_VALUE"]], ) def test_candidate_environment_preservation_requires_exact_signed_capture_keys( module, captured_keys ): private = _observation()["runtimeConfiguration"] with pytest.raises(module.RegisteredLegacyAdoptionError): module._capture_runtime_environment( {"runtimeConfiguration": private}, captured_environment_keys=captured_keys, ) @pytest.mark.parametrize( "mutate", [ lambda payload: payload.pop("expectedContainerIds"), lambda payload: payload.update(expectedSourceFingerprint="f" * 64), lambda payload: payload.update(expectedRuntimeFingerprint="f" * 64), lambda payload: payload.update(expectedStateSha256="f" * 64), lambda payload: payload.update(expectedContainerIds={"quiz-monitor-quiz-server-1": "f" * 64}), lambda payload: payload.update(unexpected="value"), ], ) def test_capture_rejects_incomplete_or_changed_cas_without_writing(module, project, tmp_path, mutate): payload = _payload() mutate(payload) with pytest.raises(module.RegisteredLegacyAdoptionError): module.baseline( payload, registry={REPOSITORY: project}, read_state=_read_state, runtime=_CaptureRuntime(), receipt_key=KEY, state_root=tmp_path, ) assert not list(tmp_path.rglob("*.baseline.json")) assert not list(tmp_path.rglob("*.private.json")) @pytest.mark.parametrize( "payload", [ {"repository": REPOSITORY, "requestId": REQUEST_ID, "expectedRuntimeFingerprint": "2" * 64}, {"repository": REPOSITORY, "requestId": REQUEST_ID, "captureRuntimeEnvironment": False, "expectedStateSha256": hashlib.sha256(STATE).hexdigest()}, ], ) def test_default_baseline_does_not_accept_capture_compare_fields(module, project, tmp_path, payload): with pytest.raises(module.RegisteredLegacyAdoptionError): module.baseline( payload, registry={REPOSITORY: project}, read_state=_read_state, runtime=_CaptureRuntime(), receipt_key=KEY, state_root=tmp_path, ) assert not list(tmp_path.rglob("*.baseline.json")) def test_capture_reobserves_exact_runtime_and_container_preimage(module, project, tmp_path): runtime = _CaptureRuntime() module.baseline( _payload(), registry={REPOSITORY: project}, read_state=_read_state, runtime=runtime, receipt_key=KEY, state_root=tmp_path, ) paths = module._paths(tmp_path, "quiz-monitor", REQUEST_ID) with pytest.raises(module.RegisteredLegacyAdoptionError, match="source or runtime changed"): changed = _observation() changed["runtimeFingerprint"] = "f" * 64 module._load_current_baseline( REPOSITORY, "quiz-monitor", REQUEST_ID, paths, KEY, _read_state, project, tmp_path, _CaptureRuntime(changed), ) changed_container = _observation() changed_container["artifacts"]["containers"]["quiz-monitor-quiz-server-1"]["container_id"] = "d" * 64 with pytest.raises(module.RegisteredLegacyAdoptionError, match="container identity changed"): module._load_current_baseline( REPOSITORY, "quiz-monitor", REQUEST_ID, paths, KEY, _read_state, project, tmp_path, _CaptureRuntime(changed_container), ) def test_private_capture_record_requires_owner_only_read_permissions(module, tmp_path): if module.os.name == "nt": pytest.skip("POSIX file-mode enforcement is not available on Windows") path = tmp_path / "private.json" module._write_signed( path, {"schema": "itworx.registered-legacy-private/v1", "repository": REPOSITORY}, KEY, ) path.chmod(0o640) with pytest.raises(module.RegisteredLegacyAdoptionError, match="permissions"): module._load_private(path, KEY) path.chmod(0o600) private, _ = module._load_private(path, KEY) assert private["repository"] == REPOSITORY def test_signed_capture_value_is_required_in_the_rendered_candidate_model(module, project): captured_key = "quiz-server/KNOWLEDGE_OPTIMIZER_MODEL" current_model = {"services": {"quiz-server": { "environment": {"KNOWLEDGE_OPTIMIZER_MODEL": SECRET_VALUE}, "image": "quiz-server:current", }}} candidate_model = {"services": {"quiz-server": { "environment": {"KNOWLEDGE_OPTIMIZER_MODEL": SECRET_VALUE}, "image": "quiz-server:candidate", }}} binding = { "schema": "itworx.registered-legacy-env-capture/v1", "sourceFingerprint": SOURCE_FP, "runtimeFingerprint": RUNTIME_FP, "expectedStateSha256": hashlib.sha256(STATE).hexdigest(), "containerIDs": {"quiz-monitor-quiz-server-1": CONTAINER_ID}, "capturedEnvironmentKeys": [captured_key], } runtime_container = { "Image": IMAGE_ID, "Config": { "Labels": {"com.docker.compose.service": "quiz-server"}, "Env": [f"KNOWLEDGE_OPTIMIZER_MODEL={SECRET_VALUE}"], }, } private = { "environmentCapture": binding, "runtimeConfiguration": { "composeModel": current_model, "containers": {"quiz-monitor-quiz-server-1": runtime_container}, "imageEnvironments": {IMAGE_ID: {}}, }, } baseline = {"environmentCapture": binding, "sourceFingerprint": SOURCE_FP, "runtimeFingerprint": RUNTIME_FP, "expectedStateSha256": hashlib.sha256(STATE).hexdigest()} normalized = module._candidate_runtime_comparison_model( candidate_model, current_model, baseline, private, project, ) assert normalized["services"]["quiz-server"]["environment"] == { "KNOWLEDGE_OPTIMIZER_MODEL": SECRET_VALUE, } def test_raw_source_comparison_allows_only_the_signed_default_or_capture_value(module, project): captured_key = "quiz-server/KNOWLEDGE_OPTIMIZER_MODEL" baseline_model = {"services": {"quiz-server": { "environment": {"KNOWLEDGE_OPTIMIZER_MODEL": SECRET_VALUE}, "image": "quiz-server:current", }}} binding = { "schema": "itworx.registered-legacy-env-capture/v1", "sourceFingerprint": SOURCE_FP, "runtimeFingerprint": RUNTIME_FP, "expectedStateSha256": hashlib.sha256(STATE).hexdigest(), "containerIDs": {"quiz-monitor-quiz-server-1": CONTAINER_ID}, "capturedEnvironmentKeys": [captured_key], } runtime_container = { "Image": IMAGE_ID, "Config": { "Labels": {"com.docker.compose.service": "quiz-server"}, "Env": [f"KNOWLEDGE_OPTIMIZER_MODEL={SECRET_VALUE}"], }, } private = { "environmentCapture": binding, "capturedEnvironmentSource": {"quiz-server": {"KNOWLEDGE_OPTIMIZER_MODEL": "gpt-5.5"}}, "runtimeConfiguration": { "composeModel": baseline_model, "containers": {"quiz-monitor-quiz-server-1": runtime_container}, "imageEnvironments": {IMAGE_ID: {}}, }, } baseline = {"environmentCapture": binding, "sourceFingerprint": SOURCE_FP, "runtimeFingerprint": RUNTIME_FP, "expectedStateSha256": hashlib.sha256(STATE).hexdigest()} for source_value in ("gpt-5.5", SECRET_VALUE): candidate_model = {"services": {"quiz-server": { "environment": {"KNOWLEDGE_OPTIMIZER_MODEL": source_value}, "image": "quiz-server:candidate", }}} normalized = module._candidate_source_comparison_model( candidate_model, baseline_model, baseline, private, project, ) assert normalized["services"]["quiz-server"]["environment"][ "KNOWLEDGE_OPTIMIZER_MODEL" ] == SECRET_VALUE third_value_model = {"services": {"quiz-server": { "environment": {"KNOWLEDGE_OPTIMIZER_MODEL": "gpt-6.2"}, "image": "quiz-server:candidate", }}} with pytest.raises(module.RegisteredLegacyAdoptionError): module._candidate_source_comparison_model( third_value_model, baseline_model, baseline, private, project, ) def test_baseline_rechecks_controller_state_before_public_signed_commit(module, project, tmp_path): calls = 0 def changing_state(_repository, _project): nonlocal calls calls += 1 if calls == 1: return {"live_status": "unknown"}, STATE return {"live_status": "changed"}, b'{"live_status":"changed"}\n' with pytest.raises(module.RegisteredLegacyAdoptionError, match="state changed before"): module.baseline( _payload(), registry={REPOSITORY: project}, read_state=changing_state, runtime=_CaptureRuntime(), receipt_key=KEY, state_root=tmp_path, ) assert calls == 2 assert not list(tmp_path.rglob("*.baseline.json")) assert not list(tmp_path.rglob("*.private.json")) @pytest.mark.parametrize("drift", ["source", "runtime", "container", "mismatch_keys"]) def test_baseline_reobserves_exact_runtime_before_public_commit(module, project, tmp_path, drift): class _DriftingRuntime(_CaptureRuntime): def observe_for_baseline_capture(self, repository, target, *, receipt_key): value = super().observe_for_baseline_capture(repository, target, receipt_key=receipt_key) if self.capture_calls == 2: if drift == "source": value["sourceFingerprint"] = "f" * 64 elif drift == "runtime": value["runtimeFingerprint"] = "f" * 64 elif drift == "container": value["artifacts"]["containers"]["quiz-monitor-quiz-server-1"][ "container_id" ] = "d" * 64 else: value["environmentMismatchKeys"] = [] value["capturedEnvironmentKeys"] = [] return value with pytest.raises(module.RegisteredLegacyAdoptionError, match="runtime changed before baseline"): module.baseline( _payload(), registry={REPOSITORY: project}, read_state=_read_state, runtime=_DriftingRuntime(), receipt_key=KEY, state_root=tmp_path, ) assert not list(tmp_path.rglob("*.baseline.json")) assert not list(tmp_path.rglob("*.private.json"))