feat(secrets): add FileBasedSecretsManager with JSONL storage

Add a file-based secrets manager that persists secrets to a JSONL file
with in-memory caching and auto-save on program exit via atexit. Supports
atomic writes using temp file + rename pattern.

Signed-off-by: Daniel Bluhm <dbluhm@pm.me>

Author: dbluhm

didcomm_messaging/crypto/backend/basic.py

@@ -1,6 +1,11 @@
 """Basic Crypto Implementations."""
 
-from typing import Optional
+import atexit
+import json
+import shutil
+from pathlib import Path
+from typing import Callable, Dict, Optional
+
 from didcomm_messaging.crypto.base import S, SecretsManager
 
 
@@ -18,3 +23,77 @@ async def get_secret_by_kid(self, kid: str) -> Optional[S]:
     async def add_secret(self, secret: S) -> None:
         """Add a secret to the secrets manager."""
         self.secrets[secret.kid] = secret
+
+
+class FileBasedSecretsManager(SecretsManager[S]):
+    """File-based Secrets Manager with in-memory caching and auto-save.
+
+    Secrets are stored in memory for fast access and persisted to a JSONL file.
+    The file is saved automatically on program exit via atexit, and can also
+    be flushed explicitly using the flush() method.
+
+    Requires serializer and deserializer callbacks to convert between SecretKey
+    objects and their JSON-serializable representation.
+    """
+
+    def __init__(
+        self,
+        path: str,
+        serializer: Callable[[S], Dict],
+        deserializer: Callable[[str, Dict], S],
+        secrets: Optional[dict] = None,
+    ):
+        """Initialize the FileBasedSecretsManager.
+
+        Args:
+            path: Full path to the JSONL file for storing secrets.
+            serializer: Callback to serialize a SecretKey to a dict.
+            deserializer: Callback to deserialize a dict to a SecretKey.
+                Takes (kid, serialized_dict) as arguments.
+            secrets: Optional initial secrets to load (file takes precedence).
+        """
+        self._path = Path(path)
+        self._serializer = serializer
+        self._deserializer = deserializer
+        self._secrets: Dict[str, S] = secrets or {}
+
+        if self._path.exists():
+            with open(self._path) as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    data = json.loads(line)
+                    kid = data.get("kid")
+                    if kid:
+                        self._secrets[kid] = self._deserializer(kid, data)
+
+        atexit.register(self._sync)
+
+    @property
+    def path(self) -> str:
+        """Return the path to the secrets file."""
+        return str(self._path)
+
+    async def get_secret_by_kid(self, kid: str) -> Optional[S]:
+        """Get a secret by its kid."""
+        return self._secrets.get(kid)
+
+    async def add_secret(self, secret: S) -> None:
+        """Add a secret to the secrets manager."""
+        self._secrets[secret.kid] = secret
+
+    async def flush(self) -> None:
+        """Explicitly save secrets to the file."""
+        self._sync()
+
+    def _sync(self) -> None:
+        """Write secrets to file (called on atexit and flush)."""
+        tmp_path = self._path.with_suffix(".tmp")
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        with open(tmp_path, "w") as f:
+            for kid, secret in self._secrets.items():
+                data = self._serializer(secret)
+                data["kid"] = kid
+                f.write(json.dumps(data) + "\n")
+        shutil.move(tmp_path, self._path)

tests/test_secrets.py

@@ -1,7 +1,14 @@
+import json
+import os
+import tempfile
+
 import pytest
 from didcomm_messaging.crypto import SecretKey
 
-from didcomm_messaging.crypto.backend.basic import InMemorySecretsManager
+from didcomm_messaging.crypto.backend.basic import (
+    FileBasedSecretsManager,
+    InMemorySecretsManager,
+)
 
 
 class MockSecretKey(SecretKey):
@@ -12,6 +19,9 @@ def __init__(self, kid) -> None:
     def kid(self):
         return self._kid
 
+    def __repr__(self):
+        return f"MockSecretKey({self._kid!r})"
+
 
 @pytest.fixture()
 def in_memory_secrets_manager():
@@ -32,3 +42,68 @@ async def test_in_memory_secrets(
     await in_memory_secrets_manager.add_secret(secret)
 
     assert await in_memory_secrets_manager.get_secret_by_kid(secret.kid) == secret
+
+
+@pytest.fixture()
+def temp_secrets_file():
+    fd, path = tempfile.mkstemp(suffix=".jsonl")
+    os.close(fd)
+    yield path
+    if os.path.exists(path):
+        os.remove(path)
+    if os.path.exists(path + ".tmp"):
+        os.remove(path + ".tmp")
+
+
+def serializer(secret: MockSecretKey):
+    return {"multikey": f"multikey:{secret.kid}"}
+
+
+def deserializer(kid: str, data: dict):
+    return MockSecretKey(kid=kid)
+
+
+@pytest.mark.asyncio
+async def test_file_based_secrets(temp_secrets_file):
+    manager = FileBasedSecretsManager(temp_secrets_file, serializer, deserializer)
+
+    secret = MockSecretKey(kid="did:example:alice#key-1")
+    await manager.add_secret(secret)
+
+    result = await manager.get_secret_by_kid(secret.kid)
+    assert result == secret
+
+    await manager.flush()
+
+    with open(temp_secrets_file) as f:
+        lines = f.readlines()
+
+    assert len(lines) == 1
+    data = json.loads(lines[0])
+    assert data["kid"] == secret.kid
+    assert data["multikey"] == f"multikey:{secret.kid}"
+
+
+@pytest.mark.asyncio
+async def test_file_based_secrets_loads_existing(temp_secrets_file):
+    initial_data = [{"kid": "did:example:alice#key-1", "multikey": "multikey:existing"}]
+    with open(temp_secrets_file, "w") as f:
+        for item in initial_data:
+            f.write(json.dumps(item) + "\n")
+
+    manager = FileBasedSecretsManager(temp_secrets_file, serializer, deserializer)
+
+    secret = await manager.get_secret_by_kid("did:example:alice#key-1")
+    assert secret is not None
+    assert secret.kid == "did:example:alice#key-1"
+
+
+@pytest.mark.asyncio
+async def test_file_based_secrets_empty_file(temp_secrets_file):
+    """Test that empty file doesn't cause error."""
+    with open(temp_secrets_file, "w") as f:
+        f.write("")
+
+    manager = FileBasedSecretsManager(temp_secrets_file, serializer, deserializer)
+    result = await manager.get_secret_by_kid("any-key")
+    assert result is None