fix(auth): key rate-limit bucket by key_id/hash, never plaintext API key (S-6)

src/serving/api/auth/manager.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import secrets
@@ -541,7 +542,21 @@ def _legacy_env_keys(self) -> list[TenantKey]:
         return items
 
     def _rate_limit_key(self, tenant_key: TenantKey) -> str:
-        return tenant_key.key or tenant_key.key_hash or tenant_key.name
+        # This string becomes a Redis sorted-set key NAME (``rate_limiter`` calls
+        # ``zadd(key, ...)``), so it must never be the plaintext API key: a Redis
+        # read (SCAN / MONITOR / a backup / a shared cache) would otherwise
+        # recover live keys straight from the bucket names. Key the bucket by the
+        # stable non-secret ``key_id``; then the stored hash; then a hash of the
+        # plaintext (never the plaintext itself); finally the non-secret name.
+        # Each distinct key still maps to its own bucket, so per-key rate-limit
+        # isolation is unchanged. (audit S-6)
+        if tenant_key.key_id:
+            return f"kid:{tenant_key.key_id}"
+        if tenant_key.key_hash:
+            return f"kh:{tenant_key.key_hash}"
+        if tenant_key.key:
+            return f"kh:{hashlib.sha256(tenant_key.key.encode()).hexdigest()}"
+        return f"name:{tenant_key.name}"
 
     def _matches_key_material(self, item: TenantKey, value: str) -> bool:
         if item.key is not None and secrets.compare_digest(item.key, value):

tests/unit/test_auth_manager_mutation.py

@@ -532,6 +532,35 @@ def test_is_rate_limited_drops_stamp_exactly_at_cutoff(self) -> None:
         # would block at exactly the window edge.
         assert m.is_rate_limited(tenant_key) is False
 
+    def test_bucket_key_never_contains_plaintext_api_key(self) -> None:
+        # The bucket key becomes a Redis sorted-set key name, so it must not leak
+        # the plaintext API key to anyone who can read Redis. (audit S-6)
+        m = _build_manager()
+        plaintext_key = "af-prod-acme-agent-super-secret-abc123xyz"  # noqa: S105
+        bucket = m._rate_limit_key(_key(key=plaintext_key, key_id=None, key_hash=None))
+        assert plaintext_key not in bucket
+        assert bucket.startswith("kh:")
+
+    def test_bucket_key_prefers_key_id_over_secret_material(self) -> None:
+        m = _build_manager()
+        bucket = m._rate_limit_key(_key(key="plaintext", key_id="kid-42", key_hash="stored"))
+        assert bucket == "kid:kid-42"
+
+    def test_bucket_key_uses_stored_hash_when_no_key_id(self) -> None:
+        m = _build_manager()
+        bucket = m._rate_limit_key(_key(key="plaintext", key_id=None, key_hash="stored-hash"))
+        assert bucket == "kh:stored-hash"
+
+    def test_bucket_key_is_deterministic_and_per_key_distinct(self) -> None:
+        # Same material -> same bucket (rate limiting still accumulates); different
+        # material -> different bucket (per-key isolation preserved).
+        m = _build_manager()
+        a1 = m._rate_limit_key(_key(key="alpha", key_id=None, key_hash=None))
+        a2 = m._rate_limit_key(_key(key="alpha", key_id=None, key_hash=None))
+        b = m._rate_limit_key(_key(key="beta", key_id=None, key_hash=None))
+        assert a1 == a2
+        assert a1 != b
+
     @pytest.mark.asyncio
     async def test_check_rate_limit_applies_local_window_when_redis_reports_full(self) -> None:
         m = _build_manager(
@@ -570,13 +599,15 @@ async def test_check_rate_limit_reset_uses_first_window_stamp(self) -> None:
 
 
 class TestKeyingAndAuthorization:
-    def test_rate_limit_key_prefers_plaintext_key(self) -> None:
+    def test_rate_limit_key_prefers_hash_over_plaintext(self) -> None:
+        # (audit S-6) the bucket key becomes a Redis key name, so keying by the
+        # stored hash in preference to the plaintext keeps live keys out of Redis.
         m = _build_manager()
-        assert m._rate_limit_key(_key(key="k", key_hash="h")) == "k"
+        assert m._rate_limit_key(_key(key="k", key_hash="h")) == "kh:h"
 
-    def test_rate_limit_key_falls_back_to_hash_when_no_plaintext(self) -> None:
+    def test_rate_limit_key_uses_hash_when_no_plaintext(self) -> None:
         m = _build_manager()
-        assert m._rate_limit_key(_key(key=None, key_hash="h")) == "h"
+        assert m._rate_limit_key(_key(key=None, key_hash="h")) == "kh:h"
 
     def test_is_entity_allowed_true_when_unrestricted(self) -> None:
         m = _build_manager()