From 18eeee031027977258583afb1ea0662b1847a05b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 16:07:52 +0000 Subject: [PATCH 1/4] feat(auth): add stdlib API credential registry Store SHA-256 verifiers in api_credentials so request-time auth no longer reads CODEC_CARVER_API_KEYS. Bootstrap stays idempotent; rotate keeps the previous key valid until revoke. Co-authored-by: Seongho Bae --- .coveragerc | 1 + .github/workflows/ci.yml | 2 +- credential_registry.py | 624 ++++++++++++++++++++++++++++++ pyproject.toml | 1 + tests/test_credential_registry.py | 306 +++++++++++++++ 5 files changed, 933 insertions(+), 1 deletion(-) create mode 100644 credential_registry.py create mode 100644 tests/test_credential_registry.py diff --git a/.coveragerc b/.coveragerc index c6decdae..f1fb18dc 100644 --- a/.coveragerc +++ b/.coveragerc @@ -2,6 +2,7 @@ source = audio_library chapters + credential_registry diarize job_store mcp_driver diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 78f2e184..eaf753e7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,7 +36,7 @@ jobs: python -m pip install --disable-pip-version-check --no-index --no-deps --no-build-isolation -e . - name: Compile check - run: python -m py_compile media_shrinker.py config_file.py presets.py saas_web.py mcp_driver.py job_store.py + run: python -m py_compile media_shrinker.py config_file.py presets.py saas_web.py mcp_driver.py job_store.py credential_registry.py - name: Run tests run: python -m unittest discover -s tests -v diff --git a/credential_registry.py b/credential_registry.py new file mode 100644 index 00000000..7d801daf --- /dev/null +++ b/credential_registry.py @@ -0,0 +1,624 @@ +"""Stdlib-only API credential registry for request-time authentication. + +Environment variables may populate this store during an explicit bootstrap +step. Request handlers must call :meth:`CredentialRegistry.verify_api_key` +and must not read ``CODEC_CARVER_API_KEYS`` themselves. + +Callers pass ``now`` explicitly. The store never calls ``datetime.now()``. +""" + +from __future__ import annotations + +import hashlib +import hmac +import sqlite3 +import threading +import unicodedata +import uuid +from collections.abc import Iterator, Mapping +from contextlib import contextmanager +from datetime import datetime + +#: Allowed credential lifecycle states. +VALID_LIFECYCLE_STATES = frozenset({"active", "rotated", "revoked"}) + +#: Hard cap on stored credentials so comparison work stays bounded. +MAX_CREDENTIAL_COUNT = 16 + +#: Maximum UTF-8 size of a plaintext key or ``X-API-Key`` header. +MAX_KEY_BYTES = 256 + +#: Hosts that count as loopback for the explicit development mode. +LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"}) + +_CREDENTIAL_SCHEMA = """ +CREATE TABLE IF NOT EXISTS api_credentials ( + credential_id TEXT PRIMARY KEY, + key_digest TEXT NOT NULL UNIQUE, + lifecycle_state TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + expires_at TEXT, + key_label TEXT NOT NULL +) +""" + +_EVENT_SCHEMA = """ +CREATE TABLE IF NOT EXISTS credential_events ( + event_id TEXT PRIMARY KEY, + credential_id TEXT, + event_type TEXT NOT NULL, + event_at TEXT NOT NULL, + actor_label TEXT NOT NULL, + FOREIGN KEY (credential_id) REFERENCES api_credentials (credential_id) +) +""" + +_POLICY_SCHEMA = """ +CREATE TABLE IF NOT EXISTS runtime_policies ( + policy_name TEXT PRIMARY KEY, + policy_value TEXT NOT NULL, + updated_at TEXT NOT NULL +) +""" + +_PUBLIC_COLUMNS = ( + "credential_id", + "lifecycle_state", + "created_at", + "updated_at", + "expires_at", + "key_label", +) + + +class CredentialValidationError(ValueError): + """Raised when bootstrap text cannot become a stored credential.""" + + +class CredentialPolicyError(ValueError): + """Raised when a listen address violates the credential policy.""" + + +def parse_transport_keys(raw: str) -> list[str]: + """Split comma-separated bootstrap text into stripped candidate keys. + + Args: + raw: Transport string, typically the value of + ``CODEC_CARVER_API_KEYS`` during startup only. + + Returns: + Non-empty key strings in the order they appeared. + """ + + if not raw: + return [] + return [part.strip() for part in raw.split(",") if part.strip()] + + +def digest_api_key(plaintext: str) -> str: + """Return the hex SHA-256 digest of ``plaintext`` encoded as UTF-8. + + Args: + plaintext: Issued API key. Must already be validated. + + Returns: + 64-character lowercase hexadecimal digest. + """ + + return hashlib.sha256(plaintext.encode("utf-8")).hexdigest() + + +def _reject_control_characters(plaintext: str) -> None: + """Reject Unicode general-category Control and format characters. + + Args: + plaintext: Candidate key. + + Raises: + CredentialValidationError: If any character is a control or format + character. The exception text never includes ``plaintext``. + """ + + for char in plaintext: + if unicodedata.category(char).startswith("C"): + raise CredentialValidationError( + "control characters are not allowed in credentials" + ) + + +def _validated_digest(plaintext: str) -> str: + """Validate one plaintext key and return its digest. + + Args: + plaintext: Candidate key after whitespace strip. + + Returns: + Hex SHA-256 digest. + + Raises: + CredentialValidationError: If the key is empty, overlong, or contains + control characters. + """ + + if not plaintext: + raise CredentialValidationError("empty credential is not allowed") + raw = plaintext.encode("utf-8") + if len(raw) > MAX_KEY_BYTES: + raise CredentialValidationError( + "credential exceeds the maximum UTF-8 length" + ) + _reject_control_characters(plaintext) + return digest_api_key(plaintext) + + +class CredentialRegistry: + """Durable, thread-safe API credential store backed by SQLite WAL. + + Args: + db_path: Filesystem path of the SQLite database. ``":memory:"`` is + rejected because each operation opens a fresh connection. + """ + + def __init__(self, db_path: str) -> None: + """Initialize the store and create the 3NF schema if needed. + + Args: + db_path: Path to the SQLite database file. + + Raises: + ValueError: If ``db_path`` is ``":memory:"``. + """ + + if db_path == ":memory:": + raise ValueError( + "CredentialRegistry requires a file path; ':memory:' " + "databases do not survive the short-lived connections " + "this store uses" + ) + self._db_path = str(db_path) + self._lock = threading.Lock() + with self._connect() as conn: + conn.execute(_CREDENTIAL_SCHEMA) + conn.execute(_EVENT_SCHEMA) + conn.execute(_POLICY_SCHEMA) + + def __repr__(self) -> str: + """Return a redacted summary that never includes key material.""" + + return f"CredentialRegistry(db_path={self._db_path!r})" + + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + """Open a short-lived WAL-mode connection. + + Yields: + A ``sqlite3.Connection`` with ``Row`` factory enabled. + """ + + conn = sqlite3.connect(self._db_path, timeout=30.0) + try: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + yield conn + conn.commit() + finally: + conn.close() + + def _record_event( + self, + conn: sqlite3.Connection, + credential_id: str | None, + event_type: str, + now: datetime, + actor_label: str, + ) -> None: + """Append one audit row that never stores a plaintext key. + + Args: + conn: Open connection inside the caller lock. + credential_id: Affected credential primary key, or ``None`` + for policy-only events. + event_type: ``imported``, ``rotated``, ``revoked``, or + ``policy_updated``. + now: Event timestamp. + actor_label: Non-secret source such as ``env`` or ``test``. + """ + + conn.execute( + "INSERT INTO credential_events (event_id, credential_id, " + "event_type, event_at, actor_label) VALUES (?, ?, ?, ?, ?)", + ( + str(uuid.uuid4()), + credential_id, + event_type, + now.isoformat(), + actor_label, + ), + ) + + def import_plaintext_keys( + self, + keys: list[str], + *, + now: datetime, + source: str, + expires_at: datetime | None = None, + ) -> int: + """Insert new active credentials from already-split plaintext keys. + + Existing digests are left unchanged so bootstrap is idempotent and + a revoked key stays revoked. + + Args: + keys: Plaintext keys. Duplicates in this list are rejected. + now: Timestamp for ``created_at`` / ``updated_at``. + source: Non-secret actor label written to ``credential_events``. + expires_at: Optional expiry applied only to newly inserted rows. + + Returns: + Count of newly inserted rows. + + Raises: + CredentialValidationError: If any key is empty, duplicated in + ``keys``, overlong, contains control characters, or the + bounded set would exceed :data:`MAX_CREDENTIAL_COUNT`. + """ + + digests: list[str] = [] + seen: set[str] = set() + for key in keys: + digest = _validated_digest(key) + if digest in seen: + raise CredentialValidationError( + "duplicate credential in the import list" + ) + seen.add(digest) + digests.append(digest) + + inserted = 0 + expiry = expires_at.isoformat() if expires_at is not None else None + with self._lock, self._connect() as conn: + existing = conn.execute( + "SELECT COUNT(*) FROM api_credentials" + ).fetchone()[0] + known = { + row["key_digest"] + for row in conn.execute("SELECT key_digest FROM api_credentials") + } + new_digests = [digest for digest in digests if digest not in known] + if existing + len(new_digests) > MAX_CREDENTIAL_COUNT: + raise CredentialValidationError( + f"at most {MAX_CREDENTIAL_COUNT} credentials may be stored" + ) + for digest in new_digests: + credential_id = str(uuid.uuid4()) + conn.execute( + "INSERT INTO api_credentials (credential_id, key_digest, " + "lifecycle_state, created_at, updated_at, expires_at, " + "key_label) VALUES (?, ?, 'active', ?, ?, ?, ?)", + ( + credential_id, + digest, + now.isoformat(), + now.isoformat(), + expiry, + digest[:8], + ), + ) + self._record_event(conn, credential_id, "imported", now, source) + inserted += 1 + return inserted + + def bootstrap_from_transport( + self, + raw: str, + *, + now: datetime, + source: str = "transport", + ) -> int: + """Parse bootstrap text and import new keys idempotently. + + Args: + raw: Comma-separated transport string. + now: Import timestamp. + source: Non-secret actor label. + + Returns: + Count of newly inserted rows. + """ + + return self.import_plaintext_keys( + parse_transport_keys(raw), + now=now, + source=source, + ) + + def rotate(self, current_plaintext: str, next_plaintext: str, *, now: datetime) -> None: + """Mark ``current_plaintext`` rotated and insert ``next_plaintext``. + + The rotated key remains valid until :meth:`revoke` so in-flight + clients can finish while operators distribute the next key. + + Args: + current_plaintext: Key already stored as ``active``. + next_plaintext: Replacement key to insert as ``active``. + now: Transition timestamp. + + Raises: + KeyError: If the current key is not stored. + CredentialValidationError: If the next key is invalid or already + stored. + """ + + current_digest = _validated_digest(current_plaintext) + next_digest = _validated_digest(next_plaintext) + with self._lock, self._connect() as conn: + current = conn.execute( + "SELECT credential_id FROM api_credentials " + "WHERE key_digest = ?", + (current_digest,), + ).fetchone() + if current is None: + raise KeyError("current credential is not in the registry") + existing_next = conn.execute( + "SELECT credential_id FROM api_credentials " + "WHERE key_digest = ?", + (next_digest,), + ).fetchone() + if existing_next is not None: + raise CredentialValidationError( + "next credential is already stored" + ) + total = conn.execute( + "SELECT COUNT(*) FROM api_credentials" + ).fetchone()[0] + if total + 1 > MAX_CREDENTIAL_COUNT: + raise CredentialValidationError( + f"at most {MAX_CREDENTIAL_COUNT} credentials may be stored" + ) + conn.execute( + "UPDATE api_credentials SET lifecycle_state = 'rotated', " + "updated_at = ? WHERE credential_id = ?", + (now.isoformat(), current["credential_id"]), + ) + self._record_event( + conn, current["credential_id"], "rotated", now, "rotate" + ) + next_id = str(uuid.uuid4()) + conn.execute( + "INSERT INTO api_credentials (credential_id, key_digest, " + "lifecycle_state, created_at, updated_at, expires_at, " + "key_label) VALUES (?, ?, 'active', ?, ?, NULL, ?)", + ( + next_id, + next_digest, + now.isoformat(), + now.isoformat(), + next_digest[:8], + ), + ) + self._record_event(conn, next_id, "imported", now, "rotate") + + def revoke(self, plaintext: str, *, now: datetime) -> None: + """Mark a stored key revoked so it no longer verifies. + + Args: + plaintext: Key to revoke. + now: Revocation timestamp. + + Raises: + KeyError: If the key is not stored. + """ + + digest = _validated_digest(plaintext) + with self._lock, self._connect() as conn: + row = conn.execute( + "SELECT credential_id FROM api_credentials WHERE key_digest = ?", + (digest,), + ).fetchone() + if row is None: + raise KeyError("credential is not in the registry") + conn.execute( + "UPDATE api_credentials SET lifecycle_state = 'revoked', " + "updated_at = ? WHERE credential_id = ?", + (now.isoformat(), row["credential_id"]), + ) + self._record_event(conn, row["credential_id"], "revoked", now, "revoke") + + def _usable_digests(self, conn: sqlite3.Connection, now: datetime) -> list[str]: + """Return digests that may still authenticate at ``now``. + + Args: + conn: Open connection. + now: Comparison timestamp. + + Returns: + Digests for ``active`` and ``rotated`` rows that have not expired. + """ + + rows = conn.execute( + "SELECT key_digest, expires_at FROM api_credentials " + "WHERE lifecycle_state IN ('active', 'rotated')" + ).fetchall() + usable: list[str] = [] + now_text = now.isoformat() + for row in rows: + expires_at = row["expires_at"] + if expires_at is not None and expires_at <= now_text: + continue + usable.append(row["key_digest"]) + return usable + + def verify_api_key(self, provided: object, *, now: datetime) -> bool: + """Return True when ``provided`` matches a usable stored digest. + + Comparison always visits every usable digest. Hostile or overlong + headers return False instead of raising. + + Args: + provided: ``X-API-Key`` value. Non-strings are rejected. + now: Comparison timestamp used for expiry. + + Returns: + True when the header matches an unexpired ``active`` or + ``rotated`` credential. + """ + + if not isinstance(provided, str): + return False + raw = provided.encode("utf-8") + if not raw or len(raw) > MAX_KEY_BYTES: + return False + try: + _reject_control_characters(provided) + except CredentialValidationError: + return False + provided_digest = digest_api_key(provided) + with self._lock, self._connect() as conn: + stored = self._usable_digests(conn, now) + matched = False + for digest in stored: + if hmac.compare_digest(provided_digest, digest): + matched = True + return matched + + def has_active_credentials(self, *, now: datetime) -> bool: + """Return True when at least one usable credential exists. + + Args: + now: Comparison timestamp used for expiry. + + Returns: + True if verify could succeed for some issued key. + """ + + with self._lock, self._connect() as conn: + return bool(self._usable_digests(conn, now)) + + def list_public_records(self) -> list[dict[str, str | None]]: + """Return non-secret credential rows for operators. + + Returns: + Dicts with ``credential_id``, lifecycle, timestamps, and + ``key_label``. Digests and plaintext are omitted. + """ + + with self._lock, self._connect() as conn: + rows = conn.execute( + "SELECT credential_id, lifecycle_state, created_at, " + "updated_at, expires_at, key_label FROM api_credentials " + "ORDER BY created_at, credential_id" + ).fetchall() + return [{key: row[key] for key in _PUBLIC_COLUMNS} for row in rows] + + def audit_events(self) -> list[dict[str, str]]: + """Return audit rows that never contain credential values. + + Returns: + Event dicts ordered by ``event_at`` then ``event_id``. + """ + + with self._lock, self._connect() as conn: + rows = conn.execute( + "SELECT event_id, credential_id, event_type, event_at, " + "actor_label FROM credential_events " + "ORDER BY event_at, event_id" + ).fetchall() + return [dict(row) for row in rows] + + def set_loopback_development(self, enabled: bool, *, now: datetime) -> None: + """Record the explicit loopback-only development policy. + + Args: + enabled: True to allow an empty registry on loopback binds. + now: Policy timestamp. + """ + + value = "1" if enabled else "0" + with self._lock, self._connect() as conn: + conn.execute( + "INSERT INTO runtime_policies (policy_name, policy_value, " + "updated_at) VALUES ('loopback_development', ?, ?) " + "ON CONFLICT(policy_name) DO UPDATE SET " + "policy_value = excluded.policy_value, " + "updated_at = excluded.updated_at", + (value, now.isoformat()), + ) + self._record_event( + conn, + None, + "policy_updated", + now, + "loopback_development", + ) + + def loopback_development_enabled(self) -> bool: + """Return True when loopback development mode is stored as enabled. + + Returns: + False when the policy row is missing or set to ``0``. + """ + + with self._lock, self._connect() as conn: + row = conn.execute( + "SELECT policy_value FROM runtime_policies " + "WHERE policy_name = 'loopback_development'" + ).fetchone() + return bool(row) and row["policy_value"] == "1" + + def ensure_listen_policy(self, host: str, *, now: datetime) -> None: + """Refuse a non-loopback bind when no usable credentials exist. + + Args: + host: Intended bind address (``0.0.0.0``, ``127.0.0.1``, …). + now: Timestamp used to decide whether credentials are usable. + + Raises: + CredentialPolicyError: If the bind is public and the registry + has no usable key, or loopback development is off. + """ + + if self.has_active_credentials(now=now): + return + normalized = (host or "").strip().lower() + if self.loopback_development_enabled() and normalized in LOOPBACK_HOSTS: + return + raise CredentialPolicyError( + "this bind requires configured API credentials; " + "import keys or enable loopback development on 127.0.0.1" + ) + + +def bootstrap_registry_from_mapping( + transport: Mapping[str, str], + *, + now: datetime, + db_path: str, +) -> CredentialRegistry: + """Create a registry and import keys from a bootstrap mapping. + + Args: + transport: Mapping that may contain ``CODEC_CARVER_API_KEYS`` and + ``CODEC_CARVER_LOOPBACK_DEV``. This is the only approved env + read surface; pass ``os.environ`` from a named startup hook. + now: Bootstrap timestamp. + db_path: SQLite file for the registry. + + Returns: + The populated :class:`CredentialRegistry`. + """ + + registry = CredentialRegistry(db_path) + registry.bootstrap_from_transport( + transport.get("CODEC_CARVER_API_KEYS", ""), + now=now, + source="transport", + ) + if transport.get("CODEC_CARVER_LOOPBACK_DEV") == "1": + registry.set_loopback_development(True, now=now) + bind_host = transport.get("CODEC_CARVER_BIND_HOST") + if bind_host: + registry.ensure_listen_policy(bind_host, now=now) + return registry diff --git a/pyproject.toml b/pyproject.toml index 91884dfe..07d1371d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -78,6 +78,7 @@ codec-carver-library = "audio_library:main" py-modules = [ "chapters", "config_file", + "credential_registry", "diarize", "job_store", "mcp_driver", diff --git a/tests/test_credential_registry.py b/tests/test_credential_registry.py new file mode 100644 index 00000000..21d96eea --- /dev/null +++ b/tests/test_credential_registry.py @@ -0,0 +1,306 @@ +"""Contract tests for the stdlib API credential registry.""" + +from __future__ import annotations + +import hmac +import os +import tempfile +import threading +import unittest +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +from credential_registry import ( + MAX_CREDENTIAL_COUNT, + MAX_KEY_BYTES, + CredentialPolicyError, + CredentialRegistry, + CredentialValidationError, + bootstrap_registry_from_mapping, + digest_api_key, + parse_transport_keys, +) + +T0 = datetime(2026, 8, 16, 12, 0, 0, tzinfo=timezone.utc) +T1 = T0 + timedelta(hours=1) +T2 = T0 + timedelta(days=30) + + +class CredentialRegistryTestCase(unittest.TestCase): + """Fresh SQLite registry on a temporary file.""" + + def setUp(self) -> None: + """Create an isolated registry file for each test.""" + + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.db_path = os.path.join(self._tmp.name, "api_credentials.db") + self.registry = CredentialRegistry(self.db_path) + + +class TestParseTransportKeys(unittest.TestCase): + """Transport text is comma-separated and never read on the request path.""" + + def test_strips_whitespace_and_drops_empty_entries(self) -> None: + """Buyers can paste `key-a, key-b` and both keys import.""" + + self.assertEqual(parse_transport_keys(" key-a , , key-b "), ["key-a", "key-b"]) + + def test_blank_transport_is_empty(self) -> None: + """Unset or whitespace-only transport leaves the registry empty.""" + + self.assertEqual(parse_transport_keys(""), []) + self.assertEqual(parse_transport_keys(" , ,"), []) + + +class TestImportAndVerify(CredentialRegistryTestCase): + """Import stores digests; verify compares UTF-8 SHA-256 without plaintext.""" + + def test_imported_key_verifies_and_wrong_key_does_not(self) -> None: + """A meeting-upload client with the issued key is accepted; a guess is not.""" + + self.registry.import_plaintext_keys(["meeting-upload-key"], now=T0, source="test") + self.assertTrue(self.registry.verify_api_key("meeting-upload-key", now=T0)) + self.assertFalse(self.registry.verify_api_key("guessed-key", now=T0)) + self.assertFalse(self.registry.verify_api_key("", now=T0)) + + def test_non_ascii_key_round_trips(self) -> None: + """UTF-8 keys used by non-English operators verify on the same code path.""" + + key = "업로드-키-αβγ" + self.registry.import_plaintext_keys([key], now=T0, source="test") + self.assertTrue(self.registry.verify_api_key(key, now=T0)) + self.assertFalse(self.registry.verify_api_key("업로드-키-αβγ\u0000", now=T0)) + + def test_hostile_header_types_and_overlong_values_are_false(self) -> None: + """A hostile X-API-Key must 401, never raise into the web worker.""" + + self.registry.import_plaintext_keys(["stable-key"], now=T0, source="test") + self.assertFalse(self.registry.verify_api_key(None, now=T0)) + self.assertFalse(self.registry.verify_api_key(b"stable-key", now=T0)) + self.assertFalse(self.registry.verify_api_key("x" * (MAX_KEY_BYTES + 1), now=T0)) + + def test_verify_compares_every_active_digest(self) -> None: + """No first-match short-circuit: every stored digest is visited.""" + + keys = ["alpha-key", "bravo-key", "charlie-key"] + self.registry.import_plaintext_keys(keys, now=T0, source="test") + calls: list[tuple[str, str]] = [] + real = hmac.compare_digest + + def counting_compare(left: str, right: str) -> bool: + """Count compare_digest visits while preserving real comparison.""" + + calls.append((left, right)) + return real(left, right) + + with patch("credential_registry.hmac.compare_digest", side_effect=counting_compare): + self.assertTrue(self.registry.verify_api_key("alpha-key", now=T0)) + + self.assertEqual(len(calls), 3) + + def test_public_records_and_repr_omit_plaintext(self) -> None: + """Listings, repr, and audit rows must not echo the issued secret.""" + + secret = "SUPER-SECRET-KEY-VALUE" + self.registry.import_plaintext_keys([secret], now=T0, source="bootstrap") + public = self.registry.list_public_records() + blob = repr(self.registry) + str(public) + str(self.registry.audit_events()) + self.assertNotIn(secret, blob) + self.assertEqual(len(public), 1) + self.assertEqual(public[0]["lifecycle_state"], "active") + self.assertNotIn("key_digest", public[0]) + self.assertTrue(self.registry.has_active_credentials(now=T0)) + + def test_bootstrap_is_idempotent(self) -> None: + """Re-running the env transport import does not duplicate or revive revoked keys.""" + + raw = "keep-key,drop-later" + first = self.registry.bootstrap_from_transport(raw, now=T0, source="env") + second = self.registry.bootstrap_from_transport(raw, now=T1, source="env") + self.assertEqual(first, 2) + self.assertEqual(second, 0) + self.registry.revoke("drop-later", now=T1) + third = self.registry.bootstrap_from_transport(raw, now=T2, source="env") + self.assertEqual(third, 0) + self.assertFalse(self.registry.verify_api_key("drop-later", now=T2)) + self.assertTrue(self.registry.verify_api_key("keep-key", now=T2)) + + +class TestValidation(CredentialRegistryTestCase): + """Bootstrap rejects empty, duplicate, overlong, over-count, and control keys.""" + + def test_empty_and_control_and_overlong_rejected_without_echo(self) -> None: + """Operators get an actionable error that does not repeat the secret.""" + + secret = "bad\x00key-material" + with self.assertRaises(CredentialValidationError) as empty: + self.registry.import_plaintext_keys([""], now=T0, source="test") + with self.assertRaises(CredentialValidationError) as control: + self.registry.import_plaintext_keys([secret], now=T0, source="test") + with self.assertRaises(CredentialValidationError) as huge: + self.registry.import_plaintext_keys(["k" * (MAX_KEY_BYTES + 1)], now=T0, source="test") + self.assertNotIn(secret, str(control.exception)) + self.assertIn("empty", str(empty.exception)) + self.assertIn("control", str(control.exception)) + self.assertIn("maximum", str(huge.exception)) + + def test_duplicates_and_over_count_rejected(self) -> None: + """A pasted list cannot silently collapse or exceed the bounded set.""" + + with self.assertRaises(CredentialValidationError) as dup: + self.registry.import_plaintext_keys(["same-key", "same-key"], now=T0, source="test") + self.assertNotIn("same-key", str(dup.exception)) + too_many = [f"issued-key-{index:02d}" for index in range(MAX_CREDENTIAL_COUNT + 1)] + with self.assertRaises(CredentialValidationError) as count: + self.registry.import_plaintext_keys(too_many, now=T0, source="test") + self.assertIn("at most", str(count.exception)) + + +class TestRotationExpiryAndRevoke(CredentialRegistryTestCase): + """Zero-downtime rotation keeps current+next valid until revoke or expiry.""" + + def test_rotated_key_still_verifies_until_revoked(self) -> None: + """Cut over to the next key without dropping in-flight clients.""" + + self.registry.import_plaintext_keys(["current-key"], now=T0, source="test") + self.registry.rotate("current-key", "next-key", now=T1) + self.assertTrue(self.registry.verify_api_key("current-key", now=T1)) + self.assertTrue(self.registry.verify_api_key("next-key", now=T1)) + states = {row["lifecycle_state"] for row in self.registry.list_public_records()} + self.assertEqual(states, {"rotated", "active"}) + self.registry.revoke("current-key", now=T2) + self.assertFalse(self.registry.verify_api_key("current-key", now=T2)) + self.assertTrue(self.registry.verify_api_key("next-key", now=T2)) + + def test_expired_key_does_not_verify(self) -> None: + """A time-bounded contractor key stops working after expires_at.""" + + self.registry.import_plaintext_keys( + ["contractor-key"], + now=T0, + source="test", + expires_at=T1, + ) + self.assertTrue(self.registry.verify_api_key("contractor-key", now=T0)) + self.assertFalse(self.registry.verify_api_key("contractor-key", now=T2)) + self.assertFalse(self.registry.has_active_credentials(now=T2)) + + +class TestListenPolicy(CredentialRegistryTestCase): + """Non-loopback binds fail closed unless credentials exist.""" + + def test_public_bind_without_keys_fails_closed(self) -> None: + """Do not publish 0.0.0.0 until an operator imported keys.""" + + with self.assertRaises(CredentialPolicyError) as ctx: + self.registry.ensure_listen_policy("0.0.0.0", now=T0) + self.assertIn("credentials", str(ctx.exception)) + + def test_loopback_development_allows_empty_registry(self) -> None: + """Local `127.0.0.1` work is explicit, not an accidental open bind.""" + + self.registry.set_loopback_development(True, now=T0) + self.registry.ensure_listen_policy("127.0.0.1", now=T0) + with self.assertRaises(CredentialPolicyError): + self.registry.ensure_listen_policy("0.0.0.0", now=T0) + + def test_public_bind_succeeds_after_import(self) -> None: + """Once keys exist, the SaaS UI may listen on all interfaces.""" + + self.registry.import_plaintext_keys(["prod-key"], now=T0, source="test") + self.registry.ensure_listen_policy("0.0.0.0", now=T0) + + +class TestConcurrencyAndStorageRules(CredentialRegistryTestCase): + """Readers stay consistent; table names stay two-word snake_case.""" + + def test_concurrent_verifies_during_rotate(self) -> None: + """In-flight uploads keep working while the next key is inserted.""" + + self.registry.import_plaintext_keys(["live-key"], now=T0, source="test") + errors: list[str] = [] + + def hammer() -> None: + """Verify the live key from a worker thread.""" + + for _ in range(40): + if not self.registry.verify_api_key("live-key", now=T0): + errors.append("live-key rejected") + + workers = [threading.Thread(target=hammer) for _ in range(4)] + for worker in workers: + worker.start() + self.registry.rotate("live-key", "next-live-key", now=T1) + for worker in workers: + worker.join() + self.assertEqual(errors, []) + self.assertTrue(self.registry.verify_api_key("next-live-key", now=T1)) + + def test_schema_uses_two_word_tables_and_rejects_memory(self) -> None: + """Org naming: api_credentials / credential_events / runtime_policies.""" + + with self.registry._connect() as conn: + names = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + self.assertIn("api_credentials", names) + self.assertIn("credential_events", names) + self.assertIn("runtime_policies", names) + self.assertTrue(all("_" in name or name.startswith("sqlite_") for name in names)) + with self.assertRaises(ValueError): + CredentialRegistry(":memory:") + + def test_digest_is_hex_sha256_of_utf8(self) -> None: + """Verification material is a 64-character SHA-256 hex digest.""" + + digest = digest_api_key("meeting-upload-key") + self.assertEqual(len(digest), 64) + self.assertTrue(all(char in "0123456789abcdef" for char in digest)) + + def test_rotate_and_revoke_unknown_keys_raise(self) -> None: + """Operators get a KeyError they can act on, not a stack of key text.""" + + self.registry.import_plaintext_keys(["known-key"], now=T0, source="test") + with self.assertRaises(KeyError): + self.registry.rotate("missing-key", "next-key", now=T1) + with self.assertRaises(KeyError): + self.registry.revoke("missing-key", now=T1) + with self.assertRaises(CredentialValidationError): + self.registry.rotate("known-key", "known-key", now=T1) + + def test_bootstrap_mapping_applies_loopback_and_bind_policy(self) -> None: + """Startup transport can enable loopback mode and check the bind host.""" + + empty_path = os.path.join(self._tmp.name, "empty.db") + empty = bootstrap_registry_from_mapping( + {"CODEC_CARVER_LOOPBACK_DEV": "1"}, + now=T0, + db_path=empty_path, + ) + self.assertTrue(empty.loopback_development_enabled()) + empty.ensure_listen_policy("127.0.0.1", now=T0) + populated = bootstrap_registry_from_mapping( + { + "CODEC_CARVER_API_KEYS": "prod-key", + "CODEC_CARVER_BIND_HOST": "0.0.0.0", + }, + now=T0, + db_path=self.db_path, + ) + self.assertTrue(populated.verify_api_key("prod-key", now=T0)) + full = [f"issued-key-{index:02d}" for index in range(MAX_CREDENTIAL_COUNT)] + capped_path = os.path.join(self._tmp.name, "capped.db") + capped = CredentialRegistry(capped_path) + capped.import_plaintext_keys(full, now=T0, source="test") + with self.assertRaises(CredentialValidationError): + capped.rotate(full[0], "overflow-next-key", now=T1) + capped.set_loopback_development(False, now=T1) + self.assertFalse(capped.loopback_development_enabled()) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() From 92bc1f54d1e677bd233fb7e3904b90b3ffa1265c Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 16:07:57 +0000 Subject: [PATCH 2/4] feat(web): authenticate uploads from the credential registry Named startup copies transport keys once. Middleware verifies X-API-Key against stored digests, ignores later env changes, and returns 401 without echoing secrets. Co-authored-by: Seongho Bae --- saas_web.py | 125 ++++++++++++++++++++++++----- tests/test_saas_web.py | 174 ++++++++++++++++++++++++++++++++--------- 2 files changed, 240 insertions(+), 59 deletions(-) diff --git a/saas_web.py b/saas_web.py index 63265e94..293e5ed3 100644 --- a/saas_web.py +++ b/saas_web.py @@ -1,17 +1,22 @@ """FastAPI upload UI for shrinking one media file through Codec Carver.""" import json -import hmac import logging import os import shutil import tempfile import uuid import zipfile +from collections.abc import Mapping from datetime import datetime, timezone from pathlib import Path from fastapi import FastAPI, UploadFile, File, BackgroundTasks, Form, Request from fastapi.responses import HTMLResponse, FileResponse, JSONResponse +from credential_registry import ( + MAX_KEY_BYTES, + CredentialRegistry, + bootstrap_registry_from_mapping, +) from job_store import JobStore import media_shrinker @@ -83,42 +88,122 @@ async def limited_receive(): except RequestTooLarge: return JSONResponse(status_code=413, content={"error": "Payload Too Large"}) -def get_configured_api_keys(): - """Return the API keys configured via the CODEC_CARVER_API_KEYS env var. +_credential_registry: CredentialRegistry | None = None - The variable holds a comma-separated list of keys. Whitespace around each - key is stripped and empty entries are ignored. Keys are read from the - environment at request time (not import time) so tests can patch the - environment easily and key rotation needs no server restart. Returns an - empty list when the variable is unset or contains no usable keys, which - leaves the service open (today's default behaviour). + +def configure_credential_registry(registry: CredentialRegistry | None) -> None: + """Install or clear the process-wide request-time credential registry. + + Args: + registry: Store used by :func:`require_api_key`, or ``None`` to + restore the unconfigured fail-open default. + """ + + global _credential_registry + _credential_registry = registry + + +def get_credential_registry() -> CredentialRegistry | None: + """Return the registry installed for this process, if any. + + Returns: + The configured :class:`CredentialRegistry`, or ``None``. + """ + + return _credential_registry + + +def bootstrap_web_credentials( + transport: Mapping[str, str], + *, + now: datetime, + db_path: str, +) -> CredentialRegistry: + """Named startup hook: copy transport keys into the registry once. + + Args: + transport: Bootstrap mapping (pass ``os.environ`` only from this + hook). Request handlers must not read it. + now: Bootstrap timestamp. + db_path: SQLite file for ``api_credentials``. + + Returns: + The installed registry. + """ + + registry = bootstrap_registry_from_mapping(transport, now=now, db_path=db_path) + configure_credential_registry(registry) + return registry + + +def get_configured_api_keys() -> list[str]: + """Return public key labels from the registry, never plaintext secrets. + + An empty list means authentication is fail-open. Labels are the first + eight hex characters of each stored digest so operators can count keys + without recovering them. """ + registry = get_credential_registry() + if registry is None: + return [] + return [ + str(row["key_label"]) + for row in registry.list_public_records() + if row["lifecycle_state"] in {"active", "rotated"} + ] + + +@app.on_event("startup") +def bootstrap_credentials_from_environ() -> None: + """Load ``CODEC_CARVER_API_KEYS`` into the registry at process start. + + This is the only approved environment read for API keys. Tests that + already called :func:`configure_credential_registry` are left unchanged. + """ + + if get_credential_registry() is not None: + return raw = os.environ.get("CODEC_CARVER_API_KEYS", "") - return [key.strip() for key in raw.split(",") if key.strip()] + if not raw.strip(): + return + db_path = os.environ.get( + "CODEC_CARVER_CREDENTIAL_DB", + str(Path(tempfile.gettempdir()) / "codec-carver-api-credentials.db"), + ) + bootstrap_web_credentials( + os.environ, + now=datetime.now(timezone.utc), + db_path=db_path, + ) @app.middleware("http") async def require_api_key(request: Request, call_next): - """Enforce opt-in API-key authentication on all endpoints except GET /. + """Enforce registry authentication on all endpoints except GET /. - When one or more keys are configured via CODEC_CARVER_API_KEYS, every + When the process registry has at least one usable credential, every request other than GET / (the upload UI page) must carry an X-API-Key - header matching a configured key; comparison uses hmac.compare_digest to - stay constant-time. Requests failing the check receive a 401 JSON error - without echoing any key material. When no keys are configured, all - requests pass through unchanged. + header that verifies against stored digests. Failures return 401 JSON + without echoing key material. An unconfigured or empty registry leaves + the service open so local CLI-adjacent use still works. """ - configured_keys = get_configured_api_keys() - if configured_keys and not (request.method == "GET" and request.url.path == "/"): + registry = get_credential_registry() + now = datetime.now(timezone.utc) + if ( + registry is not None + and registry.has_active_credentials(now=now) + and not (request.method == "GET" and request.url.path == "/") + ): provided_key = request.headers.get("x-api-key", "") - if not any( - hmac.compare_digest(provided_key, key) for key in configured_keys + if len(provided_key.encode("utf-8")) > MAX_KEY_BYTES or not registry.verify_api_key( + provided_key, now=now ): return JSONResponse( status_code=401, content={"error": "Invalid or missing API key"}, + headers={"Cache-Control": "no-store"}, ) return await call_next(request) diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py index 3b57e033..c5ca8e00 100644 --- a/tests/test_saas_web.py +++ b/tests/test_saas_web.py @@ -5,14 +5,17 @@ import tempfile import unittest import zipfile +from datetime import datetime, timezone from unittest.mock import patch, MagicMock from pathlib import Path from types import SimpleNamespace +from credential_registry import CredentialRegistry, digest_api_key + try: from fastapi import BackgroundTasks from fastapi.testclient import TestClient - from fastapi.responses import Response + from fastapi.responses import JSONResponse, Response import saas_web from saas_web import app @@ -676,7 +679,23 @@ def test_get_ui_includes_batch_upload_form(self): _HAS_FASTAPI, "fastapi not installed (optional integration dependency)" ) class TestApiKeyAuth(unittest.TestCase): - """Tests for the opt-in CODEC_CARVER_API_KEYS authentication middleware.""" + """Tests for registry-backed API-key authentication middleware.""" + + def setUp(self) -> None: + """Install an isolated registry and always clear it afterwards.""" + + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.addCleanup(lambda: saas_web.configure_credential_registry(None)) + self.now = datetime(2026, 8, 16, 12, 0, 0, tzinfo=timezone.utc) + self.db_path = os.path.join(self._tmp.name, "api_credentials.db") + self.registry = CredentialRegistry(self.db_path) + saas_web.configure_credential_registry(self.registry) + + def _load(self, *keys: str) -> None: + """Import plaintext keys through the named bootstrap path.""" + + self.registry.import_plaintext_keys(list(keys), now=self.now, source="test") def _post_shrink(self, headers=None): """POST a minimal /shrink request and return the response.""" @@ -689,9 +708,7 @@ def _post_shrink(self, headers=None): ) def test_no_env_var_leaves_endpoints_open(self): - with patch.dict(os.environ): - os.environ.pop("CODEC_CARVER_API_KEYS", None) - response = self._post_shrink() + response = self._post_shrink() self.assertEqual(response.status_code, 200) self.assertEqual( @@ -700,24 +717,25 @@ def test_no_env_var_leaves_endpoints_open(self): ) def test_missing_header_rejected_when_keys_configured(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}): - response = self._post_shrink() + self._load("secret-key") + response = self._post_shrink() self.assertEqual(response.status_code, 401) self.assertEqual(response.json(), {"error": "Invalid or missing API key"}) + self.assertEqual(response.headers["Cache-Control"], "no-store") self.assertNotIn("secret-key", response.text) def test_wrong_key_rejected(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}): - response = self._post_shrink(headers={"X-API-Key": "wrong-key"}) + self._load("secret-key") + response = self._post_shrink(headers={"X-API-Key": "wrong-key"}) self.assertEqual(response.status_code, 401) self.assertEqual(response.json(), {"error": "Invalid or missing API key"}) self.assertNotIn("secret-key", response.text) def test_correct_key_reaches_handler(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}): - response = self._post_shrink(headers={"X-API-Key": "secret-key"}) + self._load("secret-key") + response = self._post_shrink(headers={"X-API-Key": "secret-key"}) self.assertEqual(response.status_code, 200) self.assertEqual( @@ -726,53 +744,63 @@ def test_correct_key_reaches_handler(self): ) def test_get_ui_always_open_without_key(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}): - response = client.get("/") + self._load("secret-key") + response = client.get("/") self.assertEqual(response.status_code, 200) self.assertIn(b"Codec Carver SaaS", response.content) def test_job_api_requires_key_when_configured(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}): - response = client.get("/jobs/missing") - allowed = client.get("/jobs/missing", headers={"X-API-Key": "secret-key"}) + self._load("secret-key") + response = client.get("/jobs/missing") + allowed = client.get("/jobs/missing", headers={"X-API-Key": "secret-key"}) self.assertEqual(response.status_code, 401) self.assertEqual(response.json(), {"error": "Invalid or missing API key"}) self.assertEqual(allowed.status_code, 404) def test_multiple_comma_separated_keys_all_valid(self): - with patch.dict( - os.environ, {"CODEC_CARVER_API_KEYS": "key-one,key-two,key-three"} - ): - for key in ("key-one", "key-two", "key-three"): - response = self._post_shrink(headers={"X-API-Key": key}) - self.assertEqual(response.status_code, 200, key) - rejected = self._post_shrink(headers={"X-API-Key": "key-four"}) + self._load("key-one", "key-two", "key-three") + for key in ("key-one", "key-two", "key-three"): + response = self._post_shrink(headers={"X-API-Key": key}) + self.assertEqual(response.status_code, 200, key) + rejected = self._post_shrink(headers={"X-API-Key": "key-four"}) self.assertEqual(rejected.status_code, 401) def test_whitespace_around_keys_is_stripped(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " key-one , key-two "}): - response = self._post_shrink(headers={"X-API-Key": "key-one"}) - self.assertEqual(response.status_code, 200) - response = self._post_shrink(headers={"X-API-Key": "key-two"}) - self.assertEqual(response.status_code, 200) - rejected = self._post_shrink(headers={"X-API-Key": " key-one "}) + saas_web.bootstrap_web_credentials( + {"CODEC_CARVER_API_KEYS": " key-one , key-two "}, + now=self.now, + db_path=self.db_path, + ) + response = self._post_shrink(headers={"X-API-Key": "key-one"}) + self.assertEqual(response.status_code, 200) + response = self._post_shrink(headers={"X-API-Key": "key-two"}) + self.assertEqual(response.status_code, 200) + rejected = self._post_shrink(headers={"X-API-Key": " key-one "}) self.assertEqual(rejected.status_code, 401) def test_empty_entries_are_ignored(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "key-one,, ,"}): - response = self._post_shrink(headers={"X-API-Key": "key-one"}) - self.assertEqual(response.status_code, 200) - rejected = self._post_shrink(headers={"X-API-Key": ""}) + saas_web.bootstrap_web_credentials( + {"CODEC_CARVER_API_KEYS": "key-one,, ,"}, + now=self.now, + db_path=self.db_path, + ) + response = self._post_shrink(headers={"X-API-Key": "key-one"}) + self.assertEqual(response.status_code, 200) + rejected = self._post_shrink(headers={"X-API-Key": ""}) self.assertEqual(rejected.status_code, 401) def test_only_empty_entries_leave_endpoints_open(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " , ,"}): - response = self._post_shrink() + saas_web.bootstrap_web_credentials( + {"CODEC_CARVER_API_KEYS": " , ,"}, + now=self.now, + db_path=self.db_path, + ) + response = self._post_shrink() self.assertEqual(response.status_code, 200) self.assertEqual( @@ -780,12 +808,80 @@ def test_only_empty_entries_leave_endpoints_open(self): {"error": "Invalid target_bytes value. Must be greater than 0."}, ) - def test_get_configured_api_keys_parsing(self): - with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " a ,, b ,"}): - self.assertEqual(saas_web.get_configured_api_keys(), ["a", "b"]) + def test_get_configured_api_keys_returns_labels_not_secrets(self): + self._load("secret-key") + labels = saas_web.get_configured_api_keys() + self.assertEqual(labels, [digest_api_key("secret-key")[:8]]) + self.assertNotIn("secret-key", labels) + saas_web.configure_credential_registry(None) + self.assertEqual(saas_web.get_configured_api_keys(), []) + + def test_request_path_ignores_environment_after_bootstrap(self): + self._load("registry-key") + with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "env-only-key"}): + rejected = self._post_shrink(headers={"X-API-Key": "env-only-key"}) + allowed = self._post_shrink(headers={"X-API-Key": "registry-key"}) + + self.assertEqual(rejected.status_code, 401) + self.assertEqual(allowed.status_code, 200) + + def test_non_ascii_header_verifies_and_does_not_echo(self): + key = "업로드-키-αβγ" + self._load(key) + + async def _next(_request): + return JSONResponse({"ok": True}) + + allowed_request = SimpleNamespace( + method="POST", + url=SimpleNamespace(path="/shrink"), + headers={"x-api-key": key}, + ) + rejected_request = SimpleNamespace( + method="POST", + url=SimpleNamespace(path="/shrink"), + headers={"x-api-key": "다른-키"}, + ) + allowed = asyncio.run(saas_web.require_api_key(allowed_request, _next)) + rejected = asyncio.run(saas_web.require_api_key(rejected_request, _next)) + + self.assertEqual(allowed.status_code, 200) + self.assertEqual(rejected.status_code, 401) + self.assertNotIn(key, rejected.body.decode("utf-8")) + + overlong = SimpleNamespace( + method="POST", + url=SimpleNamespace(path="/shrink"), + headers={"x-api-key": "k" * 300}, + ) + blocked = asyncio.run(saas_web.require_api_key(overlong, _next)) + self.assertEqual(blocked.status_code, 401) + + def test_startup_hook_skips_when_registry_already_configured(self): + self._load("secret-key") + saas_web.bootstrap_credentials_from_environ() + self.assertIs(saas_web.get_credential_registry(), self.registry) + + def test_startup_hook_skips_blank_transport(self): + saas_web.configure_credential_registry(None) with patch.dict(os.environ): os.environ.pop("CODEC_CARVER_API_KEYS", None) - self.assertEqual(saas_web.get_configured_api_keys(), []) + saas_web.bootstrap_credentials_from_environ() + self.assertIsNone(saas_web.get_credential_registry()) + + def test_startup_hook_imports_transport_keys(self): + saas_web.configure_credential_registry(None) + with patch.dict( + os.environ, + { + "CODEC_CARVER_API_KEYS": "startup-key", + "CODEC_CARVER_CREDENTIAL_DB": self.db_path, + }, + ): + saas_web.bootstrap_credentials_from_environ() + registry = saas_web.get_credential_registry() + self.assertIsNotNone(registry) + self.assertTrue(registry.verify_api_key("startup-key", now=self.now)) @unittest.skipUnless( From 5a80b18b69a3f6758f544c75b5e055af9db92ea3 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 16:08:02 +0000 Subject: [PATCH 3/4] docs: record the API credential registry contract Add APA 7 doctoring, ERD, and the operator next-action for public binds, rotation, and closing issues #329 and #373. Co-authored-by: Seongho Bae --- .jules/sentinel.md | 5 + AGENTS.md | 7 +- CHANGELOG.md | 1 + CLAUDE.md | 5 +- docs/doctoring/api-credential-registry.md | 106 ++++++++++++++++++++++ 5 files changed, 118 insertions(+), 6 deletions(-) create mode 100644 docs/doctoring/api-credential-registry.md diff --git a/.jules/sentinel.md b/.jules/sentinel.md index 9c9d083b..fad93cfa 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -61,6 +61,11 @@ **Learning:** To enhance security in FastAPI applications, missing HTTP response headers could leak referrers or give access to APIs (e.g. geolocation) without explicit intent. **Prevention:** Implement an `@app.middleware('http')` function to globally inject defense-in-depth security headers such as `Content-Security-Policy`, `X-Frame-Options`, `Strict-Transport-Security`, `X-Content-Type-Options`, `X-XSS-Protection`, `Referrer-Policy` (e.g., `strict-origin-when-cross-origin`), and `Permissions-Policy` (e.g., `geolocation=(), microphone=(), camera=()`). +## 2026-08-16 - [Sentinel: Request-time API key environment reads] +**Vulnerability:** API keys compared from `os.environ` on every request, with first-match `hmac.compare_digest` on raw strings and fail-open public binds. +**Learning:** Environment transport is not a verifier store. Hostile Unicode or overlong `X-API-Key` values must stay on a bounded 401 path. Listing APIs that return plaintext recreate the secret. +**Prevention:** Bootstrap keys once into `credential_registry` (SHA-256 digests, two-word `api_credentials` table). Compare every usable digest without short-circuit. Never echo secrets in errors, repr, or public listings. Fail-closed on `0.0.0.0` unless keys exist. + ## 2026-07-10 - [Sentinel: Media Source Path Traversal] **Vulnerability:** Path traversal in `media_shrinker.py` via unresolved `..` segments or symlink escapes before deriving conversion output paths. **Learning:** `Path.relative_to()` is only a lexical containment check unless both the source and root have first been resolved into canonical absolute paths. Relative paths and symlinks can otherwise bypass root-boundary assumptions. diff --git a/AGENTS.md b/AGENTS.md index 090d7d2d..7796f54d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,10 +43,9 @@ repo. job store, and open PRs adding API-key auth and usage metering), so it *will* read runtime secrets/config (API keys, DB creds, endpoints). When you add them, source them from the KV, not `os.getenv`. -- **Known deviation to migrate:** the in-flight API-key auth work reads keys from - a `CODEC_CARVER_API_KEYS` environment variable — that is exactly the anti-pattern - above. Move it to read from the credential registry (env may still be the - bootstrap transport that *populates* the KV, never the runtime source). +- **API keys:** `credential_registry.py` is the request-time source. Env + (`CODEC_CARVER_API_KEYS`) is bootstrap transport into `api_credentials` + only. See [`docs/doctoring/api-credential-registry.md`](docs/doctoring/api-credential-registry.md). ### Code exploration - There is no `.codegraph/` index in this repo today, so use normal search diff --git a/CHANGELOG.md b/CHANGELOG.md index 9313538b..13ccefbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] ### Added +- API 키는 SQLite `api_credentials` 레지스트리에 검증 재료만 저장하고, `CODEC_CARVER_API_KEYS`는 기동 시 수송만 합니다. 공개 바인드는 키가 없으면 실패합니다. 근거는 [`docs/doctoring/api-credential-registry.md`](docs/doctoring/api-credential-registry.md)에 있습니다. - 다중 파일 업로드 선택 시 즉각적인 파일 개수 피드백 및 제한 초과 경고 메시지 추가 - 일괄 업로드 폼에 대상 바이트 프리셋 버튼과 총 파일 크기 미리보기를 추가하여 사용성을 개선했습니다. - 클라이언트 측 폼 검증 시 하드코딩된 '5 GiB' 텍스트를 동적으로 변환되도록 수정하고 일괄 업로드 폼에 최대 크기(MAX_UPLOAD_BYTES) 검증 피드백을 추가했습니다. diff --git a/CLAUDE.md b/CLAUDE.md index cc7870fb..16e459fa 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -23,7 +23,7 @@ python3 -m unittest tests.test_media_shrinker -v python3 -m unittest tests.test_job_store.TestCreateAndGet.test_create_get_roundtrip # Compile check (CI runs this on all four modules) -python -m py_compile media_shrinker.py saas_web.py mcp_driver.py job_store.py +python -m py_compile media_shrinker.py saas_web.py mcp_driver.py job_store.py credential_registry.py # CLI (omit --execute for a dry run that only lists candidates) codec-carver /path/to/recordings --execute --output-dir under_2gb @@ -50,6 +50,7 @@ Four flat top-level modules (declared as `py-modules` in `pyproject.toml`; there - **`saas_web.py`** — single-file FastAPI upload UI (the `[web]` extra; what the Docker image serves). Streams one upload into a temp workspace, calls `media_shrinker.convert_file`, and returns the first generated output as a download. Middleware enforces a 5 GiB upload cap and security headers. Processing is synchronous per request. - **`mcp_driver.py`** — FastMCP server (the `[mcp]` extra) exposing a single `shrink_media` tool that wraps `convert_file`. - **`job_store.py`** — stdlib-only SQLite (WAL) durable job store intended for async/worker job tracking. It is tested but not yet wired into `saas_web.py`. Callers pass `now` explicitly; the store never calls `datetime.now()` itself. +- **`credential_registry.py`** — stdlib-only SQLite verifier store for API keys. Request-time auth reads digests only; `CODEC_CARVER_API_KEYS` is startup transport. Callers pass `now` explicitly. Supporting directories: `fuzz/` holds Atheris harnesses plus seed corpora for the three untrusted-input parsing surfaces (`parse_silencedetect_intervals`, `_parse_probe_payload`, `build_segments`); the same invariants run as Hypothesis property tests in `tests/test_fuzz_properties.py` so they execute in the normal suite. `docs/papers/` holds the fuzzing survey the harness design references. @@ -61,7 +62,7 @@ Supporting directories: `fuzz/` holds Atheris harnesses plus seed corpora for th ## Key conventions - **Never endanger sources.** The scan's selected sources are protected from deletion/overwrite (`protected_sources` / `_ensure_not_protected_source_path`). Generated names keep the full original filename plus a new suffix (`clip.wav.flac`, `meeting.wav.part0001.flac`) so same-stem inputs cannot collide. Keep `--output-dir` a generated-only directory. -- **Stdlib-only core.** `media_shrinker.py` and `job_store.py` must not grow third-party imports; FastAPI/MCP dependencies belong to the optional `web`/`mcp` extras. Tests guard optional imports with `skipUnless` so the suite passes without extras installed. +- **Stdlib-only core.** `media_shrinker.py`, `job_store.py`, and `credential_registry.py` must not grow third-party imports; FastAPI/MCP dependencies belong to the optional `web`/`mcp` extras. Tests guard optional imports with `skipUnless` so the suite passes without extras installed. - **Docstring coverage is 100%.** `interrogate` is configured with `fail-under = 100` (excluding `scripts`, `tests`, `fuzz`) — every module and function, including private helpers, needs a docstring. `.coveragerc` likewise sets `fail_under = 100` over `media_shrinker`, `saas_web`, and `mcp_driver`. - **Security posture.** ffmpeg/ffprobe are always invoked with `-nostdin` and `-protocol_whitelist file,crypto,data` (SSRF/LFI hardening); uploaded filenames are sanitized to a safe basename; temp files use `tempfile` APIs, not predictable names; copied permissions are masked to drop setuid/setgid/sticky bits. `.jules/sentinel.md` logs past vulnerabilities and their prevention rules — check it before touching subprocess invocation, temp-file, or metadata-copy code. `.jules/bolt.md` records performance lessons (pre-resolve paths once, prune walks, avoid repeated `stat`). - **Fuzzing-first for parsers.** Anything that parses ffmpeg/ffprobe output is an untrusted-input surface: parsers must never raise unexpected exception types on arbitrary input (raise `MediaShrinkerError` for invalid payloads). If you change one, update the matching harness in `fuzz/` and its Hypothesis mirror in `tests/test_fuzz_properties.py`. diff --git a/docs/doctoring/api-credential-registry.md b/docs/doctoring/api-credential-registry.md new file mode 100644 index 00000000..0dff0ebc --- /dev/null +++ b/docs/doctoring/api-credential-registry.md @@ -0,0 +1,106 @@ +# API credential registry + +## Decision + +Codec Carver stores API-key verification material in a stdlib SQLite +registry (`credential_registry.py`). `CODEC_CARVER_API_KEYS` is bootstrap +transport only. Request handlers call `CredentialRegistry.verify_api_key` +and never read the process environment. + +Operators should: + +1. Put issued keys in `CODEC_CARVER_API_KEYS` (comma-separated) **or** + call `import_plaintext_keys` against a durable `CODEC_CARVER_CREDENTIAL_DB`. +2. Start the SaaS process so `bootstrap_credentials_from_environ` copies + those keys into `api_credentials` once. +3. Send `X-API-Key` on every upload, job, and download request. `GET /` + stays open so a browser can load the form. +4. For a public bind (`0.0.0.0`), import keys first or the listen policy + fails closed. Local empty-registry work requires explicit loopback + development mode on `127.0.0.1`. +5. Rotate by inserting the next key, distributing it, then revoking the + previous key. Rotated keys stay valid until revoke so in-flight + clients are not dropped. + +## Technical basis + +OWASP API2 (Broken Authentication) requires rejecting unauthenticated +API access and avoiding ad hoc secret comparison on the request path +(OWASP Foundation, 2023). NIST SP 800-63B-4 treats shared secrets as +verifiers: store a non-reversible verifier, compare in a bounded way, and +support authenticator replacement without publishing the secret +(National Institute of Standards and Technology, 2025). HMAC comparison +of equal-length SHA-256 digests follows the keyed-hash compare contract +in RFC 2104 so a guess cannot short-circuit the remaining stored +verifiers (Krawczyk et al., 1997). Fail-closed production binds follow +NIST SSDF PW.1 / PW.5: do not accept network traffic against an empty +authenticator policy (Souppaya et al., 2022). + +PII in recordings is the product. This registry authenticates callers; it +does not mask audio. Protect recordings with access control, retention +(#367), and encryption at rest instead of redacting speech. + +```mermaid +erDiagram + API_CREDENTIALS ||--o{ CREDENTIAL_EVENTS : records + RUNTIME_POLICIES ||--o{ CREDENTIAL_EVENTS : may_audit + API_CREDENTIALS { + text credential_id PK + text key_digest UK + text lifecycle_state + text created_at + text updated_at + text expires_at + text key_label + } + CREDENTIAL_EVENTS { + text event_id PK + text credential_id FK + text event_type + text event_at + text actor_label + } + RUNTIME_POLICIES { + text policy_name PK + text policy_value + text updated_at + } +``` + +Tables use two-or-more-word snake_case. `key_digest` is functionally +dependent on the issued key and replaces it (3NF). Events depend on +`event_id`. Policies depend on `policy_name`. + +## Verification and rollback + +- `tests/test_credential_registry.py` covers import, UTF-8 keys, hostile + headers, constant-work compare, rotation, expiry, revoke, idempotent + bootstrap, public-bind fail-closed, and concurrent verify during rotate. +- `TestApiKeyAuth` proves the request path ignores `CODEC_CARVER_API_KEYS` + after bootstrap and never echoes secrets in 401 bodies. +- Roll back by restoring `get_configured_api_keys()` env reads only if + the registry file is unreadable; do not return plaintext from list APIs. + +## Next action + +After this lands, wire `usage_metering` to `credential_id` instead of +storing plaintext in the single-word `usage` table, and close #329/#373. + +## References + +Krawczyk, H., Bellare, M., & Canetti, R. (1997). *HMAC: Keyed-hashing +for message authentication* (RFC 2104). Internet Engineering Task Force. +https://doi.org/10.17487/RFC2104 + +National Institute of Standards and Technology. (2025). *Digital identity +guidelines: Authentication and authenticator management* (NIST Special +Publication 800-63B-4). https://doi.org/10.6028/NIST.SP.800-63B-4 + +OWASP Foundation. (2023). *API2:2023 Broken authentication*. +https://owasp.org/API-Security/editions/2023/en/0xa2-broken-authentication/ + +Souppaya, M., Scarfone, K., & Dodson, D. (2022). *Secure Software +Development Framework (SSDF) Version 1.1: Recommendations for mitigating +the risk of software vulnerabilities* (NIST Special Publication 800-218). +National Institute of Standards and Technology. +https://doi.org/10.6028/NIST.SP.800-218 From fe1023c708541d9f39dbfda051e000728414265b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 16:10:33 +0000 Subject: [PATCH 4/4] feat(auth): return credential_id from verify and use lifespan Port the two #429 ideas onto this head: verify yields the stable credential_id for usage metering, and FastAPI lifespan runs the named bootstrap instead of a deprecated startup event. Co-authored-by: Seongho Bae --- credential_registry.py | 36 +++++++++++-------- docs/doctoring/api-credential-registry.md | 3 +- saas_web.py | 19 +++++++--- tests/test_credential_registry.py | 44 ++++++++++++----------- tests/test_saas_web.py | 12 ++++++- 5 files changed, 74 insertions(+), 40 deletions(-) diff --git a/credential_registry.py b/credential_registry.py index 7d801daf..90b1082a 100644 --- a/credential_registry.py +++ b/credential_registry.py @@ -451,38 +451,46 @@ def _usable_digests(self, conn: sqlite3.Connection, now: datetime) -> list[str]: usable.append(row["key_digest"]) return usable - def verify_api_key(self, provided: object, *, now: datetime) -> bool: - """Return True when ``provided`` matches a usable stored digest. + def verify_api_key(self, provided: object, *, now: datetime) -> str | None: + """Return the matching ``credential_id``, or ``None``. Comparison always visits every usable digest. Hostile or overlong - headers return False instead of raising. + headers return ``None`` instead of raising. The identifier is the + stable handle usage metering should store instead of plaintext. Args: provided: ``X-API-Key`` value. Non-strings are rejected. now: Comparison timestamp used for expiry. Returns: - True when the header matches an unexpired ``active`` or - ``rotated`` credential. + The matching credential primary key, or ``None``. """ if not isinstance(provided, str): - return False + return None raw = provided.encode("utf-8") if not raw or len(raw) > MAX_KEY_BYTES: - return False + return None try: _reject_control_characters(provided) except CredentialValidationError: - return False + return None provided_digest = digest_api_key(provided) with self._lock, self._connect() as conn: - stored = self._usable_digests(conn, now) - matched = False - for digest in stored: - if hmac.compare_digest(provided_digest, digest): - matched = True - return matched + rows = conn.execute( + "SELECT credential_id, key_digest, expires_at " + "FROM api_credentials " + "WHERE lifecycle_state IN ('active', 'rotated')" + ).fetchall() + now_text = now.isoformat() + matched_id: str | None = None + for row in rows: + expires_at = row["expires_at"] + if expires_at is not None and expires_at <= now_text: + continue + if hmac.compare_digest(provided_digest, row["key_digest"]): + matched_id = row["credential_id"] + return matched_id def has_active_credentials(self, *, now: datetime) -> bool: """Return True when at least one usable credential exists. diff --git a/docs/doctoring/api-credential-registry.md b/docs/doctoring/api-credential-registry.md index 0dff0ebc..092541aa 100644 --- a/docs/doctoring/api-credential-registry.md +++ b/docs/doctoring/api-credential-registry.md @@ -14,7 +14,8 @@ Operators should: 2. Start the SaaS process so `bootstrap_credentials_from_environ` copies those keys into `api_credentials` once. 3. Send `X-API-Key` on every upload, job, and download request. `GET /` - stays open so a browser can load the form. + stays open so a browser can load the form. A successful verify returns + `credential_id` — store that on usage rows, never the plaintext key. 4. For a public bind (`0.0.0.0`), import keys first or the listen policy fails closed. Local empty-registry work requires explicit loopback development mode on `127.0.0.1`. diff --git a/saas_web.py b/saas_web.py index 293e5ed3..5e018a15 100644 --- a/saas_web.py +++ b/saas_web.py @@ -7,7 +7,8 @@ import tempfile import uuid import zipfile -from collections.abc import Mapping +from collections.abc import AsyncIterator, Mapping +from contextlib import asynccontextmanager from datetime import datetime, timezone from pathlib import Path from fastapi import FastAPI, UploadFile, File, BackgroundTasks, Form, Request @@ -154,7 +155,6 @@ def get_configured_api_keys() -> list[str]: ] -@app.on_event("startup") def bootstrap_credentials_from_environ() -> None: """Load ``CODEC_CARVER_API_KEYS`` into the registry at process start. @@ -178,6 +178,17 @@ def bootstrap_credentials_from_environ() -> None: ) +@asynccontextmanager +async def _app_lifespan(_app: FastAPI) -> AsyncIterator[None]: + """Copy bootstrap transport into the registry once per process.""" + + bootstrap_credentials_from_environ() + yield + + +app.router.lifespan_context = _app_lifespan + + @app.middleware("http") async def require_api_key(request: Request, call_next): """Enforce registry authentication on all endpoints except GET /. @@ -197,9 +208,9 @@ async def require_api_key(request: Request, call_next): and not (request.method == "GET" and request.url.path == "/") ): provided_key = request.headers.get("x-api-key", "") - if len(provided_key.encode("utf-8")) > MAX_KEY_BYTES or not registry.verify_api_key( + if len(provided_key.encode("utf-8")) > MAX_KEY_BYTES or registry.verify_api_key( provided_key, now=now - ): + ) is None: return JSONResponse( status_code=401, content={"error": "Invalid or missing API key"}, diff --git a/tests/test_credential_registry.py b/tests/test_credential_registry.py index 21d96eea..3323ae2b 100644 --- a/tests/test_credential_registry.py +++ b/tests/test_credential_registry.py @@ -60,25 +60,29 @@ def test_imported_key_verifies_and_wrong_key_does_not(self) -> None: """A meeting-upload client with the issued key is accepted; a guess is not.""" self.registry.import_plaintext_keys(["meeting-upload-key"], now=T0, source="test") - self.assertTrue(self.registry.verify_api_key("meeting-upload-key", now=T0)) - self.assertFalse(self.registry.verify_api_key("guessed-key", now=T0)) - self.assertFalse(self.registry.verify_api_key("", now=T0)) + credential_id = self.registry.verify_api_key("meeting-upload-key", now=T0) + self.assertIsNotNone(credential_id) + self.assertEqual( + credential_id, self.registry.list_public_records()[0]["credential_id"] + ) + self.assertIsNone(self.registry.verify_api_key("guessed-key", now=T0)) + self.assertIsNone(self.registry.verify_api_key("", now=T0)) def test_non_ascii_key_round_trips(self) -> None: """UTF-8 keys used by non-English operators verify on the same code path.""" key = "업로드-키-αβγ" self.registry.import_plaintext_keys([key], now=T0, source="test") - self.assertTrue(self.registry.verify_api_key(key, now=T0)) - self.assertFalse(self.registry.verify_api_key("업로드-키-αβγ\u0000", now=T0)) + self.assertIsNotNone(self.registry.verify_api_key(key, now=T0)) + self.assertIsNone(self.registry.verify_api_key("업로드-키-αβγ\u0000", now=T0)) def test_hostile_header_types_and_overlong_values_are_false(self) -> None: """A hostile X-API-Key must 401, never raise into the web worker.""" self.registry.import_plaintext_keys(["stable-key"], now=T0, source="test") - self.assertFalse(self.registry.verify_api_key(None, now=T0)) - self.assertFalse(self.registry.verify_api_key(b"stable-key", now=T0)) - self.assertFalse(self.registry.verify_api_key("x" * (MAX_KEY_BYTES + 1), now=T0)) + self.assertIsNone(self.registry.verify_api_key(None, now=T0)) + self.assertIsNone(self.registry.verify_api_key(b"stable-key", now=T0)) + self.assertIsNone(self.registry.verify_api_key("x" * (MAX_KEY_BYTES + 1), now=T0)) def test_verify_compares_every_active_digest(self) -> None: """No first-match short-circuit: every stored digest is visited.""" @@ -95,7 +99,7 @@ def counting_compare(left: str, right: str) -> bool: return real(left, right) with patch("credential_registry.hmac.compare_digest", side_effect=counting_compare): - self.assertTrue(self.registry.verify_api_key("alpha-key", now=T0)) + self.assertIsNotNone(self.registry.verify_api_key("alpha-key", now=T0)) self.assertEqual(len(calls), 3) @@ -123,8 +127,8 @@ def test_bootstrap_is_idempotent(self) -> None: self.registry.revoke("drop-later", now=T1) third = self.registry.bootstrap_from_transport(raw, now=T2, source="env") self.assertEqual(third, 0) - self.assertFalse(self.registry.verify_api_key("drop-later", now=T2)) - self.assertTrue(self.registry.verify_api_key("keep-key", now=T2)) + self.assertIsNone(self.registry.verify_api_key("drop-later", now=T2)) + self.assertIsNotNone(self.registry.verify_api_key("keep-key", now=T2)) class TestValidation(CredentialRegistryTestCase): @@ -165,13 +169,13 @@ def test_rotated_key_still_verifies_until_revoked(self) -> None: self.registry.import_plaintext_keys(["current-key"], now=T0, source="test") self.registry.rotate("current-key", "next-key", now=T1) - self.assertTrue(self.registry.verify_api_key("current-key", now=T1)) - self.assertTrue(self.registry.verify_api_key("next-key", now=T1)) + self.assertIsNotNone(self.registry.verify_api_key("current-key", now=T1)) + self.assertIsNotNone(self.registry.verify_api_key("next-key", now=T1)) states = {row["lifecycle_state"] for row in self.registry.list_public_records()} self.assertEqual(states, {"rotated", "active"}) self.registry.revoke("current-key", now=T2) - self.assertFalse(self.registry.verify_api_key("current-key", now=T2)) - self.assertTrue(self.registry.verify_api_key("next-key", now=T2)) + self.assertIsNone(self.registry.verify_api_key("current-key", now=T2)) + self.assertIsNotNone(self.registry.verify_api_key("next-key", now=T2)) def test_expired_key_does_not_verify(self) -> None: """A time-bounded contractor key stops working after expires_at.""" @@ -182,8 +186,8 @@ def test_expired_key_does_not_verify(self) -> None: source="test", expires_at=T1, ) - self.assertTrue(self.registry.verify_api_key("contractor-key", now=T0)) - self.assertFalse(self.registry.verify_api_key("contractor-key", now=T2)) + self.assertIsNotNone(self.registry.verify_api_key("contractor-key", now=T0)) + self.assertIsNone(self.registry.verify_api_key("contractor-key", now=T2)) self.assertFalse(self.registry.has_active_credentials(now=T2)) @@ -225,7 +229,7 @@ def hammer() -> None: """Verify the live key from a worker thread.""" for _ in range(40): - if not self.registry.verify_api_key("live-key", now=T0): + if self.registry.verify_api_key("live-key", now=T0) is None: errors.append("live-key rejected") workers = [threading.Thread(target=hammer) for _ in range(4)] @@ -235,7 +239,7 @@ def hammer() -> None: for worker in workers: worker.join() self.assertEqual(errors, []) - self.assertTrue(self.registry.verify_api_key("next-live-key", now=T1)) + self.assertIsNotNone(self.registry.verify_api_key("next-live-key", now=T1)) def test_schema_uses_two_word_tables_and_rejects_memory(self) -> None: """Org naming: api_credentials / credential_events / runtime_policies.""" @@ -291,7 +295,7 @@ def test_bootstrap_mapping_applies_loopback_and_bind_policy(self) -> None: now=T0, db_path=self.db_path, ) - self.assertTrue(populated.verify_api_key("prod-key", now=T0)) + self.assertIsNotNone(populated.verify_api_key("prod-key", now=T0)) full = [f"issued-key-{index:02d}" for index in range(MAX_CREDENTIAL_COUNT)] capped_path = os.path.join(self._tmp.name, "capped.db") capped = CredentialRegistry(capped_path) diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py index c5ca8e00..9d581324 100644 --- a/tests/test_saas_web.py +++ b/tests/test_saas_web.py @@ -862,6 +862,16 @@ def test_startup_hook_skips_when_registry_already_configured(self): saas_web.bootstrap_credentials_from_environ() self.assertIs(saas_web.get_credential_registry(), self.registry) + def test_lifespan_invokes_named_bootstrap(self): + self._load("secret-key") + + async def _run() -> None: + async with saas_web._app_lifespan(saas_web.app): + pass + + asyncio.run(_run()) + self.assertIs(saas_web.get_credential_registry(), self.registry) + def test_startup_hook_skips_blank_transport(self): saas_web.configure_credential_registry(None) with patch.dict(os.environ): @@ -881,7 +891,7 @@ def test_startup_hook_imports_transport_keys(self): saas_web.bootstrap_credentials_from_environ() registry = saas_web.get_credential_registry() self.assertIsNotNone(registry) - self.assertTrue(registry.verify_api_key("startup-key", now=self.now)) + self.assertIsNotNone(registry.verify_api_key("startup-key", now=self.now)) @unittest.skipUnless(