feat: add worker AES-256-GCM crypto matching the web envelope

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")

worker/requirements.txt

@@ -1,2 +1,3 @@
 psycopg[binary]>=3.2,<4
 pytest>=8.3,<9
+cryptography>=42,<46

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))