diff --git a/backend/requirements.txt b/backend/requirements.txt index 89748e34..0c650250 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -10,5 +10,4 @@ python-multipart>=0.0.9 xhtml2pdf>=0.2.17 aiosqlite>=0.20.0 python-whois>=0.9.4 -httpx>=0.28.1 -openai>=1.0.0 +cryptography>=43.0.0 diff --git a/backend/secuscan/vault.py b/backend/secuscan/vault.py index 9ec5908b..fc80c231 100644 --- a/backend/secuscan/vault.py +++ b/backend/secuscan/vault.py @@ -6,65 +6,40 @@ import hashlib import os +from cryptography.exceptions import InvalidTag from cryptography.hazmat.primitives.ciphers.aead import AESGCM class VaultCrypto: - """AES-256-GCM authenticated encryption for stored credentials. + """Symmetric encryption helper backed by AES-256-GCM. - Each call to encrypt() generates a fresh random 12-byte nonce so no two - ciphertexts ever share a nonce under the same key. The GCM auth tag - (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) + Uses a proper authenticated stream cipher instead of a hand-rolled + XOR keystream, so encrypting messages longer than one block does not + reuse keystream bytes and cannot be attacked via crib-dragging. """ - _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. - _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 + _NONCE_SIZE = 12 # recommended nonce size for AES-GCM def __init__(self, key: bytes): - """ - Args: - key: 44-byte base64url-encoded representation of a 32-byte AES-256 key, - as produced by ``settings.resolved_vault_key``. - """ - try: - raw = base64.urlsafe_b64decode(key) - except Exception as exc: - raise ValueError("Vault key must be base64url-encoded") from exc - if len(raw) != 32: - 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) + # Normalize the provided key to exactly 32 bytes (AES-256) regardless + # of the length of the key material passed in. + self._aead = AESGCM(hashlib.sha256(key).digest()) def encrypt(self, plaintext: str) -> str: - nonce = os.urandom(self._NONCE_LEN) - ciphertext = self._aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None) + raw = plaintext.encode("utf-8") + nonce = os.urandom(self._NONCE_SIZE) + ciphertext = self._aead.encrypt(nonce, raw, None) blob = nonce + ciphertext return base64.urlsafe_b64encode(blob).decode("ascii") def decrypt(self, payload: str) -> str: - try: - blob = base64.urlsafe_b64decode(payload.encode("ascii")) - except Exception as exc: - raise ValueError("Vault payload is not valid base64url") from exc - - nonce = blob[: self._NONCE_LEN] - ciphertext = blob[self._NONCE_LEN :] + blob = base64.urlsafe_b64decode(payload.encode("ascii")) + nonce = blob[: self._NONCE_SIZE] + ciphertext = blob[self._NONCE_SIZE :] try: - raw = self._aesgcm.decrypt(nonce, ciphertext, None) - except Exception as exc: + raw = self._aead.decrypt(nonce, ciphertext, None) + except InvalidTag as exc: raise ValueError("Vault payload integrity verification failed") from exc return raw.decode("utf-8") diff --git a/testing/backend/unit/test_vault.py b/testing/backend/unit/test_vault.py index 187bf064..247f8fa5 100644 --- a/testing/backend/unit/test_vault.py +++ b/testing/backend/unit/test_vault.py @@ -1,3 +1,9 @@ +import base64 +import os + +import pytest + +from backend.secuscan.config import settings from backend.secuscan.vault import VaultCrypto @@ -91,61 +97,62 @@ def test_decrypt_roundtrip_empty_string(): encrypted = crypto.encrypt("") assert encrypted decrypted = crypto.decrypt(encrypted) - assert decrypted == "" - - -class TestKeyFingerprint: - def test_key_fingerprint_is_accessible(self): - """key_fingerprint property is accessible on VaultCrypto instance.""" - import base64 - key_bytes = b"a1" + b"b2" * 15 # exactly 32 bytes - key = base64.urlsafe_b64encode(key_bytes).decode("ascii") - crypto = VaultCrypto(key) - assert hasattr(crypto, "key_fingerprint") - assert crypto.key_fingerprint is not None - - def test_key_fingerprint_matches_compute_fingerprint(self): - """key_fingerprint returns the same value as _compute_fingerprint on the raw key.""" - import base64 - key_bytes = b"c3" + b"d4" * 15 # exactly 32 bytes - key = base64.urlsafe_b64encode(key_bytes).decode("ascii") - crypto = VaultCrypto(key) - expected = VaultCrypto._compute_fingerprint(key_bytes) - assert crypto.key_fingerprint == expected - - def test_key_fingerprint_is_stable(self): - """Calling key_fingerprint multiple times returns the same value.""" - key = _make_key() - crypto = VaultCrypto(key) - fp1 = crypto.key_fingerprint - fp2 = crypto.key_fingerprint - fp3 = crypto.key_fingerprint - assert fp1 == fp2 == fp3 - - def test_key_fingerprint_is_colon_separated_hex(self): - """Fingerprint is formatted as 8 colon-separated lowercase hex pairs.""" - import base64 - key_bytes = b"0123456789abcdef0123456789abcdef" - key = base64.urlsafe_b64encode(key_bytes).decode("ascii") - crypto = VaultCrypto(key) - fp = crypto.key_fingerprint - parts = fp.split(":") - assert len(parts) == 8 - for part in parts: - assert len(part) == 2 - int(part, 16) # raises ValueError if not valid hex - - def test_different_keys_produce_different_fingerprints(self): - """Two distinct keys produce two distinct fingerprints.""" - import base64 - key1 = base64.urlsafe_b64encode(b"11111111111111111111111111111111").decode("ascii") - key2 = base64.urlsafe_b64encode(b"22222222222222222222222222222222").decode("ascii") - crypto1 = VaultCrypto(key1) - crypto2 = VaultCrypto(key2) - assert crypto1.key_fingerprint != crypto2.key_fingerprint - - def test_key_fingerprint_property_is_read_only(self): - """key_fingerprint has no setter — it cannot be overwritten.""" - key = _make_key() - crypto = VaultCrypto(key) - assert not hasattr(type(crypto).key_fingerprint, "fset") or type(crypto).key_fingerprint.fset is None + assert decrypted == secret + + +@pytest.fixture +def crypto(): + return VaultCrypto(b"test-vault-key-for-unit-tests") + + +def test_round_trip_plaintext_longer_than_one_block(crypto): + """Regression test for #302: plaintext longer than the old 32-byte + repeating keystream must still round-trip correctly under AES-GCM, + which has no fixed-size repeating keystream to exploit.""" + plaintext = "a" * 500 + encrypted = crypto.encrypt(plaintext) + assert crypto.decrypt(encrypted) == plaintext + + +def test_encrypt_is_nondeterministic(crypto): + """Each encryption should use a fresh random nonce, so the same + plaintext never produces the same ciphertext twice.""" + plaintext = "same-secret-every-time" + first = crypto.encrypt(plaintext) + second = crypto.encrypt(plaintext) + assert first != second + assert crypto.decrypt(first) == plaintext + assert crypto.decrypt(second) == plaintext + + +def test_tampered_ciphertext_is_rejected(crypto): + plaintext = "do-not-tamper-with-me" + encrypted = crypto.encrypt(plaintext) + + blob = bytearray(base64.urlsafe_b64decode(encrypted)) + blob[-1] ^= 0x01 + tampered_str = base64.urlsafe_b64encode(bytes(blob)).decode("ascii") + + with pytest.raises(ValueError, match="integrity verification failed"): + crypto.decrypt(tampered_str) + + +def test_wrong_key_cannot_decrypt(crypto): + plaintext = "only-the-right-key-should-work" + encrypted = crypto.encrypt(plaintext) + + wrong_crypto = VaultCrypto(b"a-completely-different-key") + with pytest.raises(ValueError, match="integrity verification failed"): + wrong_crypto.decrypt(encrypted) + + +def test_old_xor_format_payload_is_rejected_not_silently_garbled(crypto): + """Ciphertext produced by the old hand-rolled XOR scheme (nonce + + 32-byte HMAC signature + ciphertext) must fail loudly under the new + AES-GCM implementation rather than decrypting to garbage silently. + Documents the breaking change for anyone migrating stored secrets.""" + fake_old_blob = os.urandom(16) + os.urandom(32) + b"some-old-ciphertext-bytes" + fake_old_payload = base64.urlsafe_b64encode(fake_old_blob).decode("ascii") + + with pytest.raises(ValueError, match="integrity verification failed"): + crypto.decrypt(fake_old_payload)