feat(claude): add forward_host_credentials support

Reads the host's Claude OAuth session key from ~/.claude.json at launch and forwards it only to the egress sidecar (never to the agent), placing a placeholder CLAUDE_CODE_OAUTH_TOKEN in the agent env so Claude Code starts without seeing the real credential. Mirrors the existing Codex forward_host_credentials flow (PRD 0029). Adds claude_auth.py to extract and validate the sessionKey, a CLAUDE_HOST_CREDENTIAL_TOKEN_REF constant in egress.py, and updates manifest_agent.py to allow the flag for both 'codex' and 'claude' templates. Also adds a mutual-exclusion check that rejects setting both auth_token and forward_host_credentials together. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-01 21:14:37 +00:00
parent 71699b3ecd
commit f0d27863c2
10 changed files with 423 additions and 13 deletions
@@ -0,0 +1,106 @@
+"""Unit: host Claude auth extraction."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+
+from bot_bottle.contrib.claude.claude_auth import (
+    claude_auth_path,
+    claude_host_access_token,
+)
+from bot_bottle.log import Die
+
+
+class TestClaudeHostAccessToken(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="bb-claude-auth.")
+        self.home = Path(self.tmp.name)
+        self.auth_path = self.home / ".claude.json"
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _write(self, payload: dict) -> None:  # type: ignore
+        self.auth_path.write_text(json.dumps(payload))
+
+    def test_auth_path_uses_home_env(self):
+        self.assertEqual(
+            self.auth_path,
+            claude_auth_path({"HOME": str(self.home)}),
+        )
+
+    def test_returns_session_key(self):
+        key = "sk-ant-oat01-real-key"
+        self._write({"oauthAccount": {"sessionKey": key}})
+        out = claude_host_access_token({"HOME": str(self.home)})
+        self.assertEqual(key, out)
+
+    def test_missing_auth_file_dies(self):
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_missing_oauth_account_dies(self):
+        self._write({"hasCompletedOnboarding": True})
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_missing_session_key_dies(self):
+        self._write({"oauthAccount": {"expiresAt": 2000000000}})
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_empty_session_key_dies(self):
+        self._write({"oauthAccount": {"sessionKey": ""}})
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_expired_token_dies(self):
+        self._write({
+            "oauthAccount": {
+                "sessionKey": "sk-ant-oat01-x",
+                "expiresAt": 1000,
+            },
+        })
+        with self.assertRaises(Die):
+            claude_host_access_token(
+                {"HOME": str(self.home)},
+                now=datetime(2026, 1, 1, tzinfo=timezone.utc),
+            )
+
+    def test_future_expiry_is_accepted(self):
+        key = "sk-ant-oat01-y"
+        self._write({
+            "oauthAccount": {
+                "sessionKey": key,
+                "expiresAt": 2000000000,
+            },
+        })
+        out = claude_host_access_token(
+            {"HOME": str(self.home)},
+            now=datetime(2026, 1, 1, tzinfo=timezone.utc),
+        )
+        self.assertEqual(key, out)
+
+    def test_absent_expiry_field_is_accepted(self):
+        key = "sk-ant-oat01-z"
+        self._write({"oauthAccount": {"sessionKey": key}})
+        out = claude_host_access_token({"HOME": str(self.home)})
+        self.assertEqual(key, out)
+
+    def test_non_json_file_dies(self):
+        self.auth_path.write_text("not json {{{")
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_json_array_root_dies(self):
+        self.auth_path.write_text("[]")
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+
+if __name__ == "__main__":
+    unittest.main()