diff --git a/worker/lyra_worker/crypto.py b/worker/lyra_worker/crypto.py new file mode 100644 index 0000000..7a9dd13 --- /dev/null +++ b/worker/lyra_worker/crypto.py @@ -0,0 +1,37 @@ +import base64 +import json +import os + +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + +def _key() -> bytes: + b64 = os.environ.get("LYRA_SECRET_KEY") + if not b64: + raise RuntimeError("LYRA_SECRET_KEY is not set") + k = base64.b64decode(b64) + if len(k) != 32: + raise RuntimeError("LYRA_SECRET_KEY must decode to 32 bytes") + return k + + +def encrypt_secret(plaintext: str) -> str: + iv = os.urandom(12) + full = AESGCM(_key()).encrypt(iv, plaintext.encode("utf-8"), None) # ct || tag(16) + ct, tag = full[:-16], full[-16:] + return json.dumps( + { + "iv": base64.b64encode(iv).decode(), + "ct": base64.b64encode(ct).decode(), + "tag": base64.b64encode(tag).decode(), + } + ) + + +def decrypt_secret(envelope: str) -> str: + e = json.loads(envelope) + iv = base64.b64decode(e["iv"]) + ct = base64.b64decode(e["ct"]) + tag = base64.b64decode(e["tag"]) + pt = AESGCM(_key()).decrypt(iv, ct + tag, None) # AESGCM expects ct || tag + return pt.decode("utf-8") diff --git a/worker/requirements.txt b/worker/requirements.txt index f05dbc0..56c6792 100644 --- a/worker/requirements.txt +++ b/worker/requirements.txt @@ -1,2 +1,3 @@ psycopg[binary]>=3.2,<4 pytest>=8.3,<9 +cryptography>=42,<46 diff --git a/worker/tests/test_crypto.py b/worker/tests/test_crypto.py new file mode 100644 index 0000000..75174f0 --- /dev/null +++ b/worker/tests/test_crypto.py @@ -0,0 +1,31 @@ +import pytest + +from lyra_worker.crypto import decrypt_secret, encrypt_secret + +TEST_KEY = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=" # 32 bytes, base64 +# Produced by web/src/lib/crypto.ts encryptSecret("cross-lang-secret") with TEST_KEY (see Step 2). +NODE_FIXTURE = '{"iv":"j9RH6WtmcmkD716H","ct":"5nWNQDDhIHJJVgvEzrO+b8M=","tag":"MBtkQ+ygwAeXwLQLpTsvWw=="}' + + +@pytest.fixture(autouse=True) +def _key(monkeypatch): + monkeypatch.setenv("LYRA_SECRET_KEY", TEST_KEY) + + +def test_round_trip(): + env = encrypt_secret("hunter2") + assert "hunter2" not in env + assert decrypt_secret(env) == "hunter2" + + +def test_decrypts_node_produced_envelope(): + # Cross-language contract: the worker must decrypt what the web app encrypted. + assert decrypt_secret(NODE_FIXTURE) == "cross-lang-secret" + + +def test_rejects_tampered_envelope(): + import json + env = json.loads(encrypt_secret("secret")) + env["ct"] = "Z2FyYmFnZQ==" # "garbage" + with pytest.raises(Exception): + decrypt_secret(json.dumps(env))