From 955fca181f4b3eac9732c94835add054499964c8 Mon Sep 17 00:00:00 2001
From: "Danilo M." <danix@danix.xyz>
Date: Sun, 9 Aug 2026 14:38:31 +0200
Subject: feat: resolve provider API keys from pass, env or literal

One prefix-dispatched field. Resolution is lazy so a local-only session
never triggers a pinentry, cached for the process lifetime, and bounded
by a timeout so a stuck pinentry surfaces as an error instead of a frozen
send. Failures name the provider.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 test_llamachat.py | 103 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 103 insertions(+)

(limited to 'test_llamachat.py')

diff --git a/test_llamachat.py b/test_llamachat.py
index 8b722ae..08f31c9 100755
--- a/test_llamachat.py
+++ b/test_llamachat.py
@@ -16,6 +16,7 @@
 
 import json
 import os
+import subprocess
 import sys
 import tempfile
 from pathlib import Path
@@ -1064,6 +1065,107 @@ def test_model_ids_and_filtering():
     print("ok  model ids and filtering")
 
 
+def test_key_resolution():
+    """api_key is prefix-dispatched, resolved lazily and cached."""
+    from llamachat import providers
+
+    resolver = providers.KeyResolver()
+
+    # No key configured: no Authorization header, and nothing is run.
+    empty = providers.Provider(name="local", base_url="http://x.example.org")
+    assert resolver.resolve(empty) == ""
+
+    # A literal key is used as-is.
+    literal = providers.Provider(
+        name="p", base_url="http://x.example.org", api_key="sk-test-not-a-real-key"
+    )
+    assert resolver.resolve(literal) == "sk-test-not-a-real-key"
+
+    # env: reads the environment.
+    os.environ["LLAMACHAT_TEST_KEY"] = "from-env"
+    env = providers.Provider(
+        name="e", base_url="http://x.example.org",
+        api_key="env:LLAMACHAT_TEST_KEY",
+    )
+    assert resolver.resolve(env) == "from-env"
+    del os.environ["LLAMACHAT_TEST_KEY"]
+
+    # A missing env var is an error naming the provider, not a silent "".
+    missing = providers.Provider(
+        name="gone", base_url="http://x.example.org",
+        api_key="env:LLAMACHAT_ABSENT_VAR",
+    )
+    try:
+        resolver.resolve(missing)
+        assert False, "a missing env var must raise"
+    except providers.KeyError_ as exc:
+        assert "gone" in str(exc)
+
+    # pass: shells out. Substitute the runner rather than requiring gpg.
+    calls = []
+
+    def fake_run(cmd, timeout):
+        calls.append((cmd, timeout))
+        return "line-one\nline-two\n"
+
+    passed = providers.Provider(
+        name="together", base_url="http://x.example.org",
+        api_key="pass:api/together",
+    )
+    cached = providers.KeyResolver(runner=fake_run)
+    assert cached.resolve(passed) == "line-one"      # first line only
+    assert calls[0][0] == ["pass", "show", "api/together"]
+    assert calls[0][1] == providers.KEY_TIMEOUT
+
+    # Cached: a second resolve must not shell out again.
+    assert cached.resolve(passed) == "line-one"
+    assert len(calls) == 1
+
+    # But the cache follows the spec, not just the name. A reloaded config
+    # that points the same provider at a different entry must re-resolve,
+    # otherwise correcting a wrong entry appears to do nothing.
+    moved = providers.Provider(
+        name="together", base_url="http://x.example.org",
+        api_key="pass:api/together-corrected",
+    )
+    assert cached.resolve(moved) == "line-one"
+    assert len(calls) == 2
+    assert calls[1][0] == ["pass", "show", "api/together-corrected"]
+
+    # A failing pass is reported, naming the provider.
+    def boom(cmd, timeout):
+        raise OSError("pass: entry not found")
+
+    try:
+        providers.KeyResolver(runner=boom).resolve(passed)
+        assert False, "a failing pass must raise"
+    except providers.KeyError_ as exc:
+        assert "together" in str(exc)
+
+    # A non-zero exit quotes gpg's stderr, which is the only useful part, and
+    # never stdout, which is where the secret would be.
+    def refused(cmd, timeout):
+        raise subprocess.CalledProcessError(
+            2, cmd, output="sk-test-not-a-real-key\n",
+            stderr="gpg: decryption failed: No secret key\n",
+        )
+
+    try:
+        providers.KeyResolver(runner=refused).resolve(passed)
+        assert False, "a non-zero pass exit must raise"
+    except providers.KeyError_ as exc:
+        assert "No secret key" in str(exc)
+        assert "sk-test-not-a-real-key" not in str(exc)
+
+    # Empty output is a failure too: an empty key would 401 confusingly.
+    try:
+        providers.KeyResolver(runner=lambda cmd, timeout: "  \n").resolve(passed)
+        assert False, "empty pass output must raise"
+    except providers.KeyError_:
+        pass
+    print("ok  api key resolution")
+
+
 class _FakeResponse:
     """Enough of an http.client response for urlopen's context manager."""
 
@@ -1803,6 +1905,7 @@ if __name__ == "__main__":
     test_config_defaults()
     test_provider_parsing()
     test_model_ids_and_filtering()
+    test_key_resolution()
     test_search_tool_schema()
     test_search_results_sanitising()
     test_tool_call_accumulation()
-- 
cgit v1.2.3