diff --git a/backend/secuscan/vault.py b/backend/secuscan/vault.py index 9ec5908ba..1f10eea2a 100644 --- a/backend/secuscan/vault.py +++ b/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: diff --git a/testing/backend/unit/test_saved_views.py b/testing/backend/unit/test_saved_views.py index 62c615051..0db335362 100644 --- a/testing/backend/unit/test_saved_views.py +++ b/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 diff --git a/testing/backend/unit/test_vault.py b/testing/backend/unit/test_vault.py index 187bf0645..b6d6910dc 100644 --- a/testing/backend/unit/test_vault.py +++ b/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) diff --git a/testing/backend/unit/test_vault_security.py b/testing/backend/unit/test_vault_security.py index 82b31ed5c..9270b2121 100644 --- a/testing/backend/unit/test_vault_security.py +++ b/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"))