1 files changed, 72 insertions, 0 deletions
diff --git a/llamachat/providers.py b/llamachat/providers.py
index 0d2d9f7..051a931 100644
--- a/llamachat/providers.py
+++ b/llamachat/providers.py
@@ -155,3 +155,75 @@ def apply_filter(provider: Provider, listed: list[str]) -> list[str]:
     if not needles:
         return list(listed)
     return [m for m in listed if any(n in m.lower() for n in needles)]
+
+
+def _run_pass(cmd: list[str], timeout: int) -> str:
+    """Run `pass show NAME` and return its stdout."""
+    result = subprocess.run(
+        cmd, capture_output=True, text=True, timeout=timeout, check=True
+    )
+    return result.stdout
+
+
+class KeyResolver:
+    """Resolves api_key fields, lazily and once per process.
+
+    Lazy matters: `pass` needs the GPG key, so a session that only touches
+    local models must never trigger a pinentry. Resolved values stay in
+    memory and are never written anywhere.
+    """
+
+    def __init__(self, runner=_run_pass):
+        self._runner = runner
+        self._cache: dict[tuple[str, str], str] = {}
+
+    def resolve(self, provider: Provider) -> str:
+        """The bearer token for this provider, or '' when it needs none."""
+        spec = provider.api_key
+        if not spec:
+            return ""
+        # Keyed by the spec as well as the name: the config can be reloaded
+        # onto the same resolver, and keying by name alone would keep serving
+        # the old key after the user edits the entry to fix it, so the fix
+        # would look like it did nothing until a restart.
+        cache_key = (provider.name, spec)
+        if cache_key in self._cache:
+            return self._cache[cache_key]
+
+        if spec.startswith("env:"):
+            value = os.environ.get(spec[4:], "")
+            if not value:
+                raise KeyError_(
+                    f"{provider.name}: environment variable {spec[4:]} is not set"
+                )
+        elif spec.startswith("pass:"):
+            value = self._from_pass(provider, spec[5:])
+        else:
+            value = spec
+
+        self._cache[cache_key] = value
+        return value
+
+    def _from_pass(self, provider: Provider, entry: str) -> str:
+        try:
+            out = self._runner(["pass", "show", entry], timeout=KEY_TIMEOUT)
+        except subprocess.TimeoutExpired:
+            raise KeyError_(
+                f"{provider.name}: `pass show {entry}` timed out after "
+                f"{KEY_TIMEOUT}s. Is a pinentry waiting for input?"
+            )
+        except subprocess.CalledProcessError as exc:
+            # stderr is captured, so without this it is swallowed and the user
+            # gets an exit code where gpg told them exactly what was wrong.
+            # Only stderr is quoted, never stdout, which holds the secret.
+            detail = (exc.stderr or "").strip().splitlines()
+            reason = detail[-1] if detail else f"exit status {exc.returncode}"
+            raise KeyError_(f"{provider.name}: `pass show {entry}` failed: {reason}")
+        except Exception as exc:
+            raise KeyError_(f"{provider.name}: `pass show {entry}` failed: {exc}")
+        # A password store entry keeps the secret on the first line and
+        # metadata below it.
+        first = (out or "").strip().splitlines()
+        if not first or not first[0].strip():
+            raise KeyError_(f"{provider.name}: `pass show {entry}` returned nothing")
+        return first[0].strip()