feat(fc3b): CredentialCrypto — Fernet with on-disk system key

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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

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)