Security: fail-closed expiry verification + constant-time key compare

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}
 

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"