Gitea MCP OAuth 브리지 (DCR, Gitea 로그인)
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>
This commit is contained in:
commit
916d149ab5
18 files changed
+1494
No files matched your search
+55
@@ -0,0 +1,55 @@
|
||||
"""RFC 7591 dynamic client registration helpers (G3X, 2026-10-19)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import time
|
||||
from collections import defaultdict, deque
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from fastapi import Request
|
||||
|
||||
_SCHEME_RE = re.compile(r"^[a-z][a-z0-9+.\-]*$")
|
||||
_BAD_SCHEMES = {"javascript", "data", "file", "vbscript", "about", "blob", "ftp", "ws", "wss"}
|
||||
_LOOPBACK = {"localhost", "127.0.0.1", "::1"}
|
||||
SUPPORTED_GRANTS = {"authorization_code", "refresh_token"}
|
||||
|
||||
|
||||
def valid_redirect_uri(uri: object) -> bool:
|
||||
"""https URLs, http loopback URLs, or private app schemes (cursor://...)."""
|
||||
if not isinstance(uri, str) or not uri or len(uri) > 2000 or "*" in uri:
|
||||
return False
|
||||
if any(c.isspace() or ord(c) < 32 for c in uri):
|
||||
return False
|
||||
p = urlparse(uri)
|
||||
if p.fragment or not p.scheme:
|
||||
return False
|
||||
if p.scheme == "https":
|
||||
return bool(p.hostname)
|
||||
if p.scheme == "http":
|
||||
return p.hostname in _LOOPBACK
|
||||
return bool(_SCHEME_RE.match(p.scheme)) and p.scheme not in _BAD_SCHEMES
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
def __init__(self, max_events: int, window_seconds: float) -> None:
|
||||
self.max_events = max_events
|
||||
self.window = window_seconds
|
||||
self._events: dict[str, deque[float]] = defaultdict(deque)
|
||||
|
||||
def hit(self, key: str) -> bool:
|
||||
"""Record an event; return True when the limit is exceeded."""
|
||||
now = time.time()
|
||||
q = self._events[key]
|
||||
while q and q[0] < now - self.window:
|
||||
q.popleft()
|
||||
if len(q) >= self.max_events:
|
||||
return True
|
||||
q.append(now)
|
||||
return False
|
||||
|
||||
|
||||
def client_ip(request: Request) -> str:
|
||||
fwd = request.headers.get("x-forwarded-for", "")
|
||||
if fwd:
|
||||
return fwd.split(",")[0].strip()
|
||||
return request.client.host if request.client else "unknown"
|
||||
Reference in new issue
Block a user