ysj7705andClaude Opus 5.5
916d149ab5
Gitea에 동적 클라이언트 등록이 없어 MCP 클라이언트가 URL만으로 OAuth 연결을 못 하는 문제를 이 브리지가 인가 서버 역할을 대신해 해결한다. G3X의 plane-mcp-oauth-bridge(upstream_oauth 모드)에서 가져와 Gitea 전용으로 정리했다: Plane PAT 연결 페이지·워크스페이스 헤더·bridge_mode 분기를 없앴다. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import base64
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from app.crypto import encrypt_token
|
|
from app.config import get_settings
|
|
|
|
|
|
def _pkce_pair() -> tuple[str, str]:
|
|
verifier = "test-verifier-string-long-enough-1234567890"
|
|
digest = hashlib.sha256(verifier.encode("ascii")).digest()
|
|
challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
|
|
return verifier, challenge
|
|
|
|
|
|
def test_health(client):
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"ok": True}
|
|
|
|
|
|
def test_registration_disabled_by_flag(client, monkeypatch):
|
|
monkeypatch.setenv("DCR_ENABLED", "false")
|
|
get_settings.cache_clear()
|
|
resp = client.post("/register", json={"client_name": "x", "redirect_uris": ["http://localhost:1/cb"]})
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error"] == "registration_not_supported"
|
|
|
|
|
|
def test_authorize_rejects_unknown_client(client):
|
|
_, challenge = _pkce_pair()
|
|
resp = client.get(
|
|
"/authorize",
|
|
params={
|
|
"response_type": "code",
|
|
"client_id": "unknown",
|
|
"redirect_uri": "http://localhost:8787/callback",
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_authorize_starts_for_allowed_client(client, database, monkeypatch):
|
|
settings = get_settings()
|
|
user = database.upsert_user("sub-1", "u@example.com")
|
|
database.set_upstream_tokens(user.id, encrypt_token(settings.token_encryption_key, "gitea_tok"), None, time.time() + 3600)
|
|
|
|
import app.main as main
|
|
|
|
monkeypatch.setattr(main, "session_user_id", lambda request: user.id)
|
|
|
|
_, challenge = _pkce_pair()
|
|
resp = client.get(
|
|
"/authorize",
|
|
params={
|
|
"response_type": "code",
|
|
"client_id": "cursor",
|
|
"redirect_uri": "http://localhost:8787/callback",
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
"state": "abc",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert resp.status_code == 302
|
|
assert resp.headers["location"].startswith("http://localhost:8787/callback?")
|
|
assert "code=" in resp.headers["location"]
|
|
assert "state=abc" in resp.headers["location"]
|
|
|
|
|
|
def test_mcp_without_token_returns_401(client):
|
|
resp = client.post("/mcp", json={"jsonrpc": "2.0", "method": "ping", "id": 1})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_mcp_proxies_with_gitea_token(client, database):
|
|
settings = get_settings()
|
|
user = database.upsert_user("sub-2", "u2@example.com")
|
|
database.set_upstream_tokens(user.id, encrypt_token(settings.token_encryption_key, "gitea_secret"), None, time.time() + 3600)
|
|
token = database.create_access_token("cursor", user.id)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'{"ok":true}'
|
|
mock_response.headers = {"content-type": "application/json"}
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.request = AsyncMock(return_value=mock_response)
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("app.main.httpx.AsyncClient", return_value=mock_client):
|
|
resp = client.post(
|
|
"/mcp",
|
|
json={"jsonrpc": "2.0", "method": "tools/list", "id": 1},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
mock_client.request.assert_awaited_once()
|
|
call_kwargs = mock_client.request.await_args.kwargs
|
|
assert call_kwargs["headers"]["Authorization"] == "Bearer gitea_secret"
|