Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 48 additions & 1 deletion core/providers/credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,25 +3,48 @@
from __future__ import annotations

import hashlib
import importlib
import json
import os
import secrets
import threading
from pathlib import Path
from typing import Any

from loguru import logger

from core.config import deepcode_home
from core.file_lock import exclusive_file_lock
from core.private_storage import (
atomic_write_private_json,
open_existing_private_file,
)

# Opt-in OS-keychain backend for reads. ``keyring`` is deliberately not a
# declared dependency, so this is off by default and every import/backend
# failure degrades to the credential file. The file stays the source of truth
# and the only write target — ``set``/``clear`` never touch the keychain, which
# keeps ``revision``'s mtime/size fingerprint meaningful.
KEYRING_ENV_VAR = "DEEPCODE_KEYRING"
KEYRING_SERVICE = "deepcode"
_TRUTHY = frozenset({"1", "true", "yes", "on"})


def default_credentials_path() -> Path:
return deepcode_home() / "credentials.json"


def _keyring_backend() -> Any:
"""Return the ``keyring`` module when opted in and importable, else ``None``."""
if os.environ.get(KEYRING_ENV_VAR, "").strip().lower() not in _TRUTHY:
return None
try:
return importlib.import_module("keyring")
except Exception: # pragma: no cover - depends on the host's backends
logger.debug("{} is set but keyring is unavailable", KEYRING_ENV_VAR)
return None


class CredentialStore:
"""Atomic 0600 JSON storage whose values are never projected to clients."""

Expand All @@ -39,6 +62,25 @@ def __init__(self, path: Path | str | None = None) -> None:

def get(self, connection_id: str) -> str | None:
value = self._read().get("connections", {}).get(connection_id)
if isinstance(value, str) and value:
return value
return self._keyring_value(connection_id)

def _keyring_value(self, connection_id: str) -> str | None:
"""Read-only OS-keychain fallback consulted only when the file has none.

Every failure mode — package absent, no backend, locked or denied
keychain — returns ``None`` so the resolver simply falls through to
``legacy_config``; turning the knob on cannot break a working setup.
"""
backend = _keyring_backend()
if backend is None:
return None
try:
value = backend.get_password(KEYRING_SERVICE, connection_id)
except Exception: # pragma: no cover - depends on the host's backends
logger.debug("keyring lookup failed for {}", connection_id, exc_info=True)
return None
return value if isinstance(value, str) and value else None

def configured(self, connection_id: str) -> bool:
Expand Down Expand Up @@ -187,4 +229,9 @@ def _connections(data: dict[str, Any]) -> dict[str, str]:
return dict(value) if isinstance(value, dict) else {}


__all__ = ["CredentialStore", "default_credentials_path"]
__all__ = [
"KEYRING_ENV_VAR",
"KEYRING_SERVICE",
"CredentialStore",
"default_credentials_path",
]
216 changes: 216 additions & 0 deletions tests/test_provider_credentials.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,216 @@
"""The credential store's optional OS-keychain read fallback.

``keyring`` is deliberately not a declared dependency, so the keychain backend is
opt-in (``DEEPCODE_KEYRING``), consulted only when the credential file holds no
value for a connection, and never written to — the file stays the source of truth
and the only write target, which keeps ``revision``'s mtime/size fingerprint
meaningful. These tests inject a fake ``keyring`` module, which is also how a
default install behaves: the package is simply absent.
"""

from __future__ import annotations

import sys
from pathlib import Path
from types import ModuleType

import pytest

from core.application.config_store import ConfigStore
from core.application.llm_configuration_service import LLMConfigurationService
from core.config import load_config
from core.providers.catalog_service import ModelCatalogService
from core.providers.credentials import (
KEYRING_ENV_VAR,
KEYRING_SERVICE,
CredentialStore,
)
from core.providers.profiles import ConnectionResolver


class _FakeKeyring(ModuleType):
"""Minimal stand-in for the optional ``keyring`` package."""

def __init__(
self,
secrets: dict[str, str] | None = None,
error: Exception | None = None,
) -> None:
super().__init__("keyring")
self.secrets = dict(secrets or {})
self.error = error
self.reads: list[tuple[str, str]] = []
self.writes: list[tuple[str, str, str]] = []

def get_password(self, service: str, username: str) -> str | None:
self.reads.append((service, username))
if self.error is not None:
raise self.error
return self.secrets.get(f"{service}/{username}")

def set_password(self, service: str, username: str, password: str) -> None:
self.writes.append((service, username, password))
self.secrets[f"{service}/{username}"] = password


