feat(management): add Fernet encrypted credential storage (AIHCM-182)

src/api/infrastructure/migrations/versions/d5e6f7a8b9c0_create_encrypted_credentials_table.py

@@ -0,0 +1,39 @@
+"""create encrypted_credentials table
+
+Adds the encrypted_credentials table for Fernet-encrypted credential
+storage in the Management bounded context.
+
+Revision ID: d5e6f7a8b9c0
+Revises: c4d5e6f7a8b9
+Create Date: 2026-03-17
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = "d5e6f7a8b9c0"
+down_revision: Union[str, Sequence[str], None] = "c4d5e6f7a8b9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create encrypted_credentials table with composite PK (path, tenant_id)."""
+    op.create_table(
+        "encrypted_credentials",
+        sa.Column("path", sa.String(500), primary_key=True),
+        sa.Column("tenant_id", sa.String(26), primary_key=True),
+        sa.Column("encrypted_value", sa.LargeBinary, nullable=False),
+        sa.Column("key_version", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    """Drop encrypted_credentials table."""
+    op.drop_table("encrypted_credentials")

src/api/infrastructure/settings.py

@@ -387,3 +387,31 @@ def get_oidc_settings() -> OIDCSettings:
     Uses lru_cache to ensure settings are only loaded once.
     """
     return OIDCSettings()
+
+
+class ManagementSettings(BaseSettings):
+    """Management bounded context settings.
+
+    Environment variables:
+        KARTOGRAPH_MGMT_ENCRYPTION_KEY: Comma-separated Fernet keys for MultiFernet
+    """
+
+    model_config = SettingsConfigDict(
+        env_prefix="KARTOGRAPH_MGMT_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    encryption_key: SecretStr = Field(
+        description="Comma-separated Fernet keys for MultiFernet",
+    )
+
+
+@lru_cache
+def get_management_settings() -> ManagementSettings:
+    """Get cached Management settings.
+
+    Uses lru_cache to ensure settings are only loaded once.
+    """
+    return ManagementSettings()  # type: ignore[call-arg]

src/api/management/infrastructure/models/__init__.py

@@ -5,10 +5,14 @@
 
 from management.infrastructure.models.data_source import DataSourceModel
 from management.infrastructure.models.data_source_sync_run import DataSourceSyncRunModel
+from management.infrastructure.models.encrypted_credential import (
+    EncryptedCredentialModel,
+)
 from management.infrastructure.models.knowledge_graph import KnowledgeGraphModel
 
 __all__ = [
     "DataSourceModel",
     "DataSourceSyncRunModel",
+    "EncryptedCredentialModel",
     "KnowledgeGraphModel",
 ]

src/api/management/infrastructure/models/encrypted_credential.py

@@ -0,0 +1,35 @@
+"""SQLAlchemy ORM model for the encrypted_credentials table.
+
+Stores Fernet-encrypted credentials in PostgreSQL, scoped by path
+and tenant_id. Part of the Management bounded context.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import Integer, LargeBinary, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from infrastructure.database.models import Base, TimestampMixin
+
+
+class EncryptedCredentialModel(Base, TimestampMixin):
+    """ORM model for encrypted_credentials table.
+
+    Stores encrypted credential blobs keyed by (path, tenant_id).
+    The key_version column tracks which Fernet key was used for
+    encryption to support key rotation.
+    """
+
+    __tablename__ = "encrypted_credentials"
+
+    path: Mapped[str] = mapped_column(String(500), primary_key=True)
+    tenant_id: Mapped[str] = mapped_column(String(26), primary_key=True)
+    encrypted_value: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
+    key_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return (
+            f"<EncryptedCredentialModel(path={self.path}, "
+            f"tenant_id={self.tenant_id}, key_version={self.key_version})>"
+        )

src/api/management/infrastructure/observability/__init__.py

@@ -12,12 +12,18 @@
     KnowledgeGraphRepositoryProbe,
     SyncRunRepositoryProbe,
 )
+from management.infrastructure.observability.secret_store_probe import (
+    DefaultSecretStoreProbe,
+    SecretStoreProbe,
+)
 
 __all__ = [
     "DataSourceRepositoryProbe",
     "DefaultDataSourceRepositoryProbe",
     "DefaultKnowledgeGraphRepositoryProbe",
+    "DefaultSecretStoreProbe",
     "DefaultSyncRunRepositoryProbe",
     "KnowledgeGraphRepositoryProbe",
+    "SecretStoreProbe",
     "SyncRunRepositoryProbe",
 ]

src/api/management/infrastructure/observability/secret_store_probe.py

@@ -0,0 +1,90 @@
+"""Domain probes for secret store operations.
+
+Following Domain-Oriented Observability patterns, these probes capture
+domain-significant events related to credential storage operations.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Protocol
+
+import structlog
+
+if TYPE_CHECKING:
+    from shared_kernel.observability_context import ObservationContext
+
+
+class SecretStoreProbe(Protocol):
+    """Domain probe for secret store operations."""
+
+    def credential_stored(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were successfully stored."""
+        ...
+
+    def credential_retrieved(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were successfully retrieved."""
+        ...
+
+    def credential_not_found(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were not found."""
+        ...
+
+    def credential_deleted(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were successfully deleted."""
+        ...
+
+    def with_context(self, context: ObservationContext) -> SecretStoreProbe:
+        """Create a new probe with observation context bound."""
+        ...
+
+
+class DefaultSecretStoreProbe:
+    """Default implementation of SecretStoreProbe using structlog."""
+
+    def __init__(
+        self,
+        logger: structlog.stdlib.BoundLogger | None = None,
+        context: ObservationContext | None = None,
+    ) -> None:
+        self._logger = logger or structlog.get_logger()
+        self._context = context
+
+    def _get_context_kwargs(self) -> dict[str, Any]:
+        if self._context is None:
+            return {}
+        return self._context.as_dict()
+
+    def with_context(self, context: ObservationContext) -> DefaultSecretStoreProbe:
+        return DefaultSecretStoreProbe(logger=self._logger, context=context)
+
+    def credential_stored(self, path: str, tenant_id: str) -> None:
+        self._logger.info(
+            "credential_stored",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )
+
+    def credential_retrieved(self, path: str, tenant_id: str) -> None:
+        self._logger.debug(
+            "credential_retrieved",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )
+
+    def credential_not_found(self, path: str, tenant_id: str) -> None:
+        self._logger.debug(
+            "credential_not_found",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )
+
+    def credential_deleted(self, path: str, tenant_id: str) -> None:
+        self._logger.info(
+            "credential_deleted",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )

src/api/management/infrastructure/repositories/__init__.py

@@ -9,12 +9,16 @@
 from management.infrastructure.repositories.data_source_sync_run_repository import (
     DataSourceSyncRunRepository,
 )
+from management.infrastructure.repositories.fernet_secret_store import (
+    FernetSecretStore,
+)
 from management.infrastructure.repositories.knowledge_graph_repository import (
     KnowledgeGraphRepository,
 )
 
 __all__ = [
     "DataSourceRepository",
     "DataSourceSyncRunRepository",
+    "FernetSecretStore",
     "KnowledgeGraphRepository",
 ]

src/api/management/infrastructure/repositories/fernet_secret_store.py

@@ -0,0 +1,112 @@
+"""Fernet-encrypted secret store implementation.
+
+Uses cryptography.fernet.MultiFernet to encrypt credentials at rest
+in PostgreSQL, supporting key rotation via multiple Fernet keys.
+"""
+
+from __future__ import annotations
+
+import json
+
+from cryptography.fernet import Fernet, MultiFernet
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from management.infrastructure.models.encrypted_credential import (
+    EncryptedCredentialModel,
+)
+from management.infrastructure.observability.secret_store_probe import (
+    DefaultSecretStoreProbe,
+    SecretStoreProbe,
+)
+
+
+def _validate_inputs(path: str, tenant_id: str) -> None:
+    """Validate that path and tenant_id are non-empty."""
+    if not path or not path.strip():
+        raise ValueError("path must not be empty or whitespace-only")
+    if not tenant_id or not tenant_id.strip():
+        raise ValueError("tenant_id must not be empty or whitespace-only")
+
+
+class FernetSecretStore:
+    """Fernet-encrypted credential storage backed by PostgreSQL.
+
+    Implements both ISecretStoreRepository (management port) and
+    ICredentialReader (shared kernel) for a single implementation
+    that satisfies both read-write and read-only consumers.
+
+    Uses MultiFernet to support key rotation: the first key in the
+    list is used for encryption, and all keys are tried for decryption.
+    """
+
+    def __init__(
+        self,
+        session: AsyncSession,
+        encryption_keys: list[str],
+        probe: SecretStoreProbe | None = None,
+    ) -> None:
+        self._session = session
+        self._multi_fernet = MultiFernet([Fernet(key) for key in encryption_keys])
+        self._probe: SecretStoreProbe = probe or DefaultSecretStoreProbe()
+
+    async def store(
+        self, path: str, tenant_id: str, credentials: dict[str, str]
+    ) -> None:
+        """Encrypt and persist credentials via upsert."""
+        _validate_inputs(path, tenant_id)
+
+        plaintext = json.dumps(credentials).encode("utf-8")
+        encrypted = self._multi_fernet.encrypt(plaintext)
+
+        model = EncryptedCredentialModel(
+            path=path,
+            tenant_id=tenant_id,
+            encrypted_value=encrypted,
+            key_version=0,
+        )
+        await self._session.merge(model)
+        await self._session.flush()
+        self._probe.credential_stored(path, tenant_id)
+
+    async def retrieve(self, path: str, tenant_id: str) -> dict[str, str]:
+        """Decrypt and return stored credentials.
+
+        Raises:
+            KeyError: If no credentials exist at the given path for the tenant.
+        """
+        _validate_inputs(path, tenant_id)
+
+        stmt = select(EncryptedCredentialModel).where(
+            EncryptedCredentialModel.path == path,
+            EncryptedCredentialModel.tenant_id == tenant_id,
+        )
+        result = await self._session.execute(stmt)
+        model = result.scalar_one_or_none()
+
+        if model is None:
+            self._probe.credential_not_found(path, tenant_id)
+            raise KeyError("Credentials not found")
+
+        decrypted = self._multi_fernet.decrypt(model.encrypted_value)
+        self._probe.credential_retrieved(path, tenant_id)
+        return json.loads(decrypted.decode("utf-8"))
+
+    async def delete(self, path: str, tenant_id: str) -> bool:
+        """Remove stored credentials. Returns True if deleted."""
+        _validate_inputs(path, tenant_id)
+
+        stmt = select(EncryptedCredentialModel).where(
+            EncryptedCredentialModel.path == path,
+            EncryptedCredentialModel.tenant_id == tenant_id,
+        )
+        result = await self._session.execute(stmt)
+        model = result.scalar_one_or_none()
+
+        if model is None:
+            return False
+
+        await self._session.delete(model)
+        await self._session.flush()
+        self._probe.credential_deleted(path, tenant_id)
+        return True