diff --git a/.github/workflows/constitution.yml b/.github/workflows/constitution.yml index 2a71826..ae34fdd 100644 --- a/.github/workflows/constitution.yml +++ b/.github/workflows/constitution.yml @@ -97,13 +97,17 @@ jobs: - run: pip install pytest - # `domain` added by M2/E4/#12, which places the pure domain model's unit tests beside - # the code they test. Without it those 244 tests would exist and never run — the exact - # defect recorded as EPD-004 when the kernel landed, repeating one issue later. - # Reported as EPD-006: the Engineering Plan assigns unit tests to #12 and no issue owns - # extending the runner that executes them. + # `domain` added by M2/E4/#12 and `contexts` by M3/E5/#15, each placing its unit tests + # beside the code they test. Without the path here those tests would exist and never + # run — the exact defect recorded as EPD-004 when the kernel landed. Reported as + # EPD-006 (domain) and EPD-007 (contexts): the Engineering Plan assigns unit tests to + # these issues and no issue owns extending the runner that executes them. + # + # `contexts` will keep gaining packages through E5–E11. It is named as a root rather + # than per-context so that a new context's tests run the day they are written, instead + # of on the day someone remembers this file. - name: Unit tests beside their code - run: pytest kernel domain -q -rs + run: pytest kernel domain contexts -q -rs # Integration tests cross a real boundary, and some of them assert on repository refs # rather than on files. Without this job they would never execute — the same gap that left diff --git a/contexts/ingest/__init__.py b/contexts/ingest/__init__.py new file mode 100644 index 0000000..e0c9e30 --- /dev/null +++ b/contexts/ingest/__init__.py @@ -0,0 +1,89 @@ +"""Ingest — acquire from a Source, and canonicalise via an Instrument parser. + +M3/E5/#15 delivers the **contract** half of this context: what an acquisition channel must +provide, what it may never let out, and where the one sanctioned clock read lives. It delivers +no channel. The ISSDC-PRADAN adapter is #16; the SoLEXS and HEL1OS parsers are #17 and #18; +the bitemporal write path is #19. + +Public interface (E5 §4): + + SourceAdapter the protocol a channel satisfies + verify_conformance what an adapter must pass before its data enters the platform + SourceDescriptor source_id · authority · latency_class · granularity · retrieval + RawArtifact a Tier 0 artifact, identified rather than held + AcquisitionProvenance who fetched what, from where, and when the system learned it + Acquisition the two together, which is what `acquire` returns + boundary.stamp the single sanctioned clock read (ADR-0004, TIS §0.4) + Credential a secret that cannot be printed or serialised (STD-19) + assert_credential_free refuse to let one cross the boundary + +THE THREE RULES THIS CONTEXT EXISTS TO KEEP +-------------------------------------------- +**Credentials never leave.** E5 §13 makes this the highest-sensitivity context — it holds the +only secrets in the system. `Credential` redacts itself in `repr`, `str` and `format`, so the +common leak routes are closed by the type; `assert_credential_free` closes the rest by walking +anything about to cross outward. + +**Tier 0 bytes are never redistributed.** `RawArtifact` has a digest, a size, a way to +re-acquire, and no field that could hold the bytes. ADR-0023 and E5 §12 both land here. + +**Both times, and the second one never fabricated.** `ingest_time` is stamped at this +boundary and nowhere else (ADR-0004). Nothing here supplies one for a historical row — +ADR-0022 gives `null` exactly one meaning, and E5 §11(ii) forbids any code path from writing +a non-null value for data that predates bitemporal capture. + +IMPORTS. `contracts`, `domain`, `kernel` and the standard library — exactly what +`contexts.ingest`'s policy grants (ADR-0026, enforced by M2/E4/#13). No other context. +""" + +from contexts.ingest.acquisition import ( + Acquisition, + AcquisitionProvenance, + RawArtifact, +) +from contexts.ingest.contract import ( + CONFORMANCE_GATE, + SourceAdapter, + verify_acquisition, + verify_conformance, + verify_descriptor, +) +from contexts.ingest.credentials import ( + CREDENTIAL_BOUNDARY, + Credential, + assert_credential_free, +) +from contexts.ingest.descriptor import ( + LatencyClass, + RetrievalDescriptor, + SourceDescriptor, +) +from contexts.ingest.errors import ( + ContractViolation, + IngestError, + IntegrityFailure, + PolicyRejection, + UnavailableResource, +) + +__all__ = [ + "Acquisition", + "AcquisitionProvenance", + "CONFORMANCE_GATE", + "CREDENTIAL_BOUNDARY", + "ContractViolation", + "Credential", + "IngestError", + "IntegrityFailure", + "LatencyClass", + "PolicyRejection", + "RawArtifact", + "RetrievalDescriptor", + "SourceAdapter", + "SourceDescriptor", + "UnavailableResource", + "assert_credential_free", + "verify_acquisition", + "verify_conformance", + "verify_descriptor", +] diff --git a/contexts/ingest/acquisition.py b/contexts/ingest/acquisition.py new file mode 100644 index 0000000..3dcd6f2 --- /dev/null +++ b/contexts/ingest/acquisition.py @@ -0,0 +1,190 @@ +"""What an acquisition produces: a referenced artifact and the provenance of getting it. + +E5 §4: `acquire(selector) -> RawArtifact + AcquisitionProvenance`. + +THE ARTIFACT HOLDS NO BYTES, AND THAT IS THE POINT +--------------------------------------------------- +`RawArtifact` has a digest, a size, a retrieval descriptor and an optional cache path. It has +no `bytes` field and no `content` field, and `test_a_raw_artifact_cannot_carry_bytes` asserts +that it never gains one. Two separate rules land on the same design: + + ADR-0023 / STD-23 / E5 §11(iv) Tier 0 bytes are referenced, never redistributed. Local + copies are evictable caches, never the system of record. + E5 §12 Streaming parse; a whole-day archive must not be fully + materialised. + +A type that could hold the bytes would make both violations one attribute access away, and +neither would be visible in review — the object would simply be larger. + +`cache_path` is the evictable copy. It is optional because a cache that has been evicted is +the normal case, not an error: what makes the artifact usable again is `retrieval`, which is +why that field is required and the path is not. + +THE CLOCK READ +-------------- +`AcquisitionProvenance.ingest_time` is stamped by `contexts.ingest.boundary.stamp`, the single +sanctioned clock read in the system (ADR-0004, TIS §0.4). It is required and non-null here, +because this type describes an acquisition that *just happened*. Historical rows that predate +bitemporal capture carry `ingest_time = None` on the Observation (ADR-0022) and have no +AcquisitionProvenance at all — there is no acquisition to describe, and E5 §11(ii) forbids any +code path from writing a non-null ingest_time for them. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path + +from contexts.ingest.descriptor import RetrievalDescriptor +from domain.errors import ContractViolation +from domain.values import Digest, Identifier, RunId, Timestamp + + +@dataclass(frozen=True) +class RawArtifact: + """A Tier 0 artifact, identified rather than held (ADR-0023).""" + + digest: Digest + size_bytes: int + retrieval: RetrievalDescriptor + #: Where an evictable local copy currently sits, if one does. Never the system of record. + cache_path: Path | None = None + + def __post_init__(self) -> None: + if not isinstance(self.digest, Digest): + raise ContractViolation( + "/digest", + f"digest must be a Digest, got {type(self.digest).__name__}. Only the " + f"provenance kernel mints one (ADR-0005); this records the one it minted.", + ) + if isinstance(self.size_bytes, bool) or not isinstance(self.size_bytes, int): + raise ContractViolation( + "/size_bytes", + f"size_bytes must be an integer, got {type(self.size_bytes).__name__}", + ) + if self.size_bytes < 0: + raise ContractViolation( + "/size_bytes", f"size_bytes must be non-negative, got {self.size_bytes}" + ) + if not isinstance(self.retrieval, RetrievalDescriptor): + raise ContractViolation( + "/retrieval", + f"retrieval must be a RetrievalDescriptor, got " + f"{type(self.retrieval).__name__}. Without one these bytes cannot be " + f"re-acquired, and ADR-0023 requires that any stage can.", + ) + if self.cache_path is not None and not isinstance(self.cache_path, Path): + raise ContractViolation( + "/cache_path", + f"cache_path must be a Path or None, got {type(self.cache_path).__name__}", + ) + + @property + def is_cached(self) -> bool: + """Whether a local copy is currently recorded. An evicted cache is not an error.""" + return self.cache_path is not None + + +@dataclass(frozen=True) +class AcquisitionProvenance: + """The record of one acquisition, for the provenance kernel to store (E5 §7). + + Carries the run that performed it, the source it came from, the digest it produced and + the instant the system learned of it. Nothing else — and specifically no credential, no + URL that might embed a token, and no response body (E5 §13, §14). + """ + + run_id: RunId + source_id: Identifier + artifact_digest: Digest + #: The sanctioned clock read (ADR-0004). Required and non-null: this describes an + #: acquisition that happened, so the moment is known by construction. + ingest_time: Timestamp + #: Instruments whose products this artifact contains, if the source states them. Empty + #: where the channel is instrument-agnostic; a daily archive may carry several. + instruments: tuple[Identifier, ...] = field(default=()) + + def __post_init__(self) -> None: + if not isinstance(self.run_id, RunId): + raise ContractViolation( + "/run_id", f"run_id must be a RunId, got {type(self.run_id).__name__}" + ) + if not isinstance(self.source_id, Identifier): + raise ContractViolation( + "/source_id", + f"source_id must be an Identifier, got {type(self.source_id).__name__}", + ) + if not isinstance(self.artifact_digest, Digest): + raise ContractViolation( + "/artifact_digest", + f"artifact_digest must be a Digest, got " + f"{type(self.artifact_digest).__name__}", + ) + if not isinstance(self.ingest_time, Timestamp): + raise ContractViolation( + "/ingest_time", + f"ingest_time must be a Timestamp, got {type(self.ingest_time).__name__}. " + f"An acquisition that happened has a known moment; ADR-0022's null is for " + f"historical rows, which have no acquisition to describe.", + ) + if not isinstance(self.instruments, tuple): + raise ContractViolation( + "/instruments", + f"instruments must be a tuple, got {type(self.instruments).__name__}", + ) + if any(not isinstance(i, Identifier) for i in self.instruments): + raise ContractViolation( + "/instruments", "every instrument must be an Identifier" + ) + if len(set(self.instruments)) != len(self.instruments): + raise ContractViolation("/instruments", "instruments must be unique") + + def to_dict(self) -> dict[str, object]: + """A structured record, suitable for logging (TIS §0.3) and for the kernel. + + Contains `source_id`, `instrument_id`s and the artifact digest — exactly the fields + E5 §14 says to log — and none of the fields it says never to log. + """ + return { + "run_id": str(self.run_id), + "source_id": str(self.source_id), + "artifact_digest": str(self.artifact_digest), + "ingest_time": str(self.ingest_time), + "instruments": [str(i) for i in self.instruments], + } + + +@dataclass(frozen=True) +class Acquisition: + """The pair `acquire` returns (E5 §4), kept together because neither is usable alone. + + An artifact without its provenance is bytes nobody can attribute; provenance without its + artifact is a claim about nothing. + """ + + artifact: RawArtifact + provenance: AcquisitionProvenance + + def __post_init__(self) -> None: + if not isinstance(self.artifact, RawArtifact): + raise ContractViolation( + "/artifact", + f"artifact must be a RawArtifact, got {type(self.artifact).__name__}", + ) + if not isinstance(self.provenance, AcquisitionProvenance): + raise ContractViolation( + "/provenance", + f"provenance must be an AcquisitionProvenance, got " + f"{type(self.provenance).__name__}", + ) + if self.artifact.digest != self.provenance.artifact_digest: + raise ContractViolation( + "", + f"the provenance records digest {self.provenance.artifact_digest.short} " + f"but the artifact is {self.artifact.digest.short}. A record that points at " + f"different bytes than the artifact it accompanies is worse than no record: " + f"it looks like attribution.", + ) + + +__all__ = ["Acquisition", "AcquisitionProvenance", "RawArtifact"] diff --git a/contexts/ingest/boundary.py b/contexts/ingest/boundary.py new file mode 100644 index 0000000..6a65c4e --- /dev/null +++ b/contexts/ingest/boundary.py @@ -0,0 +1,78 @@ +"""The acquisition boundary — the single sanctioned clock read in the system. + +ADR-0004: *"`ingest_time` is stamped at the acquisition boundary. This is the single +sanctioned clock read in the system; determinism elsewhere is governed by ADR-0021."* +TIS §0.4 states the same rule from the other side: no context may read wall-clock time, +*"Exception: `ingest_time` capture in E5, which is definitionally a clock read at the +acquisition boundary."* + +WHY THE EXCEPTION LIVES IN ITS OWN MODULE +------------------------------------------ +`domain/values/Timestamp` deliberately has no `now()`, and `test_timestamp_has_no_clock_reader` +(M2/E4/#12) pins that. The domain is pure, so the moment the system learned something cannot +originate there. It has to originate somewhere, and putting it in one named module with one +exported function means the exception is enumerable: `grep -rn 'ingest.boundary' .` finds +every place in the repository that reads a clock. + +An exception spread across several call sites is not an exception, it is a convention. + +WHAT THIS DOES NOT DO +--------------------- +It does not stamp historical data. E5 §11(ii) is unambiguous: *"No code path writes a non-null +`ingest_time` for historical data."* There is no function here that takes a row and supplies a +time for it, and no default. A row that predates bitemporal capture carries `None`, which +ADR-0022 gives exactly one meaning, and the only way to obtain a stamp is to call `stamp()` at +the moment of an actual acquisition. Backfilling would require calling it and then attaching +the result to old rows — which `domain.invariants.ingest_time_is_not_backfilled` detects, and +which no code here makes convenient. +""" + +from __future__ import annotations + +import time +from datetime import datetime, timezone + +from domain.values import Timestamp + + +def stamp() -> Timestamp: + """Read the clock at the acquisition boundary and return the instant as a Timestamp. + + Always UTC with an explicit offset, because a local-time stamp is ambiguous twice a year + and a naive one is ambiguous always. Microsecond precision is kept: two acquisitions in + the same second are two acquisitions, and collapsing them to one instant would make an + ordering question unanswerable later. + """ + now = datetime.now(timezone.utc) + return Timestamp(now.isoformat().replace("+00:00", "Z")) + + +def stamp_from_epoch(seconds: float) -> Timestamp: + """The same stamp, from a caller-supplied epoch instant. + + Exists for the acquisition path that already has the instant a transfer completed — an + adapter that records the completion time of a download should record *that* moment, not + the slightly later moment it got around to calling `stamp()`. + + It is not a way to fabricate a time. It takes a number a caller must already possess, and + supplying an arbitrary one is indistinguishable from lying about when data arrived — + which is what the no-backfill invariant checks for, not what a type can prevent. + """ + if isinstance(seconds, bool) or not isinstance(seconds, (int, float)): + raise TypeError(f"epoch seconds must be a number, got {type(seconds).__name__}") + moment = datetime.fromtimestamp(float(seconds), tz=timezone.utc) + return Timestamp(moment.isoformat().replace("+00:00", "Z")) + + +def monotonic() -> float: + """A monotonic reading for measuring elapsed time within one acquisition. + + Not a clock in the ADR-0004 sense: it has no epoch, cannot be converted to a Timestamp, + and cannot be recorded as when anything happened. It exists so an adapter can implement + the retry and timeout policy E5 §15 makes adapter-local without reaching for + `time.time()`, which would be a second, unsanctioned wall-clock read. + """ + return time.monotonic() + + +__all__ = ["monotonic", "stamp", "stamp_from_epoch"] diff --git a/contexts/ingest/contract.py b/contexts/ingest/contract.py new file mode 100644 index 0000000..f8f0e95 --- /dev/null +++ b/contexts/ingest/contract.py @@ -0,0 +1,212 @@ +"""The `SourceAdapter` contract, and the conformance check every adapter must pass. + +E5 §4 fixes the interface: + + SourceAdapter: + descriptor() -> SourceDescriptor + acquire(selector) -> RawArtifact + AcquisitionProvenance + +WHY A PROTOCOL AND NOT A BASE CLASS +------------------------------------ +ADR-0003 is explicit about what it does *not* authorise: *"a source-plugin registry, a +dispatch layer, or a configurable adapter framework. Those are paid abstractions and remain +forbidden until a second source exists."* ADR-0025 lists "generic base classes" in the same +category. + +A `typing.Protocol` is none of those. It declares what a caller may rely on; it is not +inherited from, it constructs nothing, it dispatches nothing, and deleting it would not change +one line of an adapter's behaviour. Under ADR-0025's test it is a free seam: stating what an +acquisition channel must provide would still be the correct design with exactly one source +forever, because the parser downstream has to be able to trust its input regardless of which +channel produced it. + +There is no registry here. No `register()`, no adapter table, no lookup by `source_id`, no +configuration. `verify_conformance` takes an adapter a caller already holds. + +WHAT CONFORMANCE MEANS +---------------------- +Structural conformance — having the right methods — is the least of it, and is the part a type +checker already handles. `verify_conformance` checks the parts that decide whether the rest of +the platform can trust the adapter's output, which no type system checks: + + 1. `descriptor()` returns a real `SourceDescriptor`, and returns the *same* one every time. + A descriptor that varies between calls means the channel's latency or authority is + whatever it happened to be when someone asked. + 2. Nothing the adapter publishes carries a credential (STD-19, E5 §13). + 3. `acquire()` returns an `Acquisition` whose provenance names the same digest as its + artifact, with an `ingest_time` that is present and non-null (E5 §11(i), ADR-0004). + 4. The artifact carries no bytes and does carry a retrieval descriptor, so Tier 0 is + referenced rather than redistributed (ADR-0023, STD-23, E5 §11(iv)). + +The check is offered as a function rather than imposed as a base class for the same reason the +interface is a Protocol: an adapter that must inherit something in order to be checked is an +adapter the framework owns. +""" + +from __future__ import annotations + +from typing import Protocol, runtime_checkable + +from contexts.ingest.acquisition import Acquisition, AcquisitionProvenance, RawArtifact +from contexts.ingest.credentials import assert_credential_free +from contexts.ingest.descriptor import SourceDescriptor +from contexts.ingest.errors import PolicyRejection +from domain.errors import ContractViolation +from domain.values import Timestamp + +#: The gate named when an adapter fails conformance. One string, so a caller matching on it +#: and a test asserting it cannot drift. +CONFORMANCE_GATE = "source-adapter-conformance" + +#: Fields a `RawArtifact` may never gain. Checked by name because the rule is about the type's +#: shape rather than about any particular value: an artifact that *can* hold bytes will hold +#: them, and Tier 0 redistribution becomes one attribute away (ADR-0023, E5 §12). +FORBIDDEN_ARTIFACT_FIELDS = frozenset({"bytes", "content", "data", "body", "payload", "raw"}) + + +@runtime_checkable +class SourceAdapter(Protocol): + """A channel of acquisition (ADR-0003). + + `runtime_checkable` so `isinstance` can report a missing method as a conformance failure + rather than as an `AttributeError` three frames later. It checks method *presence* only, + which is why `verify_conformance` exists and does the rest. + """ + + def descriptor(self) -> SourceDescriptor: + """What this channel is: authority, latency class, granularity, how to retrieve. + + Must be constant for the life of the adapter. A channel whose declared latency + changes between calls has not declared a latency. + """ + ... + + def acquire(self, selector: str) -> Acquisition: + """Retrieve one artifact, returning it with the provenance of retrieving it. + + `selector` names what to fetch in the channel's own terms — a date for a daily + archive, a product identifier elsewhere. It is a plain string because interpreting it + is the adapter's business; a structured selector type would be a framework deciding + for adapters that do not yet exist (ADR-0003, ADR-0025). + + Raises `UnavailableResource` when the channel cannot be reached and retries are + exhausted, `IntegrityFailure` when the retrieved bytes do not match their digest. + Neither leaves partial state: E5 §9 states that `aborted` leaves no observations. + """ + ... + + +def _fail(message: str) -> None: + raise PolicyRejection(CONFORMANCE_GATE, message) + + +def verify_descriptor(adapter: object) -> SourceDescriptor: + """Check `descriptor()` and return what it produced. + + Separated from the full check because an adapter can be asked what it is without being + asked to fetch anything, and E5 §18's acceptance criterion — the ISSDC channel registered + with latency class `~33d`, granularity `daily-archive` — is exactly that question. + """ + if not hasattr(adapter, "descriptor"): + _fail(f"{type(adapter).__name__} has no descriptor(); E5 §4 requires one") + + descriptor = adapter.descriptor() + if not isinstance(descriptor, SourceDescriptor): + _fail( + f"descriptor() returned {type(descriptor).__name__}, not a SourceDescriptor. " + f"A channel that describes itself in its own vocabulary cannot be compared to " + f"another one." + ) + + again = adapter.descriptor() + if again != descriptor: + _fail( + "descriptor() is not constant between calls. A channel whose declared latency " + "or authority varies has declared neither." + ) + + assert_credential_free(descriptor, what="the source descriptor") + return descriptor + + +def verify_acquisition(acquisition: object) -> Acquisition: + """Check one `acquire()` result against everything the platform will assume about it.""" + if not isinstance(acquisition, Acquisition): + _fail( + f"acquire() returned {type(acquisition).__name__}, not an Acquisition. E5 §4 " + f"requires a RawArtifact and its AcquisitionProvenance together." + ) + + artifact: RawArtifact = acquisition.artifact + provenance: AcquisitionProvenance = acquisition.provenance + + # Tier 0 is referenced, never redistributed (ADR-0023, STD-23, E5 §11(iv)). + carried = FORBIDDEN_ARTIFACT_FIELDS & set(vars(artifact)) + if carried: + _fail( + f"the raw artifact carries {sorted(carried)}. Tier 0 bytes are referenced, never " + f"held: local copies are evictable caches and a whole-day archive must not be " + f"materialised (ADR-0023, E5 §12)." + ) + if artifact.retrieval is None: + _fail( + "the raw artifact has no retrieval descriptor, so these bytes cannot be " + "re-acquired. ADR-0023 requires that any stage can." + ) + + # Both times, and the ingest one non-null (E5 §11(i), ADR-0004). + if not isinstance(provenance.ingest_time, Timestamp): + _fail( + "the acquisition provenance has no ingest_time. It is stamped at this boundary " + "and nowhere else (ADR-0004); ADR-0022's null is for historical rows, which have " + "no acquisition to describe." + ) + + # The record must point at the artifact it accompanies. `Acquisition` enforces this at + # construction, so reaching here means it was built some other way. + if artifact.digest != provenance.artifact_digest: + _fail( + f"the provenance records {provenance.artifact_digest.short} but the artifact is " + f"{artifact.digest.short}." + ) + + assert_credential_free(provenance, what="the acquisition provenance") + assert_credential_free(artifact, what="the raw artifact") + return acquisition + + +def verify_conformance(adapter: object, *, selector: str) -> Acquisition: + """Run the full check against a real adapter, and return what it acquired. + + This is what M3/E5/#16 will run against the ISSDC-PRADAN adapter, and what any future + channel will have to satisfy before its data can enter the platform. It performs a real + acquisition — a conformance check that stubbed out the fetch would verify the parts that + were never in doubt. + """ + if not isinstance(adapter, SourceAdapter): + _fail( + f"{type(adapter).__name__} does not satisfy the SourceAdapter protocol; " + f"E5 §4 requires descriptor() and acquire()." + ) + + verify_descriptor(adapter) + + try: + acquired = adapter.acquire(selector) + except (PolicyRejection, ContractViolation): + # A refusal is conformant behaviour — an adapter that aborts rather than returning + # something wrong is doing exactly what E5 §9 requires. Re-raised unchanged so the + # caller sees the real reason rather than a conformance message wrapping it. + raise + + return verify_acquisition(acquired) + + +__all__ = [ + "CONFORMANCE_GATE", + "FORBIDDEN_ARTIFACT_FIELDS", + "SourceAdapter", + "verify_acquisition", + "verify_conformance", + "verify_descriptor", +] diff --git a/contexts/ingest/credentials.py b/contexts/ingest/credentials.py new file mode 100644 index 0000000..81ef2f7 --- /dev/null +++ b/contexts/ingest/credentials.py @@ -0,0 +1,159 @@ +"""The credential boundary — the one thing this context must never leak. + +E5 §13 calls Ingest the highest-sensitivity context: it holds the only secrets in the system. +STD-19 states the rule, and E5 §11(iii) states it as an invariant: + + Secrets exist only within the Ingest context's source adapters. They are never logged, + never persisted to an artifact, and never cross a context boundary. + +A rule of that shape is normally enforced by discipline, and discipline fails silently — the +leak is a log line nobody reads until someone else does. So it is enforced here in three ways +that do not depend on anyone remembering: + + 1. `Credential` cannot be printed. `repr` and `str` return a redaction, so a secret cannot + reach a log through the single most common route: an f-string, a `print`, or a traceback + that includes local variables. + 2. `Credential` cannot be serialised. It has no `to_dict`, and `json.dumps` refuses it + because it is not a JSON type. An artifact cannot silently acquire one. + 3. `assert_credential_free` inspects any object about to cross the boundary and refuses it + if a `Credential` is reachable from it, at any depth. + +Reading the secret requires calling `reveal()` explicitly. That is deliberate: an adapter +genuinely needs the value to authenticate, and making the read a named, greppable call means +`grep -rn 'reveal()' contexts/` enumerates every place a secret is touched — which is a +shorter list than every place a string is used. + +WHAT THIS DOES NOT DO. It does not manage, store, rotate or fetch credentials. There is no +credential store, no provider abstraction and no configuration layer: ADR-0003 forbids a +configurable adapter framework until a second source exists, and ADR-0025 classifies exactly +that as a paid abstraction. Obtaining the secret is the adapter's business (M3/E5/#16). +""" + +from __future__ import annotations + +from dataclasses import dataclass, fields, is_dataclass + +from contexts.ingest.errors import PolicyRejection + +#: The gate named in a `PolicyRejection` raised from here. One string, so a caller matching +#: on it and a test asserting it cannot drift apart. +CREDENTIAL_BOUNDARY = "credential-boundary" + +REDACTED = "" + +#: Field names that would carry a secret in plain form. Used only to catch a raw string that +#: was never wrapped in `Credential` — the wrapped case is caught by type, which is exact. +#: This list is a backstop for the mistake of not using the type at all, and is deliberately +#: about names rather than values: a heuristic over values would have to guess what a secret +#: looks like, and would be both wrong and unfalsifiable. +SECRET_FIELD_NAMES = frozenset({ + "credential", "credentials", "secret", "password", "passwd", "token", + "api_key", "apikey", "access_key", "secret_key", "cookie", "cookies", + "session", "session_id", "auth", "authorization", "bearer", "private_key", +}) + + +@dataclass(frozen=True) +class Credential: + """A secret held inside Ingest and nowhere else. + + Frozen, unprintable, unserialisable. The value is reachable only through `reveal()`. + """ + + #: What this secret is for, e.g. "issdc-pradan-session". Not itself secret — a purpose + #: has to be nameable in a log for an operator to diagnose an authentication failure. + purpose: str + _value: str + + def __post_init__(self) -> None: + if not isinstance(self.purpose, str) or not self.purpose: + raise ValueError("a credential must state its purpose") + if not isinstance(self._value, str) or not self._value: + # An empty secret is almost always an unset environment variable read without + # checking. Failing here turns a silent unauthenticated request into an abort. + raise ValueError(f"credential {self.purpose!r} has no value") + + def reveal(self) -> str: + """Return the secret. The only way to read it, and deliberately greppable.""" + return self._value + + def __repr__(self) -> str: + return f"Credential(purpose={self.purpose!r}, value={REDACTED})" + + def __str__(self) -> str: + return REDACTED + + def __format__(self, spec: str) -> str: + # Without this, `f"{credential:>20}"` would bypass __str__ for some format specs. + return REDACTED + + +def _reachable_credentials(value: object, trail: str, seen: set[int]) -> list[str]: + """Every path within `value` at which a `Credential` or secret-named field is reachable. + + Walks dataclasses, mappings and sequences. Recursion is guarded by object identity so a + cyclic structure terminates rather than exhausting the stack — a leak check that crashes + is a leak check that gets removed. + """ + if id(value) in seen: + return [] + seen.add(id(value)) + + if isinstance(value, Credential): + return [trail or "/"] + + found: list[str] = [] + + if is_dataclass(value) and not isinstance(value, type): + for field in fields(value): + attribute = getattr(value, field.name, None) + if field.name.strip("_").lower() in SECRET_FIELD_NAMES and isinstance( + attribute, str + ): + found.append(f"{trail}/{field.name}") + found += _reachable_credentials(attribute, f"{trail}/{field.name}", seen) + return found + + if isinstance(value, dict): + for key, item in value.items(): + here = f"{trail}/{key}" + if isinstance(key, str) and key.strip("_").lower() in SECRET_FIELD_NAMES: + found.append(here) + found += _reachable_credentials(item, here, seen) + return found + + if isinstance(value, (list, tuple, set, frozenset)): + for index, item in enumerate(value): + found += _reachable_credentials(item, f"{trail}/{index}", seen) + return found + + return found + + +def assert_credential_free(value: object, *, what: str) -> None: + """Refuse to let `value` cross the boundary if a secret is reachable from it. + + Applied to everything Ingest hands outward: the descriptor it publishes, the provenance + it records, the artifact it produces. `what` names the thing being checked so the + rejection says which boundary crossing was refused rather than only that one was. + + Raises `PolicyRejection` naming `credential-boundary` (TIS §0.2 — a gate refusing an + operation aborts with the gate's identity). + """ + leaks = _reachable_credentials(value, "", set()) + if leaks: + raise PolicyRejection( + CREDENTIAL_BOUNDARY, + f"{what} would carry a credential across the context boundary at " + f"{sorted(leaks)}. STD-19 confines secrets to Ingest; E5 §13 forbids " + f"persisting one to an artifact or logging one.", + ) + + +__all__ = [ + "CREDENTIAL_BOUNDARY", + "Credential", + "REDACTED", + "SECRET_FIELD_NAMES", + "assert_credential_free", +] diff --git a/contexts/ingest/descriptor.py b/contexts/ingest/descriptor.py new file mode 100644 index 0000000..d829010 --- /dev/null +++ b/contexts/ingest/descriptor.py @@ -0,0 +1,185 @@ +"""What a Source publishes about itself. + +E5 §4 fixes the shape: `SourceDescriptor {source_id, authority, latency_class, granularity}`. +ADR-0003 says why a Source is a separate concept from an Instrument — a Source is a *channel* +(authority, latency class, granularity, retrieval descriptor), an Instrument is a *sensor*. +Conflating a channel with a sensor is a modelling error independent of how many channels +exist, which is what makes the separation a free seam under ADR-0025 rather than an +abstraction bought on speculation. + +WHY `latency_class` IS VALIDATED RATHER THAN FREE TEXT +------------------------------------------------------ +ADR-0001 lists as a binding non-goal: *"Real-time Aditya-L1 services while the archive remains +~33 days latent."* That non-goal is only checkable if a source's latency is a quantity rather +than a phrase. `~33d` and `about a month` and `slow` are all true descriptions and only one of +them can be compared to anything. + +So the field takes a duration — an optional `~` for "approximately", a count, and a unit — and +exposes it in seconds. E5 §18 requires the ISSDC adapter to be registered with latency class +`~33d`, which is a claim a test can now check rather than read. + +No enumeration of latency *categories* is defined here. "realtime", "near-real-time" and +"archival" are words this repository has not agreed on, and inventing a vocabulary for a +second source that does not exist is the paid abstraction ADR-0003 explicitly refuses to +authorise. + +WHAT THE DESCRIPTOR MUST NOT CARRY +---------------------------------- +Credentials. ADR-0003 lists credentials among a Source's properties, and STD-19 and E5 §13 +confine them to the adapter. Both hold because the descriptor records *whether* +authentication is needed, never the secret — the same division `manifest.schema.json` makes +for a Tier 0 retrieval descriptor (M2/E4/#14). `requires_credentials` is the whole of what +crosses. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from domain.errors import ContractViolation +from domain.values import Identifier + +#: `~33d` — optional approximation marker, a count, and a unit. Seconds through days; a +#: latency measured in weeks is expressed in days, because two units for one duration is how +#: `~1w` and `~7d` come to look like different classes. +LATENCY_PATTERN = re.compile(r"^(~?)(\d+)([smhd])$") + +_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400} + + +@dataclass(frozen=True) +class LatencyClass: + """How far behind the phenomenon a Source's data arrives. + + Parsed, not merely stored, so that ADR-0001's non-goal about real-time services is a + comparison rather than a reading. + """ + + text: str + + def __post_init__(self) -> None: + if not isinstance(self.text, str) or not self.text: + raise ContractViolation("/latency_class", "latency class must be a non-empty string") + if LATENCY_PATTERN.match(self.text) is None: + raise ContractViolation( + "/latency_class", + f"latency class {self.text!r} is not a duration such as '~33d' or '5m'. " + f"A latency that cannot be compared cannot be checked against ADR-0001's " + f"non-goal on real-time services.", + ) + + @property + def seconds(self) -> int: + match = LATENCY_PATTERN.match(self.text) + assert match is not None # guaranteed by __post_init__ + return int(match.group(2)) * _UNIT_SECONDS[match.group(3)] + + @property + def is_approximate(self) -> bool: + """True where the value is prefixed `~`. + + Recorded because ~33 days is a characterisation of a manual archive process, not a + measurement of one, and a reader should be able to tell which they are looking at. + """ + return self.text.startswith("~") + + def __str__(self) -> str: + return self.text + + +@dataclass(frozen=True) +class RetrievalDescriptor: + """How to re-acquire Tier 0 bytes this repository does not hold. + + Mirrors `manifest.schema.json#/properties/retrieval` field for field, deliberately: the + descriptor an adapter publishes and the descriptor a manifest records are the same fact, + and two shapes for one fact is how they come to disagree (ADR-0023, STD-23). + """ + + provider: str + locator: str + requires_credentials: bool = False + + def __post_init__(self) -> None: + for name in ("provider", "locator"): + value = getattr(self, name) + if not isinstance(value, str) or not value: + raise ContractViolation(f"/{name}", f"{name} must be a non-empty string") + if not isinstance(self.requires_credentials, bool): + raise ContractViolation( + "/requires_credentials", + f"requires_credentials must be a bool, got " + f"{type(self.requires_credentials).__name__}", + ) + + def to_dict(self) -> dict[str, object]: + """The shape `manifest.schema.json` expects under `retrieval`.""" + return { + "provider": self.provider, + "locator": self.locator, + "requires_credentials": self.requires_credentials, + } + + +@dataclass(frozen=True) +class SourceDescriptor: + """What a Source publishes about itself (E5 §4). + + Four fields, exactly as the TIS names them, plus the retrieval descriptor ADR-0003 lists + among a Source's properties and ADR-0023 requires for Tier 0. + """ + + source_id: Identifier + authority: str + latency_class: LatencyClass + granularity: Identifier + retrieval: RetrievalDescriptor + + def __post_init__(self) -> None: + if not isinstance(self.source_id, Identifier): + raise ContractViolation( + "/source_id", + f"source_id must be an Identifier, got {type(self.source_id).__name__}", + ) + if not isinstance(self.authority, str) or not self.authority: + raise ContractViolation( + "/authority", + "authority must be named — who publishes this channel is part of what the " + "data means (ADR-0003)", + ) + if not isinstance(self.latency_class, LatencyClass): + raise ContractViolation( + "/latency_class", + f"latency_class must be a LatencyClass, got " + f"{type(self.latency_class).__name__}", + ) + if not isinstance(self.granularity, Identifier): + raise ContractViolation( + "/granularity", + f"granularity must be an Identifier, got {type(self.granularity).__name__}. " + f"E5 §18 registers the ISSDC channel as 'daily-archive'.", + ) + if not isinstance(self.retrieval, RetrievalDescriptor): + raise ContractViolation( + "/retrieval", + f"retrieval must be a RetrievalDescriptor, got " + f"{type(self.retrieval).__name__}", + ) + + def to_dict(self) -> dict[str, object]: + return { + "source_id": str(self.source_id), + "authority": self.authority, + "latency_class": str(self.latency_class), + "granularity": str(self.granularity), + "retrieval": self.retrieval.to_dict(), + } + + +__all__ = [ + "LATENCY_PATTERN", + "LatencyClass", + "RetrievalDescriptor", + "SourceDescriptor", +] diff --git a/contexts/ingest/errors.py b/contexts/ingest/errors.py new file mode 100644 index 0000000..9d03a2a --- /dev/null +++ b/contexts/ingest/errors.py @@ -0,0 +1,65 @@ +"""Ingest's members of the failure taxonomy. + +TIS §0.2 fixes five failure classes for the whole system and permits no others and no +catch-all. E5 §8 names the four Ingest raises: + + UnavailableResource the portal or feed is unreachable + IntegrityFailure digest mismatch — NO PARTIAL INGEST + ContractViolation structure violates the spec — fail loud, never coerce + PolicyRejection an adapter attempted to redistribute Tier 0 bytes + +Two of those already have owners and are re-exported rather than redefined. +`IntegrityFailure` belongs to the provenance kernel, which is the only thing that computes a +digest and so the only thing that can find one wrong (ADR-0005). `ContractViolation` belongs +to the domain, which is where the shape of an Observation is decided (ADR-0019). Redefining +either here would give one concept two identities, and a caller catching the wrong one would +see an abort pass through as an unhandled error. + +The two defined here are the ones with no other home: an unreachable archive and a refused +operation are facts about acquisition, not about digests or shapes. +""" + +from __future__ import annotations + +from domain.errors import ContractViolation +from kernel.provenance.errors import IntegrityFailure + + +class IngestError(Exception): + """Root of the errors Ingest itself defines. + + Exists so a caller can catch them without a bare `except`, which TIS §0.2 forbids + universally. + """ + + +class UnavailableResource(IngestError): + """An external source could not be reached (TIS §0.2, E5 §8). + + Retry policy is adapter-local (E5 §15). Exhaustion raises this, the run aborts, and + nothing partial is written — E5 §9 states that `aborted` leaves no observations. + """ + + +class PolicyRejection(IngestError): + """A gate refused the operation, and the gate is named (TIS §0.2). + + In this context the gate is almost always Tier 0 non-redistribution (ADR-0023, STD-23, + E5 §11(iv)) or the credential boundary (STD-19, E5 §13). The gate's identity is carried + rather than described, because a refusal nobody can attribute is a refusal nobody can + act on. + """ + + def __init__(self, gate: str, message: str) -> None: + self.gate = gate + self.message = message + super().__init__(f"{gate}: {message}") + + +__all__ = [ + "ContractViolation", + "IngestError", + "IntegrityFailure", + "PolicyRejection", + "UnavailableResource", +] diff --git a/contexts/ingest/tests/__init__.py b/contexts/ingest/tests/__init__.py new file mode 100644 index 0000000..fc6e345 --- /dev/null +++ b/contexts/ingest/tests/__init__.py @@ -0,0 +1,6 @@ +"""Marks the ingest test directory as a package. + +Bare `pytest` and `python -m pytest` differ in `sys.path`: the latter inserts the current +directory, the former does not. Without this file the suite passes locally and fails in CI +(IMPL-007, M2/E3). The marker makes both invocations resolve `contexts.ingest.*` the same way. +""" diff --git a/contexts/ingest/tests/adapters.py b/contexts/ingest/tests/adapters.py new file mode 100644 index 0000000..689bbcb --- /dev/null +++ b/contexts/ingest/tests/adapters.py @@ -0,0 +1,259 @@ +"""In-test adapters: one that conforms, and one per rule that does not. + +NOT SOURCE ADAPTERS, AND THE DISTINCTION IS THE ISSUE BOUNDARY +--------------------------------------------------------------- +M3/E5/#16 delivers the ISSDC-PRADAN adapter: it reaches a real archive, holds a real session, +and retrieves real bytes. Nothing here does any of that. These are the *subjects* of the +conformance check — the smallest objects that satisfy or violate the contract, so that +`verify_conformance` can be watched accepting and refusing. + +A conformance checker tested only against a conforming adapter is a checker nobody has seen +reject anything, and this repository has already shipped one gate with that defect. So each +violation below is a separate class that breaks exactly one rule and keeps the others, which +is what makes a failing assertion attributable to the rule it names. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from contexts.ingest.acquisition import Acquisition, AcquisitionProvenance, RawArtifact +from contexts.ingest.credentials import Credential +from contexts.ingest.descriptor import ( + LatencyClass, + RetrievalDescriptor, + SourceDescriptor, +) +from domain.values import Digest, Identifier, RunId, Timestamp + +RUN = RunId("01ARZ3NDEKTSV4RRFFQ69G5FAV") +DIGEST = Digest("a" * 64) +OTHER_DIGEST = Digest("b" * 64) + + +def descriptor(**overrides) -> SourceDescriptor: + """The ISSDC channel as E5 §18 describes it: latency `~33d`, granularity `daily-archive`.""" + fields = { + "source_id": Identifier("issdc-pradan"), + "authority": "ISSDC, Indian Space Research Organisation", + "latency_class": LatencyClass("~33d"), + "granularity": Identifier("daily-archive"), + "retrieval": RetrievalDescriptor( + provider="ISSDC PRADAN", + locator="aditya-l1/solexs/l1/2024-03-01", + requires_credentials=True, + ), + } + fields.update(overrides) + return SourceDescriptor(**fields) + + +def acquisition(**overrides) -> Acquisition: + fields = { + "artifact": RawArtifact( + digest=DIGEST, + size_bytes=4096, + retrieval=descriptor().retrieval, + cache_path=Path("/tmp/cache/aaaa.fits"), + ), + "provenance": AcquisitionProvenance( + run_id=RUN, + source_id=Identifier("issdc-pradan"), + artifact_digest=DIGEST, + ingest_time=Timestamp("2024-04-03T12:00:00Z"), + instruments=(Identifier("solexs"),), + ), + } + fields.update(overrides) + return Acquisition(**fields) + + +class ConformingAdapter: + """Satisfies every rule. The positive case, without which the checker could reject all.""" + + def descriptor(self) -> SourceDescriptor: + return descriptor() + + def acquire(self, selector: str) -> Acquisition: + return acquisition() + + +class AdapterWithNoDescriptor: + """Missing half the interface E5 §4 requires.""" + + def acquire(self, selector: str) -> Acquisition: + return acquisition() + + +class AdapterWithNoAcquire: + def descriptor(self) -> SourceDescriptor: + return descriptor() + + +class AdapterReturningTheWrongDescriptorType: + """Describes itself in its own vocabulary, so nothing can compare it to another channel.""" + + def descriptor(self) -> dict: + return {"source_id": "issdc-pradan", "latency": "about a month"} + + def acquire(self, selector: str) -> Acquisition: + return acquisition() + + +class AdapterWithAVaryingDescriptor: + """Declares a different latency every time it is asked, which is to declare none.""" + + def __init__(self) -> None: + self._calls = 0 + + def descriptor(self) -> SourceDescriptor: + self._calls += 1 + return descriptor(latency_class=LatencyClass(f"~{30 + self._calls}d")) + + def acquire(self, selector: str) -> Acquisition: + return acquisition() + + +@dataclass(frozen=True) +class LeakyDescriptorHolder: + """A descriptor-shaped object that also carries the session cookie.""" + + source_id: Identifier + authority: str + session: Credential + + +class AdapterLeakingACredentialInItsDescriptor: + """STD-19, E5 §13: the descriptor records *how* to acquire, never the secret.""" + + def descriptor(self) -> SourceDescriptor: + return LeakyDescriptorHolder( # type: ignore[return-value] + source_id=Identifier("issdc-pradan"), + authority="ISSDC", + session=Credential("issdc-session", "PRADAN-SESSION-abc123"), + ) + + def acquire(self, selector: str) -> Acquisition: + return acquisition() + + +@dataclass(frozen=True) +class ProvenanceCarryingASecret: + """Provenance-shaped, and would persist a secret to an artifact (E5 §13).""" + + run_id: RunId + source_id: Identifier + artifact_digest: Digest + ingest_time: Timestamp + credential: Credential + + +class AdapterPersistingACredentialInProvenance: + def descriptor(self) -> SourceDescriptor: + return descriptor() + + def acquire(self, selector: str) -> Acquisition: + provenance = ProvenanceCarryingASecret( + run_id=RUN, + source_id=Identifier("issdc-pradan"), + artifact_digest=DIGEST, + ingest_time=Timestamp("2024-04-03T12:00:00Z"), + credential=Credential("issdc-session", "PRADAN-SESSION-abc123"), + ) + return _unchecked_acquisition( + RawArtifact(DIGEST, 4096, descriptor().retrieval), provenance + ) + + +@dataclass(frozen=True) +class ArtifactHoldingBytes: + """Would materialise a whole-day archive and could redistribute Tier 0 (ADR-0023, E5 §12).""" + + digest: Digest + size_bytes: int + retrieval: RetrievalDescriptor + cache_path: Path | None + bytes: bytes + + +class AdapterHoldingRawBytes: + def descriptor(self) -> SourceDescriptor: + return descriptor() + + def acquire(self, selector: str) -> Acquisition: + artifact = ArtifactHoldingBytes( + digest=DIGEST, + size_bytes=4, + retrieval=descriptor().retrieval, + cache_path=None, + bytes=b"\x00\x01\x02\x03", + ) + return _unchecked_acquisition(artifact, acquisition().provenance) + + +class AdapterReturningTheWrongAcquisitionType: + def descriptor(self) -> SourceDescriptor: + return descriptor() + + def acquire(self, selector: str) -> RawArtifact: + return acquisition().artifact + + +@dataclass(frozen=True) +class ProvenanceWithoutAnIngestTime: + run_id: RunId + source_id: Identifier + artifact_digest: Digest + ingest_time: None + + +class AdapterOmittingTheIngestTime: + """E5 §11(i): every newly ingested Observation carries valid_time AND ingest_time.""" + + def descriptor(self) -> SourceDescriptor: + return descriptor() + + def acquire(self, selector: str) -> Acquisition: + provenance = ProvenanceWithoutAnIngestTime( + run_id=RUN, + source_id=Identifier("issdc-pradan"), + artifact_digest=DIGEST, + ingest_time=None, + ) + return _unchecked_acquisition( + RawArtifact(DIGEST, 4096, descriptor().retrieval), provenance + ) + + +class AdapterMisattributingItsArtifact: + """Provenance pointing at bytes other than the artifact it accompanies.""" + + def descriptor(self) -> SourceDescriptor: + return descriptor() + + def acquire(self, selector: str) -> Acquisition: + return _unchecked_acquisition( + RawArtifact(DIGEST, 4096, descriptor().retrieval), + AcquisitionProvenance( + run_id=RUN, + source_id=Identifier("issdc-pradan"), + artifact_digest=OTHER_DIGEST, + ingest_time=Timestamp("2024-04-03T12:00:00Z"), + ), + ) + + +def _unchecked_acquisition(artifact: object, provenance: object) -> Acquisition: + """Build an `Acquisition` around parts its constructor would refuse. + + `Acquisition.__post_init__` rejects most of these violations outright, which is the + correct behaviour and is tested directly. But `verify_conformance` must also reject them, + because an adapter written before this contract existed — or one that builds its result + some other way — can hand the checker an object the constructor never saw. Bypassing + validation here is what lets the checker be tested rather than the constructor. + """ + instance = object.__new__(Acquisition) + object.__setattr__(instance, "artifact", artifact) + object.__setattr__(instance, "provenance", provenance) + return instance diff --git a/contexts/ingest/tests/test_contract.py b/contexts/ingest/tests/test_contract.py new file mode 100644 index 0000000..0455eae --- /dev/null +++ b/contexts/ingest/tests/test_contract.py @@ -0,0 +1,401 @@ +"""Contract conformance — the unit test §15 row 14+1 names. + +`| 15 | 15 | E5 | 200 | M | 11 | contract conformance | — | credential-boundary |` + +Every rule in the contract is exercised from both sides: an adapter that satisfies it, and an +adapter that breaks exactly that rule and keeps the others. The second half is the half that +matters — a checker seen only accepting is a checker nobody has watched reject anything. +""" + +from __future__ import annotations + +from dataclasses import FrozenInstanceError, fields + +import pytest + +from contexts.ingest import boundary +from contexts.ingest.acquisition import Acquisition, AcquisitionProvenance, RawArtifact +from contexts.ingest.contract import ( + CONFORMANCE_GATE, + FORBIDDEN_ARTIFACT_FIELDS, + SourceAdapter, + verify_conformance, + verify_descriptor, +) +from contexts.ingest.credentials import Credential +from contexts.ingest.descriptor import LatencyClass, RetrievalDescriptor, SourceDescriptor +from contexts.ingest.errors import PolicyRejection, UnavailableResource +from contexts.ingest.tests import adapters +from domain.errors import ContractViolation +from domain.values import Digest, Identifier, Timestamp + +SELECTOR = "2024-03-01" + + +# ═══════════════════════════════════════════ the contract accepts what conforms + + +def test_a_conforming_adapter_passes_every_check(): + """The positive case. Without it, a checker that rejected everything would look correct.""" + acquired = verify_conformance(adapters.ConformingAdapter(), selector=SELECTOR) + assert isinstance(acquired, Acquisition) + assert acquired.artifact.digest == acquired.provenance.artifact_digest + + +def test_the_protocol_recognises_a_structurally_complete_adapter(): + assert isinstance(adapters.ConformingAdapter(), SourceAdapter) + + +def test_the_descriptor_check_returns_what_it_verified(): + """E5 §18's acceptance criterion is a question about the descriptor alone. + + Separating it means an adapter can be asked what it is without being asked to fetch, + which is what lets #16 assert `~33d` / `daily-archive` without touching the archive. + """ + described = verify_descriptor(adapters.ConformingAdapter()) + assert str(described.source_id) == "issdc-pradan" + assert str(described.latency_class) == "~33d" + assert str(described.granularity) == "daily-archive" + + +# ═══════════════════════════════════════════ E5 §4 — the interface itself + + +@pytest.mark.parametrize( + "adapter", + [adapters.AdapterWithNoDescriptor(), adapters.AdapterWithNoAcquire()], + ids=["no-descriptor", "no-acquire"], +) +def test_an_incomplete_adapter_is_refused(adapter): + with pytest.raises(PolicyRejection) as caught: + verify_conformance(adapter, selector=SELECTOR) + assert caught.value.gate == CONFORMANCE_GATE + + +def test_a_descriptor_of_the_wrong_type_is_refused(): + """A channel describing itself in its own vocabulary cannot be compared to another.""" + with pytest.raises(PolicyRejection) as caught: + verify_conformance(adapters.AdapterReturningTheWrongDescriptorType(), selector=SELECTOR) + assert "SourceDescriptor" in caught.value.message + + +def test_a_descriptor_that_varies_between_calls_is_refused(): + """A channel whose declared latency changes has declared no latency.""" + with pytest.raises(PolicyRejection) as caught: + verify_conformance(adapters.AdapterWithAVaryingDescriptor(), selector=SELECTOR) + assert "not constant" in caught.value.message + + +def test_an_acquisition_of_the_wrong_type_is_refused(): + with pytest.raises(PolicyRejection) as caught: + verify_conformance(adapters.AdapterReturningTheWrongAcquisitionType(), selector=SELECTOR) + assert "Acquisition" in caught.value.message + + +# ═══════════════════════════════════════════ STD-19 / E5 §13 — credentials never leave + + +def test_an_adapter_leaking_a_credential_in_its_descriptor_is_refused(): + """The descriptor records *how* to acquire, never the secret. + + Refused by the CONFORMANCE gate rather than the credential gate, and that ordering is + the stronger result: a real `SourceDescriptor` has no field a credential could occupy + (see the test below), so the only way to publish one is to return something that is not + a SourceDescriptor at all — which the type check catches first. The runtime credential + scan is the backstop for the duck-typed case, exercised on provenance below. + """ + with pytest.raises(PolicyRejection) as caught: + verify_conformance( + adapters.AdapterLeakingACredentialInItsDescriptor(), selector=SELECTOR + ) + assert caught.value.gate == CONFORMANCE_GATE + assert "SourceDescriptor" in caught.value.message + + +def test_the_descriptor_type_has_no_field_a_credential_could_occupy(): + """STD-19 made structural rather than checked. + + Every `SourceDescriptor` field is either a typed domain value or a plain string that the + constructor requires — there is no `object`-typed field, no free-form mapping, and no + optional extras dict. A secret cannot be placed in a conforming descriptor at all, which + is a stronger guarantee than scanning for one after the fact. + """ + typed = {f.name: f.type for f in fields(SourceDescriptor)} + assert set(typed) == { + "source_id", "authority", "latency_class", "granularity", "retrieval" + } + for name, annotation in typed.items(): + assert "dict" not in str(annotation).lower(), f"{name} could carry arbitrary keys" + assert "Any" not in str(annotation), f"{name} is untyped" + + # And a credential offered to any of them is refused by the constructor. + secret = Credential("issdc-session", "PRADAN-SESSION-abc123") + for name in typed: + with pytest.raises(ContractViolation): + adapters.descriptor(**{name: secret}) + + +def test_an_adapter_persisting_a_credential_in_provenance_is_refused(): + """E5 §13: never persisted to an artifact. Provenance is persisted through E3.""" + with pytest.raises(PolicyRejection) as caught: + verify_conformance( + adapters.AdapterPersistingACredentialInProvenance(), selector=SELECTOR + ) + assert caught.value.gate == "credential-boundary" + + +def test_the_conforming_descriptor_records_that_credentials_are_needed_without_carrying_one(): + """The whole of what may cross: a boolean, mirroring the Tier 0 retrieval descriptor.""" + described = verify_descriptor(adapters.ConformingAdapter()) + assert described.retrieval.requires_credentials is True + assert "credential" not in described.to_dict() + assert set(described.retrieval.to_dict()) == { + "provider", "locator", "requires_credentials" + } + + +# ═══════════════════════════════════════════ ADR-0023 / E5 §12 — Tier 0 is referenced + + +def test_an_artifact_holding_raw_bytes_is_refused(): + """ADR-0023, STD-23, E5 §11(iv), E5 §12 all land on this one refusal.""" + with pytest.raises(PolicyRejection) as caught: + verify_conformance(adapters.AdapterHoldingRawBytes(), selector=SELECTOR) + assert caught.value.gate == CONFORMANCE_GATE + assert "referenced, never" in caught.value.message + + +def test_a_raw_artifact_cannot_carry_bytes(): + """The type has no such field, and this asserts it never gains one. + + An artifact that *can* hold the bytes will hold them, and a whole-day archive would be + materialised without anyone deciding to (E5 §12). + """ + names = {f.name for f in fields(RawArtifact)} + assert not (names & FORBIDDEN_ARTIFACT_FIELDS), ( + f"RawArtifact gained {sorted(names & FORBIDDEN_ARTIFACT_FIELDS)}" + ) + assert names == {"digest", "size_bytes", "retrieval", "cache_path"} + + +def test_an_artifact_requires_a_way_to_re_acquire_it(): + """ADR-0023: any stage must be able to re-acquire from the descriptor.""" + with pytest.raises(ContractViolation) as caught: + RawArtifact(digest=Digest("a" * 64), size_bytes=1, retrieval=None) + assert caught.value.pointer == "/retrieval" + + +def test_an_evicted_cache_is_not_an_error(): + """Local copies are evictable caches, never the system of record (ADR-0023).""" + artifact = RawArtifact( + digest=Digest("a" * 64), size_bytes=1, retrieval=adapters.descriptor().retrieval + ) + assert artifact.cache_path is None + assert artifact.is_cached is False + assert adapters.acquisition().artifact.is_cached is True + + +# ═══════════════════════════════════════════ ADR-0004 / E5 §11 — both times + + +def test_an_adapter_omitting_the_ingest_time_is_refused(): + """E5 §11(i): every newly ingested Observation carries valid_time AND ingest_time.""" + with pytest.raises(PolicyRejection) as caught: + verify_conformance(adapters.AdapterOmittingTheIngestTime(), selector=SELECTOR) + assert "ingest_time" in caught.value.message + + +def test_acquisition_provenance_requires_a_non_null_ingest_time(): + """ADR-0022's null is for historical rows, which have no acquisition to describe.""" + with pytest.raises(ContractViolation) as caught: + AcquisitionProvenance( + run_id=adapters.RUN, + source_id=Identifier("issdc-pradan"), + artifact_digest=Digest("a" * 64), + ingest_time=None, + ) + assert caught.value.pointer == "/ingest_time" + + +def test_there_is_no_way_to_stamp_a_historical_row(): + """E5 §11(ii): no code path writes a non-null ingest_time for historical data. + + The boundary module exports a clock read and a monotonic reading, and nothing that takes + a row, a dataset or a date and returns a time for it. Backfilling would require calling + `stamp()` and attaching the result to old rows — which is a deliberate act that + `domain.invariants.ingest_time_is_not_backfilled` detects. + """ + exported = set(boundary.__all__) + assert exported == {"monotonic", "stamp", "stamp_from_epoch"} + for forbidden in ("backfill", "infer", "default_ingest_time", "for_row", "impute"): + assert not hasattr(boundary, forbidden) + + +def test_a_misattributed_artifact_is_refused_by_the_checker(): + """Provenance pointing at other bytes is worse than none: it looks like attribution.""" + with pytest.raises(PolicyRejection) as caught: + verify_conformance(adapters.AdapterMisattributingItsArtifact(), selector=SELECTOR) + assert caught.value.gate == CONFORMANCE_GATE + + +def test_acquisition_refuses_mismatched_parts_at_construction(): + """The constructor enforces it too, so the checker is a second line rather than the only.""" + with pytest.raises(ContractViolation) as caught: + Acquisition( + artifact=adapters.acquisition().artifact, + provenance=AcquisitionProvenance( + run_id=adapters.RUN, + source_id=Identifier("issdc-pradan"), + artifact_digest=Digest("b" * 64), + ingest_time=Timestamp("2024-04-03T12:00:00Z"), + ), + ) + assert "looks like attribution" in caught.value.message + + +# ═══════════════════════════════════════════ E5 §9 — a refusal is conformant + + +def test_an_adapter_that_aborts_is_not_a_conformance_failure(): + """E5 §9: `aborted` leaves no observations. Aborting is what an adapter should do. + + The original error is re-raised unchanged rather than wrapped, so the caller sees that + the archive was unreachable rather than that conformance failed. + """ + + class UnreachableAdapter: + def descriptor(self): + return adapters.descriptor() + + def acquire(self, selector): + raise UnavailableResource("PRADAN portal did not respond after 3 attempts") + + with pytest.raises(UnavailableResource): + verify_conformance(UnreachableAdapter(), selector=SELECTOR) + + +# ═══════════════════════════════════════════ no registry, no dispatch, no framework + + +def test_the_contract_registers_nothing(): + """ADR-0003 does not authorise a source-plugin registry, dispatch layer or framework. + + ADR-0025 classifies all three as paid abstractions, forbidden until a second source + exists. A Protocol declares what a caller may rely on and constructs nothing; this + asserts the module never grows the thing the ADR refuses. + """ + import contexts.ingest.contract as module + + for forbidden in ("register", "REGISTRY", "ADAPTERS", "get_adapter", "load_adapter", + "dispatch", "AdapterFactory", "SourceAdapterBase"): + assert not hasattr(module, forbidden), f"contract.py grew {forbidden}" + + +def test_the_protocol_is_not_a_base_class(): + """A conforming adapter inherits nothing, which is what makes the seam free (ADR-0025). + + `isinstance` succeeds by structure, not by ancestry. An adapter that had to inherit + something in order to be recognised would be an adapter the framework owns. + """ + assert SourceAdapter not in adapters.ConformingAdapter.__mro__ + assert adapters.ConformingAdapter.__mro__ == (adapters.ConformingAdapter, object) + assert isinstance(adapters.ConformingAdapter(), SourceAdapter) + + +# ═══════════════════════════════════════════ the descriptor's own rules + + +def test_the_latency_class_is_a_comparable_quantity(): + """ADR-0001's non-goal on real-time services is only checkable if latency is a number.""" + assert LatencyClass("~33d").seconds == 33 * 86400 + assert LatencyClass("5m").seconds == 300 + assert LatencyClass("~33d").is_approximate is True + assert LatencyClass("5m").is_approximate is False + + +@pytest.mark.parametrize( + "bad", ["about a month", "slow", "33", "d", "", "33 days", "~33w", "-5m", None] +) +def test_a_latency_that_cannot_be_compared_is_refused(bad): + with pytest.raises(ContractViolation): + LatencyClass(bad) + + +def test_the_issdc_channel_is_far_outside_anything_real_time(): + """The measurement behind ADR-0001's non-goal, stated as a test rather than as prose.""" + assert LatencyClass("~33d").seconds == 2_851_200 + + +@pytest.mark.parametrize( + "field, bad", + [ + ("source_id", "issdc-pradan"), + ("authority", ""), + ("latency_class", "~33d"), + ("granularity", "daily-archive"), + ("retrieval", {"provider": "x", "locator": "y"}), + ], +) +def test_the_descriptor_refuses_raw_values_where_a_typed_one_is_required(field, bad): + with pytest.raises(ContractViolation) as caught: + adapters.descriptor(**{field: bad}) + assert caught.value.pointer == f"/{field}" + + +def test_the_descriptor_serialises_to_the_shape_the_manifest_expects(): + """The retrieval descriptor an adapter publishes and the one a manifest records are the + same fact; two shapes for one fact is how they come to disagree (M2/E4/#14).""" + assert adapters.descriptor().retrieval.to_dict() == { + "provider": "ISSDC PRADAN", + "locator": "aditya-l1/solexs/l1/2024-03-01", + "requires_credentials": True, + } + + +@pytest.mark.parametrize("cls", [SourceDescriptor, RetrievalDescriptor, RawArtifact, + AcquisitionProvenance, Acquisition]) +def test_every_contract_type_is_frozen(cls): + """A descriptor that can be edited after publication has not been published.""" + instance = { + SourceDescriptor: adapters.descriptor(), + RetrievalDescriptor: adapters.descriptor().retrieval, + RawArtifact: adapters.acquisition().artifact, + AcquisitionProvenance: adapters.acquisition().provenance, + Acquisition: adapters.acquisition(), + }[cls] + first = fields(instance)[0].name + with pytest.raises(FrozenInstanceError): + setattr(instance, first, None) + + +def test_acquisition_provenance_logs_exactly_what_e5_permits(): + """E5 §14: log source_id, instrument_id, artifact_digest, counts. Never cookies or URLs.""" + record = adapters.acquisition().provenance.to_dict() + assert set(record) == { + "run_id", "source_id", "artifact_digest", "ingest_time", "instruments" + } + for forbidden in ("cookie", "session", "url", "token", "bytes", "credential"): + assert forbidden not in record + + +def test_the_boundary_produces_a_utc_timestamp_the_domain_accepts(): + """The one sanctioned clock read (ADR-0004), returning a value the pure domain validates.""" + stamped = boundary.stamp() + assert isinstance(stamped, Timestamp) + assert stamped.is_utc + assert str(stamped).endswith("Z") + + +def test_the_epoch_stamp_and_the_clock_stamp_agree_on_form(): + fixed = boundary.stamp_from_epoch(1_709_251_200) + assert str(fixed) == "2024-03-01T00:00:00Z" + assert fixed.is_utc + + +def test_monotonic_cannot_become_a_timestamp(): + """It has no epoch, so it can measure elapsed time and cannot record when.""" + first = boundary.monotonic() + assert isinstance(first, float) + assert boundary.monotonic() >= first + with pytest.raises(ContractViolation): + Timestamp(str(first)) diff --git a/tests/integration/test_ingest_contract_with_platform.py b/tests/integration/test_ingest_contract_with_platform.py new file mode 100644 index 0000000..5ae730b --- /dev/null +++ b/tests/integration/test_ingest_contract_with_platform.py @@ -0,0 +1,386 @@ +"""A complete acquisition path, across every component the platform has. + +This is the first test in the repository that follows real scientific data from an +acquisition channel to a contract-valid, provenance-linked Observation. Everything before M3 +built the vocabulary and the guarantees; this exercises the path they exist for. + + #10 kernel/provenance digests the raw bytes and chains the derivation + #11 contracts/ says what an Observation and a manifest must look like + #12 domain/ builds the Observation, and refuses a fabricated ingest_time + #13 context import rules say that Ingest may do all of this and may not reach elsewhere + #14 manifest records where the Tier 0 bytes stay, since they are not kept here + #15 ingest contract the channel, the boundary stamp, and the credential confinement + +Two paths are followed, and the difference between them is the whole of ADR-0022: + + a NEW acquisition both times present, the ingest one stamped at the boundary + a HISTORICAL row valid_time only; ingest_time null, meaning "unknown — predates + bitemporal capture", with no acquisition to describe and nothing + anywhere that would fill it in + +SCOPE. No source adapter, no instrument parser, no registry. The adapter here is the in-test +conforming one from `contexts/ingest/tests/adapters.py`; the ISSDC-PRADAN adapter is M3/E5/#16 +and the SoLEXS and HEL1OS parsers are #17 and #18. The Observation is constructed directly +from bytes the test wrote, because parsing a FITS product is precisely what #17 owns. +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import jsonschema +import pytest +from referencing import Registry, Resource + +from contexts.ingest import ( + Acquisition, + AcquisitionProvenance, + RawArtifact, + RetrievalDescriptor, + assert_credential_free, + verify_conformance, +) +from contexts.ingest import boundary +from contexts.ingest.descriptor import LatencyClass, SourceDescriptor +from domain.entities import Observation +from domain.invariants import ( + ingest_time_is_not_backfilled, + observation_is_admissible_under, + observation_is_wellformed, +) +from domain.values import Digest, Identifier, RunId, Timestamp +from kernel.provenance import Digest as KernelDigest, ProvenanceStore, begin_run + +REPO_ROOT = Path(__file__).resolve().parents[2] +CONTRACTS = REPO_ROOT / "contracts" +GATE_PATH = REPO_ROOT / "tools" / "gates" / "imports.py" + +#: One minute of SoLEXS-shaped light-curve data. Not a FITS product — parsing one is #17's +#: work — but real bytes, digested by the real kernel, so the digests below are genuine. +RAW_FRAME = b"# solexs l1 lightcurve\n2024-03-01T00:00:00Z,12.5\n2024-03-01T00:01:00Z,\n" + + +def registry() -> Registry: + built = Registry() + for path in sorted(CONTRACTS.glob("*.schema.json")): + if path.name.startswith("._"): + continue + built = Resource.from_contents(json.loads(path.read_text())) @ built + return built + + +def validator_for(name: str): + schema = json.loads((CONTRACTS / f"{name}.schema.json").read_text()) + return jsonschema.Draft202012Validator(schema, registry=registry()) + + +class IssdcShapedAdapter: + """A channel with the ISSDC channel's declared shape, acquiring from a local file. + + It is not the ISSDC adapter (#16): it reaches no portal and holds no session. What it does + reproduce faithfully is the *contract* — the descriptor E5 §18 specifies, a digest minted + by the kernel, and an ingest_time stamped at the boundary. + """ + + def __init__(self, store: ProvenanceStore, source_file: Path) -> None: + self._store = store + self._file = source_file + + def descriptor(self) -> SourceDescriptor: + return SourceDescriptor( + source_id=Identifier("issdc-pradan"), + authority="ISSDC, Indian Space Research Organisation", + latency_class=LatencyClass("~33d"), + granularity=Identifier("daily-archive"), + retrieval=RetrievalDescriptor( + provider="ISSDC PRADAN", + locator="aditya-l1/solexs/l1/2024-03-01", + requires_credentials=True, + ), + ) + + def acquire(self, selector: str) -> Acquisition: + run = begin_run(context="ingest", event="acquire") + + # The kernel mints the digest. Ingest could not: ADR-0005 reserves minting to it. + artifact = self._store.put_file(self._file) + + # The one sanctioned clock read in the system (ADR-0004, TIS §0.4). + stamped = boundary.stamp() + + return Acquisition( + artifact=RawArtifact( + digest=Digest(artifact.digest.hex), + size_bytes=artifact.size_bytes, + retrieval=self.descriptor().retrieval, + cache_path=self._file, + ), + provenance=AcquisitionProvenance( + run_id=RunId(run.run_id), + source_id=Identifier("issdc-pradan"), + artifact_digest=Digest(artifact.digest.hex), + ingest_time=stamped, + instruments=(Identifier("solexs"),), + ), + ) + + +@pytest.fixture +def acquired(tmp_path): + """Perform one conforming acquisition and return everything it produced.""" + source = tmp_path / "solexs_20240301.lc" + source.write_bytes(RAW_FRAME) + store = ProvenanceStore(tmp_path / "store") + adapter = IssdcShapedAdapter(store, source) + return store, adapter, verify_conformance(adapter, selector="2024-03-01") + + +# ═══════════════════════════════════════════ the whole path + + +def test_a_new_acquisition_produces_a_contract_valid_observation_with_both_times(acquired): + """The path M3 exists to open, end to end. + + Channel → kernel digest → boundary stamp → domain Observation → contract validation → + provenance record. Every step is performed by the component that owns it, and the + Observation that comes out carries both times with nothing fabricated. + """ + store, adapter, acquisition = acquired + + # An Observation canonicalised from the acquired artifact. The parse itself is #17's; + # what matters here is that the times and the provenance column come from the acquisition. + observation = Observation( + source_id=acquisition.provenance.source_id, + instrument_id=Identifier("solexs"), + quantity="count_rate", + unit="counts/s", + valid_time=Timestamp("2024-03-01T00:00:00Z"), + ingest_time=acquisition.provenance.ingest_time, + value=12.5, + source_digest=acquisition.artifact.digest, + ) + + validator_for("observation").validate(observation.to_dict()) + assert observation_is_wellformed(observation) + + # E5 §11(i): every newly ingested Observation carries both times. + assert observation.valid_time is not None + assert observation.ingest_time is not None + assert observation.ingest_time.instant >= observation.valid_time.instant + + # The provenance column points at the bytes the kernel actually holds. + assert store.has_artifact(KernelDigest(str(observation.source_digest))) + + # And the derivation is recorded, so the row is answerable back to the raw frame. + run = begin_run(context="ingest", event="parse") + parsed = store.put_bytes(json.dumps(observation.to_dict(), sort_keys=True).encode()) + store.record(run, inputs=[KernelDigest(str(observation.source_digest))], + outputs=[parsed.digest]) + assert KernelDigest(str(observation.source_digest)) in store.ancestors(parsed.digest) + + +def test_the_digest_is_the_same_digest_at_every_layer(acquired): + """Kernel, ingest artifact, acquisition provenance, domain Observation, manifest. + + Five representations of one content address. Each is a separate type in a separate + component, and nothing but a test holds them to the same value. + """ + store, adapter, acquisition = acquired + minted = store.put_file(acquisition.artifact.cache_path) + + assert acquisition.artifact.digest.hex == minted.digest.hex + assert acquisition.provenance.artifact_digest.hex == minted.digest.hex + + manifest = { + "kind": "dataset", + "digest": minted.digest.hex, + "tier": 0, + "recorded_at": str(acquisition.provenance.ingest_time), + "retention": {"class": "permanent"}, + "retrieval": acquisition.artifact.retrieval.to_dict(), + } + validator_for("manifest").validate(manifest) + assert manifest["digest"] == acquisition.artifact.digest.hex + + +def test_the_ingest_descriptor_is_exactly_a_manifest_retrieval_descriptor(acquired): + """One fact, one shape (M2/E4/#14, ADR-0023). + + The descriptor an adapter publishes and the descriptor a Tier 0 manifest records are the + same statement about how to re-acquire. If their shapes ever diverge, a manifest written + from an adapter's descriptor stops validating — so this is checked rather than assumed. + """ + _, adapter, _ = acquired + published = adapter.descriptor().retrieval.to_dict() + + manifest = { + "kind": "dataset", "digest": "a" * 64, "tier": 0, + "recorded_at": "2024-04-03T12:00:00Z", + "retention": {"class": "permanent"}, + "retrieval": published, + } + validator_for("manifest").validate(manifest) + + schema = json.loads((CONTRACTS / "manifest.schema.json").read_text()) + assert set(published) == set(schema["properties"]["retrieval"]["properties"]) + + +def test_tier_0_bytes_are_referenced_and_the_manifest_cannot_say_otherwise(acquired): + """ADR-0023 and STD-23, enforced at both ends of the same path. + + Ingest cannot hand out the bytes — `RawArtifact` has no field for them — and the manifest + cannot record them as deposited. Two independent refusals of the same thing. + """ + _, _, acquisition = acquired + assert not hasattr(acquisition.artifact, "bytes") + assert acquisition.artifact.retrieval is not None + + redistributing = { + "kind": "dataset", "digest": acquisition.artifact.digest.hex, "tier": 0, + "recorded_at": "2024-04-03T12:00:00Z", + "retention": {"class": "permanent"}, + "retrieval": acquisition.artifact.retrieval.to_dict(), + "deposition": {"provider": "Zenodo", "url": "https://zenodo.org/records/1", + "doi": None}, + } + assert list(validator_for("manifest").iter_errors(redistributing)) + + +# ═══════════════════════════════════════════ ADR-0022 — the historical path + + +def test_a_historical_row_carries_a_null_ingest_time_and_no_acquisition(): + """The other half of the bitemporal story, and the one that is easy to get wrong. + + A row that predates bitemporal capture has no acquisition to describe. It gets + `ingest_time = None` — ADR-0022's single meaning — and E5 §11(ii) forbids any code path + from writing a non-null value for it. The contract accepts it, the invariant confirms it + was not backfilled, and there is nothing in `contexts.ingest.boundary` that would supply + one. + """ + historical = Observation( + source_id=Identifier("issdc-pradan"), + instrument_id=Identifier("solexs"), + quantity="count_rate", + unit="counts/s", + valid_time=Timestamp("2023-01-15T00:00:00Z"), + ingest_time=None, + value=8.25, + source_digest=Digest("c" * 64), + ) + + validator_for("observation").validate(historical.to_dict()) + assert historical.to_dict()["ingest_time"] is None + assert observation_is_wellformed(historical) + assert historical.ingest_time_is_unknown + + freeze = Timestamp("2024-05-01T00:00:00Z") + assert ingest_time_is_not_backfilled(historical, freeze) + + # Stamping it would be a deliberate act, and the invariant catches the specific + # fabrication ADR-0022 was written to reject. + fabricated = Observation( + source_id=historical.source_id, + instrument_id=historical.instrument_id, + quantity=historical.quantity, + unit=historical.unit, + valid_time=historical.valid_time, + ingest_time=freeze, + value=historical.value, + source_digest=historical.source_digest, + ) + validator_for("observation").validate(fabricated.to_dict()) # schema cannot see it + assert not ingest_time_is_not_backfilled(fabricated, freeze) # the invariant can + + +def test_a_bitemporal_protocol_excludes_the_historical_row_and_admits_the_acquired_one(acquired): + """Why the distinction is load-bearing (ADR-0022 §2). + + A Protocol with `requires_bitemporal = true` excludes rows whose availability cannot be + established. The newly acquired row survives it; the historical one does not. That is the + leakage gate becoming enforceable by construction rather than by trust — and it only + works because nothing filled in the missing time. + """ + from domain.tests import build + + _, _, acquisition = acquired + strict = build.protocol(requires_bitemporal=True) + permissive = build.protocol(requires_bitemporal=False) + + fresh = build.observation(ingest_time=acquisition.provenance.ingest_time) + historical = build.observation(ingest_time=None) + + assert observation_is_admissible_under(fresh, strict) + assert not observation_is_admissible_under(historical, strict) + assert observation_is_admissible_under(historical, permissive) + + +# ═══════════════════════════════════════════ #13 — the rules still hold + + +def test_the_ingest_context_imports_only_what_adr_0026_grants(): + """The real tree, under the shipped policy — not a fixture. + + M3/E5/#15 is the first issue to put code in a context, so this is the first time the + ingest rule binds against anything. It asserts the policy is applied to the real package + and that the package satisfies it. + """ + from tools.gates.imports import POLICIES, run + + ingest = next(p for p in POLICIES if p.package == "contexts.ingest") + assert ingest.populated is True, "the ingest policy still declares itself unpopulated" + assert ingest.allow == frozenset({"contracts", "domain", "kernel"}) + + report, code = run(POLICIES) + assert code == 0, report.violations + assert report.modules >= 7, f"only {report.modules} modules scanned" + + +def test_ingest_reaches_no_other_context(tmp_path): + """The deliberate violation, against the shipped rule rather than a synthetic one.""" + spec = importlib.util.spec_from_file_location(f"ing_gate_{tmp_path.name}", GATE_PATH) + assert spec is not None and spec.loader is not None + gate = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = gate + spec.loader.exec_module(gate) + gate.REPO_ROOT = tmp_path + + for dotted in ("contracts", "domain", "kernel", "contexts.curation"): + directory = tmp_path / Path(*dotted.split(".")) + directory.mkdir(parents=True, exist_ok=True) + (directory / "__init__.py").write_text("") + ingest_dir = tmp_path / "contexts" / "ingest" + ingest_dir.mkdir(parents=True) + (ingest_dir / "__init__.py").write_text("") + (ingest_dir / "acquire.py").write_text( + "from domain.values import Digest\n" + "from kernel.provenance import ProvenanceStore\n" + "from contexts.curation.freeze import freeze\n" + ) + + from tools.gates.imports import POLICIES as SHIPPED + + shipped = next(p for p in SHIPPED if p.package == "contexts.ingest") + report, code = gate.run([ + gate.Policy(package="contexts.ingest", allow=shipped.allow, populated=True), + gate.Policy(package="contexts.curation", allow=shipped.allow, populated=False), + *[gate.Policy(package=n) for n in ("contracts", "domain", "kernel")], + ]) + assert code == 1 + assert any( + "contexts.curation" in v and "not permitted" in v for v in report.violations + ), report.violations + + +def test_no_credential_reaches_anything_the_platform_stores(acquired): + """The boundary holds across the whole path, checked on what was actually produced.""" + store, adapter, acquisition = acquired + assert_credential_free(adapter.descriptor(), what="the source descriptor") + assert_credential_free(acquisition, what="the acquisition") + + serialised = json.dumps(acquisition.provenance.to_dict(), sort_keys=True) + for leak in ("session", "cookie", "token", "password", "credential"): + assert leak not in serialised diff --git a/tests/integration/test_ingest_credential_boundary.py b/tests/integration/test_ingest_credential_boundary.py new file mode 100644 index 0000000..e09dd67 --- /dev/null +++ b/tests/integration/test_ingest_credential_boundary.py @@ -0,0 +1,249 @@ +"""The credential boundary — the integration test §15 row 15 names. + +`| 15 | 15 | E5 | 200 | M | 11 | contract conformance | — | credential-boundary |` + +E5 §13 makes Ingest the highest-sensitivity context in the system: it holds the only secrets. +STD-19 and E5 §11(iii) say what must never happen — a secret logged, persisted to an artifact, +or crossing a context boundary. + +This is an integration test rather than a unit test because a leak is not a property of the +`Credential` type. It is a property of a *path*: the secret is fine sitting in memory and +becomes a defect the moment something writes it somewhere. So each test below follows a real +path outward — into a log record, into a provenance artifact registered with the kernel, into +a manifest, into a formatted string — and checks what arrives at the far end. + +The bytes on disk are read back and searched for the secret. Not the object, not the +serialisation function's return value: the file. That is the only check that would have caught +the class of leak this is written against, where a redaction works everywhere except the one +path nobody tested. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path + +import pytest + +from contexts.ingest import ( + Acquisition, + AcquisitionProvenance, + CREDENTIAL_BOUNDARY, + Credential, + PolicyRejection, + RawArtifact, + RetrievalDescriptor, + assert_credential_free, + verify_conformance, +) +from contexts.ingest.credentials import REDACTED, SECRET_FIELD_NAMES +from contexts.ingest.tests import adapters +from domain.values import Digest, Identifier, RunId, Timestamp +from kernel.provenance import ProvenanceStore, begin_run + +#: A value distinctive enough that finding it anywhere is unambiguous. +SECRET = "PRADAN-SESSION-3f9c1a7e-do-not-log-me" + + +def a_credential() -> Credential: + return Credential("issdc-pradan-session", SECRET) + + +# ═══════════════════════════════════════════ the secret does not reach a log + + +def test_a_credential_cannot_be_logged_through_any_common_route(tmp_path, caplog): + """The routes a secret actually escapes by: an f-string, a %-format, a repr, a traceback. + + E5 §14 forbids logging cookies. Enforcement is on the type rather than on every call site, + because "remember not to log this" is the instruction that fails. + """ + secret = a_credential() + log_file = tmp_path / "ingest.log" + + handler = logging.FileHandler(log_file) + logger = logging.getLogger("ingest.boundary.test") + logger.setLevel(logging.DEBUG) + logger.addHandler(handler) + try: + logger.info("acquiring with %s", secret) + logger.info(f"acquiring with {secret}") + logger.info("acquiring with %r", secret) + logger.info("padded: %s", f"{secret:>40}") + logger.debug("structured: %s", {"session": str(secret)}) + try: + raise RuntimeError(f"auth failed for {secret}") + except RuntimeError: + logger.exception("acquisition failed") + finally: + logger.removeHandler(handler) + handler.close() + + written = log_file.read_text() + assert SECRET not in written, "the secret reached the log file" + assert REDACTED in written, "nothing was written; the test would pass vacuously" + + +def test_the_secret_is_still_readable_where_it_is_supposed_to_be(): + """Redaction that also hid the value from the adapter would make the type unusable. + + `reveal()` is the single named accessor, so `grep -rn 'reveal()' contexts/` enumerates + every place in the repository where a secret is touched. + """ + assert a_credential().reveal() == SECRET + assert str(a_credential()) != SECRET + assert repr(a_credential()) != SECRET + + +# ═══════════════════════════════════════════ the secret does not reach an artifact + + +def test_a_credential_cannot_be_serialised_into_an_artifact(): + """E5 §13: never persisted to an artifact. `json.dumps` refuses it outright.""" + with pytest.raises(TypeError): + json.dumps({"session": a_credential()}) + assert not hasattr(Credential, "to_dict") + + +def test_provenance_registered_with_the_kernel_contains_no_secret(tmp_path): + """The full path: acquire → serialise provenance → register → read the file back. + + Reading the bytes from disk is the point. A check on the dict would pass even if the + serialiser had been changed to include the secret somewhere the dict did not show. + """ + store = ProvenanceStore(tmp_path / "store") + run = begin_run(context="ingest", event="acquire") + + # An adapter authenticates with a real secret and publishes provenance that must not + # carry it. + secret = a_credential() + assert secret.reveal() == SECRET # the adapter genuinely used it + + provenance = AcquisitionProvenance( + run_id=RunId(run.run_id.value if hasattr(run.run_id, "value") else str(run.run_id)), + source_id=Identifier("issdc-pradan"), + artifact_digest=Digest("a" * 64), + ingest_time=Timestamp("2024-04-03T12:00:00Z"), + instruments=(Identifier("solexs"),), + ) + assert_credential_free(provenance, what="the acquisition provenance") + + payload = json.dumps(provenance.to_dict(), sort_keys=True).encode() + artifact = store.put_bytes(payload) + + on_disk = (store.root / "artifacts" / f"{artifact.digest.hex}.json").read_text() + assert SECRET not in on_disk + assert SECRET not in payload.decode() + assert "issdc-pradan" in payload.decode(), "nothing was recorded; the check is vacuous" + + +def test_a_manifest_built_from_a_descriptor_carries_no_secret(): + """The descriptor an adapter publishes becomes a Tier 0 manifest's retrieval descriptor. + + M2/E4/#14's `manifest.schema.json` records *how* to re-acquire and has no field for a + secret; this shows the ingest side produces exactly that shape and nothing more. + """ + retrieval = adapters.descriptor().retrieval.to_dict() + assert set(retrieval) == {"provider", "locator", "requires_credentials"} + assert retrieval["requires_credentials"] is True + assert SECRET not in json.dumps(retrieval) + for name in SECRET_FIELD_NAMES: + assert name not in retrieval + + +# ═══════════════════════════════════════════ the boundary refuses the crossing + + +def test_the_boundary_refuses_an_object_carrying_a_credential(): + with pytest.raises(PolicyRejection) as caught: + assert_credential_free({"session": a_credential()}, what="a log record") + assert caught.value.gate == CREDENTIAL_BOUNDARY + assert SECRET not in str(caught.value), "the rejection message leaked the secret" + + +@pytest.mark.parametrize( + "carrier, description", + [ + (lambda c: c, "the credential itself"), + (lambda c: {"session": c}, "a mapping value"), + (lambda c: [1, {"deep": {"inner": c}}], "nested three levels down"), + (lambda c: (c,), "inside a tuple"), + (lambda c: {"a": [{"b": [c]}]}, "alternating mappings and sequences"), + ], + ids=["direct", "mapping", "nested", "tuple", "alternating"], +) +def test_a_credential_is_found_wherever_it_is_hidden(carrier, description): + """A shallow check would pass on four of these five.""" + with pytest.raises(PolicyRejection): + assert_credential_free(carrier(a_credential()), what=description) + + +def test_a_raw_string_in_a_secret_named_field_is_also_refused(): + """The backstop for the mistake of never using the type at all. + + Caught by field *name*, not by guessing what a secret looks like: a heuristic over values + would be both wrong and unfalsifiable. + """ + with pytest.raises(PolicyRejection): + assert_credential_free({"cookie": "sessionid=abc123"}, what="a request record") + + +def test_the_boundary_permits_what_should_cross(): + """A checker that refused everything would be satisfied by every test above.""" + assert_credential_free(adapters.descriptor(), what="the source descriptor") + assert_credential_free(adapters.acquisition(), what="an acquisition") + assert_credential_free( + {"source_id": "issdc-pradan", "requires_credentials": True}, + what="a manifest retrieval descriptor", + ) + + +def test_a_cyclic_structure_does_not_crash_the_check(): + """A leak check that raises RecursionError is a leak check that gets removed.""" + loop: dict = {"self": None} + loop["self"] = loop + assert_credential_free(loop, what="a cyclic record") + + loop_with_secret: dict = {"self": None, "session": a_credential()} + loop_with_secret["self"] = loop_with_secret + with pytest.raises(PolicyRejection): + assert_credential_free(loop_with_secret, what="a cyclic record") + + +# ═══════════════════════════════════════════ conformance enforces it end to end + + +def test_the_conformance_check_refuses_an_adapter_that_would_leak(tmp_path): + """The boundary is not advisory: an adapter that leaks cannot pass conformance. + + This is what #16's ISSDC adapter will be run against before its data is allowed in. + """ + with pytest.raises(PolicyRejection) as caught: + verify_conformance( + adapters.AdapterPersistingACredentialInProvenance(), selector="2024-03-01" + ) + assert caught.value.gate == CREDENTIAL_BOUNDARY + + +def test_a_conforming_adapter_passes_and_leaks_nothing(tmp_path): + """The positive path, checked at the far end rather than at the type.""" + acquired: Acquisition = verify_conformance( + adapters.ConformingAdapter(), selector="2024-03-01" + ) + serialised = json.dumps(acquired.provenance.to_dict(), sort_keys=True) + assert SECRET not in serialised + for name in SECRET_FIELD_NAMES: + assert name not in serialised + + +def test_an_artifact_path_is_a_cache_not_a_secret(tmp_path): + """A cache path is recorded; it is not sensitive and must not be redacted into uselessness.""" + artifact = RawArtifact( + digest=Digest("a" * 64), + size_bytes=1, + retrieval=RetrievalDescriptor("ISSDC PRADAN", "aditya-l1/solexs", True), + cache_path=Path(tmp_path / "cache" / "frame.fits"), + ) + assert_credential_free(artifact, what="a raw artifact") + assert artifact.is_cached diff --git a/tools/gates/imports.py b/tools/gates/imports.py index ba25065..39595a0 100644 --- a/tools/gates/imports.py +++ b/tools/gates/imports.py @@ -318,7 +318,12 @@ def run(policies: list[Policy]) -> tuple[Report, int]: # R1 — Ingest. Acquires and canonicalises. Needs the vocabulary, the domain model and # the kernel to register raw artifacts as it acquires them (TIS E5 §19). - Policy(package="contexts.ingest", populated=False, + # + # `populated` flipped to True by M3/E5/#15, the first issue to put code in a context. The + # staleness check did its job on the way: with the modules added and the flag still False + # the gate went red — "declared unpopulated but contains 7 module(s)" — which is exactly + # what M2/E4/#13 built it for. The rule now binds against real code rather than a fixture. + Policy(package="contexts.ingest", populated=True, allow=frozenset({"contracts", "domain", "kernel"})), # R2 — Curation. Freezes observations into digest-addressed releases; the kernel is what