security: encrypt idempotency ledger at rest (Closes #1122)

tests/test_idempotency_ledger.py

@@ -2,7 +2,132 @@ import sqlite3
 from concurrent.futures import ThreadPoolExecutor
 from threading import Barrier
 
+import pytest
+
 from src.idempotency import IdempotencyLedger
+from src.state_encryption import PrivateStateEncryptionError
+
+
+PRIVATE_KEY = b"i" * 32
+
+
+def test_authored_payloads_are_encrypted_at_rest_and_replay_after_restart(tmp_path):
+    database = tmp_path / "idempotency.sqlite3"
+    fingerprint_canary = "private-issue-body-canary"
+    response_canary = "private-gitea-response-canary"
+    ledger = IdempotencyLedger(
+        database,
+        ttl_seconds=600,
+        max_entries=256,
+        encryption_key=PRIVATE_KEY,
+    )
+
+    assert ledger.reserve("create-201", ("issue-create", fingerprint_canary)).state == "reserved"
+    ledger.complete("create-201", {"title": response_canary})
+
+    persisted = database.read_bytes()
+    assert fingerprint_canary.encode() not in persisted
+    assert response_canary.encode() not in persisted
+    with sqlite3.connect(database) as connection:
+        fingerprint, response = connection.execute(
+            "SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
+            ("create-201",),
+        ).fetchone()
+    assert fingerprint.startswith("v1:")
+    assert response.startswith("v1:")
+
+    reopened = IdempotencyLedger(
+        database,
+        ttl_seconds=600,
+        max_entries=256,
+        encryption_key=PRIVATE_KEY,
+    )
+    replay = reopened.reserve("create-201", ("issue-create", fingerprint_canary))
+    assert replay.state == "completed"
+    assert replay.response == {"title": response_canary}
+
+
+def test_legacy_plaintext_completed_row_is_migrated_without_losing_replay(tmp_path):
+    database = tmp_path / "legacy.sqlite3"
+    ledger = IdempotencyLedger(
+        database,
+        ttl_seconds=600,
+        max_entries=256,
+        encryption_key=PRIVATE_KEY,
+        clock=lambda: 100.0,
+    )
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            "INSERT INTO idempotency_operations "
+            "(key, fingerprint, status, response_json, created_at, completed_at) "
+            "VALUES (?, ?, 'completed', ?, ?, ?)",
+            (
+                "legacy-comment",
+                '["issue-comment","stackchain/api",7,"legacy secret"]',
+                '{"id":42,"body":"legacy response"}',
+                90.0,
+                95.0,
+            ),
+        )
+
+    replay = ledger.reserve(
+        "legacy-comment",
+        ("issue-comment", "stackchain/api", 7, "legacy secret"),
+    )
+
+    assert replay.state == "completed"
+    assert replay.response == {"id": 42, "body": "legacy response"}
+    with sqlite3.connect(database) as connection:
+        fingerprint, response = connection.execute(
+            "SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
+            ("legacy-comment",),
+        ).fetchone()
+    assert fingerprint.startswith("v1:")
+    assert response.startswith("v1:")
+
+
+def test_wrong_key_fails_closed_before_a_completed_result_can_replay(tmp_path):
+    database = tmp_path / "wrong-key.sqlite3"
+    ledger = IdempotencyLedger(
+        database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
+    )
+    fingerprint = ("issue-comment", "stackchain/api", 7, "secret")
+    assert ledger.reserve("comment-7", fingerprint).state == "reserved"
+    ledger.complete("comment-7", {"id": 42})
+
+    wrong_key = IdempotencyLedger(
+        database, ttl_seconds=600, max_entries=256, encryption_key=b"x" * 32
+    )
+    with pytest.raises(
+        PrivateStateEncryptionError, match="private state could not be decrypted"
+    ):
+        wrong_key.reserve("comment-7", fingerprint)
+
+
+def test_completed_ciphertext_cannot_be_substituted_between_operation_keys(tmp_path):
+    database = tmp_path / "substitution.sqlite3"
+    ledger = IdempotencyLedger(
+        database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
+    )
+    fingerprint = ("issue-create", "stackchain/api", "same authored request")
+    for key, number in (("create-a", 41), ("create-b", 42)):
+        assert ledger.reserve(key, fingerprint).state == "reserved"
+        ledger.complete(key, {"number": number})
+    with sqlite3.connect(database) as connection:
+        responses = dict(
+            connection.execute(
+                "SELECT key, response_json FROM idempotency_operations"
+            ).fetchall()
+        )
+        connection.execute(
+            "UPDATE idempotency_operations SET response_json = ? WHERE key = ?",
+            (responses["create-b"], "create-a"),
+        )
+
+    with pytest.raises(
+        PrivateStateEncryptionError, match="private state could not be decrypted"
+    ):
+        ledger.reserve("create-a", fingerprint)
 
 
 def test_completed_result_replays_after_ledger_reconstruction(tmp_path):