feat(vault): add key-version header and rotation keyring support (#1770)

backend/secuscan/vault.py

@@ -5,6 +5,7 @@
 import base64
 import hashlib
 import os
+from typing import Dict, List, Optional
 
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 
@@ -17,23 +18,56 @@ class VaultCrypto:
     (16 bytes, appended by AESGCM) provides both confidentiality and integrity -
     any tampering causes decrypt() to raise ValueError.
 
-    Wire format (base64url): nonce(12) || ciphertext || auth_tag(16)
+    Wire format v1 (base64url):
+        header_prefix(4) || key_id_bytes(8) || nonce(12) || ciphertext || auth_tag(16)
+
+    Legacy wire format (base64url):
+        nonce(12) || ciphertext || auth_tag(16)
     """
 
+    _HEADER_PREFIX = b"SV1:"
     _NONCE_LEN = 12
-
-    # Domain-separation prefix so the fingerprint can never collide with any other use of the key material as a hash input.
-    # So the digest is a dedicated identifier rather than a reusable oracle.
+    _TAG_LEN = 16
     _FINGERPRINT_DOMAIN = b"secuscan/vault-key-fingerprint/v1"
-    # 8 bytes (64 bits) is plenty to distinguish keys for rotation checks while keeping the value short and obviously non-recoverable.
     _FINGERPRINT_BYTES = 8
 
-    def __init__(self, key: bytes):
+    # Minimum payload lengths:
+    # Legacy: 12 (nonce) + 16 (tag) = 28 bytes
+    # Versioned: 4 (prefix) + 8 (key_id) + 12 (nonce) + 16 (tag) = 40 bytes
+    _MIN_LEGACY_LEN = _NONCE_LEN + _TAG_LEN
+    _MIN_VERSIONED_LEN = len(_HEADER_PREFIX) + _FINGERPRINT_BYTES + _NONCE_LEN + _TAG_LEN
+
+    def __init__(self, key: bytes, fallback_keys: Optional[List[bytes]] = None):
         """
         Args:
             key: 44-byte base64url-encoded representation of a 32-byte AES-256 key,
                  as produced by ``settings.resolved_vault_key``.
+            fallback_keys: Optional list of secondary/old 44-byte base64url-encoded AES-256 keys
+                 used to support decryption during key rotation.
         """
+        raw_primary = self._decode_key(key)
+        self._primary_key_id_bytes = self._compute_fingerprint_bytes(raw_primary)
+        self._key_fingerprint = self._format_fingerprint(self._primary_key_id_bytes)
+
+        # Map key_id_bytes -> AESGCM cipher
+        self._keyring: Dict[bytes, AESGCM] = {}
+        self._keyring_list: List[AESGCM] = []
+
+        primary_cipher = AESGCM(raw_primary)
+        self._keyring[self._primary_key_id_bytes] = primary_cipher
+        self._keyring_list.append(primary_cipher)
+
+        if fallback_keys:
+            for f_key in fallback_keys:
+                raw_fb = self._decode_key(f_key)
+                fb_key_id = self._compute_fingerprint_bytes(raw_fb)
+                fb_cipher = AESGCM(raw_fb)
+                if fb_key_id not in self._keyring:
+                    self._keyring[fb_key_id] = fb_cipher
+                    self._keyring_list.append(fb_cipher)
+
+    @classmethod
+    def _decode_key(cls, key: bytes) -> bytes:
         try:
             raw = base64.urlsafe_b64decode(key)
         except Exception as exc:
@@ -42,15 +76,14 @@ def __init__(self, key: bytes):
             raise ValueError(
                 f"Vault key must decode to exactly 32 bytes (AES-256); got {len(raw)}"
             )
-        self._aesgcm = AESGCM(raw)
-        # Compute the fingerprint at construction and retain only the resulting string, never the raw key bytes.
-        # So the instance keeps no extra copy of the key material beyond what AESGCM already holds internally.
-        self._key_fingerprint = self._compute_fingerprint(raw)
+        return raw
 
     def encrypt(self, plaintext: str) -> str:
         nonce = os.urandom(self._NONCE_LEN)
-        ciphertext = self._aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
-        blob = nonce + ciphertext
+        primary_cipher = self._keyring[self._primary_key_id_bytes]
+        ciphertext = primary_cipher.encrypt(nonce, plaintext.encode("utf-8"), None)
+        header = self._HEADER_PREFIX + self._primary_key_id_bytes
+        blob = header + nonce + ciphertext
         return base64.urlsafe_b64encode(blob).decode("ascii")
 
     def decrypt(self, payload: str) -> str:
@@ -59,22 +92,71 @@ def decrypt(self, payload: str) -> str:
         except Exception as exc:
             raise ValueError("Vault payload is not valid base64url") from exc
 
+        if len(blob) < self._MIN_LEGACY_LEN:
+            raise ValueError("Vault payload is too short")
+
+        # Attempt versioned decryption if header is present
+        if blob.startswith(self._HEADER_PREFIX) and len(blob) >= self._MIN_VERSIONED_LEN:
+            prefix_len = len(self._HEADER_PREFIX)
+            key_id = blob[prefix_len : prefix_len + self._FINGERPRINT_BYTES]
+            nonce_start = prefix_len + self._FINGERPRINT_BYTES
+            nonce = blob[nonce_start : nonce_start + self._NONCE_LEN]
+            ciphertext = blob[nonce_start + self._NONCE_LEN :]
+
+            cipher = self._keyring.get(key_id)
+            if cipher is not None:
+                try:
+                    raw = cipher.decrypt(nonce, ciphertext, None)
+                    return raw.decode("utf-8")
+                except Exception:
+                    # GCM failure on versioned payload; will attempt legacy fallback below
+                    pass
+
+        # Legacy unversioned format fallback (nonce(12) || ciphertext) or 1-in-2^32 header collision fallback
         nonce = blob[: self._NONCE_LEN]
         ciphertext = blob[self._NONCE_LEN :]
 
-        try:
-            raw = self._aesgcm.decrypt(nonce, ciphertext, None)
-        except Exception as exc:
-            raise ValueError("Vault payload integrity verification failed") from exc
+        for cipher in self._keyring_list:
+            try:
+                raw = cipher.decrypt(nonce, ciphertext, None)
+                return raw.decode("utf-8")
+            except Exception:
+                continue
+
+        raise ValueError("Vault payload integrity verification failed")
 
-        return raw.decode("utf-8")
+    @classmethod
+    def _compute_fingerprint_bytes(cls, raw_key: bytes) -> bytes:
+        """Derive the 8-byte raw fingerprint for 32-byte key material."""
+        digest = hashlib.sha256(cls._FINGERPRINT_DOMAIN + raw_key).digest()
+        return digest[: cls._FINGERPRINT_BYTES]
+
+    @classmethod
+    def _format_fingerprint(cls, fingerprint_bytes: bytes) -> str:
+        """Format 8-byte raw fingerprint as colon-separated hex pairs."""
+        return ":".join(f"{byte:02x}" for byte in fingerprint_bytes)
 
     @classmethod
     def _compute_fingerprint(cls, raw_key: bytes) -> str:
         """Derive the colon-separated hex fingerprint for raw 32-byte key material."""
-        digest = hashlib.sha256(cls._FINGERPRINT_DOMAIN + raw_key).digest()
-        truncated = digest[: cls._FINGERPRINT_BYTES]
-        return ":".join(f"{byte:02x}" for byte in truncated)
+        return cls._format_fingerprint(cls._compute_fingerprint_bytes(raw_key))
+
+    @classmethod
+    def extract_key_id(cls, payload: str) -> Optional[str]:
+        """Extract the colon-separated hex key fingerprint from a versioned vault blob payload if present.
+
+        Returns None if the payload is legacy/unversioned or invalid.
+        """
+        try:
+            blob = base64.urlsafe_b64decode(payload.encode("ascii"))
+        except Exception:
+            return None
+
+        if blob.startswith(cls._HEADER_PREFIX) and len(blob) >= cls._MIN_VERSIONED_LEN:
+            prefix_len = len(cls._HEADER_PREFIX)
+            key_id_bytes = blob[prefix_len : prefix_len + cls._FINGERPRINT_BYTES]
+            return cls._format_fingerprint(key_id_bytes)
+        return None
 
     @property
     def key_fingerprint(self) -> str:

testing/backend/unit/test_saved_views.py

@@ -357,19 +357,37 @@ async def test_filter_json_with_null_values_rejected(app_client: AsyncClient):
 @pytest.mark.asyncio
 async def test_unauthenticated_request_rejected(app_client: AsyncClient):
     """Requests without a valid API key/session are rejected, not served."""
-    res = await app_client.get(
-        "/api/v1/saved-views", headers={"X-Api-Key": ""}
-    )
-    assert res.status_code == 401
+    app = app_client.test_transport.app
+    orig_override = app.dependency_overrides.get(require_api_key)
+    if require_api_key in app.dependency_overrides:
+        del app.dependency_overrides[require_api_key]
+
+    try:
+        res = await app_client.get(
+            "/api/v1/saved-views", headers={"X-Api-Key": ""}
+        )
+        assert res.status_code == 401
+    finally:
+        if orig_override is not None:
+            app.dependency_overrides[require_api_key] = orig_override
 
 
 @pytest.mark.asyncio
 async def test_wrong_api_key_rejected(app_client: AsyncClient):
     """A malformed/incorrect API key is rejected."""
-    res = await app_client.get(
-        "/api/v1/saved-views", headers={"X-Api-Key": "not-the-real-key"}
-    )
-    assert res.status_code == 401
+    app = app_client.test_transport.app
+    orig_override = app.dependency_overrides.get(require_api_key)
+    if require_api_key in app.dependency_overrides:
+        del app.dependency_overrides[require_api_key]
+
+    try:
+        res = await app_client.get(
+            "/api/v1/saved-views", headers={"X-Api-Key": "not-the-real-key"}
+        )
+        assert res.status_code == 401
+    finally:
+        if orig_override is not None:
+            app.dependency_overrides[require_api_key] = orig_override
 
 
 @pytest.mark.asyncio

testing/backend/unit/test_vault.py

@@ -149,3 +149,96 @@ def test_key_fingerprint_property_is_read_only(self):
         key = _make_key()
         crypto = VaultCrypto(key)
         assert not hasattr(type(crypto).key_fingerprint, "fset") or type(crypto).key_fingerprint.fset is None
+
+
+def test_encrypt_includes_header_prefix_and_key_id():
+    """encrypt() output wire format starts with SV1: header prefix and key ID bytes."""
+    import base64
+    key = _make_key()
+    crypto = VaultCrypto(key)
+    encrypted = crypto.encrypt("my_secret")
+    raw = base64.urlsafe_b64decode(encrypted.encode("ascii"))
+    assert raw.startswith(b"SV1:")
+    extracted_key_id = crypto.extract_key_id(encrypted)
+    assert extracted_key_id == crypto.key_fingerprint
+
+
+def test_extract_key_id_returns_none_for_legacy_payload():
+    """extract_key_id returns None when passed a legacy unversioned payload."""
+    import base64
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    key = _make_key()
+    crypto = VaultCrypto(key)
+    raw_key = base64.urlsafe_b64decode(key)
+
+    # Construct legacy blob manually: nonce(12) + ciphertext + tag(16)
+    nonce = b"123456789012"
+    aesgcm = AESGCM(raw_key)
+    ct = aesgcm.encrypt(nonce, b"secret", None)
+    legacy_payload = base64.urlsafe_b64encode(nonce + ct).decode("ascii")
+
+    assert crypto.extract_key_id(legacy_payload) is None
+    # Decrypting legacy payload must still succeed
+    assert crypto.decrypt(legacy_payload) == "secret"
+
+
+def test_vault_keyring_rotation():
+    """VaultCrypto with fallback_keys decrypts blobs encrypted with primary or fallback keys."""
+    import base64
+    key_old = base64.urlsafe_b64encode(b"11111111111111111111111111111111").decode("ascii")
+    key_new = base64.urlsafe_b64encode(b"22222222222222222222222222222222").decode("ascii")
+
+    crypto_old = VaultCrypto(key_old)
+    encrypted_old = crypto_old.encrypt("old_secret")
+
+    # New active crypto instance with fallback_keys
+    crypto_new = VaultCrypto(key_new, fallback_keys=[key_old])
+
+    encrypted_new = crypto_new.encrypt("new_secret")
+
+    # Should decrypt both old and new encrypted payloads
+    assert crypto_new.decrypt(encrypted_new) == "new_secret"
+    assert crypto_new.decrypt(encrypted_old) == "old_secret"
+
+
+def test_header_prefix_collision_fallback():
+    """A legacy blob whose nonce starts with b'SV1:' is correctly decrypted via fallback."""
+    import base64
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    key = _make_key()
+    crypto = VaultCrypto(key)
+    raw_key = base64.urlsafe_b64decode(key)
+
+    # Construct legacy nonce that starts with b"SV1:"
+    nonce = b"SV1:12345678"  # exactly 12 bytes
+    aesgcm = AESGCM(raw_key)
+    ct = aesgcm.encrypt(nonce, b"collision_secret", None)
+    legacy_collision_payload = base64.urlsafe_b64encode(nonce + ct).decode("ascii")
+
+    # decrypt must not raise ValueError, but successfully decrypt via legacy fallback
+    assert crypto.decrypt(legacy_collision_payload) == "collision_secret"
+
+
+def test_payload_length_guards():
+    """decrypt raises ValueError when payload length is below minimum required bytes."""
+    import base64
+    key = _make_key()
+    crypto = VaultCrypto(key)
+
+    # 1. Below 28 bytes (below minimum legacy length)
+    short_raw = b"x" * 20
+    short_payload = base64.urlsafe_b64encode(short_raw).decode("ascii")
+    try:
+        crypto.decrypt(short_payload)
+        raise AssertionError("Expected ValueError for short payload")
+    except ValueError as exc:
+        assert "too short" in str(exc)
+
+    # 2. Versioned header present (SV1:) but total length < 40 bytes
+    short_versioned_raw = b"SV1:" + b"x" * 25
+    short_versioned_payload = base64.urlsafe_b64encode(short_versioned_raw).decode("ascii")
+    try:
+        crypto.decrypt(short_versioned_payload)
+        raise AssertionError("Expected ValueError for short versioned payload")
+    except ValueError as exc:
+        assert "verification failed" in str(exc) or "too short" in str(exc)

testing/backend/unit/test_vault_security.py

@@ -35,11 +35,11 @@ def test_unique_ciphertexts_per_call(self):
         assert c1 != c2, "Each call must use a fresh nonce"
 
     def test_blob_structure_has_12_byte_nonce(self):
-        """Blob starts with a 12-byte nonce (AES-GCM standard)."""
+        """Blob has versioned header SV1:(4) + key ID(8) + 12-byte nonce (AES-GCM standard)."""
         crypto = VaultCrypto(_make_key("test"))
         blob = base64.urlsafe_b64decode(crypto.encrypt("x").encode())
-        # nonce(12) + 1-byte plaintext + 16-byte auth_tag = 29 bytes total
-        assert len(blob) == 29
+        # header(4) + key_id(8) + nonce(12) + 1-byte plaintext + 16-byte auth_tag = 41 bytes total
+        assert len(blob) == 41
 
     def test_tamper_ciphertext_raises(self):
         crypto = VaultCrypto(_make_key("test"))