diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 271d7ee..5cbd7c6 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -410,7 +410,7 @@ - _Depends: 10.6_ - [ ] 11. Validate readiness, failure visibility, and implementation boundaries -- [ ] 11.1 Run both reference families through the complete synthetic readiness path +- [x] 11.1 Run both reference families through the complete synthetic readiness path - Start from a verified nonzero input and run recognition, transitions, security object/constraints, pair, development, reviewed synthetic admission, guarded spend, and consumer verdict. - Produce violation, bounded satisfaction, and bounded silence for both unlike families. - Require current prerequisite readiness identities at every later stage. diff --git a/tests/detection_core_end_to_end.py b/tests/detection_core_end_to_end.py new file mode 100644 index 0000000..fbd14af --- /dev/null +++ b/tests/detection_core_end_to_end.py @@ -0,0 +1,430 @@ +"""Test-owned composition of the production detection-core readiness contracts.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, replace +from pathlib import Path +from typing import cast + +from detection_core_fixtures import InMemoryContentResolver +from detection_core_source_authority import SourceAuthority, bind_source_inputs, bind_source_record, load_source_authority +from test_core_admission_context import _resign_manifest +from test_core_admission_metrics import ( + _case_result, + _MetricSetup, + _rename_named_inputs, + _resign_census, + _resolver, + _setup, +) +from test_core_reference_families import _contract, _record +from test_core_validation_v2 import _catalog, _named_inputs, _with_fact_set +from test_detection_core_admission_adapter import _custodian, _reviewer, _Store + +from benchmarks.valid import readiness_contract +from benchmarks.valid.admission_run import V2AdmissionRunInput, run_v2_admission +from benchmarks.valid.population_custody import AdmissionSpendAuthority +from openultrasast.core import admission_context, canonical, consumer, evidence, family, validation +from openultrasast.core._admission_funnel import derive_case_funnel +from openultrasast.core._admission_replay import AdmissionReplay +from openultrasast.core.admission_case import derive_admission_case_result_digest, derive_case_coverage_digest +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_metrics import AdmissionCaseResult, AdmissionCaseRole +from openultrasast.core.admission_report import AdmissionDecision +from openultrasast.core.admission_review import ScopeCensus +from openultrasast.core.coverage import InputManifest, InputReadReceipt, validate_instrument +from openultrasast.core.custody import PopulationStatus +from openultrasast.core.family_catalog import FamilyCatalog +from openultrasast.core.measurement import BaselineIdentity, DenominatorPolicy, IntervalMethod, ScopeKey, standard_metric_rules +from openultrasast.core.readiness import ( + AnalysisIdentity, + FixtureResult, + FixtureRole, + ReadinessResult, + ReadinessStage, + StageResult, +) +from openultrasast.core.verdict import derive_verdict_outputs + + +@dataclass(frozen=True) +class ReferencePathResult: + family_id: str + ecosystem: str + repository_bytes: int + opened_input_count: int + source_authority_digest: str + source_content_digests: tuple[str, ...] + inputs: tuple[validation.EvidenceValidationInput, ...] + readiness: tuple[ReadinessResult, ...] + admission: AdmissionReplay + verdicts: tuple[consumer.PushVerdict, ...] + population_status_at_label_read: PopulationStatus + label_read_count: int + + +@dataclass(frozen=True) +class _PreparedFamily: + contract: family.FamilyContract + ecosystem: str + inputs: tuple[validation.EvidenceValidationInput, ...] + resolver: InMemoryContentResolver + manifest: InputManifest + receipts: tuple[InputReadReceipt, ...] + authority: SourceAuthority + + +def _load_object(path: Path) -> dict[str, object]: + payload = path.read_bytes() + if not payload: + raise ValueError(f"zero-byte fixture: {path}") + value = json.loads(payload) + if not isinstance(value, dict): + raise ValueError(f"fixture is not an object: {path}") + return cast(dict[str, object], value) + + +def _prepare_family(fixture_root: Path, family_file: str, cases_file: str, repository: str) -> _PreparedFamily: + contract = _contract(_load_object(fixture_root / family_file)) + raw_cases = cast(list[dict[str, object]], _load_object(fixture_root / cases_file)["cases"]) + expected_cases = tuple((str(item["case_id"]), str(item["path"])) for item in raw_cases) + authority = load_source_authority(fixture_root / repository, repository, expected_cases) + inputs: list[validation.EvidenceValidationInput] = [] + resolvers: list[InMemoryContentResolver] = [] + tracked: list[tuple[str, Path]] = [] + for raw, source in zip(raw_cases, authority.sources, strict=True): + case_id = str(raw["case_id"]) + raw_record = _with_fact_set(_record(contract, raw), ()) + base_record = bind_source_record( + raw_record, + canonical.content_digest("acquisition"), + canonical.content_digest("change-input"), + ) + named, resolver = _named_inputs(base_record, contract, ()) + named, resolver = _rename_named_inputs(named, resolver, case_id) + record = bind_source_record(base_record, authority.authority_digest, source.content_digest) + named, resolver = bind_source_inputs(named, resolver, authority.manifest_content, source) + inputs.append(validation.make_evidence_validation_input(record=record, named_inputs=named)) + resolvers.append(resolver) + tracked.append((case_id, fixture_root / repository / source.path)) + manifest, receipts = readiness_contract.read_tracked_inputs(tuple(tracked)) + validate_instrument(manifest, receipts) + expected_reads = tuple((item.case_id, len(item.content), item.content_digest) for item in authority.sources) + actual_reads = tuple((item.input_id, item.byte_count, item.content_digest) for item in receipts) + if actual_reads != expected_reads: + raise ValueError("source authority does not match instrument read receipts") + ecosystem = inputs[0].record.packet.ecosystem + return _PreparedFamily(contract, ecosystem, tuple(inputs), _resolver(*resolvers), manifest, receipts, authority) + + +def _evidence_context(base: _MetricSetup, catalog: FamilyCatalog) -> AdmissionValidationContext: + program_lineages = {item.lineage_manifest_digest for item in base.context.evidence.model_programs} + lineages = tuple(item for item in base.context.evidence.lineages if item.manifest_digest not in program_lineages) + evidence_context = validation.make_evidence_validation_context( + family_catalog=catalog, + fact_packs=(), + model_programs=(), + lineages=lineages, + fact_review_policy=base.context.evidence.fact_review_policy, + provenance_trust_policy=base.context.evidence.provenance_trust_policy, + custody_policy=base.context.evidence.custody_policy, + ) + return admission_context.make_admission_validation_context( + evidence=evidence_context, + admission_review_policy=base.context.admission_review_policy, + proof_trust_policy=None, + ) + + +def _baseline( + base: _MetricSetup, + catalog: FamilyCatalog, + scope: ScopeKey, + denominator: DenominatorPolicy, + acquisition_record_digest: str, +) -> BaselineIdentity: + entry = catalog.entries[0] + return replace( + base.manifest.identity, + claimed_scopes=(scope,), + family_contract_digests=(entry.contract_digest,), + family_contract_set_digest=catalog.catalog_digest, + fact_schema_digest=admission_context.derive_fact_schema_digest(), + fact_set_digest=validation.derive_fact_set_digest(()), + acquisition_record_digest=acquisition_record_digest, + analyzer_revision="reference-analyzer@v1", + configuration_digest=canonical.content_digest("configuration"), + model_program_digests=(), + model_program_set_digest=None, + denominator_policy_digest=denominator.policy_digest, + outcome_mapping_digest=admission_context.derive_outcome_mapping_digest(catalog), + ) + + +def _fixtures(inputs: tuple[validation.EvidenceValidationInput, ...]) -> tuple[FixtureResult, ...]: + return tuple(FixtureResult(role, item.record.record_id, True, ()) for role, item in zip(FixtureRole, inputs, strict=True)) + + +def _measurement(stage: ReadinessStage, prepared: _PreparedFamily) -> str: + records = tuple(item.record for item in prepared.inputs) + if stage is ReadinessStage.INSTRUMENT: + validate_instrument(prepared.manifest, prepared.receipts) + return canonical.content_digest( + { + "stage": stage.value, + "manifest": prepared.manifest.manifest_digest, + "source_authority": prepared.authority.authority_digest, + "reads": tuple((item.input_id, item.byte_count, item.content_digest) for item in prepared.receipts), + } + ) + for record in records: + family.validate_record(prepared.contract, record) + if stage is ReadinessStage.RECOGNITION and any(not item.packet.security_object_id for item in records): + raise ValueError("reference recognition did not construct every security object") + if stage is ReadinessStage.TRANSITIONS and any(not item.packet.transitions for item in records): + raise ValueError("reference transition stage produced an empty relation set") + if stage is ReadinessStage.CONSTRAINTS and any(not item.packet.observations for item in records): + raise ValueError("reference constraint stage produced no observation") + outcomes = tuple(item.decision.outcome.value for item in records) + if stage is ReadinessStage.PAIRS and outcomes != ("violation", "satisfied", "indeterminate"): + raise ValueError("reference pair stage did not preserve the outcome triplet") + return canonical.content_digest({"stage": stage.value, "records": tuple(item.record_id for item in records)}) + + +def _prerequisite_chain(prepared: _PreparedFamily, identity: AnalysisIdentity) -> tuple[ReadinessResult, ...]: + results: list[ReadinessResult] = [] + fixtures = _fixtures(prepared.inputs) + for stage in tuple(ReadinessStage)[:5]: + if results: + readiness_contract.require_stage_authorized(stage, tuple(results), identity) + measurement = _measurement(stage, prepared) + results.append( + readiness_contract.build_readiness_result( + arm_id="reference-arm", + family_id=prepared.contract.family_id, + ecosystem=prepared.ecosystem, + stage=stage, + result=StageResult.PASSED, + identity=identity, + predecessor_digests=tuple(item.result_digest for item in results), + fixture_results=fixtures, + measurement_digest=measurement, + correction_lever=None, + population_digest=None, + population_freeze_digest=None, + population_use_receipt_digests=(), + ) + ) + return tuple(results) + + +def _development( + prepared: _PreparedFamily, + identity: AnalysisIdentity, + prior: tuple[ReadinessResult, ...], +) -> ReadinessResult: + readiness_contract.require_stage_authorized(ReadinessStage.DEVELOPMENT, prior, identity) + observation = readiness_contract.build_development_observation( + arm_id="reference-arm", + family_id=prepared.contract.family_id, + ecosystem=prepared.ecosystem, + identity=identity, + population_digest=canonical.content_digest(f"{prepared.contract.family_id}:development"), + population_freeze_digest=canonical.content_digest(f"{prepared.contract.family_id}:development-freeze"), + population_use_receipt_digests=(canonical.content_digest(f"{prepared.contract.family_id}:development-use"),), + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=_fixtures(prepared.inputs), + input_manifest=prepared.manifest, + input_read_receipts=prepared.receipts, + ) + return observation.readiness_result + + +def _canonical_case( + value: AdmissionCaseResult, + record: evidence.EvidenceRecord, + contract: family.FamilyContract, +) -> AdmissionCaseResult: + pending = derive_verdict_outputs((record,), contract, canonical.content_digest("admission-coverage-pending")) + funnel = derive_case_funnel(value, (record,), len(pending.defects)) + coverage_digest = derive_case_coverage_digest(value.case_id_digest, value.scope, funnel) + outputs = derive_verdict_outputs((record,), contract, coverage_digest) + changed = replace( + value, + defect_claims=outputs.defects, + satisfactions=outputs.satisfactions, + silence=outputs.silence, + coverage=funnel, + result_digest="0" * 64, + ) + return replace(changed, result_digest=derive_admission_case_result_digest(changed)) + + +def _admission_setup(prepared: _PreparedFamily) -> _MetricSetup: + base = _setup() + catalog = _catalog(prepared.contract) + context = _evidence_context(base, catalog) + scope = ScopeKey(prepared.contract.family_id, prepared.ecosystem) + denominator = DenominatorPolicy(standard_metric_rules((scope,)), IntervalMethod.WILSON_95, "") + identity = _baseline(base, catalog, scope, denominator, prepared.authority.authority_digest) + current = admission_context.derive_analysis_identity(identity, catalog, scope, None) + prior = _prerequisite_chain(prepared, current) + readiness = (*prior, _development(prepared, current, prior)) + resolver = _resolver(base.resolver, prepared.resolver) + pair = canonical.content_digest(f"{prepared.contract.family_id}:pair") + roles = (AdmissionCaseRole.VULNERABLE, AdmissionCaseRole.FIXED, AdmissionCaseRole.ORDINARY) + pairs = (pair, pair, None) + cases = [] + for value, role, pair_id in zip(prepared.inputs, roles, pairs, strict=True): + case, resolver = _case_result(value, role, pair_id, resolver, prepared.contract) + cases.append(_canonical_case(case, value.record, prepared.contract)) + semantic_sites = sum(item.coverage.sites_considered for item in cases) + row = ScopeCensus(scope, 3, 1, 1, 1, 1, semantic_sites, 0) + census = _resign_census(base.manifest.census, row, base.manifest.identity.population_digest) + manifest = _resign_manifest( + replace( + base.manifest, + identity=identity, + census=census, + readiness_digests=tuple(sorted(item.result_digest for item in readiness)), + denominator_policy_digest=denominator.policy_digest, + validation_context_digest=context.context_digest, + ), + census.review.conflict_disclosure, + ) + return _MetricSetup(manifest, denominator, tuple(cases), context, base.custody, resolver, readiness) + + +def _run_admission(setup: _MetricSetup, scratch: Path) -> tuple[AdmissionReplay, PopulationStatus, int]: + readiness_contract.require_stage_authorized(ReadinessStage.ADMISSION, setup.readiness, setup.readiness[0].identity) + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + label_path = scratch / f"{setup.manifest.identity.claimed_scopes[0].family_id}-labels.json" + label_path.write_bytes(b'{"roles":["vulnerable","fixed","ordinary"]}') + status_at_read: list[PopulationStatus] = [] + + def load_labels() -> bytes: + status_at_read.append(store.log.status) + return label_path.read_bytes() + + value = V2AdmissionRunInput( + AdmissionSpendAuthority(setup.manifest, setup.denominator, setup.readiness, setup.context, setup.resolver), + custodian, + load_labels, + lambda _labels: setup.cases, + _reviewer(setup), + ) + replay = run_v2_admission(value) + if replay.report.decision is not AdmissionDecision.GO: + raise ValueError("synthetic reference admission did not pass its reviewed threshold") + return replay, status_at_read[0], len(status_at_read) + + +def _admission_result( + setup: _MetricSetup, + replay: AdmissionReplay, + prior: tuple[ReadinessResult, ...], +) -> ReadinessResult: + baseline = setup.manifest.identity + return readiness_contract.build_readiness_result( + arm_id=baseline.arm_id, + family_id=baseline.claimed_scopes[0].family_id, + ecosystem=baseline.claimed_scopes[0].ecosystem, + stage=ReadinessStage.ADMISSION, + result=StageResult.PASSED, + identity=prior[0].identity, + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=prior[0].fixture_results, + measurement_digest=replay.report.report_digest, + correction_lever=None, + population_digest=baseline.population_digest, + population_freeze_digest=baseline.freeze_digest, + population_use_receipt_digests=(replay.label_release_receipt.receipt_digest,), + ) + + +def _consume( + setup: _MetricSetup, + prepared: _PreparedFamily, + readiness: tuple[ReadinessResult, ...], +) -> tuple[tuple[consumer.PushVerdict, ...], InMemoryContentResolver]: + from test_core_consumer_v2 import _request + + readiness_contract.require_stage_authorized(ReadinessStage.INTEGRATION, readiness, readiness[0].identity) + resolver = setup.resolver + verdicts = [] + for value in prepared.inputs: + request, resolver = _request(value, resolver) + verdicts.append( + consumer.consume_records( + request, + (value,), + setup.context.evidence, + setup.custody, + 1, + resolver, + ) + ) + return tuple(verdicts), resolver + + +def _integration_result( + prior: tuple[ReadinessResult, ...], + verdicts: tuple[consumer.PushVerdict, ...], +) -> ReadinessResult: + first = prior[0] + measurement = canonical.content_digest( + {"requests": tuple(item.request_digest for item in verdicts), "records": tuple(item.input_record_digests for item in verdicts)} + ) + return readiness_contract.build_readiness_result( + arm_id=first.arm_id, + family_id=first.family_id, + ecosystem=first.ecosystem, + stage=ReadinessStage.INTEGRATION, + result=StageResult.PASSED, + identity=first.identity, + predecessor_digests=tuple(item.result_digest for item in prior), + fixture_results=first.fixture_results, + measurement_digest=measurement, + correction_lever=None, + population_digest=None, + population_freeze_digest=None, + population_use_receipt_digests=(), + ) + + +def run_reference_family_path( + *, + fixture_root: Path, + family_file: str, + cases_file: str, + repository: str, + scratch: Path, +) -> ReferencePathResult: + prepared = _prepare_family(fixture_root, family_file, cases_file, repository) + setup = _admission_setup(prepared) + replay, status_at_read, read_count = _run_admission(setup, scratch) + admission = _admission_result(setup, replay, setup.readiness) + through_admission = (*setup.readiness, admission) + verdicts, _resolver_value = _consume(setup, prepared, through_admission) + integration = _integration_result(through_admission, verdicts) + readiness = (*through_admission, integration) + return ReferencePathResult( + prepared.contract.family_id, + prepared.ecosystem, + sum(item.byte_count for item in prepared.receipts), + sum(item.opened for item in prepared.receipts), + prepared.authority.authority_digest, + tuple(item.content_digest for item in prepared.authority.sources), + prepared.inputs, + readiness, + replay, + verdicts, + status_at_read, + read_count, + ) + + +__all__ = ["ReferencePathResult", "run_reference_family_path"] diff --git a/tests/detection_core_source_authority.py b/tests/detection_core_source_authority.py new file mode 100644 index 0000000..db50f8d --- /dev/null +++ b/tests/detection_core_source_authority.py @@ -0,0 +1,291 @@ +"""Frozen source authority for detection-core end-to-end fixtures.""" + +from __future__ import annotations + +import json +import os +import stat +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, replace +from pathlib import Path, PurePosixPath +from typing import cast + +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical, evidence, provenance + +_SCHEMA = "detection-core-source-authority-v1" +_MANIFEST = "source_authority.json" +_HEADER_KEYS = {"schema_version", "repository", "producer_identity", "reviewer_identity", "cases"} +_CASE_KEYS = {"case_id", "path", "content_digest"} +_DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC +_FILE_FLAGS = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC + + +@dataclass(frozen=True) +class ReviewedSource: + case_id: str + path: str + content: bytes + content_digest: str + + +@dataclass(frozen=True) +class SourceAuthority: + repository: str + manifest_content: bytes + authority_digest: str + sources: tuple[ReviewedSource, ...] + + +def bind_source_record( + value: evidence.EvidenceRecord, + acquisition_digest: str, + change_input_digest: str, +) -> evidence.EvidenceRecord: + stamp = replace( + value.packet.provenance, + analyzer_revision=canonical.content_digest("analyzer"), + acquisition_record_digest=acquisition_digest, + configuration_digest=canonical.content_digest("configuration"), + population_digest=canonical.content_digest("population"), + change_input_digest=change_input_digest, + ) + packet = replace(value.packet, provenance=stamp) + decision = replace(value.decision, packet_digest=evidence.derive_packet_digest(packet)) + return evidence.EvidenceRecord(packet, decision, "") + + +def bind_source_inputs( + named: provenance.NamedInputs, + resolver: InMemoryContentResolver, + authority_content: bytes, + source: ReviewedSource, +) -> tuple[provenance.NamedInputs, InMemoryContentResolver]: + artifacts = dict(resolver.artifacts) + items = [] + replacements = { + provenance.ProvenanceInputRole.ACQUISITION: authority_content, + provenance.ProvenanceInputRole.CHANGE_INPUT: source.content, + } + for item in named.inputs: + content = replacements.get(item.role) + if content is not None: + reference, replacement = stored_artifact(content, artifact_id=item.content.artifact_id) + artifacts.update(replacement.artifacts) + item = replace(item, identity=reference.content_digest, content=reference) + items.append(item) + values = tuple(items) + digest = provenance.derive_named_inputs_digest(values) + return provenance.NamedInputs(values, digest), InMemoryContentResolver(artifacts) + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict): + raise ValueError(f"source authority {label} must be an object") + return cast(dict[str, object], value) + + +def _text(value: object, label: str) -> str: + if not isinstance(value, str) or not value: + raise ValueError(f"source authority {label} must be nonempty text") + return value + + +def _canonical_path(value: object) -> str: + raw = _text(value, "path") + path = PurePosixPath(raw) + forbidden = "\x00" in raw or "\\" in raw or any(part in ("", ".", "..") for part in path.parts) + if path.is_absolute() or path.as_posix() != raw or forbidden: + raise ValueError(f"source authority path is not canonical: {raw}") + return raw + + +@contextmanager +def _repository_descriptor(repository_root: Path) -> Iterator[int]: + try: + descriptor = os.open(repository_root, _DIR_FLAGS) + except (OSError, ValueError) as exc: + raise ValueError("source authority repository root is unsafe") from exc + try: + if not stat.S_ISDIR(os.fstat(descriptor).st_mode): + raise ValueError("source authority repository root is not a directory") + yield descriptor + finally: + os.close(descriptor) + + +def _open_directory(parent: int, name: str) -> int: + try: + descriptor = os.open(name, _DIR_FLAGS, dir_fd=parent) + except (OSError, ValueError) as exc: + raise ValueError(f"source authority directory component is unsafe: {name}") from exc + try: + if stat.S_ISDIR(os.fstat(descriptor).st_mode): + return descriptor + raise ValueError("source authority path component is not a directory") + except (OSError, ValueError) as exc: + if descriptor >= 0: + os.close(descriptor) + raise ValueError(f"source authority directory component is unsafe: {name}") from exc + + +def _read_regular(root: int, relative: str) -> bytes: + parts = PurePosixPath(relative).parts + directory = os.dup(root) + try: + for part in parts[:-1]: + child = _open_directory(directory, part) + os.close(directory) + directory = child + return _read_final_regular(directory, parts[-1], relative) + finally: + os.close(directory) + + +def _read_final_regular(parent: int, name: str, relative: str) -> bytes: + try: + descriptor = os.open(name, _FILE_FLAGS, dir_fd=parent) + except (OSError, ValueError) as exc: + raise ValueError(f"source authority entry is unsafe: {relative}") from exc + try: + before = os.fstat(descriptor) + if not stat.S_ISREG(before.st_mode): + raise ValueError(f"source authority entry is not a regular file: {relative}") + content = _read_descriptor(descriptor, relative) + after = os.fstat(descriptor) + if len(content) != before.st_size or after.st_size != before.st_size: + raise ValueError(f"source authority entry changed while reading: {relative}") + return content + finally: + os.close(descriptor) + + +def _read_descriptor(descriptor: int, relative: str) -> bytes: + chunks: list[bytes] = [] + try: + while chunk := os.read(descriptor, 65536): + chunks.append(chunk) + except OSError as exc: + raise ValueError(f"source authority entry could not be read: {relative}") from exc + return b"".join(chunks) + + +def _read_source(root: int, case: dict[str, object]) -> ReviewedSource: + if set(case) != _CASE_KEYS: + raise ValueError("source authority case fields do not match the closed schema") + case_id = _text(case["case_id"], "case_id") + relative = _canonical_path(case["path"]) + expected_digest = _text(case["content_digest"], "content_digest") + content = _read_regular(root, relative) + if not content: + raise ValueError(f"source authority input is empty: {case_id}") + if canonical.bytes_digest(content) != expected_digest: + raise ValueError(f"source authority content mismatch: {case_id}") + return ReviewedSource(case_id, relative, content, expected_digest) + + +def _allowed_directories(files: set[str]) -> set[str]: + directories: set[str] = set() + for value in files: + parts = PurePosixPath(value).parts + directories.update("/".join(parts[:index]) for index in range(1, len(parts))) + return directories + + +def _entry_mode(entry: os.DirEntry[str], relative: str) -> int: + try: + return entry.stat(follow_symlinks=False).st_mode + except OSError as exc: + raise ValueError(f"source authority entry cannot be inspected: {relative}") from exc + + +def _scan_child(parent: int, entry: os.DirEntry[str], relative: str, files: set[str], directories: set[str]) -> set[str]: + if relative not in directories: + raise ValueError(f"source authority contains an additional directory: {relative}") + descriptor = _open_directory(parent, entry.name) + try: + return _scan_directory(descriptor, relative, files, directories) + finally: + os.close(descriptor) + + +def _scan_entry(parent: int, entry: os.DirEntry[str], prefix: str, files: set[str], directories: set[str]) -> set[str]: + relative = f"{prefix}/{entry.name}" if prefix else entry.name + mode = _entry_mode(entry, relative) + if stat.S_ISLNK(mode): + raise ValueError(f"source authority contains a symlink: {relative}") + if stat.S_ISDIR(mode): + return _scan_child(parent, entry, relative, files, directories) + if stat.S_ISREG(mode) and relative in files: + return {relative} + kind = "additional file" if stat.S_ISREG(mode) else "special entry" + raise ValueError(f"source authority contains an {kind}: {relative}") + + +def _scan_directory(root: int, prefix: str, files: set[str], directories: set[str]) -> set[str]: + found: set[str] = set() + try: + with os.scandir(root) as entries: + for entry in entries: + found.update(_scan_entry(root, entry, prefix, files, directories)) + except OSError as exc: + raise ValueError("source authority repository census failed") from exc + return found + + +def _validate_repository_entries(root: int, source_paths: tuple[str, ...]) -> None: + allowed_files = {_MANIFEST, *source_paths} + found = _scan_directory(root, "", allowed_files, _allowed_directories(allowed_files)) + if found != allowed_files: + raise ValueError("source authority entries do not match the exact repository") + + +def load_source_authority( + repository_root: Path, + repository: str, + expected_cases: tuple[tuple[str, str], ...], +) -> SourceAuthority: + with _repository_descriptor(repository_root) as root: + manifest_content = _read_regular(root, _MANIFEST) + manifest = _validate_manifest(manifest_content, repository) + cases = _manifest_cases(manifest, expected_cases) + _validate_repository_entries(root, tuple(path for _, path in expected_cases)) + sources = tuple(_read_source(root, item) for item in cases) + return SourceAuthority(repository, manifest_content, canonical.bytes_digest(manifest_content), sources) + + +def _validate_manifest(content: bytes, repository: str) -> dict[str, object]: + if not content: + raise ValueError("source authority manifest is empty") + manifest = _object(json.loads(content), "manifest") + if set(manifest) != _HEADER_KEYS or manifest.get("schema_version") != _SCHEMA: + raise ValueError("source authority manifest does not match the closed schema") + if manifest.get("repository") != repository: + raise ValueError("source authority repository identity does not match") + producer = _text(manifest["producer_identity"], "producer_identity") + reviewer = _text(manifest["reviewer_identity"], "reviewer_identity") + if producer == reviewer: + raise ValueError("source authority requires an independent reviewer") + return manifest + + +def _manifest_cases(manifest: dict[str, object], expected_cases: tuple[tuple[str, str], ...]) -> tuple[dict[str, object], ...]: + raw_cases = manifest["cases"] + if not isinstance(raw_cases, list): + raise ValueError("source authority cases must be a list") + cases = tuple(_object(item, "case") for item in raw_cases) + declared = tuple((_text(item.get("case_id"), "case_id"), _canonical_path(item.get("path"))) for item in cases) + if declared != expected_cases or len(set(declared)) != len(declared): + raise ValueError("source authority cases do not match the exact authored cases") + return cases + + +__all__ = [ + "ReviewedSource", + "SourceAuthority", + "bind_source_inputs", + "bind_source_record", + "load_source_authority", +] diff --git a/tests/fixtures/detection_core/obligation_repository/reference/obligation_missing.py b/tests/fixtures/detection_core/obligation_repository/reference/obligation_missing.py new file mode 100644 index 0000000..14be4ba --- /dev/null +++ b/tests/fixtures/detection_core/obligation_repository/reference/obligation_missing.py @@ -0,0 +1,3 @@ +def publish(request: object, store: object, policy: object) -> None: + if policy.permits(request): + store.publish(request) diff --git a/tests/fixtures/detection_core/obligation_repository/reference/obligation_negative.py b/tests/fixtures/detection_core/obligation_repository/reference/obligation_negative.py new file mode 100644 index 0000000..c0a1c1f --- /dev/null +++ b/tests/fixtures/detection_core/obligation_repository/reference/obligation_negative.py @@ -0,0 +1,3 @@ +def publish(request: object, store: object, allowed: bool) -> None: + if allowed: + store.publish(request) diff --git a/tests/fixtures/detection_core/obligation_repository/reference/obligation_positive.py b/tests/fixtures/detection_core/obligation_repository/reference/obligation_positive.py new file mode 100644 index 0000000..0493fd4 --- /dev/null +++ b/tests/fixtures/detection_core/obligation_repository/reference/obligation_positive.py @@ -0,0 +1,2 @@ +def publish(request: object, store: object) -> None: + store.publish(request) diff --git a/tests/fixtures/detection_core/obligation_repository/source_authority.json b/tests/fixtures/detection_core/obligation_repository/source_authority.json new file mode 100644 index 0000000..6343d8f --- /dev/null +++ b/tests/fixtures/detection_core/obligation_repository/source_authority.json @@ -0,0 +1,23 @@ +{ + "schema_version": "detection-core-source-authority-v1", + "repository": "obligation_repository", + "producer_identity": "reference-fixture-author@v1", + "reviewer_identity": "independent-reference-review@v1", + "cases": [ + { + "case_id": "obligation-positive", + "path": "reference/obligation_positive.py", + "content_digest": "e29307cd0ae6d1d817c19fd79bfa4d494d602fc5daf9aad4e6da9caa1ee31e29" + }, + { + "case_id": "obligation-negative", + "path": "reference/obligation_negative.py", + "content_digest": "223e1065ef30aa0cb1073125e2f845c1b2b5a498400ac34cdc5f78bb0621cd7a" + }, + { + "case_id": "obligation-missing", + "path": "reference/obligation_missing.py", + "content_digest": "c119b846e6797a9e1b6c96aff73426c1f53a46053ab074f88931e7d07383652c" + } + ] +} diff --git a/tests/fixtures/detection_core/taint_repository/reference/flow_missing.py b/tests/fixtures/detection_core/taint_repository/reference/flow_missing.py new file mode 100644 index 0000000..283b23d --- /dev/null +++ b/tests/fixtures/detection_core/taint_repository/reference/flow_missing.py @@ -0,0 +1,2 @@ +def handle(value: str, effect: object, transform: object) -> None: + effect.write(transform.apply(value)) diff --git a/tests/fixtures/detection_core/taint_repository/reference/flow_negative.py b/tests/fixtures/detection_core/taint_repository/reference/flow_negative.py new file mode 100644 index 0000000..df22a09 --- /dev/null +++ b/tests/fixtures/detection_core/taint_repository/reference/flow_negative.py @@ -0,0 +1,2 @@ +def handle(value: str, effect: object) -> None: + effect.write(value.strip()) diff --git a/tests/fixtures/detection_core/taint_repository/reference/flow_positive.py b/tests/fixtures/detection_core/taint_repository/reference/flow_positive.py new file mode 100644 index 0000000..288fb80 --- /dev/null +++ b/tests/fixtures/detection_core/taint_repository/reference/flow_positive.py @@ -0,0 +1,2 @@ +def handle(value: str, effect: object) -> None: + effect.write(value) diff --git a/tests/fixtures/detection_core/taint_repository/source_authority.json b/tests/fixtures/detection_core/taint_repository/source_authority.json new file mode 100644 index 0000000..277c648 --- /dev/null +++ b/tests/fixtures/detection_core/taint_repository/source_authority.json @@ -0,0 +1,23 @@ +{ + "schema_version": "detection-core-source-authority-v1", + "repository": "taint_repository", + "producer_identity": "reference-fixture-author@v1", + "reviewer_identity": "independent-reference-review@v1", + "cases": [ + { + "case_id": "flow-positive", + "path": "reference/flow_positive.py", + "content_digest": "ece3add1d60cf01527cb83746b20c3be1f9468c589677932830efbb6a5d67a48" + }, + { + "case_id": "flow-negative", + "path": "reference/flow_negative.py", + "content_digest": "2413778080ea6626b47dff8ee82a54f88902707a90e1f41c67a7380bc8784a5f" + }, + { + "case_id": "flow-missing", + "path": "reference/flow_missing.py", + "content_digest": "662b99c3c093b38de535ab21b17a1acfe10bb70911c8b83768844b607424c345" + } + ] +} diff --git a/tests/test_detection_core_end_to_end.py b/tests/test_detection_core_end_to_end.py new file mode 100644 index 0000000..ef1356e --- /dev/null +++ b/tests/test_detection_core_end_to_end.py @@ -0,0 +1,230 @@ +"""Two unlike families traverse the complete synthetic readiness path.""" + +from __future__ import annotations + +import os +import shutil +from dataclasses import replace +from pathlib import Path + +import pytest +from detection_core_end_to_end import ReferencePathResult, run_reference_family_path + +from benchmarks.valid import readiness_contract +from openultrasast.core import canonical +from openultrasast.core.admission_report import AdmissionDecision +from openultrasast.core.custody import PopulationStatus +from openultrasast.core.evidence import NeutralOutcome +from openultrasast.core.readiness import ReadinessEvidenceUse, ReadinessStage, StageResult + +FIXTURES = Path(__file__).parent / "fixtures" / "detection_core" + + +def _copied_obligation_fixture(tmp_path: Path) -> Path: + fixture_root = tmp_path / "detection-core" + shutil.copytree(FIXTURES, fixture_root) + return fixture_root + + +def _run_obligation(fixture_root: Path, scratch: Path) -> ReferencePathResult: + return run_reference_family_path( + fixture_root=fixture_root, + family_file="obligation_family.json", + cases_file="obligation_cases.json", + repository="obligation_repository", + scratch=scratch, + ) + + +def _assert_obligation_refused(fixture_root: Path, scratch: Path) -> None: + with pytest.raises(ValueError, match="source authority"): + _run_obligation(fixture_root, scratch) + assert not tuple(scratch.glob("*-labels.json")) + + +@pytest.mark.parametrize( + ("family_file", "cases_file", "repository", "family_id", "ecosystem", "silence_reason"), + ( + ( + "obligation_family.json", + "obligation_cases.json", + "obligation_repository", + "operation-constraint", + "reference-obligation", + "missing_constraint_relation", + ), + ( + "taint_family.json", + "taint_cases.json", + "taint_repository", + "ordered-taint-flow", + "reference-flow", + "missing_flow_segment", + ), + ), +) +def test_reference_family_completes_synthetic_readiness_path( + tmp_path: Path, + family_file: str, + cases_file: str, + repository: str, + family_id: str, + ecosystem: str, + silence_reason: str, +) -> None: + result = run_reference_family_path( + fixture_root=FIXTURES, + family_file=family_file, + cases_file=cases_file, + repository=repository, + scratch=tmp_path, + ) + + _assert_instrument_and_readiness(result, family_id, ecosystem) + _assert_bounded_outputs(result, silence_reason) + _assert_reviewed_guarded_admission(result) + + +def test_replaced_authored_source_is_refused_before_admission(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + source = fixture_root / "obligation_repository/reference/obligation_positive.py" + source.write_text("def unrelated() -> int:\n return 7\n") + + _assert_obligation_refused(fixture_root, tmp_path) + + +def test_swapped_authored_sources_are_refused_before_admission(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + positive = fixture_root / "obligation_repository/reference/obligation_positive.py" + negative = fixture_root / "obligation_repository/reference/obligation_negative.py" + positive_bytes, negative_bytes = positive.read_bytes(), negative.read_bytes() + positive.write_bytes(negative_bytes) + negative.write_bytes(positive_bytes) + + _assert_obligation_refused(fixture_root, tmp_path) + + +def test_nested_authority_named_file_is_refused_before_admission(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + nested = fixture_root / "obligation_repository/reference/source_authority.json" + nested.write_bytes(b"") + + _assert_obligation_refused(fixture_root, tmp_path) + + +def test_outside_root_authority_symlink_is_refused_without_descriptor_leak(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + manifest = fixture_root / "obligation_repository/source_authority.json" + outside = tmp_path / "outside-authority.json" + manifest.rename(outside) + manifest.symlink_to(outside) + descriptor_root = Path("/proc/self/fd") + before = len(tuple(descriptor_root.iterdir())) + + for _attempt in range(5): + _assert_obligation_refused(fixture_root, tmp_path) + + assert len(tuple(descriptor_root.iterdir())) == before + + +def test_inside_root_source_symlink_is_refused_before_admission(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + repository = fixture_root / "obligation_repository" + source = repository / "reference/obligation_positive.py" + hidden = repository / "reference/source_authority.json" + source.rename(hidden) + source.symlink_to(hidden.name) + + _assert_obligation_refused(fixture_root, tmp_path) + + +def test_outside_root_source_symlink_is_refused_before_admission(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + source = fixture_root / "obligation_repository/reference/obligation_positive.py" + outside = tmp_path / "outside-source.py" + source.rename(outside) + source.symlink_to(outside) + + _assert_obligation_refused(fixture_root, tmp_path) + + +def test_additional_fifo_is_refused_without_opening_it(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + fifo = fixture_root / "obligation_repository/reference/untracked.pipe" + os.mkfifo(fifo) + + _assert_obligation_refused(fixture_root, tmp_path) + + +def test_declared_fifo_is_refused_without_opening_it(tmp_path: Path) -> None: + fixture_root = _copied_obligation_fixture(tmp_path) + source = fixture_root / "obligation_repository/reference/obligation_positive.py" + source.unlink() + os.mkfifo(source) + + _assert_obligation_refused(fixture_root, tmp_path) + + +def _assert_instrument_and_readiness( + result: ReferencePathResult, + family_id: str, + ecosystem: str, +) -> None: + assert result.family_id == family_id + assert result.ecosystem == ecosystem + assert result.repository_bytes > 0 + assert result.opened_input_count == 3 + identity = result.readiness[0].identity + assert identity.acquisition_record_digest == result.source_authority_digest + assert {item.record.packet.provenance.acquisition_record_digest for item in result.inputs} == {result.source_authority_digest} + assert tuple(item.record.packet.provenance.change_input_digest for item in result.inputs) == result.source_content_digests + assert tuple(item.stage for item in result.readiness) == tuple(ReadinessStage) + assert tuple(item.result for item in result.readiness) == (StageResult.PASSED,) * len(ReadinessStage) + assert tuple(item.evidence_use for item in result.readiness) == ( + ReadinessEvidenceUse.REGRESSION_FIXTURE, + ReadinessEvidenceUse.REGRESSION_FIXTURE, + ReadinessEvidenceUse.REGRESSION_FIXTURE, + ReadinessEvidenceUse.REGRESSION_FIXTURE, + ReadinessEvidenceUse.REGRESSION_FIXTURE, + ReadinessEvidenceUse.DEVELOPMENT_POPULATION, + ReadinessEvidenceUse.FRESH_ADMISSION, + ReadinessEvidenceUse.CONSUMER_INTEGRATION, + ) + for index, item in enumerate(result.readiness): + assert item.predecessor_digests == tuple(prior.result_digest for prior in result.readiness[:index]) + assert item.identity == result.readiness[0].identity + changed_identity = replace(identity, acquisition_record_digest=canonical.content_digest("different-source-authority")) + with pytest.raises(readiness_contract.ReadinessBlocked, match="readiness is stale"): + readiness_contract.require_stage_authorized(ReadinessStage.ADMISSION, result.readiness[:6], changed_identity) + + +def _assert_bounded_outputs(result: ReferencePathResult, silence_reason: str) -> None: + assert tuple(item.record.decision.outcome for item in result.inputs) == ( + NeutralOutcome.VIOLATION, + NeutralOutcome.SATISFIED, + NeutralOutcome.INDETERMINATE, + ) + assert tuple((len(item.defects), len(item.satisfactions), len(item.silence)) for item in result.verdicts) == ( + (1, 0, 0), + (0, 1, 0), + (0, 0, 1), + ) + assert result.verdicts[2].silence[0].reasons == (silence_reason,) + assert all(item.total_reply_ms > 0 for item in result.verdicts) + + +def _assert_reviewed_guarded_admission(result: ReferencePathResult) -> None: + replay = result.admission + assert replay.manifest.label_isolated is True + assert replay.manifest.labels_opened is False + assert replay.manifest.review.producer_identity != replay.manifest.review.reviewer_identity + assert replay.label_release_receipt.status is PopulationStatus.SPENT + assert replay.report.decision is AdmissionDecision.GO + assert result.population_status_at_label_read is PopulationStatus.SPENT + assert result.label_read_count == 1 + assert replay.readiness == result.readiness[:6] + references = tuple( + (reference.stage, reference.result, reference.result_digest) for reference in replay.report.measurement.prerequisite_readiness + ) + expected = tuple((item.stage, item.result, item.result_digest) for item in replay.readiness) + assert references == expected