diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 7fc0680..b46398c 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -24,7 +24,8 @@ - Preserve existing v1 behavior as migration input without treating it as v2 conformance. - Done when fixture tests detect any byte change and the import/writer census identifies every cutover target. - _Requirements: 14.3, 15.3_ -- [ ] 1.2 Build the strict canonical codec and resolver-backed artifact boundary + - _Blocked: debug attempted twice, still failing — a bounded AST census with exact provenance registries still omits active hunter and Semgrep scorer-record writer families, so “every cutover target” is not yet mechanically demonstrated._ +- [x] 1.2 Build the strict canonical codec and resolver-backed artifact boundary - Produce deterministic UTF-8 JSON with full SHA-256 identities, stable mapping order, and retained sequence order. - Refuse non-finite numbers, unsupported values, unknown schemas, duplicate logical identities, and malformed digests. - Validate resolved bytes, length, storage receipt, retention identity, trusted store, and configured size limits. @@ -377,3 +378,7 @@ - Prove each test instrument read nonzero fixture bytes before interpreting zero findings or failures. - Done when the complete suite passes, boundary/import censuses are clean, historical bytes are unchanged, and all declared adapter and consumer seams pass conformance. - _Requirements: 1.4, 10.4, 14.3, 15.3, 15.4_ + +## Implementation Notes + +- Task 1.1: repository-wide traversal alone is insufficient for a complete writer census; active writers require bounded provenance across direct schema construction, local loader/producer flow, imported producers, and scorer-envelope aggregation. Two debug rounds closed loader and imported-prover gaps, but hunter/Semgrep scorer writers remain unresolved. diff --git a/src/openultrasast/core/canonical.py b/src/openultrasast/core/canonical.py new file mode 100644 index 0000000..6db47cd --- /dev/null +++ b/src/openultrasast/core/canonical.py @@ -0,0 +1,263 @@ +"""Strict canonical serialization and read-only artifact validation for core v2.""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from collections.abc import Collection, Mapping, Sequence +from dataclasses import dataclass +from typing import Protocol, TypeAlias + +DETECTION_CORE_SCHEMA = "openultrasast.detection-core.v2" +STORAGE_RECEIPT_SCHEMA = "openultrasast.detection-core.storage-receipt.v2" + +JsonScalar: TypeAlias = str | int | float | bool | None +JsonValue: TypeAlias = JsonScalar | tuple["JsonValue", ...] | Mapping[str, "JsonValue"] +_NormalizedJson: TypeAlias = JsonScalar | list["_NormalizedJson"] | dict[str, "_NormalizedJson"] +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class CanonicalizationError(ValueError): + """A value is outside the canonical JSON profile.""" + + +class SchemaVersionError(CanonicalizationError): + """A digestible record lacks a recognized schema version.""" + + +class DuplicateIdentityError(CanonicalizationError): + """A logical identity occurs more than once in an aggregate.""" + + +class DigestFormatError(ValueError): + """A v2 digest is not a full lowercase SHA-256 identity.""" + + +class ArtifactResolutionError(LookupError): + """A referenced artifact could not be read.""" + + +class ArtifactValidationError(ValueError): + """Resolved artifact bytes or their storage receipt are untrusted.""" + + +def _normalize(value: object) -> _NormalizedJson: + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise CanonicalizationError("numbers must be finite") + return value + if isinstance(value, tuple): + return [_normalize(item) for item in value] + if isinstance(value, Mapping): + if not all(isinstance(key, str) for key in value): + raise CanonicalizationError("mapping keys must be strings") + return {key: _normalize(item) for key, item in value.items()} + raise CanonicalizationError(f"unsupported canonical value: {type(value).__name__}") + + +def canonical_bytes(value: JsonValue) -> bytes: + """Return deterministic compact UTF-8 JSON for the closed canonical profile.""" + normalized = _normalize(value) + text = json.dumps(normalized, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":")) + try: + return text.encode("utf-8") + except UnicodeEncodeError as exc: + raise CanonicalizationError("strings must be valid UTF-8") from exc + + +def canonical_record_bytes( + value: Mapping[str, JsonValue], + *, + known_schema_versions: Collection[str] = (DETECTION_CORE_SCHEMA,), +) -> bytes: + """Canonicalize a record only when its top-level schema is explicitly known.""" + schema = value.get("schema_version") + if not isinstance(schema, str): + raise SchemaVersionError("schema_version must be present as text") + if schema not in known_schema_versions: + raise SchemaVersionError(f"unknown schema_version: {schema}") + return canonical_bytes(value) + + +def bytes_digest(content: bytes) -> str: + """Return the full lowercase SHA-256 identity of exact bytes.""" + if not isinstance(content, bytes): + raise TypeError("content must be bytes") + return hashlib.sha256(content).hexdigest() + + +def content_digest(value: JsonValue) -> str: + """Return the full lowercase SHA-256 identity of a canonical value.""" + return bytes_digest(canonical_bytes(value)) + + +def require_unique_identities(identities: Sequence[str], *, kind: str) -> None: + """Refuse duplicate logical identities before an aggregate is hashed.""" + seen: set[str] = set() + for identity in identities: + if identity in seen: + raise DuplicateIdentityError(f"duplicate {kind} identity: {identity}") + seen.add(identity) + + +def _require_digest(value: object, field_name: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise DigestFormatError(f"{field_name} must be a full lowercase SHA-256 digest") + + +def _require_nonempty(value: object, field_name: str) -> None: + if not isinstance(value, str) or not value: + raise ValueError(f"{field_name} must be nonempty text") + + +def _require_length(value: int, field_name: str) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"{field_name} must be a non-negative integer") + + +@dataclass(frozen=True) +class ContentRef: + """Opaque artifact identity plus the identities needed to validate its bytes.""" + + artifact_id: str + content_digest: str + media_type: str + byte_length: int + retention_policy_digest: str + storage_receipt_digest: str + + def __post_init__(self) -> None: + _require_nonempty(self.artifact_id, "artifact_id") + _require_nonempty(self.media_type, "media_type") + _require_length(self.byte_length, "byte_length") + _require_digest(self.content_digest, "content_digest") + _require_digest(self.retention_policy_digest, "retention_policy_digest") + _require_digest(self.storage_receipt_digest, "storage_receipt_digest") + + +@dataclass(frozen=True) +class StorageReceipt: + """Store attestation over content, length, and retention identity.""" + + store_identity: str + content_digest: str + byte_length: int + retention_policy_digest: str + receipt_digest: str + + def __post_init__(self) -> None: + _require_nonempty(self.store_identity, "store_identity") + _require_length(self.byte_length, "byte_length") + _require_digest(self.content_digest, "content_digest") + _require_digest(self.retention_policy_digest, "retention_policy_digest") + _require_digest(self.receipt_digest, "receipt_digest") + + +@dataclass(frozen=True) +class ResolvedContent: + """Bytes and the immutable store receipt returned by a resolver.""" + + content: bytes + receipt: StorageReceipt + + def __post_init__(self) -> None: + if not isinstance(self.content, bytes): + raise TypeError("resolved content must be bytes") + + +class ContentResolver(Protocol): + """Injected read-only artifact resolver; core never writes through this port.""" + + def resolve(self, reference: ContentRef) -> ResolvedContent: ... + + +def derive_storage_receipt_digest( + *, + store_identity: str, + content_digest: str, + byte_length: int, + retention_policy_digest: str, +) -> str: + """Derive a receipt identity from every attested storage property.""" + _require_nonempty(store_identity, "store_identity") + _require_digest(content_digest, "content_digest") + _require_length(byte_length, "byte_length") + _require_digest(retention_policy_digest, "retention_policy_digest") + payload: JsonValue = { + "schema_version": STORAGE_RECEIPT_SCHEMA, + "store_identity": store_identity, + "content_digest": content_digest, + "byte_length": byte_length, + "retention_policy_digest": retention_policy_digest, + } + return bytes_digest(canonical_bytes(payload)) + + +def _validate_receipt(reference: ContentRef, receipt: StorageReceipt, trusted_stores: Collection[str]) -> None: + if receipt.content_digest != reference.content_digest: + raise ArtifactValidationError("receipt content digest does not match reference") + if receipt.byte_length != reference.byte_length: + raise ArtifactValidationError("receipt byte length does not match reference") + if receipt.retention_policy_digest != reference.retention_policy_digest: + raise ArtifactValidationError("receipt retention policy does not match reference") + if receipt.store_identity not in trusted_stores: + raise ArtifactValidationError(f"store is not trusted: {receipt.store_identity}") + if receipt.receipt_digest != reference.storage_receipt_digest: + raise ArtifactValidationError("receipt digest does not match reference") + expected = derive_storage_receipt_digest( + store_identity=receipt.store_identity, + content_digest=receipt.content_digest, + byte_length=receipt.byte_length, + retention_policy_digest=receipt.retention_policy_digest, + ) + if receipt.receipt_digest != expected: + raise ArtifactValidationError("receipt digest cannot be re-derived") + + +def resolve_content( + reference: ContentRef, + resolver: ContentResolver, + *, + trusted_store_identities: Collection[str], + maximum_bytes: int, +) -> bytes: + """Resolve and validate exact bytes and receipt identities before parsing.""" + _require_length(maximum_bytes, "maximum_bytes") + if reference.byte_length > maximum_bytes: + raise ArtifactValidationError("reference byte length exceeds configured maximum") + resolved = resolver.resolve(reference) + if len(resolved.content) != reference.byte_length: + raise ArtifactValidationError("resolved byte length does not match reference") + if bytes_digest(resolved.content) != reference.content_digest: + raise ArtifactValidationError("resolved content digest does not match reference") + _validate_receipt(reference, resolved.receipt, trusted_store_identities) + return resolved.content + + +__all__ = [ + "ArtifactResolutionError", + "ArtifactValidationError", + "CanonicalizationError", + "ContentRef", + "ContentResolver", + "DETECTION_CORE_SCHEMA", + "DigestFormatError", + "DuplicateIdentityError", + "JsonScalar", + "JsonValue", + "ResolvedContent", + "STORAGE_RECEIPT_SCHEMA", + "SchemaVersionError", + "StorageReceipt", + "bytes_digest", + "canonical_bytes", + "canonical_record_bytes", + "content_digest", + "derive_storage_receipt_digest", + "require_unique_identities", + "resolve_content", +] diff --git a/tests/detection_core_fixtures.py b/tests/detection_core_fixtures.py new file mode 100644 index 0000000..ceb3956 --- /dev/null +++ b/tests/detection_core_fixtures.py @@ -0,0 +1,61 @@ +"""Deterministic content-resolver fixtures shared by detection-core contract tests.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType + +from openultrasast.core import canonical + + +@dataclass(frozen=True) +class InMemoryContentResolver: + """Read-only resolver keyed by opaque artifact identity.""" + + artifacts: Mapping[str, canonical.ResolvedContent] + + def __post_init__(self) -> None: + object.__setattr__(self, "artifacts", MappingProxyType(dict(self.artifacts))) + + def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent: + artifact_id = reference.artifact_id + try: + return self.artifacts[artifact_id] + except KeyError as exc: + raise canonical.ArtifactResolutionError(f"artifact not found: {artifact_id}") from exc + + +def stored_artifact( + content: bytes, + *, + artifact_id: str = "artifact-1", + store_identity: str = "trusted-store", + retention_policy_digest: str | None = None, +) -> tuple[canonical.ContentRef, InMemoryContentResolver]: + """Create a reference and resolver whose receipt identities are reproducible.""" + retention = retention_policy_digest or canonical.content_digest("retain-30-days") + digest = canonical.bytes_digest(content) + receipt_digest = canonical.derive_storage_receipt_digest( + store_identity=store_identity, + content_digest=digest, + byte_length=len(content), + retention_policy_digest=retention, + ) + receipt = canonical.StorageReceipt( + store_identity=store_identity, + content_digest=digest, + byte_length=len(content), + retention_policy_digest=retention, + receipt_digest=receipt_digest, + ) + reference = canonical.ContentRef( + artifact_id=artifact_id, + content_digest=digest, + media_type="application/json", + byte_length=len(content), + retention_policy_digest=retention, + storage_receipt_digest=receipt_digest, + ) + resolver = InMemoryContentResolver({artifact_id: canonical.ResolvedContent(content=content, receipt=receipt)}) + return reference, resolver diff --git a/tests/test_core_canonical.py b/tests/test_core_canonical.py new file mode 100644 index 0000000..a532523 --- /dev/null +++ b/tests/test_core_canonical.py @@ -0,0 +1,239 @@ +"""Strict canonical codec and resolver-backed artifact boundary (Req 1.4, 10.4).""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import MutableMapping +from dataclasses import replace +from enum import Enum +from typing import cast + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical + + +class _UnknownEnum(Enum): + VALUE = "value" + + +def test_canonical_bytes_are_deterministic_utf8_and_compact() -> None: + left: canonical.JsonValue = {"schema_version": canonical.DETECTION_CORE_SCHEMA, "z": "café", "a": (2, 1)} + right: canonical.JsonValue = {"a": (2, 1), "z": "café", "schema_version": canonical.DETECTION_CORE_SCHEMA} + expected = b'{"a":[2,1],"schema_version":"openultrasast.detection-core.v2","z":"caf\xc3\xa9"}' + + assert canonical.canonical_bytes(left) == expected + assert canonical.canonical_bytes(right) == expected + + +def test_sequence_order_is_retained() -> None: + assert canonical.canonical_bytes(("second", "first")) == b'["second","first"]' + assert canonical.canonical_bytes(("first", "second")) != canonical.canonical_bytes(("second", "first")) + + +@pytest.mark.parametrize("value", [float("nan"), float("inf"), -float("inf")]) +def test_non_finite_numbers_are_refused_before_hashing(value: float) -> None: + with pytest.raises(canonical.CanonicalizationError, match="finite"): + canonical.content_digest({"schema_version": canonical.DETECTION_CORE_SCHEMA, "value": value}) + + +@pytest.mark.parametrize("value", [[1], {1}, b"bytes", _UnknownEnum.VALUE, {1: "not-string"}]) +def test_unsupported_values_are_refused(value: object) -> None: + with pytest.raises(canonical.CanonicalizationError): + canonical.canonical_bytes(value) # type: ignore[arg-type] + + +def test_record_schema_is_explicit_and_closed() -> None: + record = {"schema_version": canonical.DETECTION_CORE_SCHEMA, "value": "ok"} + assert canonical.canonical_record_bytes(record) == canonical.canonical_bytes(record) + with pytest.raises(canonical.SchemaVersionError, match="schema_version"): + canonical.canonical_record_bytes({"value": "missing"}) + with pytest.raises(canonical.SchemaVersionError, match="unknown"): + canonical.canonical_record_bytes({"schema_version": "openultrasast.unknown.v9"}) + + +def test_sha256_identities_are_full_lowercase_hex() -> None: + payload = {"schema_version": canonical.DETECTION_CORE_SCHEMA, "value": "ok"} + digest = canonical.content_digest(payload) + assert digest == hashlib.sha256(canonical.canonical_bytes(payload)).hexdigest() + assert len(digest) == 64 + assert digest == digest.lower() + + +def test_duplicate_logical_identities_are_refused() -> None: + canonical.require_unique_identities(("relation-1", "relation-2"), kind="relation") + with pytest.raises(canonical.DuplicateIdentityError, match="relation-1"): + canonical.require_unique_identities(("relation-1", "relation-1"), kind="relation") + + +@pytest.mark.parametrize("field", ["content_digest", "retention_policy_digest", "storage_receipt_digest"]) +def test_content_reference_refuses_malformed_digests(field: str) -> None: + reference, _ = stored_artifact(b"{}") + with pytest.raises(canonical.DigestFormatError, match=field): + _replace_reference_digest(reference, field, "ABC123") + with pytest.raises(canonical.DigestFormatError, match=field): + _replace_reference_digest(reference, field, cast(str, None)) + + +@pytest.mark.parametrize("value", ["", 7, None]) +def test_content_reference_requires_nonempty_text_identities(value: object) -> None: + reference, _ = stored_artifact(b"{}") + with pytest.raises(ValueError, match="artifact_id"): + replace(reference, artifact_id=cast(str, value)) + with pytest.raises(ValueError, match="media_type"): + replace(reference, media_type=cast(str, value)) + + +@pytest.mark.parametrize("value", ["", 7, None]) +def test_storage_receipt_requires_nonempty_text_identity(value: object) -> None: + reference, resolver = stored_artifact(b"{}") + receipt = resolver.artifacts[reference.artifact_id].receipt + with pytest.raises(ValueError, match="store_identity"): + replace(receipt, store_identity=cast(str, value)) + + +def _replace_reference_digest(reference: canonical.ContentRef, field: str, value: str) -> canonical.ContentRef: + if field == "content_digest": + return replace(reference, content_digest=value) + if field == "retention_policy_digest": + return replace(reference, retention_policy_digest=value) + if field == "storage_receipt_digest": + return replace(reference, storage_receipt_digest=value) + raise AssertionError(f"unknown test field: {field}") + + +def test_resolver_validates_content_and_receipt_before_returning_bytes() -> None: + content = b'{"schema_version":"openultrasast.detection-core.v2"}' + reference, resolver = stored_artifact(content) + assert ( + canonical.resolve_content( + reference, + resolver, + trusted_store_identities=("trusted-store",), + maximum_bytes=1024, + ) + == content + ) + + +def test_missing_artifact_is_an_instrument_failure_not_empty_content() -> None: + reference, _ = stored_artifact(b"nonempty") + resolver = InMemoryContentResolver({}) + with pytest.raises(canonical.ArtifactResolutionError, match="not found"): + canonical.resolve_content( + reference, + resolver, + trusted_store_identities=("trusted-store",), + maximum_bytes=1024, + ) + + +def test_in_memory_resolver_defensively_copies_caller_mapping() -> None: + reference, original = stored_artifact(b"original") + _, replacement = stored_artifact(b"replacement") + caller_mapping = {reference.artifact_id: original.artifacts[reference.artifact_id]} + resolver = InMemoryContentResolver(caller_mapping) + + caller_mapping[reference.artifact_id] = replacement.artifacts[reference.artifact_id] + + assert resolver.resolve(reference).content == b"original" + + +def test_in_memory_resolver_exposes_read_only_state() -> None: + reference, resolver = stored_artifact(b"original") + _, replacement = stored_artifact(b"replacement") + exposed = cast(MutableMapping[str, canonical.ResolvedContent], resolver.artifacts) + + with pytest.raises(TypeError): + exposed[reference.artifact_id] = replacement.artifacts[reference.artifact_id] + assert resolver.resolve(reference).content == b"original" + + +def _resolve( + reference: canonical.ContentRef, + resolver: InMemoryContentResolver, + *, + trusted_store_identities: tuple[str, ...] = ("trusted-store",), + maximum_bytes: int = 1024, +) -> bytes: + return canonical.resolve_content( + reference, + resolver, + trusted_store_identities=trusted_store_identities, + maximum_bytes=maximum_bytes, + ) + + +def test_content_digest_mismatch_precedes_parsing() -> None: + reference, resolver = stored_artifact(b"not json") + bad_reference = replace(reference, content_digest="0" * 64) + with pytest.raises(canonical.ArtifactValidationError, match="content digest"): + _resolve(bad_reference, resolver) + + +def test_resolved_byte_length_mismatch_is_refused() -> None: + reference, resolver = stored_artifact(b"content") + bad_reference = replace(reference, byte_length=reference.byte_length + 1) + with pytest.raises(canonical.ArtifactValidationError, match="byte length"): + _resolve(bad_reference, resolver) + + +@pytest.mark.parametrize( + ("receipt_field", "value", "message"), + [ + ("content_digest", "0" * 64, "receipt content digest"), + ("byte_length", 99, "receipt byte length"), + ("retention_policy_digest", "1" * 64, "retention policy"), + ("receipt_digest", "2" * 64, "receipt digest"), + ], +) +def test_storage_receipt_mismatches_are_refused(receipt_field: str, value: object, message: str) -> None: + reference, resolver = stored_artifact(b"content") + resolved = resolver.artifacts[reference.artifact_id] + bad_receipt = _replace_receipt_field(resolved.receipt, receipt_field, value) + bad_resolver = InMemoryContentResolver({reference.artifact_id: replace(resolved, receipt=bad_receipt)}) + with pytest.raises(canonical.ArtifactValidationError, match=message): + _resolve(reference, bad_resolver) + + +def _replace_receipt_field(receipt: canonical.StorageReceipt, field: str, value: object) -> canonical.StorageReceipt: + if field == "byte_length": + if not isinstance(value, int): + raise AssertionError("byte_length test value must be an integer") + return replace(receipt, byte_length=value) + if not isinstance(value, str): + raise AssertionError(f"{field} test value must be text") + if field == "content_digest": + return replace(receipt, content_digest=value) + if field == "retention_policy_digest": + return replace(receipt, retention_policy_digest=value) + if field == "receipt_digest": + return replace(receipt, receipt_digest=value) + raise AssertionError(f"unknown test field: {field}") + + +def test_untrusted_store_and_configured_size_limit_are_refused() -> None: + reference, resolver = stored_artifact(b"content") + with pytest.raises(canonical.ArtifactValidationError, match="trusted"): + _resolve(reference, resolver, trusted_store_identities=("other-store",)) + with pytest.raises(canonical.ArtifactValidationError, match="maximum"): + _resolve(reference, resolver, maximum_bytes=3) + + +def test_receipt_identity_is_rederived_not_merely_compared() -> None: + reference, resolver = stored_artifact(b"content") + resolved = resolver.artifacts[reference.artifact_id] + forged = replace(resolved.receipt, store_identity="other-store") + forged_reference = replace(reference, storage_receipt_digest=forged.receipt_digest) + forged_resolver = InMemoryContentResolver({reference.artifact_id: replace(resolved, receipt=forged)}) + with pytest.raises(canonical.ArtifactValidationError, match="receipt digest"): + _resolve(forged_reference, forged_resolver, trusted_store_identities=("other-store",)) + + +def test_resolved_content_is_valid_json_only_after_boundary_validation() -> None: + payload = {"schema_version": canonical.DETECTION_CORE_SCHEMA, "value": "read"} + reference, resolver = stored_artifact(canonical.canonical_bytes(payload)) + validated = _resolve(reference, resolver) + assert json.loads(validated) == payload