diff --git a/backend/app/services/credential_crypto.py b/backend/app/services/credential_crypto.py new file mode 100644 index 0000000..86a45ce --- /dev/null +++ b/backend/app/services/credential_crypto.py @@ -0,0 +1,55 @@ +"""Fernet-based encryption for credential blobs. + +The key is a single 32-byte value (urlsafe-base64-encoded; what +Fernet.generate_key produces) stored at a fixed path inside the +images/data root. Created on first boot if absent; mode 0600. No KDF +needed — the file contents are already maximum-entropy random bytes. + +Operator backup procedure must include this file alongside the rest +of /images/ — losing it makes existing encrypted_blob rows +undecryptable (recovery = delete the rows and re-upload). +""" + +import os +from pathlib import Path + +from cryptography.fernet import Fernet, InvalidToken + + +class InvalidCredentialBlob(Exception): + """Raised when decryption fails (wrong key, tampered blob, …).""" + + +class CredentialCrypto: + """Fernet encrypt/decrypt with an on-disk key file. + + Instantiate with a path; the file is created on first access and + reused thereafter. Tests pass a tmp_path; production calls with + `IMAGES_ROOT / "secrets" / "credential_key.b64"`. + """ + + def __init__(self, key_path: Path): + self._key_path = Path(key_path) + self._fernet = Fernet(self._load_or_create_key()) + + def _load_or_create_key(self) -> bytes: + if self._key_path.exists(): + return self._key_path.read_bytes() + parent = self._key_path.parent + parent.mkdir(parents=True, exist_ok=True) + os.chmod(parent, 0o700) + key = Fernet.generate_key() + self._key_path.write_bytes(key) + os.chmod(self._key_path, 0o600) + return key + + def encrypt(self, plaintext: str) -> bytes: + return self._fernet.encrypt(plaintext.encode("utf-8")) + + def decrypt(self, ciphertext: bytes) -> str: + try: + return self._fernet.decrypt(ciphertext).decode("utf-8") + except InvalidToken as exc: + raise InvalidCredentialBlob( + "credential blob is corrupt or encrypted with a different key" + ) from exc diff --git a/tests/test_credential_crypto.py b/tests/test_credential_crypto.py new file mode 100644 index 0000000..5604ac3 --- /dev/null +++ b/tests/test_credential_crypto.py @@ -0,0 +1,53 @@ +import os +import stat +from pathlib import Path + +import pytest + +from backend.app.services.credential_crypto import ( + CredentialCrypto, + InvalidCredentialBlob, +) + + +pytestmark = pytest.mark.integration + + +def test_load_or_create_writes_key_with_mode_0600(tmp_path): + key_path = tmp_path / "secrets" / "credential_key.b64" + crypto = CredentialCrypto(key_path) + assert key_path.exists() + mode = stat.S_IMODE(os.stat(key_path).st_mode) + assert mode == 0o600 + # parent dir is 0o700 + parent_mode = stat.S_IMODE(os.stat(key_path.parent).st_mode) + assert parent_mode == 0o700 + + +def test_load_existing_key_is_idempotent(tmp_path): + key_path = tmp_path / "credential_key.b64" + crypto1 = CredentialCrypto(key_path) + contents_after_first = key_path.read_bytes() + crypto2 = CredentialCrypto(key_path) + contents_after_second = key_path.read_bytes() + assert contents_after_first == contents_after_second + # And both crypto instances decrypt each other's ciphertext + ct = crypto1.encrypt("hello") + assert crypto2.decrypt(ct) == "hello" + + +def test_encrypt_decrypt_round_trip(tmp_path): + crypto = CredentialCrypto(tmp_path / "k") + plaintext = "domain.com\tTRUE\t/\tTRUE\t1700000000\tname\tvalue" + ct = crypto.encrypt(plaintext) + assert isinstance(ct, bytes) + assert ct != plaintext.encode() + assert crypto.decrypt(ct) == plaintext + + +def test_decrypt_with_wrong_key_raises(tmp_path): + crypto_a = CredentialCrypto(tmp_path / "a") + crypto_b = CredentialCrypto(tmp_path / "b") + ct = crypto_a.encrypt("secret") + with pytest.raises(InvalidCredentialBlob): + crypto_b.decrypt(ct)