EvoScientist-Multi/EvoScientist/llm/secret_store.py

"""Encrypted, versioned storage for model-provider credentials.

Only secret references cross the model-route configuration boundary. The
store deliberately has no API for reading a plaintext credential after it was
written; callers receive an opaque reference and masked metadata instead.
"""

from __future__ import annotations

import base64
import hashlib
import os
import re
import sqlite3
from dataclasses import dataclass
from pathlib import Path

from cryptography.fernet import Fernet, InvalidToken
from filelock import FileLock

from ..config.settings import get_config_dir
from .configuration import ResolvedSecret, SecretReference
from .contracts import EvoRuntimeError

_SECRET_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$")


@dataclass(frozen=True, slots=True)
class SecretMetadata:
    secret_id: str
    version: int
    masked_value: str
    created_by: str
    created_at: str
    status: str = "active"
    retired_at: str | None = None
    revoked_at: str | None = None
    revoked_by: str | None = None
    revoke_reason: str | None = None

    @property
    def ref(self) -> str:
        return f"secret://{self.secret_id}#{self.version}"


class EncryptedModelSecretStore:
    """SQLite-backed Fernet store scoped to the Evo configuration directory."""

    def __init__(
        self,
        path: Path | None = None,
        *,
        master_secret: str | None = None,
    ) -> None:
        self.path = path or (get_config_dir() / "model_secrets.sqlite")
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._lock = FileLock(str(self.path) + ".lock")
        material = master_secret or os.environ.get(
            "AI4SCI_EVO_MODEL_SECRET_MASTER_KEY", ""
        )
        if not material:
            # The identity secret is already mandatory for the V3 runtime.
            # Operators can configure a dedicated secret to separate rotation.
            material = os.environ.get("AI4SCI_EVO_CONFIG_IDENTITY_SECRET", "")
        if len(material.encode("utf-8")) < 32:
            raise RuntimeError(
                "AI4SCI_EVO_MODEL_SECRET_MASTER_KEY or "
                "AI4SCI_EVO_CONFIG_IDENTITY_SECRET must contain at least 32 bytes"
            )
        key = base64.urlsafe_b64encode(hashlib.sha256(material.encode()).digest())
        self._fernet = Fernet(key)
        self._init_schema()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(self.path)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys=ON")
        connection.execute("PRAGMA busy_timeout=5000")
        connection.execute("PRAGMA synchronous=FULL")
        connection.execute("PRAGMA journal_mode=WAL")
        return connection

    def _init_schema(self) -> None:
        with self._lock, self._connect() as connection:
            connection.executescript(
                """
                CREATE TABLE IF NOT EXISTS model_secret_versions (
                    secret_id TEXT NOT NULL,
                    version INTEGER NOT NULL CHECK (version > 0),
                    ciphertext BLOB NOT NULL,
                    masked_value TEXT NOT NULL,
                    created_by TEXT NOT NULL,
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (secret_id, version)
                );
                CREATE INDEX IF NOT EXISTS idx_model_secret_versions_latest
                    ON model_secret_versions (secret_id, version DESC);
                """
            )
            columns = {
                str(row["name"])
                for row in connection.execute(
                    "PRAGMA table_info(model_secret_versions)"
                ).fetchall()
            }
            additions = {
                "status": "TEXT NOT NULL DEFAULT 'active'",
                "retired_at": "TEXT",
                "revoked_at": "TEXT",
                "revoked_by": "TEXT",
                "revoke_reason": "TEXT",
                "transition_operation_id": "TEXT",
            }
            for name, definition in additions.items():
                if name not in columns:
                    connection.execute(
                        f"ALTER TABLE model_secret_versions ADD COLUMN {name} {definition}"
                    )
            # Legacy stores treated every version as active. Keep only the latest
            # active version before installing the partial unique index.
            connection.execute(
                """UPDATE model_secret_versions AS current
                   SET status='retired', retired_at=COALESCE(retired_at, CURRENT_TIMESTAMP)
                   WHERE status='active' AND version < (
                       SELECT MAX(newer.version) FROM model_secret_versions AS newer
                       WHERE newer.secret_id=current.secret_id
                   )"""
            )
            connection.execute(
                """CREATE UNIQUE INDEX IF NOT EXISTS uq_model_secret_active
                   ON model_secret_versions(secret_id) WHERE status='active'"""
            )
            connection.execute(
                """CREATE TABLE IF NOT EXISTS model_secret_operations (
                    operation_id TEXT PRIMARY KEY,
                    action TEXT NOT NULL,
                    request_digest TEXT NOT NULL,
                    secret_id TEXT NOT NULL,
                    version INTEGER NOT NULL,
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )"""
            )
        try:
            os.chmod(self.path, 0o600)
        except OSError:
            pass

    @staticmethod
    def _validate_secret_id(secret_id: str) -> str:
        normalized = str(secret_id or "").strip()
        if not _SECRET_ID_RE.fullmatch(normalized):
            raise EvoRuntimeError("LLM_SECRET_INVALID")
        return normalized

    @staticmethod
    def _mask(value: str) -> str:
        if len(value) <= 8:
            return "*" * len(value)
        return f"{value[:4]}...{value[-4:]}"

    def put(
        self,
        secret_id: str,
        value: str,
        *,
        created_by: str,
        status: str = "active",
        operation_id: str | None = None,
    ) -> SecretMetadata:
        secret_id = self._validate_secret_id(secret_id)
        value = str(value or "")
        if not value or any(char in value for char in "\r\n\0"):
            raise EvoRuntimeError("LLM_SECRET_INVALID")
        actor = str(created_by or "unknown")[:256]
        if status not in {"pending", "active"}:
            raise EvoRuntimeError("LLM_SECRET_INVALID")
        ciphertext = self._fernet.encrypt(value.encode("utf-8"))
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT COALESCE(MAX(version), 0) AS version "
                "FROM model_secret_versions WHERE secret_id=?",
                (secret_id,),
            ).fetchone()
            version = int(row["version"]) + 1
            if status == "active":
                connection.execute(
                    """UPDATE model_secret_versions
                       SET status='retired', retired_at=CURRENT_TIMESTAMP,
                           transition_operation_id=?
                       WHERE secret_id=? AND status='active'""",
                    (operation_id, secret_id),
                )
            connection.execute(
                """INSERT INTO model_secret_versions
                   (secret_id, version, ciphertext, masked_value, created_by,
                    status, transition_operation_id)
                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
                (
                    secret_id,
                    version,
                    ciphertext,
                    self._mask(value),
                    actor,
                    status,
                    operation_id,
                ),
            )
            saved = connection.execute(
                """SELECT secret_id, version, masked_value, created_by, created_at,
                          status, retired_at, revoked_at, revoked_by, revoke_reason
                   FROM model_secret_versions WHERE secret_id=? AND version=?""",
                (secret_id, version),
            ).fetchone()
        return SecretMetadata(
            secret_id=str(saved["secret_id"]),
            version=int(saved["version"]),
            masked_value=str(saved["masked_value"]),
            created_by=str(saved["created_by"]),
            created_at=str(saved["created_at"]),
            status=str(saved["status"]),
            retired_at=saved["retired_at"],
            revoked_at=saved["revoked_at"],
            revoked_by=saved["revoked_by"],
            revoke_reason=saved["revoke_reason"],
        )

    def create_pending(
        self,
        provider_id: str,
        value: str,
        *,
        created_by: str,
        operation_id: str,
    ) -> SecretMetadata:
        secret_id = f"model-providers/{self._validate_secret_id(provider_id)}"
        digest = hashlib.sha256(
            (secret_id + "\0" + str(value)).encode("utf-8")
        ).hexdigest()
        with self._lock:
            with self._connect() as connection:
                replay = connection.execute(
                    "SELECT * FROM model_secret_operations WHERE operation_id=?",
                    (operation_id,),
                ).fetchone()
            if replay is not None:
                if (
                    str(replay["action"]) != "create_pending"
                    or str(replay["request_digest"]) != digest
                    or str(replay["secret_id"]) != secret_id
                ):
                    raise EvoRuntimeError("IDEMPOTENCY_CONFLICT")
                return self._metadata(secret_id, int(replay["version"]))
            result = self.put(
                secret_id,
                value,
                created_by=created_by,
                status="pending",
                operation_id=operation_id,
            )
            with self._connect() as connection:
                connection.execute(
                    """INSERT INTO model_secret_operations
                       (operation_id, action, request_digest, secret_id, version)
                       VALUES (?, 'create_pending', ?, ?, ?)""",
                    (operation_id, digest, secret_id, result.version),
                )
            return result

    def list_metadata(self) -> list[SecretMetadata]:
        with self._connect() as connection:
            rows = connection.execute(
                """SELECT secret_id, version, masked_value, created_by, created_at,
                          status, retired_at, revoked_at, revoked_by, revoke_reason
                   FROM model_secret_versions
                   ORDER BY secret_id ASC, version DESC"""
            ).fetchall()
        return [
            SecretMetadata(
                secret_id=str(row["secret_id"]),
                version=int(row["version"]),
                masked_value=str(row["masked_value"]),
                created_by=str(row["created_by"]),
                created_at=str(row["created_at"]),
                status=str(row["status"]),
                retired_at=row["retired_at"],
                revoked_at=row["revoked_at"],
                revoked_by=row["revoked_by"],
                revoke_reason=row["revoke_reason"],
            )
            for row in rows
        ]

    def current_provider_metadata(self, provider_id: str) -> SecretMetadata | None:
        secret_id = f"model-providers/{self._validate_secret_id(provider_id)}"
        with self._connect() as connection:
            row = connection.execute(
                """SELECT secret_id, version, masked_value, created_by, created_at,
                          status, retired_at, revoked_at, revoked_by, revoke_reason
                   FROM model_secret_versions
                   WHERE secret_id=? AND status='active'
                   ORDER BY version DESC LIMIT 1""",
                (secret_id,),
            ).fetchone()
        if row is None:
            return None
        return SecretMetadata(
            secret_id=str(row["secret_id"]),
            version=int(row["version"]),
            masked_value=str(row["masked_value"]),
            created_by=str(row["created_by"]),
            created_at=str(row["created_at"]),
            status=str(row["status"]),
            retired_at=row["retired_at"],
            revoked_at=row["revoked_at"],
            revoked_by=row["revoked_by"],
            revoke_reason=row["revoke_reason"],
        )

    def resolve(self, reference: SecretReference) -> ResolvedSecret:
        if reference.ref.startswith("provider://"):
            provider_id = self._validate_secret_id(reference.ref[11:])
            secret_id = f"model-providers/{provider_id}"
            with self._connect() as connection:
                row = connection.execute(
                    """SELECT ciphertext, version FROM model_secret_versions
                       WHERE secret_id=? AND status='active'
                       ORDER BY version DESC LIMIT 1""",
                    (secret_id,),
                ).fetchone()
            if row is None:
                raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
            try:
                value = self._fernet.decrypt(bytes(row["ciphertext"])).decode("utf-8")
            except (InvalidToken, UnicodeDecodeError) as exc:
                raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE") from exc
            fingerprint = hashlib.sha256(value.encode("utf-8")).hexdigest()
            return ResolvedSecret(
                value, reference.revision, str(row["version"]), fingerprint
            )
        if not reference.ref.startswith("secret://") or "#" not in reference.ref:
            raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
        secret_id, version_text = reference.ref[9:].rsplit("#", 1)
        try:
            version = int(version_text)
        except ValueError as exc:
            raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE") from exc
        if version != reference.revision:
            raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
        secret_id = self._validate_secret_id(secret_id)
        with self._connect() as connection:
            row = connection.execute(
                """SELECT ciphertext, status FROM model_secret_versions
                   WHERE secret_id=? AND version=?""",
                (secret_id, version),
            ).fetchone()
        if row is None:
            raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
        if str(row["status"]) in {"revoked", "destroyed"}:
            raise EvoRuntimeError("MODEL_CREDENTIAL_REVOKED")
        try:
            value = self._fernet.decrypt(bytes(row["ciphertext"])).decode("utf-8")
        except (InvalidToken, UnicodeDecodeError) as exc:
            raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE") from exc
        fingerprint = hashlib.sha256(value.encode("utf-8")).hexdigest()
        return ResolvedSecret(value, version, str(version), fingerprint)

    def activate(
        self, secret_id: str, version: int, *, operation_id: str
    ) -> SecretMetadata:
        secret_id = self._validate_secret_id(secret_id)
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT status FROM model_secret_versions WHERE secret_id=? AND version=?",
                (secret_id, version),
            ).fetchone()
            if row is None or str(row["status"]) in {"revoked", "destroyed"}:
                raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
            connection.execute(
                """UPDATE model_secret_versions SET status='retired',
                   retired_at=COALESCE(retired_at, CURRENT_TIMESTAMP),
                   transition_operation_id=?
                   WHERE secret_id=? AND status='active' AND version<>?""",
                (operation_id, secret_id, version),
            )
            connection.execute(
                """UPDATE model_secret_versions SET status='active', retired_at=NULL,
                   transition_operation_id=? WHERE secret_id=? AND version=?""",
                (operation_id, secret_id, version),
            )
        return self._metadata(secret_id, version)

    def retire(
        self, secret_id: str, version: int, *, operation_id: str
    ) -> SecretMetadata:
        secret_id = self._validate_secret_id(secret_id)
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT status FROM model_secret_versions WHERE secret_id=? AND version=?",
                (secret_id, version),
            ).fetchone()
            if row is None:
                raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
            if str(row["status"]) not in {"revoked", "destroyed"}:
                connection.execute(
                    """UPDATE model_secret_versions SET status='retired',
                       retired_at=COALESCE(retired_at, CURRENT_TIMESTAMP),
                       transition_operation_id=? WHERE secret_id=? AND version=?""",
                    (operation_id, secret_id, version),
                )
        return self._metadata(secret_id, version)

    def revoke(
        self,
        secret_id: str,
        version: int,
        *,
        revoked_by: str,
        reason: str,
        operation_id: str,
    ) -> SecretMetadata:
        secret_id = self._validate_secret_id(secret_id)
        clean_reason = str(reason or "").strip()
        if not clean_reason:
            raise EvoRuntimeError("LLM_SECRET_INVALID")
        with self._lock, self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT status FROM model_secret_versions WHERE secret_id=? AND version=?",
                (secret_id, version),
            ).fetchone()
            if row is None or str(row["status"]) == "destroyed":
                raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
            if str(row["status"]) != "revoked":
                connection.execute(
                    """UPDATE model_secret_versions SET status='revoked',
                       revoked_at=CURRENT_TIMESTAMP, revoked_by=?, revoke_reason=?,
                       transition_operation_id=? WHERE secret_id=? AND version=?""",
                    (
                        revoked_by[:256],
                        clean_reason[:1024],
                        operation_id,
                        secret_id,
                        version,
                    ),
                )
        return self._metadata(secret_id, version)

    def _metadata(self, secret_id: str, version: int) -> SecretMetadata:
        with self._connect() as connection:
            row = connection.execute(
                """SELECT secret_id, version, masked_value, created_by, created_at,
                          status, retired_at, revoked_at, revoked_by, revoke_reason
                   FROM model_secret_versions WHERE secret_id=? AND version=?""",
                (secret_id, version),
            ).fetchone()
        if row is None:
            raise EvoRuntimeError("ROUTE_SECRET_UNAVAILABLE")
        return SecretMetadata(
            secret_id=str(row["secret_id"]),
            version=int(row["version"]),
            masked_value=str(row["masked_value"]),
            created_by=str(row["created_by"]),
            created_at=str(row["created_at"]),
            status=str(row["status"]),
            retired_at=row["retired_at"],
            revoked_at=row["revoked_at"],
            revoked_by=row["revoked_by"],
            revoke_reason=row["revoke_reason"],
        )