257 lines
8.9 KiB
Python
257 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from copy import deepcopy
|
|
from datetime import UTC, datetime, timedelta
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from unittest import mock
|
|
import unittest
|
|
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[1]
|
|
ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments"
|
|
if str(ASSESSMENT_TOOLS_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ASSESSMENT_TOOLS_ROOT))
|
|
|
|
from govoplan_assessment.capability_fit import ( # noqa: E402
|
|
local_tag_provenance,
|
|
render_review,
|
|
review_capability_fit,
|
|
)
|
|
|
|
|
|
class CapabilityFitReviewTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.assessment = json.loads(
|
|
(META_ROOT / "docs" / "capability-fit-current.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
cls.schema = json.loads(
|
|
(META_ROOT / "docs" / "capability-fit.schema.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
|
|
def test_matching_signed_catalog_is_current(self) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
|
|
report = review_capability_fit(
|
|
assessment=deepcopy(self.assessment),
|
|
schema=self.schema,
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
)
|
|
|
|
self.assertEqual("current", report["status"])
|
|
self.assertEqual([], report["changes"])
|
|
self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"])
|
|
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
|
|
self.assertFalse(report["proof_scope"]["local_tag_provenance"]["checked"])
|
|
repeated = review_capability_fit(
|
|
assessment=deepcopy(self.assessment),
|
|
schema=self.schema,
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
)
|
|
self.assertEqual(
|
|
json.dumps(report, sort_keys=True), json.dumps(repeated, sort_keys=True)
|
|
)
|
|
|
|
def test_release_drift_identifies_affected_conclusions(self) -> None:
|
|
catalog, keyring = signed_catalog(
|
|
self.assessment, versions={"campaigns": "0.1.11"}, sequence=202607230001
|
|
)
|
|
|
|
report = review_capability_fit(
|
|
assessment=deepcopy(self.assessment),
|
|
schema=self.schema,
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
)
|
|
|
|
self.assertEqual("review_required", report["status"])
|
|
target_ids = {item["id"] for item in report["review_targets"]}
|
|
self.assertIn("campaign.journey", target_ids)
|
|
self.assertIn("composition.campaigns", target_ids)
|
|
self.assertIn("assessment.release", target_ids)
|
|
self.assertIn(
|
|
"composition_version_changed", {item["code"] for item in report["findings"]}
|
|
)
|
|
|
|
def test_untrusted_or_substituted_keyring_blocks_rerun(self) -> None:
|
|
catalog, _ = signed_catalog(self.assessment)
|
|
_, other_keyring = signed_catalog(self.assessment)
|
|
|
|
report = review_capability_fit(
|
|
assessment=deepcopy(self.assessment),
|
|
schema=self.schema,
|
|
catalog=catalog,
|
|
keyring=other_keyring,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
codes = {item["code"] for item in report["findings"]}
|
|
self.assertIn("catalog_trust", codes)
|
|
self.assertIn("catalog_keyring_hash_mismatch", codes)
|
|
self.assertTrue(report["proof_scope"]["assessment_schema"]["valid"])
|
|
self.assertFalse(
|
|
report["proof_scope"]["catalog_signature_and_keyring"]["valid"]
|
|
)
|
|
|
|
def test_schema_error_blocks_before_comparison(self) -> None:
|
|
assessment = deepcopy(self.assessment)
|
|
assessment.pop("scope")
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
|
|
report = review_capability_fit(
|
|
assessment=assessment,
|
|
schema=self.schema,
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertEqual(
|
|
{"assessment_schema"}, {item["code"] for item in report["findings"]}
|
|
)
|
|
self.assertFalse(report["proof_scope"]["assessment_schema"]["valid"])
|
|
self.assertFalse(
|
|
report["proof_scope"]["catalog_signature_and_keyring"]["checked"]
|
|
)
|
|
self.assertIsNone(
|
|
report["proof_scope"]["catalog_signature_and_keyring"]["valid"]
|
|
)
|
|
|
|
def test_human_report_states_proof_limit(self) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
report = review_capability_fit(
|
|
assessment=deepcopy(self.assessment),
|
|
schema=self.schema,
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
)
|
|
|
|
rendered = render_review(report)
|
|
|
|
self.assertIn("Capability fit rerun: current", rendered)
|
|
self.assertIn(
|
|
"No installed-artifact, target-provider, or production proof", rendered
|
|
)
|
|
|
|
def test_public_fetch_failure_is_generic_and_blocking(self) -> None:
|
|
script = META_ROOT / "tools" / "assessments" / "capability-fit.py"
|
|
spec = importlib.util.spec_from_file_location("capability_fit_cli", script)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
with mock.patch.object(
|
|
module,
|
|
"fetch_json",
|
|
return_value={
|
|
"ok": False,
|
|
"error": "Authorization: secret at https://internal.invalid",
|
|
},
|
|
):
|
|
with self.assertRaisesRegex(
|
|
SystemExit, r"^Could not fetch fixed public catalog endpoint[.]$"
|
|
):
|
|
module.main(["--public", "--skip-tag-provenance"])
|
|
|
|
def test_repository_path_traversal_is_not_resolved(self) -> None:
|
|
with mock.patch("govoplan_assessment.capability_fit.subprocess.run") as run:
|
|
result = local_tag_provenance(
|
|
workspace_root=META_ROOT.parent,
|
|
repository="../outside",
|
|
version="0.1.0",
|
|
assessed_commit="0123456",
|
|
)
|
|
|
|
self.assertEqual("tag_provenance_unavailable", result["kind"])
|
|
self.assertIn("invalid repository identifier", result["message"])
|
|
run.assert_not_called()
|
|
|
|
|
|
def signed_catalog(
|
|
assessment: dict[str, object],
|
|
*,
|
|
versions: dict[str, str] | None = None,
|
|
sequence: int = 202607220843,
|
|
) -> tuple[dict[str, object], dict[str, object]]:
|
|
versions = versions or {}
|
|
private_key = Ed25519PrivateKey.generate()
|
|
public_key = base64.b64encode(
|
|
private_key.public_key().public_bytes(
|
|
encoding=serialization.Encoding.Raw,
|
|
format=serialization.PublicFormat.Raw,
|
|
)
|
|
).decode("ascii")
|
|
keyring: dict[str, object] = {
|
|
"keyring_version": "1",
|
|
"purpose": "test",
|
|
"generated_at": "2026-01-01T00:00:00Z",
|
|
"keys": [{"key_id": "test-key", "status": "active", "public_key": public_key}],
|
|
}
|
|
now = datetime.now(tz=UTC)
|
|
modules: list[dict[str, object]] = []
|
|
core_release: dict[str, object] | None = None
|
|
for component in assessment["composition"]:
|
|
module_id = str(component["module_id"])
|
|
repository = str(component["repository"])
|
|
version = versions.get(module_id, str(component["manifest_version"]))
|
|
entry: dict[str, object] = {
|
|
"module_id": module_id,
|
|
"name": module_id,
|
|
"version": version,
|
|
"python_package": repository,
|
|
"python_ref": f"{repository} @ git+ssh://git@example.invalid/example/{repository}.git@v{version}",
|
|
}
|
|
if module_id == "core":
|
|
core_release = entry
|
|
else:
|
|
modules.append(entry)
|
|
assert core_release is not None
|
|
catalog: dict[str, object] = {
|
|
"catalog_version": "1",
|
|
"channel": "stable",
|
|
"sequence": sequence,
|
|
"generated_at": now.isoformat().replace("+00:00", "Z"),
|
|
"expires_at": (now + timedelta(days=30)).isoformat().replace("+00:00", "Z"),
|
|
"core_release": core_release,
|
|
"modules": modules,
|
|
"release": {"keyring_sha256": canonical_hash(keyring)},
|
|
}
|
|
signature_payload = json.dumps(
|
|
catalog, sort_keys=True, separators=(",", ":"), ensure_ascii=False
|
|
).encode("utf-8")
|
|
catalog["signatures"] = [
|
|
{
|
|
"algorithm": "ed25519",
|
|
"key_id": "test-key",
|
|
"value": base64.b64encode(private_key.sign(signature_payload)).decode(
|
|
"ascii"
|
|
),
|
|
}
|
|
]
|
|
return catalog, keyring
|
|
|
|
|
|
def canonical_hash(payload: object) -> str:
|
|
encoded = json.dumps(
|
|
payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|