def _install(monkeypatch: pytest.MonkeyPatch, module: ModuleType | None) -> None:
monkeypatch.setitem(sys.modules, "keyring", module)


def _store(tmp_path: Path) -> CredentialStore:
return CredentialStore(tmp_path / "credentials.json")


def _keyed(secret: str) -> _FakeKeyring:
return _FakeKeyring({f"{KEYRING_SERVICE}/router": secret})


def test_keychain_is_not_consulted_by_default(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
backend = _keyed("keychain-secret")
_install(monkeypatch, backend)
monkeypatch.delenv(KEYRING_ENV_VAR, raising=False)

assert _store(tmp_path).get("router") is None
assert backend.reads == []


@pytest.mark.parametrize("value", ["", "0", "false", "no", "off", " "])
def test_keychain_ignores_falsey_knob_spellings(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
value: str,
) -> None:
backend = _keyed("keychain-secret")
_install(monkeypatch, backend)
monkeypatch.setenv(KEYRING_ENV_VAR, value)

assert _store(tmp_path).get("router") is None
assert backend.reads == []


def test_keychain_supplies_an_absent_credential(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
backend = _keyed("keychain-secret")
_install(monkeypatch, backend)
monkeypatch.setenv(KEYRING_ENV_VAR, "on")

assert _store(tmp_path).get("router") == "keychain-secret"
assert backend.reads == [(KEYRING_SERVICE, "router")]


def test_credential_file_still_wins_over_the_keychain(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
backend = _keyed("keychain-secret")
_install(monkeypatch, backend)
monkeypatch.setenv(KEYRING_ENV_VAR, "1")
store = _store(tmp_path)
store.set("router", "file-secret")

assert store.get("router") == "file-secret"
assert backend.reads == []


def test_missing_keyring_package_degrades_to_the_file(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A default install has no ``keyring``; the import must not raise."""
_install(monkeypatch, None)
monkeypatch.setenv(KEYRING_ENV_VAR, "1")

assert _store(tmp_path).get("router") is None


def test_backend_failure_never_escapes(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
_install(monkeypatch, _FakeKeyring(error=RuntimeError("keychain locked")))
monkeypatch.setenv(KEYRING_ENV_VAR, "1")

assert _store(tmp_path).get("router") is None


@pytest.mark.parametrize("blank", ["", None, 123])
def test_a_non_string_file_value_falls_through_to_the_keychain(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
blank: object,
) -> None:
_install(monkeypatch, _keyed("keychain-secret"))
monkeypatch.setenv(KEYRING_ENV_VAR, "1")
store = _store(tmp_path)
monkeypatch.setattr(
store,
"_read",
lambda: {"version": 1, "connections": {"router": blank}},
)

assert store.get("router") == "keychain-secret"


def test_the_store_never_writes_to_the_keychain(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
backend = _keyed("keychain-secret")
_install(monkeypatch, backend)
monkeypatch.setenv(KEYRING_ENV_VAR, "1")
store = _store(tmp_path)

store.set("router", "file-secret")
store.clear("router")

assert backend.writes == []


def test_a_keychain_only_credential_leaves_no_revision_trace(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The keychain is a backend, not a second store: nothing is persisted."""
_install(monkeypatch, _keyed("keychain-secret"))
monkeypatch.setenv(KEYRING_ENV_VAR, "1")
store = _store(tmp_path)

assert store.revision() == "missing"
assert store.get("router") == "keychain-secret"
assert store.revision() == "missing"
assert not store.path.exists()


def test_resolution_order_gains_keychain_support(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Env → credential store (keychain-backed) → legacy config, end to end."""
home = tmp_path / "home"
monkeypatch.setenv("DEEPCODE_HOME", str(home))
config = ConfigStore(home / "deepcode_config.json")
credentials = CredentialStore(home / "credentials.json")
service = LLMConfigurationService(
config_store=config,
credential_store=credentials,
catalog=ModelCatalogService(home / "model_catalog_cache.json"),
)
service.upsert({"id": "router", "template": "openrouter", "modelCatalog": "auto"})
_install(monkeypatch, _keyed("keychain-secret"))
resolver = ConnectionResolver(load_config(config_path=config.path), credentials)

monkeypatch.setenv(KEYRING_ENV_VAR, "1")
resolved = resolver.resolve_connection("router")
assert resolved.api_key == "keychain-secret"
# The keychain is the credential store's backend, not a fifth tier.
assert resolved.credential_source == "credential_store"

monkeypatch.setenv(KEYRING_ENV_VAR, "0")
without = resolver.resolve_connection("router")
assert without.api_key is None
assert without.credential_source == "missing"
Loading