Complete default-branch CodeQL remediation

src/ledger/access/policy.py

@@ -83,46 +83,23 @@ def is_visible(
     """
     effective = PUBLIC_GRANT if grant.is_expired(now) else grant
 
-    match policy:
-        case AccessPolicy.PUBLIC:
-            return True
-        case AccessPolicy.COMMUNITY:
-            return effective.is_steward or AccessPolicy.COMMUNITY in effective.levels
-        case AccessPolicy.STEWARDS:
-            return effective.is_steward
-        case AccessPolicy.SEALED_UNTIL:
-            if unseal_at is not None:
-                # A *temporal* seal ("sealed until <date>") is an embargo: a promise
-                # made to time, not an access level. It binds EVERY tier, including
-                # stewards, until the date passes — then it opens to all. A steward
-                # who must reach embargoed content before the date does so through an
-                # explicit, logged mechanism, not by silently bypassing the seal
-                #
-                # NOTE: this is a policy check on plaintext, not an encryption
-                # boundary — a seized disk yields the value immediately regardless
-                # of how far away unseal_at is (unlike absolute SEALED, which is
-                # encrypted at rest; see ingest.py). Whether a genuine
-                # cryptographic time-lock could close that gap is explored,
-                # research-first and not yet a build decision, in
-                # docs/audits/crypto-design-review-embargo-timelock.md (EXP-12).
-                # (fail-closed; honours the date the contributor was promised).
-                return _unseal_reached(now, unseal_at)
-            # An indefinite seal (no date) is an access-level seal a steward may
-            # read, as the threat model documents.
-            return effective.is_steward
-        case AccessPolicy.SEALED_CONDITIONAL:
-            if effective.is_steward:
-                return True
-            return unseal_condition is not None and unseal_condition in conditions_met
-        case AccessPolicy.SEALED:
-            # An ABSOLUTE seal: restricted from everyone, including stewards. No
-            # grant satisfies it — there is no read path on which it is disclosed
-            # (the "seal from everyone" tier; such values are encrypted at rest).
-            return False
-        case _:
-            # Unknown enum-like values are never visible. This explicit fallback
-            # preserves the deny-by-default contract for malformed runtime input.
-            return False  # type: ignore[unreachable]
+    if policy is AccessPolicy.PUBLIC:
+        return True
+    if policy is AccessPolicy.COMMUNITY:
+        return effective.is_steward or AccessPolicy.COMMUNITY in effective.levels
+    if policy is AccessPolicy.STEWARDS:
+        return effective.is_steward
+    if policy is AccessPolicy.SEALED_UNTIL:
+        if unseal_at is not None:
+            # A temporal seal is a promise to time and binds every viewer tier.
+            return _unseal_reached(now, unseal_at)
+        return effective.is_steward
+    if policy is AccessPolicy.SEALED_CONDITIONAL:
+        return effective.is_steward or (
+            unseal_condition is not None and unseal_condition in conditions_met
+        )
+    # SEALED and malformed runtime values are both denied by default.
+    return False
 
 
 def is_listable(
@@ -263,19 +240,17 @@ def withheld_reason(policy: AccessPolicy, unseal_at: str | None, *, now: str | N
     a dated temporal seal, a live countdown ("opens in N days") is appended so the
     embargo is an honest promise to a time, not just a label (C2).
     """
-    match policy:
-        case AccessPolicy.COMMUNITY:
-            return "shared with community members"
-        case AccessPolicy.STEWARDS:
-            return "restricted to stewards"
-        case AccessPolicy.SEALED_UNTIL:
-            if unseal_at:
-                countdown = _embargo_countdown(now, unseal_at) if now else ""
-                return f"sealed until {unseal_at[:10]}{countdown}"
-            return "sealed (no opening date set)"
-        case AccessPolicy.SEALED_CONDITIONAL:
-            return "sealed until a condition is met"
-        case AccessPolicy.SEALED:
-            return "sealed from everyone, including stewards"
-        case _:
-            return "restricted"
+    if policy is AccessPolicy.COMMUNITY:
+        return "shared with community members"
+    if policy is AccessPolicy.STEWARDS:
+        return "restricted to stewards"
+    if policy is AccessPolicy.SEALED_UNTIL:
+        if unseal_at:
+            countdown = _embargo_countdown(now, unseal_at) if now else ""
+            return f"sealed until {unseal_at[:10]}{countdown}"
+        return "sealed (no opening date set)"
+    if policy is AccessPolicy.SEALED_CONDITIONAL:
+        return "sealed until a condition is met"
+    if policy is AccessPolicy.SEALED:
+        return "sealed from everyone, including stewards"
+    return "restricted"

src/ledger/acr_gen.py

@@ -631,19 +631,19 @@ def render() -> str:
         "### Evidence basis",
         "",
         "This report rests on two committed, recurring sources of evidence — neither "
-        "adds a runtime dependency, both run against the same canonical pages:",
+        + "adds a runtime dependency, both run against the same canonical pages:",
         "",
         "- **Automated.** The stdlib static gate "
-        "(`python -m ledger.accessibility_check web`) runs on every commit, and a "
-        "browser-real **axe-core** job (the `accessibility-browser` CI job) drives "
-        "the served site in a headless Chromium under both the light and dark colour "
-        "schemes, asserting no WCAG-tagged axe violations.",
+        + "(`python -m ledger.accessibility_check web`) runs on every commit, and a "
+        + "browser-real **axe-core** job (the `accessibility-browser` CI job) drives "
+        + "the served site in a headless Chromium under both the light and dark colour "
+        + "schemes, asserting no WCAG-tagged axe violations.",
         "- **Manual.** A committed quarterly (and pre-release) **NVDA and VoiceOver** "
-        "review covers what no scan can judge — reading order, content-warning "
-        "announcement, `aria-live` status, and spoken form errors. Its cadence, "
-        "checklist, and results log live in "
-        "[`MANUAL-REVIEW-CADENCE.md`](./MANUAL-REVIEW-CADENCE.md); manual findings "
-        "are reflected back into the remarks below.",
+        + "review covers what no scan can judge — reading order, content-warning "
+        + "announcement, `aria-live` status, and spoken form errors. Its cadence, "
+        + "checklist, and results log live in "
+        + "[`MANUAL-REVIEW-CADENCE.md`](./MANUAL-REVIEW-CADENCE.md); manual findings "
+        + "are reflected back into the remarks below.",
         "",
         "## Tables",
         "",

src/ledger/captions.py

@@ -96,14 +96,38 @@
 # (an optional dot-separated class list before the annotation). The annotation —
 # everything up to the closing '>' — is the voice name (WebVTT spec §"WebVTT cue
 # voice span").
-_VOICE_TAG = re.compile(r"<v(?:\.[^>.\s]+)*[ \t]+([^>]+)>")
-
 # Any WebVTT cue-span tag: <v ...>, </v>, <b>, <i>, <u>, <c>, <lang ...>, or a
 # cue-internal timestamp tag like <00:00:01.000>. Stripped from cue payload text
 # after the voice (if any) has been pulled out, so the stored text is plain prose,
 # not markup.
 _ANY_TAG = re.compile(r"<[^>]*>")
 
+
+def _voice_annotation(raw: str) -> str | None:
+    """Return the first valid ``<v[.class] annotation>`` annotation, linearly."""
+    offset = 0
+    while (start := raw.find("<v", offset)) >= 0:
+        cursor = start + 2
+        offset = cursor
+        while cursor < len(raw) and raw[cursor] == ".":
+            cursor += 1
+            class_start = cursor
+            while cursor < len(raw) and raw[cursor] not in ".> \t\r\n":
+                cursor += 1
+            if cursor == class_start:
+                break
+        if cursor >= len(raw) or raw[cursor] not in " \t":
+            continue
+        while cursor < len(raw) and raw[cursor] in " \t":
+            cursor += 1
+        end = raw.find(">", cursor)
+        if end >= 0:
+            annotation = raw[cursor:end].strip()
+            if annotation:
+                return annotation
+    return None
+
+
 # WebVTT timestamp (spec: hours optional-but-required-if-nonzero, minutes/seconds
 # always 2 digits 00-59, milliseconds always 3 digits, '.' before the milliseconds).
 _WEBVTT_TS = re.compile(r"^(?:(\d{2,}):)?([0-5]\d):([0-5]\d)\.(\d{3})$")
@@ -179,8 +203,7 @@ def _extract_voice(payload_lines: list[str]) -> tuple[str | None, str]:
     plain prose.
     """
     raw = "\n".join(payload_lines).strip()
-    match = _VOICE_TAG.search(raw)
-    speaker = match.group(1).strip() if match else None
+    speaker = _voice_annotation(raw)
     cleaned = _ANY_TAG.sub("", raw)
     text = " ".join(cleaned.split())
     return speaker, text

src/ledger/checkup.py

@@ -150,14 +150,14 @@ def to_markdown(self) -> str:
             CheckStatus.UNVERIFIED: "CHECK",
         }
         intro = " ".join(
-            [
+            (
                 "This report checks a live deployment against the operational controls in",
                 "`docs/ADOPTING.md`. It is advisory: it changes nothing and sets no defaults,",
                 "and it records only operational facts (paths, device ids, counts, a bound",
                 "host) — never a contributor identity, a sealed value, or the vault key",
                 "(no-outing rule). A `CHECK` result is a control this tool could not verify",
                 "from the box it ran on; confirm it by hand rather than assuming it is fine.",
-            ]
+            )
         )
         lines = [
             f"# ledger readiness checkup — {self.generated_date}",
@@ -175,13 +175,13 @@ def to_markdown(self) -> str:
             cell = " ".join(r.explanation.split())
             lines.append(f"| {r.title} | {symbol[r.status]} | {cell} |")
         closing = " ".join(
-            [
+            (
                 "Most residual risk in the threat model is reduced by these operational",
                 "choices. ledger holds the line it can hold in code; this checklist is the",
                 "line you hold in deployment. Consult `docs/ADOPTING.md` and",
                 "`docs/THREAT-MODEL.md` in full before trusting the system with records",
                 "that can endanger the people who made them.",
-            ]
+            )
         )
         lines.extend(
             [

src/ledger/config.py

@@ -30,7 +30,6 @@
 from pathlib import Path
 
 from ledger.errors import ConfigError
-from ledger.lockdown import LockdownConfig
 from ledger.models import AccessPolicy
 
 # Current on-disk schema version. Bumped whenever the serialized shape changes; the
@@ -41,6 +40,68 @@
 # ``mirror`` is a replica target (replication/redundancy).
 _KNOWN_KINDS: frozenset[str] = frozenset({"local", "mirror"})
 
+
+@dataclass(frozen=True)
+class LockdownConfig:
+    """Declarative duress posture, kept with configuration to avoid import cycles."""
+
+    stop_disclosure: bool = True
+    shred_vault: bool = False
+    required_replica_locations: list[str] = field(default_factory=list)
+    min_verified_replicas: int = 1
+
+    def validate(self, *, archive_locations: tuple[str | Path, ...] = ()) -> None:
+        """Reject a destructive or self-referential lockdown configuration."""
+        if self.min_verified_replicas < 1:
+            raise ConfigError("lockdown.min_verified_replicas must be at least 1")
+        if self.shred_vault:
+            if not self.required_replica_locations:
+                raise ConfigError(
+                    "lockdown.shred_vault requires at least one required_replica_locations "
+                    "entry to verify before destroying the local vault"
+                )
+            if self.min_verified_replicas > len(self.required_replica_locations):
+                raise ConfigError(
+                    "lockdown.min_verified_replicas exceeds the number of configured "
+                    "required_replica_locations; the shred could never be authorized"
+                )
+        if archive_locations and self.required_replica_locations:
+            live = {Path(loc).expanduser().resolve() for loc in archive_locations}
+            for location in self.required_replica_locations:
+                if Path(location).expanduser().resolve() in live:
+                    raise ConfigError(
+                        f"lockdown.required_replica_locations entry {location!r} is the "
+                        "live archive's own location, not an off-box replica — this "
+                        "provides no real redundancy; point it at a genuinely separate copy"
+                    )
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "stop_disclosure": self.stop_disclosure,
+            "shred_vault": self.shred_vault,
+            "required_replica_locations": list(self.required_replica_locations),
+            "min_verified_replicas": self.min_verified_replicas,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, object]) -> LockdownConfig:
+        raw_locations = data.get("required_replica_locations", [])
+        if not isinstance(raw_locations, list):
+            raise ConfigError("lockdown.required_replica_locations must be a list")
+        try:
+            min_verified = int(str(data.get("min_verified_replicas", 1)))
+        except ValueError as exc:
+            raise ConfigError("lockdown.min_verified_replicas must be an integer") from exc
+        config = cls(
+            stop_disclosure=bool(data.get("stop_disclosure", True)),
+            shred_vault=bool(data.get("shred_vault", False)),
+            required_replica_locations=[str(loc) for loc in raw_locations],
+            min_verified_replicas=min_verified,
+        )
+        config.validate()
+        return config
+
+
 # A small, opinionated starter vocabulary for content warnings. It is intentionally
 # editable: stewards extend it for their community, but a fresh archive is never
 # left with an empty controlled vocabulary (usability, care for readers).

src/ledger/ingest.py

@@ -738,7 +738,7 @@ def attested_conditions(self) -> frozenset[str]:
 
     def _record_path(self, record_id: str) -> Path:
         """The fast-lookup manifest path for ``record_id`` under ``records/``."""
-        return self.records_dir / f"{record_id}.json"
+        return self.records_dir / f"{_safe_record_component(record_id)}.json"
 
     def get(self, record_id: str) -> Record:
         """Load the stored, identity-free record manifest for ``record_id``.
@@ -751,14 +751,15 @@ def get(self, record_id: str) -> Record:
         fast = self._record_path(record_id)
         if fast.exists():
             return deserialize_record(fast.read_text(encoding="utf-8"))
-        in_bag = self.bags_dir / record_id / _RECORD_FILENAME
+        component = _safe_record_component(record_id)
+        in_bag = self.bags_dir / component / _RECORD_FILENAME
         if in_bag.exists():
             return deserialize_record(in_bag.read_text(encoding="utf-8"))
         raise ObjectNotFound(record_id)
 
     def _versions_path(self, record_id: str) -> Path:
         """The append-only version-index path for ``record_id`` under ``records/``."""
-        return self.records_dir / f"{record_id}{_VERSIONS_SUFFIX}"
+        return self.records_dir / f"{_safe_record_component(record_id)}{_VERSIONS_SUFFIX}"
 
     def apply_update(self, record: Record, event: PremisEvent) -> None:
         """Persist an updated record manifest and append a PREMIS event to its bag.
@@ -996,37 +997,32 @@ def remove_all_copies(self, record_id: str, *, now: str | None = None) -> tuple[
         turn a removal into a directory traversal that deletes outside the archive
         (defense in depth on a destructive primitive).
         """
-        if (
-            not record_id
-            or record_id in {".", ".."}
-            or any(sep in record_id for sep in ("/", "\\", "\x00"))
-        ):
-            raise LedgerError("invalid record id")
+        component = _safe_record_component(record_id)
         stamp = now if now is not None else now_iso()
-        revoked = self._revoke_identity_if_present(record_id)
+        revoked = self._revoke_identity_if_present(component)
 
         removed = 0
         cleared_locations: list[str] = []
-        bag_dir = self.bags_dir / record_id
+        bag_dir = self.bags_dir / component
         if bag_dir.exists():
             shutil.rmtree(bag_dir)
             removed += 1
-        fast = self.records_dir / f"{record_id}.json"
+        fast = self.records_dir / f"{component}.json"
         if fast.exists():
             fast.unlink()
         # Remove the append-only version index too, so a takedown leaves no dangling
         # history pointer. The snapshot bytes it referenced are content-addressed and
         # identity-free; they are left to normal store maintenance rather than chased
         # here, keeping this destructive primitive simple.
-        versions = self._versions_path(record_id)
+        versions = self._versions_path(component)
         if versions.exists():
             versions.unlink()
         # The catalog index (FIX-04) notices the fast-lookup file is gone the next
         # time it self-syncs and drops the cached row then -- no explicit call
         # needed here (see ledger.catalog_index), so a removed record cannot linger
         # in browse/search results.
         for location in self.config.locations:
-            replica = Path(location.path) / record_id
+            replica = Path(location.path) / component
             if replica.exists() and replica != bag_dir:
                 shutil.rmtree(replica)
                 removed += 1
@@ -1037,10 +1033,10 @@ def remove_all_copies(self, record_id: str, *, now: str | None = None) -> tuple[
         # reachable replica that held a copy is confirmed too. Any replica offline
         # now is left pending, to be confirmed by the reattach sweep (propagation).
         store = TombstoneStore(self.logs_dir)
-        store.add(record_id, stamp)
-        store.confirm(record_id, PRIMARY_LOCATION, stamp)
+        store.add(component, stamp)
+        store.confirm(component, PRIMARY_LOCATION, stamp)
         for name in cleared_locations:
-            store.confirm(record_id, name, stamp)
+            store.confirm(component, name, stamp)
         return removed, revoked
 
     def disclose(
@@ -1455,3 +1451,15 @@ def _vault_key_from_env() -> bytes | None:
     """
     raw = os.environ.get(_VAULT_KEY_ENV)
     return raw.encode("ascii") if raw else None
+
+
+def _safe_record_component(record_id: str) -> str:
+    """Return a single allowlisted path component or reject the identifier."""
+    component = Path(record_id).name
+    if (
+        component != record_id
+        or component in {"", ".", ".."}
+        or re.fullmatch(r"[A-Za-z0-9_-]+", component) is None
+    ):
+        raise LedgerError("invalid record id")
+    return component