mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(scripts): reject invalid Claude Keychain credential containers (#6281)
* fix(scripts): validate Claude Keychain export credentials * fix(scripts): reuse validated Keychain access token
This commit is contained in:
parent
6cdc600895
commit
2970aa4323
@ -337,6 +337,8 @@ For Google's official Gemini OpenAI-compatible endpoint, use the
|
||||
eval "$(python3 scripts/export_claude_code_oauth.py --print-export)"
|
||||
```
|
||||
|
||||
The exporter rejects malformed credential containers and non-string or blank access tokens before printing a token, emitting a shell export, or writing a credentials file. Valid tokens are exported unchanged, and file exports preserve the full credential container.
|
||||
|
||||
API keys can also be set manually in `.env` (recommended) or exported in your shell:
|
||||
|
||||
```bash
|
||||
|
||||
121
backend/tests/test_claude_keychain_export.py
Normal file
121
backend/tests/test_claude_keychain_export.py
Normal file
@ -0,0 +1,121 @@
|
||||
"""Exercise the manual OAuth exporter with synthetic Keychain responses."""
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import shlex
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def exporter(monkeypatch):
|
||||
script = Path(__file__).resolve().parents[2] / "scripts" / "export_claude_code_oauth.py"
|
||||
spec = importlib.util.spec_from_file_location("claude_keychain_export", script)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
monkeypatch.setattr(module.platform, "system", lambda: "Darwin")
|
||||
return module
|
||||
|
||||
|
||||
def mock_keychain(exporter, monkeypatch, payload):
|
||||
response = SimpleNamespace(returncode=0, stdout=json.dumps(payload), stderr="")
|
||||
monkeypatch.setattr(exporter, "subprocess", SimpleNamespace(run=lambda *args, **kwargs: response))
|
||||
|
||||
|
||||
INVALID_CONTAINERS = [
|
||||
None,
|
||||
[],
|
||||
"synthetic-secret",
|
||||
42,
|
||||
{"claudeAiOauth": None},
|
||||
{"claudeAiOauth": []},
|
||||
{"claudeAiOauth": "synthetic-secret"},
|
||||
{"claudeAiOauth": {"accessToken": None}},
|
||||
{"claudeAiOauth": {"accessToken": True}},
|
||||
{"claudeAiOauth": {"accessToken": 42}},
|
||||
{"claudeAiOauth": {"accessToken": ["synthetic-secret"]}},
|
||||
{"claudeAiOauth": {"accessToken": {"value": "synthetic-secret"}}},
|
||||
{"claudeAiOauth": {"accessToken": ""}},
|
||||
{"claudeAiOauth": {"accessToken": " \t\n"}},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", INVALID_CONTAINERS)
|
||||
@pytest.mark.parametrize("action", ["print_token", "print_export", "write_credentials"])
|
||||
def test_invalid_keychain_container_fails_without_exporting(exporter, monkeypatch, tmp_path, capsys, payload, action):
|
||||
mock_keychain(exporter, monkeypatch, payload)
|
||||
destination = tmp_path / "credentials.json"
|
||||
args = Namespace(
|
||||
service="synthetic-service",
|
||||
account="synthetic-account",
|
||||
show_target=False,
|
||||
print_token=False,
|
||||
print_export=False,
|
||||
write_credentials=None,
|
||||
)
|
||||
setattr(args, action, destination if action == "write_credentials" else True)
|
||||
monkeypatch.setattr(exporter, "parse_args", lambda: args)
|
||||
|
||||
assert exporter.main() == 1
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == ""
|
||||
assert captured.err == "Claude Code Keychain item did not contain claudeAiOauth.accessToken.\n"
|
||||
assert "synthetic-secret" not in captured.err
|
||||
assert not destination.exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("token", ["synthetic-token'with-shell-quote", " synthetic-token "])
|
||||
def test_valid_container_preserves_all_export_modes(exporter, monkeypatch, tmp_path, capsys, token):
|
||||
payload = {
|
||||
"claudeAiOauth": {
|
||||
"accessToken": token,
|
||||
"refreshToken": "synthetic-refresh",
|
||||
"expiresAt": 123,
|
||||
},
|
||||
"metadata": "preserved",
|
||||
}
|
||||
mock_keychain(exporter, monkeypatch, payload)
|
||||
destination = tmp_path / "credentials.json"
|
||||
args = Namespace(
|
||||
service="synthetic-service",
|
||||
account="synthetic-account",
|
||||
show_target=False,
|
||||
print_token=True,
|
||||
print_export=True,
|
||||
write_credentials=destination,
|
||||
)
|
||||
monkeypatch.setattr(exporter, "parse_args", lambda: args)
|
||||
|
||||
assert exporter.main() == 0
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == f"{token}\nexport CLAUDE_CODE_OAUTH_TOKEN={shlex.quote(token)}\n"
|
||||
assert captured.err == f"Wrote Claude Code credentials to {destination}\n"
|
||||
assert json.loads(destination.read_text(encoding="utf-8")) == payload
|
||||
|
||||
|
||||
def test_export_uses_validated_token_without_reextracting_container(exporter, monkeypatch, tmp_path, capsys):
|
||||
token = "synthetic-validated-token"
|
||||
payload = {"metadata": "opaque-container"}
|
||||
monkeypatch.setattr(exporter, "load_keychain_container", lambda **kwargs: (payload, token))
|
||||
destination = tmp_path / "credentials.json"
|
||||
args = Namespace(
|
||||
service="synthetic-service",
|
||||
account="synthetic-account",
|
||||
show_target=False,
|
||||
print_token=True,
|
||||
print_export=True,
|
||||
write_credentials=destination,
|
||||
)
|
||||
monkeypatch.setattr(exporter, "parse_args", lambda: args)
|
||||
|
||||
assert exporter.main() == 0
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == f"{token}\nexport CLAUDE_CODE_OAUTH_TOKEN={shlex.quote(token)}\n"
|
||||
assert captured.err == f"Wrote Claude Code credentials to {destination}\n"
|
||||
assert json.loads(destination.read_text(encoding="utf-8")) == payload
|
||||
@ -1,3 +1,13 @@
|
||||
## Manual Claude OAuth Export
|
||||
|
||||
`export_claude_code_oauth.py` validates Keychain JSON as an object containing an
|
||||
object `claudeAiOauth` and a nonblank string `accessToken` before any export action.
|
||||
Malformed containers use the existing token-missing error without exposing their
|
||||
contents. The loader returns the full container together with the validated token;
|
||||
export actions use that token without repeating credential-shape assumptions or
|
||||
trimming its contents. Offline CLI coverage:
|
||||
`backend/tests/test_claude_keychain_export.py`.
|
||||
|
||||
## Service Startup Contracts
|
||||
|
||||
Optional browser dependency detection reads the top-level `tools:` sequence
|
||||
|
||||
@ -44,7 +44,7 @@ def default_account_name() -> str:
|
||||
return os.getenv("USER") or "claude-code-user"
|
||||
|
||||
|
||||
def load_keychain_container(service: str, account: str) -> dict[str, Any]:
|
||||
def load_keychain_container(service: str, account: str) -> tuple[dict[str, Any], str]:
|
||||
if platform.system() != "Darwin":
|
||||
raise RuntimeError("Claude Code Keychain export is only supported on macOS.")
|
||||
|
||||
@ -71,11 +71,12 @@ def load_keychain_container(service: str, account: str) -> dict[str, Any]:
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RuntimeError("Claude Code Keychain item did not contain valid JSON.") from exc
|
||||
|
||||
access_token = data.get("claudeAiOauth", {}).get("accessToken", "")
|
||||
if not access_token:
|
||||
oauth = data.get("claudeAiOauth") if isinstance(data, dict) else None
|
||||
access_token = oauth.get("accessToken") if isinstance(oauth, dict) else None
|
||||
if not isinstance(access_token, str) or not access_token.strip():
|
||||
raise RuntimeError("Claude Code Keychain item did not contain claudeAiOauth.accessToken.")
|
||||
|
||||
return data
|
||||
return data, access_token
|
||||
|
||||
|
||||
def write_credentials_file(output_path: Path, data: dict[str, Any]) -> None:
|
||||
@ -141,13 +142,11 @@ def main() -> int:
|
||||
return 0
|
||||
|
||||
try:
|
||||
data = load_keychain_container(service=args.service, account=args.account)
|
||||
data, access_token = load_keychain_container(service=args.service, account=args.account)
|
||||
except RuntimeError as exc:
|
||||
print(str(exc), file=sys.stderr)
|
||||
return 1
|
||||
|
||||
access_token = data["claudeAiOauth"]["accessToken"]
|
||||
|
||||
if args.print_token:
|
||||
print(access_token)
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user