From 4aeb950dca43845b2e93c45761e2b36a328ef783 Mon Sep 17 00:00:00 2001 From: cowork-bot Date: Sat, 11 Jul 2026 18:49:56 -0400 Subject: [PATCH 1/3] cowork-bot: fail-closed expiry verification in verify_api_key/verify_jwt_token A present-but-unparseable expires_at (naive timestamp, malformed string, or non-string value) previously hit an 'except (ValueError, TypeError): pass' and fell through to status='valid' -- a silent fail-open that let a corrupted, naive, or tampered expiry bypass expiry entirely in a security library. - Add _parse_expiry(): returns an aware UTC datetime, normalizing naive values to UTC so comparisons never raise; returns None only when truly unparseable. - verify_api_key/verify_jwt_token now fail CLOSED: a set-but-unparseable expires_at yields status='invalid' instead of 'valid'. - Harden the key-hash match with hmac.compare_digest (constant-time) and skip entries whose stored hash is missing/non-string. - +7 regression tests (tests/test_verify_expiry_failclosed.py) covering naive past/future, malformed, and non-string expiries for keys and JWTs. check_expiry (display helper) behavior is unchanged (tests lock in None-on- unparseable). 76 tests pass; changed files ruff-clean. --- src/apiauth/keygen.py | 53 +++++++++++----- tests/test_verify_expiry_failclosed.py | 85 ++++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 15 deletions(-) create mode 100644 tests/test_verify_expiry_failclosed.py diff --git a/src/apiauth/keygen.py b/src/apiauth/keygen.py index 37b9ec6..27e6424 100644 --- a/src/apiauth/keygen.py +++ b/src/apiauth/keygen.py @@ -4,6 +4,7 @@ import datetime import hashlib +import hmac import secrets import uuid @@ -179,6 +180,26 @@ def rotate_key( } +def _parse_expiry(exp_str: object) -> datetime.datetime | None: + """Parse a stored ``expires_at`` value into a timezone-aware UTC datetime. + + Returns ``None`` when the value is missing or cannot be parsed. Naive + datetimes (no offset) are assumed to be UTC, so comparing them against an + aware ``now`` never raises ``TypeError``. Callers on the verification path + must treat a ``None`` result for a *present* ``expires_at`` as a failure + (fail closed) -- never as "valid". + """ + if not isinstance(exp_str, str): + return None + try: + exp = datetime.datetime.fromisoformat(exp_str.replace("Z", "+00:00")) + except (ValueError, TypeError): + return None + if exp.tzinfo is None: + exp = exp.replace(tzinfo=UTC) + return exp + + def verify_api_key(keystore: Keystore, api_key: str) -> dict | None: """Verify a plaintext API key against the keystore. @@ -189,17 +210,19 @@ def verify_api_key(keystore: Keystore, api_key: str) -> dict | None: for kid, entry in keystore.get_all().items(): if entry.get("type") != "api_key": continue - if entry.get("key_hash") == key_hash: + stored_hash = entry.get("key_hash") + if isinstance(stored_hash, str) and hmac.compare_digest(stored_hash, key_hash): if entry.get("revoked"): return {"id": kid, "status": "revoked", **entry} - # Check expiry + # Check expiry. A present-but-unparseable expires_at must fail + # CLOSED: returning "valid" here would let a corrupted, naive, or + # tampered timestamp bypass expiry entirely (silent fail-open). if entry.get("expires_at"): - try: - exp = datetime.datetime.fromisoformat(entry["expires_at"].replace("Z", "+00:00")) - if now > exp: - return {"id": kid, "status": "expired", **entry} - except (ValueError, TypeError): - pass + exp = _parse_expiry(entry["expires_at"]) + if exp is None: + return {"id": kid, "status": "invalid", **entry} + if now > exp: + return {"id": kid, "status": "expired", **entry} return {"id": kid, "status": "valid", **entry} return None @@ -228,14 +251,14 @@ def verify_jwt_token(keystore: Keystore, token: str) -> dict | None: if entry.get("revoked"): return {"id": jti, "status": "revoked", **entry} - # Check expiry + # Check expiry. A present-but-unparseable expires_at fails CLOSED + # (status "invalid") rather than silently reporting the JWT as valid. if entry.get("expires_at"): - try: - exp = datetime.datetime.fromisoformat(entry["expires_at"].replace("Z", "+00:00")) - if datetime.datetime.now(UTC) > exp: - return {"id": jti, "status": "expired", **entry} - except (ValueError, TypeError): - pass + exp = _parse_expiry(entry["expires_at"]) + if exp is None: + return {"id": jti, "status": "invalid", **entry} + if datetime.datetime.now(UTC) > exp: + return {"id": jti, "status": "expired", **entry} return {"id": jti, "status": "valid", **entry} diff --git a/tests/test_verify_expiry_failclosed.py b/tests/test_verify_expiry_failclosed.py new file mode 100644 index 0000000..9df0898 --- /dev/null +++ b/tests/test_verify_expiry_failclosed.py @@ -0,0 +1,85 @@ +"""Regression tests: expiry verification must FAIL CLOSED. + +A key or JWT whose stored ``expires_at`` is naive (no offset) or otherwise +unparseable previously slipped through the ``except: pass`` in +``verify_api_key`` / ``verify_jwt_token`` and was reported as ``"valid"`` -- +a silent fail-open in a security library. These tests lock in the corrected +behavior: naive timestamps are compared correctly, and an unparseable expiry +yields ``status == "invalid"`` (never ``"valid"``). +""" + +from __future__ import annotations + +import pytest +import tempfile +from apiauth.keygen import ( + create_api_key_entry, + create_jwt_entry, + verify_api_key, + verify_jwt_token, +) +from apiauth.keystore import Keystore + + +@pytest.fixture +def tmp_keystore(): + with tempfile.TemporaryDirectory() as tmpdir: + yield Keystore(tmpdir) + + +def _set_expiry(ks, key_id, value): + entry = ks.get(key_id) + entry["expires_at"] = value + ks.put(key_id, entry) + + +class TestApiKeyExpiryFailClosed: + def test_naive_past_expiry_is_expired_not_valid(self, tmp_keystore): + r = create_api_key_entry(tmp_keystore, "NaivePast", "api") + _set_expiry(tmp_keystore, r["id"], "2020-01-01T00:00:00") # naive, past + v = verify_api_key(tmp_keystore, r["api_key"]) + assert v is not None + assert v["status"] == "expired" # previously silently "valid" + + def test_naive_future_expiry_is_valid(self, tmp_keystore): + r = create_api_key_entry(tmp_keystore, "NaiveFuture", "api") + _set_expiry(tmp_keystore, r["id"], "2099-01-01T00:00:00") # naive, future + v = verify_api_key(tmp_keystore, r["api_key"]) + assert v is not None + assert v["status"] == "valid" + + def test_malformed_expiry_fails_closed(self, tmp_keystore): + r = create_api_key_entry(tmp_keystore, "Malformed", "api") + _set_expiry(tmp_keystore, r["id"], "not-a-real-date") + v = verify_api_key(tmp_keystore, r["api_key"]) + assert v is not None + assert v["status"] == "invalid" # must NOT be "valid" + + def test_non_string_expiry_fails_closed(self, tmp_keystore): + r = create_api_key_entry(tmp_keystore, "NonString", "api") + _set_expiry(tmp_keystore, r["id"], 1234567890) # e.g. epoch int + v = verify_api_key(tmp_keystore, r["api_key"]) + assert v is not None + assert v["status"] == "invalid" + + def test_valid_zulu_expiry_still_valid(self, tmp_keystore): + r = create_api_key_entry(tmp_keystore, "Zulu", "api") + _set_expiry(tmp_keystore, r["id"], "2099-01-01T00:00:00Z") + v = verify_api_key(tmp_keystore, r["api_key"]) + assert v["status"] == "valid" + + +class TestJwtExpiryFailClosed: + def test_naive_past_expiry_is_expired(self, tmp_keystore): + r = create_jwt_entry(tmp_keystore, "JwtNaivePast", "auth") + _set_expiry(tmp_keystore, r["id"], "2020-01-01T00:00:00") + v = verify_jwt_token(tmp_keystore, r["token"]) + assert v is not None + assert v["status"] == "expired" + + def test_malformed_expiry_fails_closed(self, tmp_keystore): + r = create_jwt_entry(tmp_keystore, "JwtMalformed", "auth") + _set_expiry(tmp_keystore, r["id"], "garbage") + v = verify_jwt_token(tmp_keystore, r["token"]) + assert v is not None + assert v["status"] == "invalid" From dd229ab655b183e977db76281d77160b2291e8c4 Mon Sep 17 00:00:00 2001 From: cowork-bot Date: Sun, 23 Aug 2026 14:50:59 -0400 Subject: [PATCH 2/3] =?UTF-8?q?cowork-bot:=20rotate=5Fkey()/rotate=5Fjwt()?= =?UTF-8?q?=20type=20guards=20=E2=80=94=20no=20more=20silent=20cross-type?= =?UTF-8?q?=20keystore=20corruption?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Calling the API-key rotation path on a JWT entry (or vice versa) silently corrupted the entry: key_hash/prefix overwritten, signing_secret_hash left stale, version bumped — a half-migrated entry neither verifier trusts. Both rotators now raise ValueError on a mismatched entry type, leaving the entry untouched; correct-type rotation behavior is unchanged. Adds tests/test_rotate_type_guard.py (3 tests). Full suite: 79 passed, ruff clean. --- src/apiauth/keygen.py | 14 ++++++++++++ tests/test_rotate_type_guard.py | 39 +++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) create mode 100644 tests/test_rotate_type_guard.py diff --git a/src/apiauth/keygen.py b/src/apiauth/keygen.py index 27e6424..ca56409 100644 --- a/src/apiauth/keygen.py +++ b/src/apiauth/keygen.py @@ -151,6 +151,15 @@ def rotate_key( entry = keystore.get(key_id) if entry is None: return None + if entry.get("type") != "api_key": + # Refuse to corrupt a differently-typed entry: running the API-key + # rotation path over a JWT entry used to overwrite key_hash/prefix, + # bump version, and leave signing_secret_hash stale -- silently + # producing a half-migrated entry that neither verifier trusts. + raise ValueError( + f"rotate_key() requires an 'api_key' entry, got type={entry.get('type')!r} " + f"(key_id={key_id!r}); use rotate_jwt() for JWT entries" + ) new_api_key = generate_api_key() new_hash = hashlib.sha256(new_api_key.encode()).hexdigest() @@ -272,6 +281,11 @@ def rotate_jwt( entry = keystore.get(key_id) if entry is None: return None + if entry.get("type") != "jwt": + raise ValueError( + f"rotate_jwt() requires a 'jwt' entry, got type={entry.get('type')!r} " + f"(key_id={key_id!r}); use rotate_key() for API-key entries" + ) signing_secret = secrets.token_hex(32) import jwt as pyjwt diff --git a/tests/test_rotate_type_guard.py b/tests/test_rotate_type_guard.py new file mode 100644 index 0000000..6cf8cfc --- /dev/null +++ b/tests/test_rotate_type_guard.py @@ -0,0 +1,39 @@ +"""rotate_key()/rotate_jwt() must refuse to operate on mismatched entry types. + +Running the API-key rotation path over a JWT entry (or vice versa) used to +silently corrupt the keystore entry: key_hash/prefix were overwritten, +signing_secret_hash left stale, version bumped. These guards make the +mismatch loud instead of silent. +""" +import pytest +from apiauth.keygen import create_api_key_entry, create_jwt_entry, rotate_jwt, rotate_key +from apiauth.keystore import Keystore + + +def test_rotate_key_on_jwt_raises(tmp_path): + tmp_keystore = Keystore(str(tmp_path / "ks")) + created = create_jwt_entry(tmp_keystore, name="svc", service="api") + with pytest.raises(ValueError, match="rotate_key"): + rotate_key(tmp_keystore, created["id"]) + # Entry is untouched. + assert tmp_keystore.get(created["id"])["type"] == "jwt" + assert "key_hash" not in tmp_keystore.get(created["id"]) + + +def test_rotate_jwt_on_api_key_raises(tmp_path): + tmp_keystore = Keystore(str(tmp_path / "ks")) + created = create_api_key_entry(tmp_keystore, name="k", service="api") + orig = dict(tmp_keystore.get(created["id"])) + with pytest.raises(ValueError, match="rotate_jwt"): + rotate_jwt(tmp_keystore, created["id"]) + assert tmp_keystore.get(created["id"]) == orig + + +def test_correct_type_rotation_still_works(tmp_path): + tmp_keystore = Keystore(str(tmp_path / "ks")) + key_entry = create_api_key_entry(tmp_keystore, name="k", service="api") + rotated = rotate_key(tmp_keystore, key_entry["id"]) + assert rotated["version"] == 2 + jwt_entry = create_jwt_entry(tmp_keystore, name="j", service="api") + rotated_jwt = rotate_jwt(tmp_keystore, jwt_entry["id"]) + assert rotated_jwt["version"] == 2 From 809a012ba0641aef7ebec5f91356b5af091aa7a1 Mon Sep 17 00:00:00 2001 From: cowork-bot Date: Tue, 25 Aug 2026 15:48:27 -0400 Subject: [PATCH 3/3] cowork-bot: atomic keystore persistence + master key corruption guard - _save() now writes to a temp file in the key dir, fsyncs, and os.replace()s into keys.json. Previously an in-place torn write would brick the whole keystore (decrypt-failure -> RuntimeError on every later load). - _get_or_create_master_key() validates a 32-byte master.key; a truncated or corrupt key now fails loudly instead of raising an opaque crypto error or being silently regenerated (which would make all entries undecryptable). - tests/test_keystore_atomic.py: 3 tests covering reloadability, no temp-file leftovers, torn-store refusal to overwrite, corrupt-master-key guard. --- src/apiauth/keystore.py | 41 ++++++++++++++++++++++++++++++--- tests/test_keystore_atomic.py | 43 +++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 3 deletions(-) create mode 100644 tests/test_keystore_atomic.py diff --git a/src/apiauth/keystore.py b/src/apiauth/keystore.py index b0cbaf0..9b8a86b 100644 --- a/src/apiauth/keystore.py +++ b/src/apiauth/keystore.py @@ -4,6 +4,7 @@ import json import os +import tempfile from cryptography.hazmat.primitives.ciphers.aead import AESGCM from pathlib import Path from typing import Any @@ -18,7 +19,19 @@ def _get_or_create_master_key(key_dir: Path) -> bytes: """Load existing master key or generate a new one.""" key_path = key_dir / _KEY_FILE if key_path.exists(): - return key_path.read_bytes() + key = key_path.read_bytes() + if len(key) != 32: + # A truncated or otherwise malformed master.key would otherwise + # surface as an opaque cryptography error (or worse, get silently + # regenerated below, bricking every existing entry). Fail loudly + # instead so the real master key can be restored. + raise RuntimeError( + f"Master key at {key_path} is corrupt: expected 32 bytes, " + f"got {len(key)}. Refusing to regenerate it, which would make " + "every existing keystore entry undecryptable. Restore the " + "original master.key from backup." + ) + return key key_dir.mkdir(parents=True, exist_ok=True) key = AESGCM.generate_key(bit_length=256) @@ -65,11 +78,33 @@ def _load(self) -> None: ) from exc def _save(self) -> None: + """Persist the store atomically. + + Writing keys.json in place meant a crash mid-write could leave a torn + ciphertext behind -- and since _load() refuses to continue on decrypt + failure (by design), that tear would brick the whole keystore. Write + to a temp file in the same directory, fsync, then os.replace() so + readers only ever see a fully-written store. + """ plaintext = json.dumps(self._entries, indent=2, default=str).encode("utf-8") nonce = os.urandom(12) ciphertext = self._aesgcm.encrypt(nonce, plaintext, None) - self._store_path.write_bytes(nonce + ciphertext) - os.chmod(str(self._store_path), 0o600) + fd, tmp_path = tempfile.mkstemp( + dir=str(self.key_dir), prefix=".keys-", suffix=".tmp" + ) + try: + with os.fdopen(fd, "wb") as fh: + fh.write(nonce + ciphertext) + fh.flush() + os.fsync(fh.fileno()) + os.chmod(tmp_path, 0o600) + os.replace(tmp_path, str(self._store_path)) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + pass + raise def get_all(self) -> dict[str, dict[str, Any]]: """Return all stored entries.""" diff --git a/tests/test_keystore_atomic.py b/tests/test_keystore_atomic.py new file mode 100644 index 0000000..12d9a3a --- /dev/null +++ b/tests/test_keystore_atomic.py @@ -0,0 +1,43 @@ +"""Atomic keystore persistence + master key validation.""" +import os + +import pytest + +from apiauth.keystore import Keystore, _get_or_create_master_key + + +def _make_keystore(tmp_path): + return Keystore(key_dir=tmp_path) + + +def test_save_is_atomic_and_reloadable(tmp_path): + ks = _make_keystore(tmp_path) + ks.put("k1", {"type": "api_key", "name": "n", "service": "s"}) + # No temp files left behind after a successful save. + leftovers = [p for p in os.listdir(tmp_path) if p.startswith(".keys-")] + assert leftovers == [] + # A fresh instance reads back exactly what was written. + ks2 = Keystore(key_dir=tmp_path) + assert ks2.get("k1")["name"] == "n" + + +def test_torn_store_does_not_silently_overwrite_entries(tmp_path): + """Simulate a torn write: garbage in keys.json must raise, never reset.""" + ks = _make_keystore(tmp_path) + ks.put("k1", {"type": "api_key", "name": "n"}) + store = tmp_path / "keys.json" + store.write_bytes(b"\x00" * 64) + with pytest.raises(RuntimeError, match="Failed to decrypt"): + Keystore(key_dir=tmp_path) + # The corrupt file was NOT replaced by an empty store. + assert store.stat().st_size == 64 + + +def test_corrupt_master_key_fails_loudly(tmp_path): + ks = _make_keystore(tmp_path) + ks.put("k1", {"type": "api_key", "name": "n"}) + (tmp_path / "master.key").write_bytes(b"short") + with pytest.raises(RuntimeError, match="corrupt: expected 32 bytes"): + _get_or_create_master_key(tmp_path) + # ...and the bad key was not silently regenerated over. + assert (tmp_path / "master.key").read_bytes() == b"short"