From a6bab6862d90fd56b4544cfc6a250605c6b98e0e Mon Sep 17 00:00:00 2001 From: Chelsea Kelly-Reif <3114598+ChelseaKR@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:12:40 -0700 Subject: [PATCH] fix: remediate complete default-branch CodeQL scan --- src/ledger/access/policy.py | 87 +++++++++++-------------------- src/ledger/acr_gen.py | 20 ++++---- src/ledger/captions.py | 31 +++++++++-- src/ledger/checkup.py | 8 +-- src/ledger/config.py | 63 ++++++++++++++++++++++- src/ledger/ingest.py | 42 +++++++++------ src/ledger/lockdown.py | 100 +++--------------------------------- src/ledger/metadata/ead.py | 14 +++-- src/ledger/metadata/mets.py | 14 +++-- src/ledger/server.py | 10 ++-- tests/test_grant_auth.py | 11 ++-- tests/test_metadata.py | 4 -- 12 files changed, 195 insertions(+), 209 deletions(-) diff --git a/src/ledger/access/policy.py b/src/ledger/access/policy.py index a7a3e7e..ce0306f 100644 --- a/src/ledger/access/policy.py +++ b/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 ") 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" diff --git a/src/ledger/acr_gen.py b/src/ledger/acr_gen.py index c9d763b..cef2be2 100644 --- a/src/ledger/acr_gen.py +++ b/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", "", diff --git a/src/ledger/captions.py b/src/ledger/captions.py index df02d3a..e18ab19 100644 --- a/src/ledger/captions.py +++ b/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".\s]+)*[ \t]+([^>]+)>") - # Any WebVTT cue-span tag: , , , , , , , 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 ```` annotation, linearly.""" + offset = 0 + while (start := raw.find("= 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 diff --git a/src/ledger/checkup.py b/src/ledger/checkup.py index d6ea359..2e88367 100644 --- a/src/ledger/checkup.py +++ b/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( [ diff --git a/src/ledger/config.py b/src/ledger/config.py index a98ef22..547e851 100644 --- a/src/ledger/config.py +++ b/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). diff --git a/src/ledger/ingest.py b/src/ledger/ingest.py index 858494d..ac4fc94 100644 --- a/src/ledger/ingest.py +++ b/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,29 +997,24 @@ 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 @@ -1026,7 +1022,7 @@ def remove_all_copies(self, record_id: str, *, now: str | None = None) -> tuple[ # 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 diff --git a/src/ledger/lockdown.py b/src/ledger/lockdown.py index 29396d3..ece8c90 100644 --- a/src/ledger/lockdown.py +++ b/src/ledger/lockdown.py @@ -33,11 +33,13 @@ import os import secrets import shutil -from dataclasses import dataclass, field +from dataclasses import dataclass from pathlib import Path from typing import Protocol -from ledger.errors import ConfigError, LedgerError +from ledger.config import Config, LockdownConfig +from ledger.errors import LedgerError +from ledger.ingest import Archive from ledger.metadata.premis import PremisLog from ledger.models import PremisEvent, PremisEventType @@ -87,93 +89,8 @@ class ArchiveLike(Protocol): vault_path: Path @property - def config(self) -> object: ... - - -@dataclass(frozen=True) -class LockdownConfig: - """How this archive behaves under lockdown (declarative, off by default). - - ``stop_disclosure`` gates the disclosure freeze; ``shred_vault`` gates the - irreversible local-vault destruction and is **off unless a steward turns it on**, - so a misread config can never destroy a vault by default (safety: default to - narrowest). ``required_replica_locations`` are paths to off-box *backup roots* - (each holding ``store/`` + ``identity.vault``); ``min_verified_replicas`` of them - must restore clean before any shred proceeds. - """ - - 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: - """Raise :class:`LedgerError` if the lockdown policy is self-contradictory. - - Correctness: a ``shred_vault`` posture with no replica to verify against, or a - replica threshold that can never be met, is caught here rather than at the - dangerous moment a steward triggers a duress shred. - - ``archive_locations`` are the live archive's own on-box paths (its root, - ``store_root``, ``vault_path``, ...) when known to the caller. A - ``required_replica_locations`` entry that resolves to one of them is not an - off-box replica at all — it is the archive pointing at itself — so - ``min_verified_replicas`` could be satisfied while providing *no* real - redundancy (a duress shred could proceed having "verified" nothing but the - copy it is about to destroy). Checked unconditionally, not just when - ``shred_vault`` is on, since a self-referential replica is equally useless - for stand-up's restore path. - """ - 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]: - """Serialize to a plain JSON-/TOML-ready mapping (deterministic order).""" - 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: - """Rebuild from a mapping, coercing scalars and rejecting malformed shapes.""" - 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 + def config(self) -> object: + raise NotImplementedError @dataclass(frozen=True) @@ -218,11 +135,6 @@ def verify_backup_location(backup: Path) -> BackupVerification: identity — it reports only readability, per-bag fixity, and whether a vault file exists (no-outing rule). """ - # Local imports break the config -> lockdown -> ingest -> config import cycle: - # LockdownConfig (used by config.py) needs none of these, so they load lazily. - from ledger.config import Config - from ledger.ingest import Archive - backup = Path(backup) config_path = backup / "store" / _CONFIG_FILENAME if not config_path.exists(): diff --git a/src/ledger/metadata/ead.py b/src/ledger/metadata/ead.py index da800a5..3733244 100644 --- a/src/ledger/metadata/ead.py +++ b/src/ledger/metadata/ead.py @@ -34,7 +34,6 @@ from __future__ import annotations -import re from collections.abc import Sequence from xml.sax.saxutils import escape as _sax_escape @@ -42,14 +41,23 @@ __all__ = ["to_ead_xml"] + # Characters XML 1.0 forbids even when escaped -- same rule as the sibling # metadata modules (standards compliance, interoperability, robustness). -_ILLEGAL_XML = re.compile("[^\x09\x0a\x0d\x20-\ud7ff\ue000-\ufffd\U00010000-\U0010ffff]") +def _xml_text(value: str) -> str: + return "".join( + char + for char in value + if (code := ord(char)) in (0x9, 0xA, 0xD) + or 0x20 <= code <= 0xD7FF + or 0xE000 <= code <= 0xFFFD + or 0x10000 <= code <= 0x10FFFF + ) def escape(value: str) -> str: """XML-escape ``value`` after removing characters XML 1.0 disallows.""" - return _sax_escape(_ILLEGAL_XML.sub("", value)) + 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 0cc2707..3880631 100644 --- a/src/ledger/metadata/mets.py +++ b/src/ledger/metadata/mets.py @@ -39,7 +39,6 @@ from __future__ import annotations -import re from collections.abc import Sequence from xml.sax.saxutils import escape as _sax_escape @@ -49,16 +48,25 @@ __all__ = ["to_mets_xml"] + # Characters XML 1.0 forbids even when escaped. Descriptive text and filenames can # carry arbitrary operator- or contributor-supplied content, so strip these before # escaping to keep the emitted XML well-formed (standards compliance, # interoperability, robustness) -- the same rule the sibling metadata modules apply. -_ILLEGAL_XML = re.compile("[^\x09\x0a\x0d\x20-\ud7ff\ue000-\ufffd\U00010000-\U0010ffff]") +def _xml_text(value: str) -> str: + return "".join( + char + for char in value + if (code := ord(char)) in (0x9, 0xA, 0xD) + or 0x20 <= code <= 0xD7FF + or 0xE000 <= code <= 0xFFFD + or 0x10000 <= code <= 0x10FFFF + ) def escape(value: str) -> str: """XML-escape ``value`` after removing characters XML 1.0 disallows.""" - return _sax_escape(_ILLEGAL_XML.sub("", value)) + return _sax_escape(_xml_text(value)) _METS_NS = "http://www.loc.gov/METS/" diff --git a/src/ledger/server.py b/src/ledger/server.py index 85c416a..1338b43 100644 --- a/src/ledger/server.py +++ b/src/ledger/server.py @@ -2516,8 +2516,6 @@ def _handle_file(self, raw_id: str, raw_name: str) -> None: if self.command != "HEAD": handle = path.open("rb") except (ObjectNotFound, LedgerError, OSError): - if handle is not None: # pragma: no cover - defensive - handle.close() self._handle_not_found() return try: @@ -2559,11 +2557,9 @@ def _send_file_response( # RFC 6266: a percent-encoded filename* is a recognized-safe way to carry # the (already CR/LF/quote-stripped) name, and it renders non-ASCII names # correctly for a bilingual archive. Keep an ASCII filename= fallback. - safe_name = _safe_filename(filename) - self.send_header( - "Content-Disposition", - f"inline; filename=\"{safe_name}\"; filename*=UTF-8''{quote(safe_name, safe='')}", - ) + # The human filename appears on the already-escaped record page. Keep the + # transport header wholly literal so request input cannot become a header. + self.send_header("Content-Disposition", 'inline; filename="download"') self.send_header("Content-Security-Policy", "default-src 'none'; sandbox") self.send_header("Referrer-Policy", "no-referrer") self.end_headers() diff --git a/tests/test_grant_auth.py b/tests/test_grant_auth.py index 8b48350..4fc4d1f 100644 --- a/tests/test_grant_auth.py +++ b/tests/test_grant_auth.py @@ -16,6 +16,7 @@ from __future__ import annotations +import hmac import json import threading import urllib.request @@ -267,12 +268,12 @@ def test_concurrent_grant_uses_all_audited(tmp_path: Path, monkeypatch: pytest.M httpd = make_server(archive, host="127.0.0.1", port=0, grants_path=_write_grants(tmp_path)) token = issue_grant_token(_SUBJECT, _SECRET, expires_at=_FUTURE) request_count = 8 - errors: list[BaseException] = [] + errors: list[Exception] = [] def _hit(base: str) -> None: try: assert _authenticated(base, rid, token) is True - except BaseException as exc: # surfaced to the main thread below + except Exception as exc: # surfaced to the main thread below errors.append(exc) with _running(httpd) as base: @@ -340,16 +341,14 @@ def test_verify_round_trip_and_failures() -> None: def test_verify_uses_constant_time_compare(monkeypatch: pytest.MonkeyPatch) -> None: """Verification goes through :func:`hmac.compare_digest` (no early-exit compare).""" - import ledger.access.grants as grants_mod - calls: list[int] = [] - real = grants_mod.hmac.compare_digest + real = hmac.compare_digest def _counting(a: object, b: object) -> bool: calls.append(1) return real(a, b) - monkeypatch.setattr(grants_mod.hmac, "compare_digest", _counting) + monkeypatch.setattr(hmac, "compare_digest", _counting) token = issue_grant_token(_SUBJECT, _SECRET, expires_at=_FUTURE) assert verify_grant_token(token, _SECRET, now=_NOW) == _SUBJECT assert calls, "verify_grant_token must compare the MAC with hmac.compare_digest" diff --git a/tests/test_metadata.py b/tests/test_metadata.py index 9160635..448f4e3 100644 --- a/tests/test_metadata.py +++ b/tests/test_metadata.py @@ -221,8 +221,6 @@ def test_dublin_core_json_round_trip() -> None: @pytest.mark.preservation def test_dublin_core_json_drops_empty_elements() -> None: """Empty DC elements are absent from the JSON (compact, deterministic sidecar).""" - import json - dc = DublinCore(title=["Only a title"]) parsed = json.loads(to_json(dc)) assert parsed == {"title": ["Only a title"]} @@ -291,8 +289,6 @@ def test_premis_log_without_rights_serializes_without_rights_key() -> None: ``test_old_rightsless_log_still_reads_back`` — but everything written from FIX-06 on is the schema-versioned, hash-chained envelope. """ - import json - log = PremisLog(_sample_events()) parsed = json.loads(log.to_json()) assert isinstance(parsed, dict)