From 40fb3cad7207d2052c668df3776335978520c0c1 Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:40:17 -0700 Subject: [PATCH 1/9] fix: close sealed-payload path traversal and ordering/atomicity bugs in ingest - Validate a payload's filename (via bag.py's existing _reject_unsafe_relpath) before writing its encrypted bytes into the SEALED-payload staging directory. Previously the write happened before any traversal check, so a payload key containing `..` or an absolute path (reachable via oralhistory.apply_session_manifest's payload_filename, which is not otherwise sanitized) could write ciphertext outside the intended temp directory. - Move the bag-collision check ahead of SEALED-field encryption, matching the ordering already used for identity sealing: a colliding ingest no longer mutates the caller's in-memory Record fields (plaintext -> ciphertext, in place, unrecoverable) before discovering the ingest will fail anyway. - Write the per-record .versions.json history index via the module's own atomic_write_text helper instead of a plain write_text call, so a crash mid-write can no longer corrupt it into something _read_versions silently reads back as "no history" (the method's own docstring already claims every write here is atomic; this one wasn't). Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/ingest.py | 36 +++++++++++++++++++++++++++--------- 1 file changed, 27 insertions(+), 9 deletions(-) diff --git a/src/ledger/ingest.py b/src/ledger/ingest.py index bd16a6a..3202a2e 100644 --- a/src/ledger/ingest.py +++ b/src/ledger/ingest.py @@ -34,7 +34,13 @@ from ledger import catalog_index from ledger.access import disclose, is_listable from ledger.attest import attested_conditions -from ledger.bag import atomic_write_text, refresh_tag_manifests, validate_bag, write_bag +from ledger.bag import ( + _reject_unsafe_relpath, + atomic_write_text, + refresh_tag_manifests, + validate_bag, + write_bag, +) from ledger.cas import ContentStore from ledger.chain import GENESIS_HASH, ChainVerification from ledger.config import Config @@ -404,6 +410,13 @@ def ingest_sip( # noqa: C901 raise LedgerError( "a 'sealed' (absolute) payload requires a vault key for at-rest encryption" ) + # Validate BEFORE joining: `filename` is a bag-relative payload key that + # can originate from caller-supplied data (e.g. a session manifest's + # `payload_filename`), not only a trusted local file path. `write_bag` + # rejects an unsafe relpath too, but only much later — after this branch + # would already have written ciphertext to an absolute or traversal + # path via plain `Path` joining (a real arbitrary-write otherwise). + _reject_unsafe_relpath(filename, context="sealed payload") if sealed_tmp is None: sealed_tmp = Path(tempfile.mkdtemp(prefix="ledger-sealed-")) store_source = sealed_tmp / filename @@ -480,6 +493,15 @@ def ingest_sip( # noqa: C901 if pid not in record.dublin_core.identifier: record.dublin_core.identifier = [*record.dublin_core.identifier, pid] + # 2. Refuse a collision BEFORE sealing any identity OR field, so a failed + # ingest cannot leave an orphaned, unreachable identity in the vault, or + # silently overwrite the caller's in-memory sealed-field plaintext with + # ciphertext it has no way to recover (#correctness, consent). + bag_dir = bags_dir / record.record_id + if bag_dir.exists(): + # An item is bagged exactly once; refuse to clobber a prior AIP silently. + raise LedgerError(f"bag already exists for record {record.record_id}") + # Absolute-SEALED fields are encrypted AT REST so a stolen disk or hostile # replica reveals nothing, not even to a steward (user research P2-4). Such a # field is never disclosed on any read path, so it is only ever encrypted, never @@ -493,13 +515,6 @@ def ingest_sip( # noqa: C901 if fld.policy is AccessPolicy.SEALED and not fld.value.startswith("enc:"): fld.value = vault.encrypt_text(fld.value) - # 2. Refuse a collision BEFORE sealing any identity, so a failed ingest cannot - # leave an orphaned, unreachable identity in the vault (#correctness, consent). - bag_dir = bags_dir / record.record_id - if bag_dir.exists(): - # An item is bagged exactly once; refuse to clobber a prior AIP silently. - raise LedgerError(f"bag already exists for record {record.record_id}") - # 3. Seal identity into the vault and replace it with an opaque ref. Everything # after this point is wrapped so any failure revokes the ref (no orphan). sealed_ref: str | None = None @@ -868,7 +883,10 @@ def _append_version(self, record_id: str, address: str, event_type: str) -> None entries = self._read_versions(path) entries.append({"address": address, "saved_at": now_iso(), "event_type": event_type}) path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(canonical_json(entries), encoding="utf-8", newline="\n") + # Atomic (temp + os.replace), like every other write in this module: a + # crash mid-write must leave the prior index intact rather than a torn + # file that `_read_versions` would silently read back as empty history. + atomic_write_text(path, canonical_json(entries)) @staticmethod def _read_versions(path: Path) -> list[dict[str, str]]: From e4005b995a42090f666ad9327d9f2abc5d0a807c Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:40:26 -0700 Subject: [PATCH 2/9] fix: lock identity vault and attestation writes against concurrent corruption IdentityVault.add/revoke/rekey mutated the in-memory store and persisted to disk with no locking at all, unlike every sibling JSON store in the codebase (ConsentRequestStore, ProposalStore, TombstoneStore, ...). Under the threaded browse server, two concurrent calls (e.g. an ingest and a takedown) open the *same* temp file -- its name is derived only from os.getpid(), identical across threads in one process -- so the second thread's truncating open can corrupt the first thread's in-flight write before either renames it into place. This is the archive's single most sensitive file. Wrapped each method's mutate-then-persist step in the existing ledger._filelock.file_lock, the same primitive already used throughout the codebase for exactly this class of race. AttestStore._record_attested had the identical unlocked read-modify-write gap on attested-conditions.json (two approvals crossing a 2-of-N threshold for different conditions near-simultaneously could race and drop one). Locked it the same way. Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/attest.py | 11 +++++-- src/ledger/identity.py | 65 +++++++++++++++++++++++++++--------------- 2 files changed, 51 insertions(+), 25 deletions(-) diff --git a/src/ledger/attest.py b/src/ledger/attest.py index 3f4a1bc..b6a79ac 100644 --- a/src/ledger/attest.py +++ b/src/ledger/attest.py @@ -34,6 +34,7 @@ import os from pathlib import Path +from ledger._filelock import file_lock from ledger.dualcontrol import ActionProposal, ProposalStore from ledger.metadata.premis import PremisLog from ledger.models import PremisEvent, PremisEventType, now_iso @@ -162,10 +163,16 @@ def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) -> The set is the machine-readable authority the access layer consults; the append-only PREMIS event is the accountable, replica-checkable *why*. Both are - no-outing safe — a condition string plus steward ids, never an identity.""" + no-outing safe — a condition string plus steward ids, never an identity. + + The attested-set read-modify-write is locked (:func:`ledger._filelock.file_lock`) + so two approvals that each cross the threshold for a *different* condition at + nearly the same moment cannot race and silently drop one (a lost attestation is + as serious a bug here as a lost consent withdrawal elsewhere in the archive).""" condition = proposal.target attested_path = self._dir / _ATTESTED_FILENAME - _write_attested(attested_path, _read_attested(attested_path) | {condition}) + with file_lock(attested_path): + _write_attested(attested_path, _read_attested(attested_path) | {condition}) event = PremisEvent( event_type=PremisEventType.POLICY_CHANGE, diff --git a/src/ledger/identity.py b/src/ledger/identity.py index bd6fe27..246dc59 100644 --- a/src/ledger/identity.py +++ b/src/ledger/identity.py @@ -56,6 +56,7 @@ from cryptography.fernet import Fernet, InvalidToken from cryptography.hazmat.primitives.kdf.scrypt import Scrypt +from ledger._filelock import file_lock from ledger.errors import AccessDenied, IdentityVaultError from ledger.models import PUBLIC_GRANT, Grant, canonical_json @@ -202,12 +203,22 @@ def add(self, identity: ContributorIdentity) -> str: Returns the new opaque ``identity_ref``. The ref is generated from a CSPRNG and is independent of the identity contents, so it carries no identifying signal and may safely live in a record -> unlinkability. + + Locked (:func:`ledger._filelock.file_lock`) around the mutate-then-persist + step: the threaded browse server can run concurrent ingests against the + same ``Archive`` (and so the same vault instance), and :meth:`_persist`'s + temp filename is derived only from ``os.getpid()`` -- identical across + threads in one process. Without a lock, two concurrent calls open and + write the *same* temp path, and the second thread's truncating open can + corrupt the first thread's in-flight write before either renames it into + place (fault tolerance, integrity of the archive's most sensitive file). """ - ref = self._new_ref() - token = self._encrypt(identity) - self._store[ref] = token - self._persist() - return ref + with file_lock(self._path): + ref = self._new_ref() + token = self._encrypt(identity) + self._store[ref] = token + self._persist() + return ref def resolve(self, ref: str, grant: Grant, now: str) -> ContributorIdentity: """Decrypt and return the identity for *ref*, gated by *grant* at *now*. @@ -239,10 +250,14 @@ def revoke(self, ref: str) -> None: Idempotent: revoking an absent ref is a no-op so a takedown can be retried safely. Honours consent revocation / takedown -> autonomy, consent. + + Locked like :meth:`add` so a concurrent takedown and ingest never race on + the same temp file (see :meth:`add`'s docstring). """ - if ref in self._store: - del self._store[ref] - self._persist() + with file_lock(self._path): + if ref in self._store: + del self._store[ref] + self._persist() def contains(self, ref: str) -> bool: """Return whether *ref* currently maps to a stored identity.""" @@ -277,21 +292,25 @@ def rekey(self, new_key: bytes) -> int: new_fernet = Fernet(new_key) except (ValueError, TypeError) as exc: raise IdentityVaultError("invalid Fernet key") from exc - # Re-encrypt into a fresh map first; only commit if every entry succeeds, so - # a mid-rotation failure cannot leave a half-rotated vault -> integrity. - rotated: dict[str, str] = {} - for ref, token in self._store.items(): - try: - raw = self._fernet.decrypt(token.encode("ascii")) - except InvalidToken as exc: - raise IdentityVaultError( - "identity decryption failed during rekey (wrong key or tampering)" - ) from exc - rotated[ref] = new_fernet.encrypt(raw).decode("ascii") - self._fernet = new_fernet - self._store = rotated - self._persist() - return len(rotated) + # Locked like :meth:`add`/:meth:`revoke` so a rotation can never race a + # concurrent add/revoke on the same vault file (see :meth:`add`'s + # docstring for why the unlocked temp-file write is unsafe). + with file_lock(self._path): + # Re-encrypt into a fresh map first; only commit if every entry succeeds, + # so a mid-rotation failure cannot leave a half-rotated vault -> integrity. + rotated: dict[str, str] = {} + for ref, token in self._store.items(): + try: + raw = self._fernet.decrypt(token.encode("ascii")) + except InvalidToken as exc: + raise IdentityVaultError( + "identity decryption failed during rekey (wrong key or tampering)" + ) from exc + rotated[ref] = new_fernet.encrypt(raw).decode("ascii") + self._fernet = new_fernet + self._store = rotated + self._persist() + return len(rotated) # --- at-rest encryption of absolute-sealed content ---------------------- From 2fa182035b6122a22387ca0572f21950a21f9765 Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:40:35 -0700 Subject: [PATCH 3/9] fix: lock lockdown log writes and correct stand-up's status reporting _record_event read-modify-wrote logs/lockdown.premis.json with no lock, so a lockdown and stand-up event (or two lockdown attempts) racing under the threaded server could lose one. Locked it like every other archive log store. execute_stand_up() always constructed its LockdownResult with disclosure_stopped=False, even after a real, successful stand-up that unlinked lockdown.flag and resumed disclosure -- so LockdownResult.summary() printed "stand-up executed -- disclosure unchanged", the opposite of what happened, directly to the steward running `ledger stand-up --execute` (cli.py prints this string verbatim). The PREMIS audit event had the same bug: its detail hardcoded "flag_removed=True" regardless of whether a flag was actually present to remove. Both now reflect whether the freeze was actually lifted, and summary() renders stand-up's own wording ("resumed" / "was already open") instead of reusing lockdown's "stopped"/"unchanged" phrasing for an action it was never written to describe. Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/lockdown.py | 45 +++++++++++++++++++++++++++++------------- 1 file changed, 31 insertions(+), 14 deletions(-) diff --git a/src/ledger/lockdown.py b/src/ledger/lockdown.py index ece8c90..5edea08 100644 --- a/src/ledger/lockdown.py +++ b/src/ledger/lockdown.py @@ -37,6 +37,7 @@ from pathlib import Path from typing import Protocol +from ledger._filelock import file_lock from ledger.config import Config, LockdownConfig from ledger.errors import LedgerError from ledger.ingest import Archive @@ -187,7 +188,13 @@ def summary(self) -> str: """A single no-outing-safe status line for the CLI/audit surface.""" if self.dry_run: return f"{self.action} DRY-RUN — {len(self.steps)} step(s) planned; nothing changed" - bits = [f"disclosure {'stopped' if self.disclosure_stopped else 'unchanged'}"] + if self.action == "stand-up": + # For stand-up, `disclosure_stopped` carries the inverse fact (did this + # run actually lift a freeze that was in place), so the wording must be + # its own, not "stopped"/"unchanged" (which describe a *lockdown*). + bits = [f"disclosure {'resumed' if self.disclosure_stopped else 'was already open'}"] + else: + bits = [f"disclosure {'stopped' if self.disclosure_stopped else 'unchanged'}"] if self.vault_shredded: bits.append(f"local vault shredded after {self.verified_replicas} verified replica(s)") elif self.action == "lockdown": @@ -235,20 +242,25 @@ def _record_event( Kept in ``logs/lockdown.premis.json`` (append-only) so the duress decision is provable after the fact. The detail carries only counts and posture — never an identity or a vault byte (no-outing rule). + + Locked (:func:`ledger._filelock.file_lock`) around the read-modify-write so a + lockdown/stand-up event can never be lost to a concurrent write to the same log + (accountability -- this is the audit trail for the archive's duress posture). """ archive.logs_dir.mkdir(parents=True, exist_ok=True) log_path = archive.logs_dir / _LOCKDOWN_PREMIS - log = PremisLog.read(log_path) if log_path.exists() else PremisLog() - log.record( - PremisEvent( - event_type=event_type, - agent=actor, - outcome=outcome, - detail=detail, - event_datetime=now, + with file_lock(log_path): + log = PremisLog.read(log_path) if log_path.exists() else PremisLog() + log.record( + PremisEvent( + event_type=event_type, + agent=actor, + outcome=outcome, + detail=detail, + event_datetime=now, + ) ) - ) - log.write(log_path) + log.write(log_path) def _verify_replicas(config: LockdownConfig) -> list[BackupVerification]: @@ -468,7 +480,8 @@ def execute_stand_up(archive: ArchiveLike, *, actor: str, now: str) -> LockdownR steps.append(f"restored local vault from verified replica at {clean[0].location}") flag = lockdown_flag_path(archive) - if flag.exists(): + flag_removed = flag.exists() + if flag_removed: flag.unlink() steps.append("removed lockdown.flag — non-PUBLIC disclosure resumes") else: @@ -479,13 +492,17 @@ def execute_stand_up(archive: ArchiveLike, *, actor: str, now: str) -> LockdownR event_type=PremisEventType.STANDUP, actor=actor, outcome="success", - detail=f"vault_restored={restored}; flag_removed=True", + detail=f"vault_restored={restored}; flag_removed={flag_removed}", now=now, ) return LockdownResult( action="stand-up", dry_run=False, - disclosure_stopped=False, + # Repurposed for this action: whether the freeze was actually lifted just + # now (the flag existed and was removed), not "did this stop disclosure" + # (that phrasing belongs to lockdown; see `summary()`, which branches on + # `action` to render the right words for whichever fact this is). + disclosure_stopped=flag_removed, vault_shredded=False, verified_replicas=0, steps=tuple(steps), From 9edca0925ac8ce2d78ecdd5630a7cad531137c00 Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:40:43 -0700 Subject: [PATCH 4/9] fix: sort Atom feed entries chronologically, not lexicographically atom_feed_xml sorted entries by each record's raw, un-normalized dc:date string. Dublin Core dates in this app are free text of varying granularity/padding ("1994", "2021-5-1", "2021-12-01", a full timestamp), and comparing them as plain strings is lexicographic, not chronological -- an unpadded date can sort after a later, zero-padded one purely because of character ordering, breaking the feed's documented "most recent first" guarantee. The sort key now runs through the same _atom_timestamp widening already used to render each entry's value, so the order a reader sees always agrees with the dates displayed. Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/oai.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/ledger/oai.py b/src/ledger/oai.py index 9e045de..6c5478c 100644 --- a/src/ledger/oai.py +++ b/src/ledger/oai.py @@ -180,11 +180,20 @@ def atom_feed_xml( Entries are ordered newest first by their Dublin Core date (``record_id`` breaks ties for a deterministic feed) and capped at ``limit``. Every value is XML-escaped through the shared :func:`escape` boundary. ``now`` is the feed's generation time - and the fallback timestamp, so output is deterministic for a given input.""" + and the fallback timestamp, so output is deterministic for a given input. + + The sort key is the same RFC 3339 instant :func:`_atom_timestamp` widens each + entry's date to for display (not the raw ``dc:date`` string): a ``dc:date`` is + free text of varying granularity and padding (``"1994"``, ``"2021-5-1"``, + ``"2021-12-01"``, ...), and comparing those as plain strings is lexicographic, + not chronological — e.g. ``"2021-5-1" > "2021-12-01"`` as strings even though May + precedes December. Normalizing both to the same padded RFC 3339 form before + comparing keeps "most recent first" true for any date shape a contributor used, + and keeps the sort key consistent with what ```` actually displays.""" root = base_url.rstrip("/") ordered = sorted( records, - key=lambda r: (_datestamp(r, now), r.record_id), + key=lambda r: (_atom_timestamp(_datestamp(r, now), now), r.record_id), reverse=True, )[: max(0, limit)] From 9c7e3f1cea9904cda54dff6b9283f208385adf02 Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:40:51 -0700 Subject: [PATCH 5/9] fix: escape quotes in METS/EAD XML attribute values Both modules' escape() helper wrapped xml.sax.saxutils.escape, which only escapes &, <, and > by default -- not the quote character delimiting an XML attribute. Both modules interpolate escaped values directly into double-quoted attributes built from contributor-suppliable text (mets.py: LABEL="...", xlink:href="...", xlink:title="..." built from record.title and payload.filename; ead.py: id="c-..." built from record.record_id). A literal `"` in that text breaks the exported document's well-formedness. record_id is always a system-generated uuid4 hex today so that specific call site isn't reachable yet, but record.title is genuine free text from the web contribute form or CLI, so this was a real, reachable correctness bug. Both helpers now also escape `"` and `'`. Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/metadata/ead.py | 10 ++++++++-- src/ledger/metadata/mets.py | 12 ++++++++++-- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/src/ledger/metadata/ead.py b/src/ledger/metadata/ead.py index 3733244..18e9eb5 100644 --- a/src/ledger/metadata/ead.py +++ b/src/ledger/metadata/ead.py @@ -56,8 +56,14 @@ def _xml_text(value: str) -> str: def escape(value: str) -> str: - """XML-escape ``value`` after removing characters XML 1.0 disallows.""" - return _sax_escape(_xml_text(value)) + """XML-escape ``value`` after removing characters XML 1.0 disallows. + + Also escapes ``"`` and ``'`` (beyond ``xml.sax.saxutils.escape``'s default + ``&``/``<``/``>``): this module interpolates escaped record ids directly into + a double-quoted ``id="c-..."`` attribute, so a literal quote in that value + must not be able to break out of the attribute and produce malformed XML. + """ + return _sax_escape(_xml_text(value), {'"': """, "'": "'"}) _EAD_NS = "urn:isbn:1-931666-22-9" # the EAD 2002 namespace, per the LC schema diff --git a/src/ledger/metadata/mets.py b/src/ledger/metadata/mets.py index 3880631..36c4a09 100644 --- a/src/ledger/metadata/mets.py +++ b/src/ledger/metadata/mets.py @@ -65,8 +65,16 @@ def _xml_text(value: str) -> str: def escape(value: str) -> str: - """XML-escape ``value`` after removing characters XML 1.0 disallows.""" - return _sax_escape(_xml_text(value)) + """XML-escape ``value`` after removing characters XML 1.0 disallows. + + Also escapes ``"`` and ``'`` (beyond ``xml.sax.saxutils.escape``'s default + ``&``/``<``/``>``): this module interpolates escaped values directly into + double-quoted attributes (``LABEL="..."``, ``xlink:href="..."``, + ``xlink:title="..."``) built from contributor-suppliable text such as a + record's title, so a literal quote in that text must not be able to break out + of the attribute and produce malformed XML. + """ + return _sax_escape(_xml_text(value), {'"': """, "'": "'"}) _METS_NS = "http://www.loc.gov/METS/" From 2fcf636af45424a3023b7f769064c7f7da29d1e6 Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 22:12:40 -0700 Subject: [PATCH 6/9] fix: refresh locked identity and attestation state Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/attest.py | 44 ++++++++++++++----- src/ledger/identity.py | 95 +++++++++++++++++++++++++++++------------- tests/test_attest.py | 15 +++++++ tests/test_identity.py | 61 +++++++++++++++++++++++++++ 4 files changed, 174 insertions(+), 41 deletions(-) diff --git a/src/ledger/attest.py b/src/ledger/attest.py index b6a79ac..927351d 100644 --- a/src/ledger/attest.py +++ b/src/ledger/attest.py @@ -52,6 +52,7 @@ _PROPOSALS_FILENAME = "attestations.json" _ATTESTED_FILENAME = "attested-conditions.json" _PREMIS_FILENAME = "attestations.premis.json" +_WORKFLOW_LOCK_FILENAME = "attestations.workflow" def _read_attested(path: Path) -> frozenset[str]: @@ -135,12 +136,17 @@ def approve( exactly on the approval that tipped the proposal over the threshold — the approval that writes the condition into the attested set and records the ``POLICY_CHANGE`` event.""" - proposal = self._proposals.approve(proposal_id, steward) - if not proposal.is_ready(threshold): - return proposal, False - self._record_attested(proposal, agent=steward, now=now or now_iso()) - self._proposals.mark(proposal.proposal_id, "executed") - return proposal, True + # Serialize approval through terminal marking, not just each individual + # JSON rewrite. Otherwise two stewards can both observe the same proposal + # as ready before either marks it executed, recording the outcome twice. + workflow_lock = self._dir / _WORKFLOW_LOCK_FILENAME + with file_lock(workflow_lock): + proposal = self._proposals.approve(proposal_id, steward) + if not proposal.is_ready(threshold): + return proposal, False + attested_now = self._record_attested(proposal, agent=steward, now=now or now_iso()) + self._proposals.mark(proposal.proposal_id, "executed") + return proposal, attested_now # --- reads -------------------------------------------------------------- @@ -158,7 +164,7 @@ def attested(self) -> frozenset[str]: # --- durable outcome ---------------------------------------------------- - def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) -> None: + def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) -> bool: """Add the condition to the attested set and append a PREMIS ``POLICY_CHANGE``. The set is the machine-readable authority the access layer consults; the @@ -172,7 +178,10 @@ def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) -> condition = proposal.target attested_path = self._dir / _ATTESTED_FILENAME with file_lock(attested_path): - _write_attested(attested_path, _read_attested(attested_path) | {condition}) + prior = _read_attested(attested_path) + attested_now = condition not in prior + if attested_now: + _write_attested(attested_path, prior | {condition}) event = PremisEvent( event_type=PremisEventType.POLICY_CHANGE, @@ -188,6 +197,19 @@ def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) -> event_datetime=now, ) log_path = self._dir / _PREMIS_FILENAME - log = PremisLog.read(log_path) if log_path.exists() else PremisLog() - log.record(event) - log.write(log_path) + # PREMIS is another whole-file read-modify-write and needs its own stable + # sibling lock. Also make this append idempotent by condition: if the set + # write succeeded but the log write failed, retry repairs the missing event; + # if both succeeded but terminal marking failed, retry never duplicates it. + event_prefix = f"condition attested and now met: {condition} (" + with file_lock(log_path): + log = PremisLog.read(log_path) if log_path.exists() else PremisLog() + already_logged = any( + candidate.event_type is PremisEventType.POLICY_CHANGE + and candidate.detail.startswith(event_prefix) + for candidate in log.events + ) + if not already_logged: + log.record(event) + log.write(log_path) + return attested_now diff --git a/src/ledger/identity.py b/src/ledger/identity.py index 246dc59..89c0707 100644 --- a/src/ledger/identity.py +++ b/src/ledger/identity.py @@ -158,10 +158,14 @@ def create(cls, path: Path, key: bytes) -> IdentityVault: """ target = Path(path) vault = cls(target, key) - if target.exists(): - raise IdentityVaultError(f"vault already exists: {target}") - vault._store = {} - vault._persist() + # Check-and-create under the same stable sibling lock used by mutations. + # Otherwise two processes can both observe an absent path and the second + # can silently replace the first process's newly-created vault. + with file_lock(target): + if target.exists(): + raise IdentityVaultError(f"vault already exists: {target}") + vault._store = {} + vault._persist() return vault @classmethod @@ -175,24 +179,12 @@ def open(cls, path: Path, key: bytes) -> IdentityVault: if not target.exists(): raise IdentityVaultError(f"vault not found: {target}") vault = cls(target, key) - try: - raw = target.read_text(encoding="utf-8") - loaded = json.loads(raw) - except (OSError, ValueError) as exc: - raise IdentityVaultError(f"vault could not be read: {target}") from exc - if not isinstance(loaded, dict): - raise IdentityVaultError(f"vault is malformed: {target}") - store: dict[str, str] = {} - for ref, ciphertext in loaded.items(): - if not isinstance(ref, str) or not isinstance(ciphertext, str): - raise IdentityVaultError(f"vault is malformed: {target}") - store[ref] = ciphertext - # Verify the key by decrypting one entry; a wrong key surfaces immediately - # rather than at first resolve -> failure transparency. - for ciphertext in store.values(): - vault._decrypt(ciphertext) - break - vault._store = store + with file_lock(target): + # Re-check after acquiring the lock in case a concurrent operation + # removed/replaced the path between the optimistic check and here. + if not target.exists(): + raise IdentityVaultError(f"vault not found: {target}") + vault._reload() return vault # --- mutation ----------------------------------------------------------- @@ -212,8 +204,14 @@ def add(self, identity: ContributorIdentity) -> str: write the *same* temp path, and the second thread's truncating open can corrupt the first thread's in-flight write before either renames it into place (fault tolerance, integrity of the archive's most sensitive file). + + The on-disk store is reloaded *inside* that lock. This matters when two + archive handles or processes have separate ``IdentityVault`` instances: + locking only each instance's stale in-memory dictionary would still let a + later writer erase the earlier writer's update. """ with file_lock(self._path): + self._reload() ref = self._new_ref() token = self._encrypt(identity) self._store[ref] = token @@ -240,10 +238,15 @@ def resolve(self, ref: str, grant: Grant, now: str) -> ContributorIdentity: if ref not in effective.identity_unseal: # Name the ref and the missing capability, never the protected value. raise AccessDenied(f"grant does not permit unsealing identity ref {ref}") - token = self._store.get(ref) - if token is None: - raise IdentityVaultError(f"unknown identity ref: {ref}") - return self._decrypt(token) + # Refresh under the same lock used by revoke. A long-lived archive handle + # must not keep disclosing an identity from stale memory after another + # process has durably revoked it. + with file_lock(self._path): + self._reload() + token = self._store.get(ref) + if token is None: + raise IdentityVaultError(f"unknown identity ref: {ref}") + return self._decrypt(token) def revoke(self, ref: str) -> None: """Remove the mapping for *ref*, persisting atomically. @@ -255,17 +258,22 @@ def revoke(self, ref: str) -> None: the same temp file (see :meth:`add`'s docstring). """ with file_lock(self._path): + self._reload() if ref in self._store: del self._store[ref] self._persist() def contains(self, ref: str) -> bool: """Return whether *ref* currently maps to a stored identity.""" - return ref in self._store + with file_lock(self._path): + self._reload() + return ref in self._store def __len__(self) -> int: """Number of sealed identities in the vault (no contents revealed).""" - return len(self._store) + with file_lock(self._path): + self._reload() + return len(self._store) # --- key rotation ------------------------------------------------------- @@ -296,6 +304,7 @@ def rekey(self, new_key: bytes) -> int: # concurrent add/revoke on the same vault file (see :meth:`add`'s # docstring for why the unlocked temp-file write is unsafe). with file_lock(self._path): + self._reload() # Re-encrypt into a fresh map first; only commit if every entry succeeds, # so a mid-rotation failure cannot leave a half-rotated vault -> integrity. rotated: dict[str, str] = {} @@ -378,6 +387,32 @@ def derive_key(passphrase: str, salt: bytes) -> bytes: # --- internals ---------------------------------------------------------- + def _reload(self) -> None: + """Replace the in-memory map with the current authenticated disk state. + + Callers that need a linearizable read or mutation hold :func:`file_lock` + around this method and the rest of their operation. Reloading inside the + critical section prevents separate ``IdentityVault`` instances from + clobbering one another with stale whole-file snapshots. + """ + try: + loaded = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise IdentityVaultError(f"vault could not be read: {self._path}") from exc + if not isinstance(loaded, dict): + raise IdentityVaultError(f"vault is malformed: {self._path}") + store: dict[str, str] = {} + for ref, ciphertext in loaded.items(): + if not isinstance(ref, str) or not isinstance(ciphertext, str): + raise IdentityVaultError(f"vault is malformed: {self._path}") + store[ref] = ciphertext + # Verify the active key against one entry, surfacing a rekey performed by + # another process (or tampering) before this instance can overwrite it. + for ciphertext in store.values(): + self._decrypt(ciphertext) + break + self._store = store + @staticmethod def _new_ref() -> str: """Generate a fresh opaque ref from a CSPRNG -> unlinkability.""" @@ -397,9 +432,9 @@ def _decrypt(self, token: str) -> ContributorIdentity: """ try: raw = self._fernet.decrypt(token.encode("ascii")) - except InvalidToken as exc: + return ContributorIdentity._from_json_bytes(raw) + except (InvalidToken, UnicodeError, ValueError, TypeError, AttributeError) as exc: raise IdentityVaultError("identity decryption failed (wrong key or tampering)") from exc - return ContributorIdentity._from_json_bytes(raw) def _persist(self) -> None: """Write the store to disk atomically (temp file then rename). diff --git a/tests/test_attest.py b/tests/test_attest.py index cb24c54..fecdda5 100644 --- a/tests/test_attest.py +++ b/tests/test_attest.py @@ -128,6 +128,21 @@ def test_attested_set_survives_a_fresh_archive_handle(tmp_path: Path) -> None: ) +def test_record_attested_is_idempotent_across_retry(tmp_path: Path) -> None: + """A retry after a partial workflow write never duplicates its PREMIS event.""" + logs_dir = tmp_path / "logs" + store = attest.AttestStore(logs_dir) + proposal = store.propose(_CONDITION, "steward-A", now=_NOW) + proposal, attested_now = store.approve(proposal.proposal_id, "steward-B", now=_NOW) + assert attested_now is True + + assert store._record_attested(proposal, agent="steward-B", now=_NOW) is False + + premis = attest.PremisLog.read(logs_dir / "attestations.premis.json") + matching = [event for event in premis.events if _CONDITION in event.detail] + assert len(matching) == 1 + + @pytest.mark.disclosure def test_cli_attest_flow_and_invalid_condition_rejected(tmp_path: Path) -> None: """The CLI drives the same flow, and a condition outside the vocabulary is rejected.""" diff --git a/tests/test_identity.py b/tests/test_identity.py index 0e8e2c8..d378316 100644 --- a/tests/test_identity.py +++ b/tests/test_identity.py @@ -254,3 +254,64 @@ def test_roundtrip_through_reopen_resolves(tmp_path: Path) -> None: reopened = IdentityVault.open(path, key) resolved = reopened.resolve(ref, _unseal_grant(ref), _NOW) assert resolved.name == _NAME + + +def test_separate_vault_handles_do_not_overwrite_each_others_adds(tmp_path: Path) -> None: + """A stale whole-file snapshot cannot erase an add made by another handle.""" + path = tmp_path / "identity.vault" + key = IdentityVault.generate_key() + first = IdentityVault.create(path, key) + second = IdentityVault.open(path, key) + + ref_a = first.add(ContributorIdentity(name="A")) + ref_b = second.add(ContributorIdentity(name="B")) + + reopened = IdentityVault.open(path, key) + assert len(reopened) == 2 + assert reopened.resolve(ref_a, _unseal_grant(ref_a), _NOW).name == "A" + assert reopened.resolve(ref_b, _unseal_grant(ref_b), _NOW).name == "B" + + +def test_stale_handle_revoke_preserves_another_handles_add(tmp_path: Path) -> None: + """A revoke reloads under lock, so it cannot resurrect/drop unrelated refs.""" + path = tmp_path / "identity.vault" + key = IdentityVault.generate_key() + first = IdentityVault.create(path, key) + revoked_ref = first.add(ContributorIdentity(name="withdrawn")) + stale = IdentityVault.open(path, key) + + retained_ref = first.add(ContributorIdentity(name="retained")) + stale.revoke(revoked_ref) + + reopened = IdentityVault.open(path, key) + assert not reopened.contains(revoked_ref) + assert reopened.resolve(retained_ref, _unseal_grant(retained_ref), _NOW).name == "retained" + + +def test_stale_handle_cannot_resolve_after_another_handle_revokes(tmp_path: Path) -> None: + """Durable consent revocation invalidates already-open vault handles too.""" + path = tmp_path / "identity.vault" + key = IdentityVault.generate_key() + first = IdentityVault.create(path, key) + ref = first.add(_identity()) + stale = IdentityVault.open(path, key) + + first.revoke(ref) + + with pytest.raises(IdentityVaultError): + stale.resolve(ref, _unseal_grant(ref), _NOW) + + +def test_stale_handle_rekey_includes_entries_added_by_another_handle(tmp_path: Path) -> None: + """Rekey authenticates/reloads disk state before rotating the whole map.""" + path = tmp_path / "identity.vault" + old_key = IdentityVault.generate_key() + new_key = IdentityVault.generate_key() + first = IdentityVault.create(path, old_key) + stale = IdentityVault.open(path, old_key) + ref = first.add(_identity()) + + assert stale.rekey(new_key) == 1 + + reopened = IdentityVault.open(path, new_key) + assert reopened.resolve(ref, _unseal_grant(ref), _NOW).name == _NAME From 007ecb33492ca9a647b84d6fda258a1c2e4470dc Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 22:28:27 -0700 Subject: [PATCH 7/9] fix: harden archival state boundaries Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/bag.py | 13 ++-- src/ledger/identity.py | 58 +++++++++++++++--- src/ledger/lockdown.py | 26 ++++++++ src/ledger/oai.py | 48 +++++++++++---- tests/test_deposit_bridge.py | 39 ++++++++++++ tests/test_feed.py | 32 ++++++++++ tests/test_identity.py | 39 ++++++++++++ tests/test_ingest_e2e.py | 83 ++++++++++++++++++++++++- tests/test_lockdown.py | 113 ++++++++++++++++++++++++++++++++++- tests/test_versions.py | 26 ++++++++ 10 files changed, 451 insertions(+), 26 deletions(-) diff --git a/src/ledger/bag.py b/src/ledger/bag.py index 75b196f..508f6c6 100644 --- a/src/ledger/bag.py +++ b/src/ledger/bag.py @@ -26,7 +26,7 @@ import shutil from collections.abc import Mapping, Sequence from dataclasses import dataclass -from pathlib import Path, PurePosixPath +from pathlib import Path, PurePosixPath, PureWindowsPath from ledger.errors import BagValidationError from ledger.fixity import CHUNK_SIZE, AuditReport, hash_file, hash_file_multi, verify_file @@ -49,12 +49,17 @@ def _reject_unsafe_relpath(relpath: str, *, context: str) -> None: or hash files outside the bag (a path-traversal vulnerability). Validation is purely lexical, so it holds before any file is touched (securability, safety). """ - pure = PurePosixPath(relpath) + posix = PurePosixPath(relpath) + windows = PureWindowsPath(relpath) if ( not relpath or relpath.startswith("/") - or pure.is_absolute() - or ".." in pure.parts + or "\\" in relpath + or posix.is_absolute() + or windows.is_absolute() + or bool(windows.drive) + or ".." in posix.parts + or ".." in windows.parts or "\x00" in relpath ): raise BagValidationError(f"unsafe path in {context}: {relpath!r}") diff --git a/src/ledger/identity.py b/src/ledger/identity.py index 89c0707..f270433 100644 --- a/src/ledger/identity.py +++ b/src/ledger/identity.py @@ -72,6 +72,15 @@ # carries no structure that could be linked back to an identity -> unlinkability. _REF_NBYTES = 32 +# Reserved on-disk entry that authenticates the active key even when the vault has +# no contributor identities. Older vaults predate this marker; opening one with at +# least one identity authenticates the supplied key against that identity before +# migrating it, while an empty legacy vault necessarily establishes the supplied +# key on its one-time migration (there was no ciphertext with which to distinguish +# keys before this format version). +_KEY_CHECK_REF = "__ledger_vault_key_check_v1__" +_KEY_CHECK_PLAINTEXT = b"ledger.identity-vault:key-check:v1" + @dataclass class ContributorIdentity: @@ -140,6 +149,7 @@ def __init__(self, path: Path, key: bytes) -> None: # Do not echo the key or its bytes -> no-outing rule. raise IdentityVaultError("invalid Fernet key") from exc self._store: dict[str, str] = {} + self._key_check = "" def __repr__(self) -> str: """Redacted representation: path and count only, never contents.""" @@ -165,6 +175,7 @@ def create(cls, path: Path, key: bytes) -> IdentityVault: if target.exists(): raise IdentityVaultError(f"vault already exists: {target}") vault._store = {} + vault._key_check = vault._new_key_check() vault._persist() return vault @@ -184,7 +195,8 @@ def open(cls, path: Path, key: bytes) -> IdentityVault: # removed/replaced the path between the optimistic check and here. if not target.exists(): raise IdentityVaultError(f"vault not found: {target}") - vault._reload() + if vault._reload(): + vault._persist() return vault # --- mutation ----------------------------------------------------------- @@ -213,6 +225,8 @@ def add(self, identity: ContributorIdentity) -> str: with file_lock(self._path): self._reload() ref = self._new_ref() + while ref == _KEY_CHECK_REF or ref in self._store: + ref = self._new_ref() token = self._encrypt(identity) self._store[ref] = token self._persist() @@ -318,6 +332,7 @@ def rekey(self, new_key: bytes) -> int: rotated[ref] = new_fernet.encrypt(raw).decode("ascii") self._fernet = new_fernet self._store = rotated + self._key_check = new_fernet.encrypt(_KEY_CHECK_PLAINTEXT).decode("ascii") self._persist() return len(rotated) @@ -387,13 +402,16 @@ def derive_key(passphrase: str, salt: bytes) -> bytes: # --- internals ---------------------------------------------------------- - def _reload(self) -> None: + def _reload(self) -> bool: """Replace the in-memory map with the current authenticated disk state. Callers that need a linearizable read or mutation hold :func:`file_lock` around this method and the rest of their operation. Reloading inside the critical section prevents separate ``IdentityVault`` instances from clobbering one another with stale whole-file snapshots. + + Returns ``True`` when a legacy marker-free vault needs to be persisted in + the authenticated v1 format by the caller holding the lock. """ try: loaded = json.loads(self._path.read_text(encoding="utf-8")) @@ -402,16 +420,34 @@ def _reload(self) -> None: if not isinstance(loaded, dict): raise IdentityVaultError(f"vault is malformed: {self._path}") store: dict[str, str] = {} + key_check: str | None = None for ref, ciphertext in loaded.items(): if not isinstance(ref, str) or not isinstance(ciphertext, str): raise IdentityVaultError(f"vault is malformed: {self._path}") + if ref == _KEY_CHECK_REF: + key_check = ciphertext + continue store[ref] = ciphertext - # Verify the active key against one entry, surfacing a rekey performed by - # another process (or tampering) before this instance can overwrite it. - for ciphertext in store.values(): - self._decrypt(ciphertext) - break + needs_migration = key_check is None + if key_check is not None: + try: + plaintext = self._fernet.decrypt(key_check.encode("ascii")) + except (InvalidToken, UnicodeError, ValueError, TypeError) as exc: + raise IdentityVaultError("vault key check failed (wrong key or tampering)") from exc + if plaintext != _KEY_CHECK_PLAINTEXT: + raise IdentityVaultError("vault key check failed (wrong key or tampering)") + self._key_check = key_check + else: + # Backward-compatible migration. A populated legacy vault can and must + # authenticate against an identity entry first. An empty legacy vault + # contains no cryptographic evidence of its former key, so its first + # successful opener establishes the marker for all subsequent opens. + for ciphertext in store.values(): + self._decrypt(ciphertext) + break + self._key_check = self._new_key_check() self._store = store + return needs_migration @staticmethod def _new_ref() -> str: @@ -423,6 +459,10 @@ def _encrypt(self, identity: ContributorIdentity) -> str: plaintext = identity._to_canonical_json().encode("utf-8") return self._fernet.encrypt(plaintext).decode("ascii") + def _new_key_check(self) -> str: + """Return the authenticated format/key marker stored beside identity refs.""" + return self._fernet.encrypt(_KEY_CHECK_PLAINTEXT).decode("ascii") + def _decrypt(self, token: str) -> ContributorIdentity: """Decrypt a Fernet token; re-raise tampering/wrong-key as vault error. @@ -444,7 +484,9 @@ def _persist(self) -> None: vault intact -> integrity, fault tolerance. """ self._path.parent.mkdir(parents=True, exist_ok=True) - payload = canonical_json(self._store) + if not self._key_check: + self._key_check = self._new_key_check() + payload = canonical_json({_KEY_CHECK_REF: self._key_check, **self._store}) tmp = self._path.with_name(f"{self._path.name}.{os.getpid()}.tmp") try: # Create owner-only (0o600) BEFORE writing any ciphertext, so the vault diff --git a/src/ledger/lockdown.py b/src/ledger/lockdown.py index 5edea08..74e549f 100644 --- a/src/ledger/lockdown.py +++ b/src/ledger/lockdown.py @@ -62,6 +62,10 @@ # The append-only PREMIS log for lockdown/stand-up decisions, kept beside the other # archive-level logs so the duress history outlives the data it protected. _LOCKDOWN_PREMIS = "lockdown.premis.json" +# One stable mutex for the *entire* lockdown/stand-up state transition. Locking +# only the PREMIS append still allowed flag/vault/log operations from opposite +# transitions to interleave into a contradictory terminal state. +_LOCKDOWN_WORKFLOW = "lockdown.workflow" _CONFIG_FILENAME = "config.json" # Overwrite the vault in fixed-size chunks so shredding a large vault never pulls it # all into memory (efficiency, minimal computing). @@ -207,6 +211,11 @@ def lockdown_flag_path(archive: ArchiveLike) -> Path: return archive.logs_dir / _FLAG_FILENAME +def _workflow_lock_path(archive: ArchiveLike) -> Path: + """Stable path whose sibling lock serializes duress state transitions.""" + return archive.logs_dir / _LOCKDOWN_WORKFLOW + + def is_locked_down(archive: ArchiveLike) -> bool: """Whether ``archive`` is currently in lockdown (the marker is present). @@ -353,6 +362,17 @@ def _recovery_runbook(archive: ArchiveLike, config: LockdownConfig, *, vault_shr def execute_lockdown(archive: ArchiveLike, *, actor: str, now: str) -> LockdownResult: + """Serialize and execute one complete lockdown transition. + + The stable workflow lock covers flag creation, replica verification, optional + vault shredding, and PREMIS recording as one transition. A concurrent stand-up + therefore runs wholly before or wholly after this operation, never through it. + """ + with file_lock(_workflow_lock_path(archive)): + return _execute_lockdown_locked(archive, actor=actor, now=now) + + +def _execute_lockdown_locked(archive: ArchiveLike, *, actor: str, now: str) -> LockdownResult: """Execute the duress posture: stop disclosure, then conditionally shred the vault. Order is deliberate and fail-safe. The disclosure freeze is applied *first* (write @@ -440,6 +460,12 @@ def execute_lockdown(archive: ArchiveLike, *, actor: str, now: str) -> LockdownR def execute_stand_up(archive: ArchiveLike, *, actor: str, now: str) -> LockdownResult: + """Serialize and execute one complete stand-up transition.""" + with file_lock(_workflow_lock_path(archive)): + return _execute_stand_up_locked(archive, actor=actor, now=now) + + +def _execute_stand_up_locked(archive: ArchiveLike, *, actor: str, now: str) -> LockdownResult: """Lift the duress posture: restore the vault from a verified replica, drop the flag. The exact inverse of :func:`execute_lockdown`. If the local vault was shredded, it diff --git a/src/ledger/oai.py b/src/ledger/oai.py index 6c5478c..c269b2b 100644 --- a/src/ledger/oai.py +++ b/src/ledger/oai.py @@ -32,6 +32,7 @@ import re from collections.abc import Sequence +from datetime import UTC, datetime from ledger.metadata.dublincore import escape, to_oai_dc_xml from ledger.models import DisclosedRecord, DublinCore @@ -140,6 +141,34 @@ def sitemap_xml(record_ids: Sequence[str], base_url: str) -> str: _ATOM_NS = "http://www.w3.org/2005/Atom" +def _parse_atom_datetime(value: str) -> datetime | None: + """Parse an accepted Dublin Core date shape into an aware UTC datetime.""" + text = value.strip() + date_match = re.fullmatch(r"(\d{4})(?:-(\d{1,2})(?:-(\d{1,2}))?)?", text) + if date_match is not None: + year, month, day = date_match.groups() + try: + return datetime(int(year), int(month or 1), int(day or 1), tzinfo=UTC) + except ValueError: + return None + + # ``datetime.fromisoformat`` validates calendar/time fields and understands + # RFC 3339 offsets. A timezone-less timestamp is interpreted as UTC for + # backward compatibility with contributor-entered ISO timestamps. + try: + parsed = datetime.fromisoformat(text[:-1] + "+00:00" if text.endswith("Z") else text) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + + +def _format_atom_datetime(value: datetime) -> str: + """Normalize an aware datetime to a UTC RFC 3339 string.""" + return value.astimezone(UTC).isoformat().replace("+00:00", "Z") + + def _atom_timestamp(value: str, fallback: str) -> str: """Coerce a Dublin Core date into an RFC 3339 instant Atom's ```` needs. @@ -148,16 +177,15 @@ def _atom_timestamp(value: str, fallback: str) -> str: requires a complete date-time, so a date-only value is widened to midnight UTC and anything unparseable falls back to ``fallback`` (the feed's generation time, already RFC 3339) rather than emitting an invalid feed.""" - text = value.strip() - if "T" in text: - return text - if re.fullmatch(r"\d{4}", text): - return f"{text}-01-01T00:00:00Z" - if re.fullmatch(r"\d{4}-\d{2}", text): - return f"{text}-01T00:00:00Z" - if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text): - return f"{text}T00:00:00Z" - return fallback + parsed = _parse_atom_datetime(value) + if parsed is None: + parsed = _parse_atom_datetime(fallback) + # Callers supply a valid RFC 3339 generation timestamp. Keep malformed + # programmatic inputs well-formed and deterministic rather than emitting a + # second invalid timestamp if both values are bad. + if parsed is None: + parsed = datetime(1970, 1, 1, tzinfo=UTC) + return _format_atom_datetime(parsed) def atom_feed_xml( diff --git a/tests/test_deposit_bridge.py b/tests/test_deposit_bridge.py index f1748c0..dc6b740 100644 --- a/tests/test_deposit_bridge.py +++ b/tests/test_deposit_bridge.py @@ -11,6 +11,7 @@ from __future__ import annotations import xml.etree.ElementTree as ET +from dataclasses import replace import pytest @@ -30,6 +31,7 @@ _METS_NS = "{http://www.loc.gov/METS/}" _EAD_NS = "{urn:isbn:1-931666-22-9}" +_XLINK_NS = "{http://www.w3.org/1999/xlink}" def _sample_record(*, with_withheld: bool = True) -> DisclosedRecord: @@ -84,6 +86,26 @@ def test_mets_xml_is_well_formed_and_namespaced() -> None: assert root.attrib["OBJID"] == "rec-0000000000000001" +@pytest.mark.preservation +def test_mets_attribute_values_escape_quotes_and_round_trip() -> None: + """Crafted ids/titles/filenames cannot break METS attributes.""" + original = _sample_record(with_withheld=False) + filename = 'scan "one\'s".jpg' + record = replace( + original, + record_id='rec-"quoted\'"', + title='A "quoted" collector\'s title', + payloads=(replace(original.payloads[0], filename=filename),), + ) + + root = ET.fromstring(to_mets_xml(record, created="2026-07-07T00:00:00Z")) # noqa: S314 + assert root.attrib["OBJID"] == record.record_id + assert root.find(f".//{_METS_NS}div").attrib["LABEL"] == record.title + flocat = root.find(f".//{_METS_NS}FLocat") + assert flocat is not None + assert flocat.attrib[f"{_XLINK_NS}title"] == filename + + @pytest.mark.preservation def test_mets_file_section_carries_payload_checksum() -> None: """Each payload becomes one ``mets:file`` with its content-address digest.""" @@ -170,6 +192,23 @@ def test_ead_xml_is_well_formed_and_namespaced() -> None: assert root.tag == f"{_EAD_NS}ead" +@pytest.mark.preservation +def test_ead_component_attribute_escapes_quotes_and_round_trips() -> None: + """A crafted record id remains one EAD attribute value, not injected markup.""" + record = replace(_sample_record(with_withheld=False), record_id='rec-"quoted\'"') + xml = to_ead_xml( + "Quoted collection", + [record], + created="2026-07-07T00:00:00Z", + collection_id="quoted", + ) + + root = ET.fromstring(xml) # noqa: S314 + component = root.find(f".//{_EAD_NS}c01") + assert component is not None + assert component.attrib["id"] == f"c-{record.record_id}" + + @pytest.mark.preservation def test_ead_has_one_component_per_record() -> None: """Each disclosed record becomes exactly one ``c01`` item component.""" diff --git a/tests/test_feed.py b/tests/test_feed.py index 76110f3..83747d6 100644 --- a/tests/test_feed.py +++ b/tests/test_feed.py @@ -88,6 +88,38 @@ def test_feed_coerces_dates_to_rfc3339() -> None: assert feed.count(f"{_NOW}") >= 2 # the feed itself + the undated entry +def test_feed_normalizes_unpadded_and_mixed_granularity_dates() -> None: + """Calendar order wins across bare year, month, and unpadded month/day shapes.""" + records = [ + _disclosed("year", "Year only", date=["2021"]), + _disclosed("may", "Unpadded May", date=["2021-5-1"]), + _disclosed("dec", "December", date=["2021-12"]), + ] + + feed = oai.atom_feed_xml(records, archive_name="A", base_url="http://x.org", now=_NOW) + + assert "2021-05-01T00:00:00Z" in feed + assert "2021-12-01T00:00:00Z" in feed + order = [feed.index(title) for title in ("December", "Unpadded May", "Year only")] + assert order == sorted(order) + + +def test_feed_normalizes_offsets_before_ordering_and_falls_back_on_invalid_dates() -> None: + """Offset instants sort chronologically; impossible/free-text dates use ``now``.""" + records = [ + _disclosed("offset", "Earlier by offset", date=["2021-01-01T00:30:00+02:00"]), + _disclosed("utc", "Later in UTC", date=["2020-12-31T23:00:00Z"]), + _disclosed("bad-calendar", "Bad calendar", date=["2021-13-40"]), + _disclosed("free-text", "Free text", date=["sometime after the march"]), + ] + + feed = oai.atom_feed_xml(records, archive_name="A", base_url="http://x.org", now=_NOW) + + assert "2020-12-31T22:30:00Z" in feed + assert feed.index("Later in UTC") < feed.index("Earlier by offset") + assert feed.count(f"{_NOW}") >= 3 # feed + both invalid values + + def test_feed_escapes_markup_in_titles_and_summaries() -> None: """Angle brackets and ampersands are escaped, so a title cannot break the XML.""" feed = oai.atom_feed_xml( diff --git a/tests/test_identity.py b/tests/test_identity.py index d378316..51baa48 100644 --- a/tests/test_identity.py +++ b/tests/test_identity.py @@ -10,6 +10,7 @@ from __future__ import annotations +import json from pathlib import Path import pytest @@ -315,3 +316,41 @@ def test_stale_handle_rekey_includes_entries_added_by_another_handle(tmp_path: P reopened = IdentityVault.open(path, new_key) assert reopened.resolve(ref, _unseal_grant(ref), _NOW).name == _NAME + + +def test_empty_vault_rejects_the_wrong_key_immediately(tmp_path: Path) -> None: + """The authenticated format marker makes even a zero-entry vault key-bound.""" + path = tmp_path / "identity.vault" + IdentityVault.create(path, IdentityVault.generate_key()) + + with pytest.raises(IdentityVaultError, match="wrong key or tampering"): + IdentityVault.open(path, IdentityVault.generate_key()) + + +def test_stale_old_key_handle_cannot_add_after_empty_vault_rekey(tmp_path: Path) -> None: + """Rotating an empty vault invalidates already-open old-key handles too.""" + path = tmp_path / "identity.vault" + old_key = IdentityVault.generate_key() + new_key = IdentityVault.generate_key() + current = IdentityVault.create(path, old_key) + stale = IdentityVault.open(path, old_key) + + assert current.rekey(new_key) == 0 + with pytest.raises(IdentityVaultError, match="wrong key or tampering"): + stale.add(ContributorIdentity(name="must not persist under the old key")) + + assert len(IdentityVault.open(path, new_key)) == 0 + + +def test_empty_legacy_vault_migrates_to_an_authenticated_marker(tmp_path: Path) -> None: + """A marker-free empty legacy map establishes its supplied key exactly once.""" + path = tmp_path / "identity.vault" + path.write_text("{}", encoding="utf-8") + key = IdentityVault.generate_key() + + migrated = IdentityVault.open(path, key) + assert len(migrated) == 0 + assert len(json.loads(path.read_text(encoding="utf-8"))) == 1 + + with pytest.raises(IdentityVaultError, match="wrong key or tampering"): + IdentityVault.open(path, IdentityVault.generate_key()) diff --git a/tests/test_ingest_e2e.py b/tests/test_ingest_e2e.py index b35ee76..05561f5 100644 --- a/tests/test_ingest_e2e.py +++ b/tests/test_ingest_e2e.py @@ -22,16 +22,21 @@ from pathlib import Path +import pytest + from ledger.access.grants import anonymous, build_grant, steward from ledger.config import Config, StorageLocation -from ledger.errors import AccessDenied -from ledger.identity import ContributorIdentity +from ledger.errors import AccessDenied, BagValidationError, LedgerError +from ledger.identity import ContributorIdentity, IdentityVault from ledger.ingest import Archive, serialize_record from ledger.metadata.premis import PremisLog from ledger.models import ( AccessPolicy, + ContentAddress, DublinCore, Field, + HashAlgo, + PayloadFile, PremisEvent, PremisEventType, Record, @@ -72,6 +77,80 @@ def _build_record(archive_name: str) -> Record: ) +@pytest.mark.parametrize( + "filename", + [r"..\escape.bin", r"C:\escape.bin", r"\rooted.bin"], +) +def test_sealed_payload_rejects_windows_escape_before_any_payload_write( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, filename: str +) -> None: + """Windows traversal/drive/root forms fail before temp or CAS writes.""" + archive = Archive.init(Config.default("Path Safety Archive", tmp_path / "arc")) + archive._open_vault(_VAULT_KEY) + source = tmp_path / "source.bin" + source.write_bytes(b"sensitive payload") + record = Record( + title="Hostile sealed path", + default_policy=AccessPolicy.PUBLIC, + dublin_core=DublinCore(title=["Hostile sealed path"]), + payloads=[ + PayloadFile( + filename=filename, + address=ContentAddress(HashAlgo.SHA256, "0" * 64), + media_type="application/octet-stream", + size_bytes=0, + policy=AccessPolicy.SEALED, + ) + ], + ) + + def unexpected_write(*_args: object, **_kwargs: object) -> None: + raise AssertionError("unsafe path reached a payload write boundary") + + monkeypatch.setattr("ledger.ingest.tempfile.mkdtemp", unexpected_write) + monkeypatch.setattr(archive.store, "put_file", unexpected_write) + + with pytest.raises(BagValidationError, match="unsafe path in sealed payload"): + archive.ingest({filename: source}, record, vault_key=_VAULT_KEY, now=_NOW_INGEST) + + assert not any(archive.bags_dir.iterdir()) + + +def test_record_collision_precedes_field_encryption_and_identity_sealing(tmp_path: Path) -> None: + """A duplicate id leaves caller plaintext and the identity vault untouched.""" + archive = Archive.init(Config.default("Collision Archive", tmp_path / "arc")) + original = Record( + title="Original", + default_policy=AccessPolicy.PUBLIC, + dublin_core=DublinCore(title=["Original"]), + ) + archive.ingest({}, original, now=_NOW_INGEST) + + vault = archive._open_vault(_VAULT_KEY) + before_count = len(vault) + plaintext = "must remain recoverable by the rejected caller" + duplicate = Record( + record_id=original.record_id, + title="Duplicate", + default_policy=AccessPolicy.PUBLIC, + dublin_core=DublinCore(title=["Duplicate"]), + fields=[Field(name="sealed", value=plaintext, policy=AccessPolicy.SEALED)], + ) + + with pytest.raises(LedgerError, match="bag already exists"): + archive.ingest( + {}, + duplicate, + identity=ContributorIdentity(name="not orphaned"), + vault_key=_VAULT_KEY, + now=_NOW_INGEST, + ) + + assert duplicate.fields[0].value == plaintext + assert duplicate.identity_ref is None + assert len(IdentityVault.open(archive.vault_path, _VAULT_KEY)) == before_count + + def test_full_lifecycle_ingest_disclose_replicate_consent(tmp_path: Path) -> None: """Ingest -> disclose -> replicate -> consent-change, asserting events + effects.""" root = tmp_path / "archive" diff --git a/tests/test_lockdown.py b/tests/test_lockdown.py index e17b4ac..70c6536 100644 --- a/tests/test_lockdown.py +++ b/tests/test_lockdown.py @@ -21,8 +21,9 @@ import json import shutil import threading +import time import urllib.request -from collections.abc import Iterator +from collections.abc import Callable, Iterator from contextlib import redirect_stderr, redirect_stdout from io import StringIO from pathlib import Path @@ -30,12 +31,20 @@ import pytest from ledger import cli, dualcontrol +from ledger import lockdown as lockdown_module from ledger.access.grants import build_grant, issue_grant_token, steward from ledger.config import Config from ledger.errors import ConfigError from ledger.identity import ContributorIdentity from ledger.ingest import Archive -from ledger.lockdown import LockdownConfig, is_locked_down, lockdown_flag_path +from ledger.lockdown import ( + LockdownConfig, + LockdownResult, + execute_lockdown, + execute_stand_up, + is_locked_down, + lockdown_flag_path, +) from ledger.metadata.premis import PremisLog from ledger.models import AccessPolicy, DublinCore, Field, PremisEventType, Record from ledger.server import make_server @@ -139,6 +148,106 @@ def _premis_event_types(root: Path) -> list[PremisEventType]: return [e.event_type for e in PremisLog.read(log_path).events] +def test_stand_up_summary_reports_resumed_or_already_open_status() -> None: + """Stand-up never reuses lockdown's misleading stopped/unchanged wording.""" + resumed = LockdownResult( + action="stand-up", + dry_run=False, + disclosure_stopped=True, + vault_shredded=False, + verified_replicas=0, + steps=(), + ) + already_open = LockdownResult( + action="stand-up", + dry_run=False, + disclosure_stopped=False, + vault_shredded=False, + verified_replicas=0, + steps=(), + ) + + assert "disclosure resumed" in resumed.summary() + assert "disclosure was already open" in already_open.summary() + assert "disclosure stopped" not in resumed.summary() + + +def test_concurrent_lockdown_and_stand_up_are_one_serial_transition( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Opposite transitions never overlap; the final flag matches the last audit event.""" + root = tmp_path / "arc" + config = Config.default("Concurrent Duress Archive", root) + config.lockdown = LockdownConfig(stop_disclosure=True, shred_vault=False) + archive = Archive.init(config) + + active = 0 + maximum_active = 0 + guard = threading.Lock() + + real_lockdown = lockdown_module._execute_lockdown_locked + real_stand_up = lockdown_module._execute_stand_up_locked + + def observed( + callable_: Callable[..., LockdownResult], *args: object, **kwargs: object + ) -> LockdownResult: + nonlocal active, maximum_active + with guard: + active += 1 + maximum_active = max(maximum_active, active) + try: + time.sleep(0.03) + return callable_(*args, **kwargs) + finally: + with guard: + active -= 1 + + monkeypatch.setattr( + lockdown_module, + "_execute_lockdown_locked", + lambda *args, **kwargs: observed(real_lockdown, *args, **kwargs), + ) + monkeypatch.setattr( + lockdown_module, + "_execute_stand_up_locked", + lambda *args, **kwargs: observed(real_stand_up, *args, **kwargs), + ) + + start = threading.Barrier(3) + errors: list[BaseException] = [] + + def run(action: str) -> None: + try: + start.wait(timeout=2) + if action == "lockdown": + execute_lockdown(archive, actor="lock", now=_NOW) + else: + execute_stand_up(archive, actor="stand", now=_NOW) + except BaseException as exc: # pragma: no cover - asserted below + errors.append(exc) + + threads = [threading.Thread(target=run, args=(action,)) for action in ("lockdown", "stand")] + for thread in threads: + thread.start() + start.wait(timeout=2) + for thread in threads: + thread.join(timeout=5) + + assert errors == [] + assert maximum_active == 1 + log = PremisLog.read(root / "store" / "logs" / "lockdown.premis.json") + assert len(log.events) == 2 + assert is_locked_down(archive) is (log.events[-1].event_type is PremisEventType.LOCKDOWN) + + # Repeating the terminal action is state-idempotent. + if is_locked_down(archive): + execute_lockdown(archive, actor="lock-again", now=_NOW) + assert is_locked_down(archive) + else: + execute_stand_up(archive, actor="stand-again", now=_NOW) + assert not is_locked_down(archive) + + @pytest.fixture def served( tmp_path: Path, monkeypatch: pytest.MonkeyPatch diff --git a/tests/test_versions.py b/tests/test_versions.py index ff68a3b..daf80a4 100644 --- a/tests/test_versions.py +++ b/tests/test_versions.py @@ -181,3 +181,29 @@ def test_version_index_is_append_only_canonical(tmp_path: Path) -> None: # Append-only: the earlier entry is preserved unchanged and a new one is added. assert second[0] == first[0] assert len(second) == 2 + + +def test_version_index_replace_failure_preserves_previous_complete_json( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failed atomic swap leaves the prior version index byte-for-byte intact.""" + archive, rid = _archive_with_record(tmp_path) + archive.apply_update( + replace(archive.get(rid), title="One"), + _event(rid, PremisEventType.MODERATION, _NOW), + ) + path = archive.records_dir / f"{rid}.versions.json" + before = path.read_bytes() + + def fail_replace(_source: object, _target: object) -> None: + raise OSError("simulated crash before atomic swap") + + monkeypatch.setattr("ledger.bag.os.replace", fail_replace) + with pytest.raises(OSError, match="simulated crash"): + archive._append_version( + rid, + "sha256:" + "f" * 64, + PremisEventType.CONSENT_CHANGE.value, + ) + + assert path.read_bytes() == before From a64143ef4f8daed00d8b7a94afc1fea13bdc6d6e Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 22:41:46 -0700 Subject: [PATCH 8/9] fix: make archival transitions fail closed Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- src/ledger/identity.py | 17 ++++++++++++++-- src/ledger/ingest.py | 22 ++++++++++---------- src/ledger/oai.py | 28 ++++++++++++++------------ tests/test_feed.py | 13 ++++++++++++ tests/test_ingest_e2e.py | 43 +++++++++++++++++++++++++++++++++------- tests/test_rekey.py | 35 ++++++++++++++++++++++++++++++++ 6 files changed, 126 insertions(+), 32 deletions(-) diff --git a/src/ledger/identity.py b/src/ledger/identity.py index f270433..2fcd4fe 100644 --- a/src/ledger/identity.py +++ b/src/ledger/identity.py @@ -319,6 +319,9 @@ def rekey(self, new_key: bytes) -> int: # docstring for why the unlocked temp-file write is unsafe). with file_lock(self._path): self._reload() + old_fernet = self._fernet + old_store = self._store + old_key_check = self._key_check # Re-encrypt into a fresh map first; only commit if every entry succeeds, # so a mid-rotation failure cannot leave a half-rotated vault -> integrity. rotated: dict[str, str] = {} @@ -333,7 +336,16 @@ def rekey(self, new_key: bytes) -> int: self._fernet = new_fernet self._store = rotated self._key_check = new_fernet.encrypt(_KEY_CHECK_PLAINTEXT).decode("ascii") - self._persist() + try: + self._persist() + except BaseException: + # The atomic replace may fail while the old file remains durable. + # Keep this live handle aligned with that old file as well; otherwise + # it would retain an uncommitted key/map until a later disk reload. + self._fernet = old_fernet + self._store = old_store + self._key_check = old_key_check + raise return len(rotated) # --- at-rest encryption of absolute-sealed content ---------------------- @@ -494,8 +506,9 @@ def _persist(self) -> None: fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as handle: handle.write(payload) + # The temp inode was created owner-only; replacing the target preserves + # that mode and leaves no fallible mutation after the commit point. os.replace(tmp, self._path) - os.chmod(self._path, 0o600) except OSError as exc: tmp.unlink(missing_ok=True) raise IdentityVaultError(f"vault could not be written: {self._path}") from exc diff --git a/src/ledger/ingest.py b/src/ledger/ingest.py index 3202a2e..74e5f96 100644 --- a/src/ledger/ingest.py +++ b/src/ledger/ingest.py @@ -367,6 +367,12 @@ def ingest_sip( # noqa: C901 clock, so a golden ingest is byte-reproducible (determinism). """ record = sip.record + bag_dir = bags_dir / record.record_id + if bag_dir.exists(): + # Refuse a duplicate before inspecting, encrypting, or storing any payload. + # A collision is a record-level precondition failure, so it must leave CAS, + # temporary ciphertext, caller-owned fields, and the identity vault untouched. + raise LedgerError(f"bag already exists for record {record.record_id}") # 1. Fixity + store. Preserve any per-file policy already declared on the # record; otherwise default to the record's narrowest policy. @@ -493,16 +499,7 @@ def ingest_sip( # noqa: C901 if pid not in record.dublin_core.identifier: record.dublin_core.identifier = [*record.dublin_core.identifier, pid] - # 2. Refuse a collision BEFORE sealing any identity OR field, so a failed - # ingest cannot leave an orphaned, unreachable identity in the vault, or - # silently overwrite the caller's in-memory sealed-field plaintext with - # ciphertext it has no way to recover (#correctness, consent). - bag_dir = bags_dir / record.record_id - if bag_dir.exists(): - # An item is bagged exactly once; refuse to clobber a prior AIP silently. - raise LedgerError(f"bag already exists for record {record.record_id}") - - # Absolute-SEALED fields are encrypted AT REST so a stolen disk or hostile + # 2. Absolute-SEALED fields are encrypted AT REST so a stolen disk or hostile # replica reveals nothing, not even to a steward (user research P2-4). Such a # field is never disclosed on any read path, so it is only ever encrypted, never # decrypted here. It requires the vault, like an identity. @@ -706,6 +703,11 @@ def ingest( of the stored record manifest is also written under ``records/`` for fast lookup by :meth:`get` without unpacking a bag (efficiency). """ + if (self.bags_dir / record.record_id).exists(): + # Preflight before even opening/creating the vault. ``ingest_sip`` + # repeats this guard for direct callers, but the facade owns this + # earlier boundary because vault creation is itself a durable effect. + raise LedgerError(f"bag already exists for record {record.record_id}") stamp = now if now is not None else now_iso() self.bags_dir.mkdir(parents=True, exist_ok=True) self.records_dir.mkdir(parents=True, exist_ok=True) diff --git a/src/ledger/oai.py b/src/ledger/oai.py index c269b2b..28f50c0 100644 --- a/src/ledger/oai.py +++ b/src/ledger/oai.py @@ -169,6 +169,16 @@ def _format_atom_datetime(value: datetime) -> str: return value.astimezone(UTC).isoformat().replace("+00:00", "Z") +def _atom_datetime(value: str, fallback: str) -> datetime: + """Coerce a Dublin Core date into the aware instant Atom ordering needs.""" + parsed = _parse_atom_datetime(value) + if parsed is None: + parsed = _parse_atom_datetime(fallback) + # Callers supply a valid RFC 3339 generation timestamp. Keep malformed + # programmatic inputs deterministic if both values are bad. + return parsed if parsed is not None else datetime(1970, 1, 1, tzinfo=UTC) + + def _atom_timestamp(value: str, fallback: str) -> str: """Coerce a Dublin Core date into an RFC 3339 instant Atom's ```` needs. @@ -177,15 +187,7 @@ def _atom_timestamp(value: str, fallback: str) -> str: requires a complete date-time, so a date-only value is widened to midnight UTC and anything unparseable falls back to ``fallback`` (the feed's generation time, already RFC 3339) rather than emitting an invalid feed.""" - parsed = _parse_atom_datetime(value) - if parsed is None: - parsed = _parse_atom_datetime(fallback) - # Callers supply a valid RFC 3339 generation timestamp. Keep malformed - # programmatic inputs well-formed and deterministic rather than emitting a - # second invalid timestamp if both values are bad. - if parsed is None: - parsed = datetime(1970, 1, 1, tzinfo=UTC) - return _format_atom_datetime(parsed) + return _format_atom_datetime(_atom_datetime(value, fallback)) def atom_feed_xml( @@ -210,18 +212,18 @@ def atom_feed_xml( through the shared :func:`escape` boundary. ``now`` is the feed's generation time and the fallback timestamp, so output is deterministic for a given input. - The sort key is the same RFC 3339 instant :func:`_atom_timestamp` widens each + The sort key is the same aware instant :func:`_atom_datetime` widens each entry's date to for display (not the raw ``dc:date`` string): a ``dc:date`` is free text of varying granularity and padding (``"1994"``, ``"2021-5-1"``, ``"2021-12-01"``, ...), and comparing those as plain strings is lexicographic, not chronological — e.g. ``"2021-5-1" > "2021-12-01"`` as strings even though May - precedes December. Normalizing both to the same padded RFC 3339 form before - comparing keeps "most recent first" true for any date shape a contributor used, + precedes December. Parsing both as aware UTC datetimes before comparing keeps + "most recent first" true for any date shape or fractional precision in use, and keeps the sort key consistent with what ```` actually displays.""" root = base_url.rstrip("/") ordered = sorted( records, - key=lambda r: (_atom_timestamp(_datestamp(r, now), now), r.record_id), + key=lambda r: (_atom_datetime(_datestamp(r, now), now), r.record_id), reverse=True, )[: max(0, limit)] diff --git a/tests/test_feed.py b/tests/test_feed.py index 83747d6..1c72a61 100644 --- a/tests/test_feed.py +++ b/tests/test_feed.py @@ -120,6 +120,19 @@ def test_feed_normalizes_offsets_before_ordering_and_falls_back_on_invalid_dates assert feed.count(f"{_NOW}") >= 3 # feed + both invalid values +def test_feed_orders_fractional_instants_after_the_same_whole_second() -> None: + """Datetime order wins where RFC 3339 string order puts ``Z`` before a fraction.""" + records = [ + _disclosed("whole", "Whole second", date=["2021-01-01T00:00:00Z"]), + _disclosed("fraction", "Fraction later", date=["2021-01-01T00:00:00.900000Z"]), + ] + + feed = oai.atom_feed_xml(records, archive_name="A", base_url="http://x.org", now=_NOW) + + assert feed.index("Fraction later") < feed.index("Whole second") + assert "2021-01-01T00:00:00.900000Z" in feed + + def test_feed_escapes_markup_in_titles_and_summaries() -> None: """Angle brackets and ampersands are escaped, so a title cannot break the XML.""" feed = oai.atom_feed_xml( diff --git a/tests/test_ingest_e2e.py b/tests/test_ingest_e2e.py index 05561f5..29a4a46 100644 --- a/tests/test_ingest_e2e.py +++ b/tests/test_ingest_e2e.py @@ -27,7 +27,7 @@ from ledger.access.grants import anonymous, build_grant, steward from ledger.config import Config, StorageLocation from ledger.errors import AccessDenied, BagValidationError, LedgerError -from ledger.identity import ContributorIdentity, IdentityVault +from ledger.identity import ContributorIdentity from ledger.ingest import Archive, serialize_record from ledger.metadata.premis import PremisLog from ledger.models import ( @@ -116,8 +116,10 @@ def unexpected_write(*_args: object, **_kwargs: object) -> None: assert not any(archive.bags_dir.iterdir()) -def test_record_collision_precedes_field_encryption_and_identity_sealing(tmp_path: Path) -> None: - """A duplicate id leaves caller plaintext and the identity vault untouched.""" +def test_record_collision_precedes_all_sealed_payload_and_identity_side_effects( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A duplicate id leaves payload temp/CAS, caller plaintext, and vault untouched.""" archive = Archive.init(Config.default("Collision Archive", tmp_path / "arc")) original = Record( title="Original", @@ -126,20 +128,42 @@ def test_record_collision_precedes_field_encryption_and_identity_sealing(tmp_pat ) archive.ingest({}, original, now=_NOW_INGEST) - vault = archive._open_vault(_VAULT_KEY) - before_count = len(vault) + assert not archive.vault_path.exists() + before_tree = { + path.relative_to(archive.store_root): None if path.is_dir() else path.read_bytes() + for path in sorted(archive.store_root.rglob("*")) + } plaintext = "must remain recoverable by the rejected caller" + source = tmp_path / "sealed-source.bin" + source.write_bytes(b"must never be encrypted or copied on collision") duplicate = Record( record_id=original.record_id, title="Duplicate", default_policy=AccessPolicy.PUBLIC, dublin_core=DublinCore(title=["Duplicate"]), fields=[Field(name="sealed", value=plaintext, policy=AccessPolicy.SEALED)], + payloads=[ + PayloadFile( + filename=source.name, + address=ContentAddress(HashAlgo.SHA256, "0" * 64), + media_type="application/octet-stream", + size_bytes=0, + policy=AccessPolicy.SEALED, + ) + ], ) + def unexpected_side_effect(*_args: object, **_kwargs: object) -> None: + raise AssertionError("collision reached sealed-payload processing") + + monkeypatch.setattr(archive, "_open_vault", unexpected_side_effect) + monkeypatch.setattr("ledger.ingest.identify_file", unexpected_side_effect) + monkeypatch.setattr("ledger.ingest.tempfile.mkdtemp", unexpected_side_effect) + monkeypatch.setattr(archive.store, "put_file", unexpected_side_effect) + with pytest.raises(LedgerError, match="bag already exists"): archive.ingest( - {}, + {source.name: source}, duplicate, identity=ContributorIdentity(name="not orphaned"), vault_key=_VAULT_KEY, @@ -148,7 +172,12 @@ def test_record_collision_precedes_field_encryption_and_identity_sealing(tmp_pat assert duplicate.fields[0].value == plaintext assert duplicate.identity_ref is None - assert len(IdentityVault.open(archive.vault_path, _VAULT_KEY)) == before_count + assert not archive.vault_path.exists() + after_tree = { + path.relative_to(archive.store_root): None if path.is_dir() else path.read_bytes() + for path in sorted(archive.store_root.rglob("*")) + } + assert after_tree == before_tree def test_full_lifecycle_ingest_disclose_replicate_consent(tmp_path: Path) -> None: diff --git a/tests/test_rekey.py b/tests/test_rekey.py index cce2f36..a88b0b1 100644 --- a/tests/test_rekey.py +++ b/tests/test_rekey.py @@ -63,6 +63,41 @@ def test_vault_rekey_is_atomic_on_a_bad_new_key(tmp_path: Path) -> None: assert reopened.resolve(ref, grant, _NOW).name == _SENTINEL +@pytest.mark.disclosure +def test_vault_rekey_replace_failure_restores_live_handle_and_disk( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failed atomic replace keeps both disk and the current handle on the old key.""" + path = tmp_path / "identity.vault" + old = IdentityVault.generate_key() + new = IdentityVault.generate_key() + vault = IdentityVault.create(path, old) + ref = vault.add(ContributorIdentity(name=_SENTINEL)) + old_disk = path.read_bytes() + old_fernet = vault._fernet + old_store = dict(vault._store) + old_key_check = vault._key_check + old_token = vault.encrypt_text("still decryptable after rollback") + + def fail_replace(*_args: object, **_kwargs: object) -> None: + raise OSError("injected replace failure") + + with monkeypatch.context() as patch: + patch.setattr("ledger.identity.os.replace", fail_replace) + with pytest.raises(IdentityVaultError, match="vault could not be written"): + vault.rekey(new) + + assert path.read_bytes() == old_disk + assert vault._fernet is old_fernet + assert vault._store == old_store + assert vault._key_check == old_key_check + assert vault.decrypt_text(old_token) == "still decryptable after rollback" + grant = build_grant("unsealer", identity_unseal=[ref]) + assert IdentityVault.open(path, old).resolve(ref, grant, _NOW).name == _SENTINEL + with pytest.raises(IdentityVaultError): + IdentityVault.open(path, new) + + def _ingest_identity_only(archive: Archive, payload: Path) -> Record: """Ingest one record whose only sensitive content is a sealed identity.""" record = Record( From 8a4e287b1b56353a1ab957a939e1a7e3d6365cae Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sun, 12 Jul 2026 22:45:59 -0700 Subject: [PATCH 9/9] test: separate CodeQL-visible assertions Signed-off-by: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> --- tests/test_identity.py | 3 ++- tests/test_lockdown.py | 4 ++-- tests/test_rekey.py | 4 +++- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/test_identity.py b/tests/test_identity.py index 51baa48..8f7272a 100644 --- a/tests/test_identity.py +++ b/tests/test_identity.py @@ -339,7 +339,8 @@ def test_stale_old_key_handle_cannot_add_after_empty_vault_rekey(tmp_path: Path) with pytest.raises(IdentityVaultError, match="wrong key or tampering"): stale.add(ContributorIdentity(name="must not persist under the old key")) - assert len(IdentityVault.open(path, new_key)) == 0 + reopened = IdentityVault.open(path, new_key) + assert len(reopened) == 0 def test_empty_legacy_vault_migrates_to_an_authenticated_marker(tmp_path: Path) -> None: diff --git a/tests/test_lockdown.py b/tests/test_lockdown.py index 70c6536..05c565c 100644 --- a/tests/test_lockdown.py +++ b/tests/test_lockdown.py @@ -214,7 +214,7 @@ def observed( ) start = threading.Barrier(3) - errors: list[BaseException] = [] + errors: list[Exception] = [] def run(action: str) -> None: try: @@ -223,7 +223,7 @@ def run(action: str) -> None: execute_lockdown(archive, actor="lock", now=_NOW) else: execute_stand_up(archive, actor="stand", now=_NOW) - except BaseException as exc: # pragma: no cover - asserted below + except Exception as exc: # pragma: no cover - asserted below errors.append(exc) threads = [threading.Thread(target=run, args=(action,)) for action in ("lockdown", "stand")] diff --git a/tests/test_rekey.py b/tests/test_rekey.py index a88b0b1..3f297d8 100644 --- a/tests/test_rekey.py +++ b/tests/test_rekey.py @@ -93,7 +93,9 @@ def fail_replace(*_args: object, **_kwargs: object) -> None: assert vault._key_check == old_key_check assert vault.decrypt_text(old_token) == "still decryptable after rollback" grant = build_grant("unsealer", identity_unseal=[ref]) - assert IdentityVault.open(path, old).resolve(ref, grant, _NOW).name == _SENTINEL + reopened = IdentityVault.open(path, old) + resolved = reopened.resolve(ref, grant, _NOW) + assert resolved.name == _SENTINEL with pytest.raises(IdentityVaultError): IdentityVault.open(path, new)