diff --git a/src/ledger/attest.py b/src/ledger/attest.py index 3f4a1bc..927351d 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 @@ -51,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]: @@ -134,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 -------------------------------------------------------------- @@ -157,15 +164,24 @@ 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 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): + 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, @@ -181,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/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 bd6fe27..2fcd4fe 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 @@ -71,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: @@ -139,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.""" @@ -157,10 +168,15 @@ 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._key_check = vault._new_key_check() + vault._persist() return vault @classmethod @@ -174,24 +190,13 @@ 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}") + if vault._reload(): + vault._persist() return vault # --- mutation ----------------------------------------------------------- @@ -202,12 +207,30 @@ 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). + + 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. """ - ref = self._new_ref() - token = self._encrypt(identity) - self._store[ref] = token - self._persist() - return ref + 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() + return ref def resolve(self, ref: str, grant: Grant, now: str) -> ContributorIdentity: """Decrypt and return the identity for *ref*, gated by *grant* at *now*. @@ -229,28 +252,42 @@ 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. 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): + 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 ------------------------------------------------------- @@ -277,21 +314,39 @@ 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(): + # 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): + 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] = {} + 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._key_check = new_fernet.encrypt(_KEY_CHECK_PLAINTEXT).decode("ascii") 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) + 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 ---------------------- @@ -359,6 +414,53 @@ def derive_key(passphrase: str, salt: bytes) -> bytes: # --- internals ---------------------------------------------------------- + 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")) + 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] = {} + 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 + 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: """Generate a fresh opaque ref from a CSPRNG -> unlinkability.""" @@ -369,6 +471,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. @@ -378,9 +484,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). @@ -390,7 +496,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 @@ -398,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 bd16a6a..74e5f96 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 @@ -361,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. @@ -404,6 +416,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,7 +499,7 @@ def ingest_sip( # noqa: C901 if pid not in record.dublin_core.identifier: record.dublin_core.identifier = [*record.dublin_core.identifier, pid] - # 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. @@ -493,13 +512,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 @@ -691,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) @@ -868,7 +885,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]]: diff --git a/src/ledger/lockdown.py b/src/ledger/lockdown.py index ece8c90..74e549f 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 @@ -61,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). @@ -187,7 +192,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": @@ -200,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). @@ -235,20 +251,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]: @@ -341,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 @@ -428,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 @@ -468,7 +506,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 +518,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), 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/" diff --git a/src/ledger/oai.py b/src/ledger/oai.py index 9e045de..28f50c0 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,44 @@ 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_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. @@ -148,16 +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.""" - 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 + return _format_atom_datetime(_atom_datetime(value, fallback)) def atom_feed_xml( @@ -180,11 +210,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 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. 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: (_datestamp(r, 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_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_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..1c72a61 100644 --- a/tests/test_feed.py +++ b/tests/test_feed.py @@ -88,6 +88,51 @@ 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_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_identity.py b/tests/test_identity.py index 0e8e2c8..8f7272a 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 @@ -254,3 +255,103 @@ 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 + + +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")) + + reopened = IdentityVault.open(path, new_key) + assert len(reopened) == 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..29a4a46 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.errors import AccessDenied, BagValidationError, LedgerError from ledger.identity import ContributorIdentity 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,109 @@ 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_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", + default_policy=AccessPolicy.PUBLIC, + dublin_core=DublinCore(title=["Original"]), + ) + archive.ingest({}, original, now=_NOW_INGEST) + + 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, + now=_NOW_INGEST, + ) + + assert duplicate.fields[0].value == plaintext + assert duplicate.identity_ref is None + 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: """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..05c565c 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[Exception] = [] + + 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 Exception 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_rekey.py b/tests/test_rekey.py index cce2f36..3f297d8 100644 --- a/tests/test_rekey.py +++ b/tests/test_rekey.py @@ -63,6 +63,43 @@ 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]) + reopened = IdentityVault.open(path, old) + resolved = reopened.resolve(ref, grant, _NOW) + assert resolved.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( 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