add common/baserow-oauth-stub/oauth_app.py (fix connecteur Claude.ai Baserow MCP)

2026-07-28 12:25:40 +00:00
parent 071be110a8
commit 0f11f07567
1 changed files with 115 additions and 0 deletions
@@ -0,0 +1,115 @@
+"""
+Stub OAuth minimal pour satisfaire le handshake de connecteur Claude.ai
+devant Baserow (image officielle, non patchable directement).
+
+Ne fait AUCUNE verification de securite reelle : la vraie protection
+reste le token secret deja present dans l'URL /mcp/<token>/sse de Baserow.
+Ce stub sert uniquement a repondre aux sondes de decouverte OAuth/DCR
+que Claude.ai effectue systematiquement lors de l'ajout d'un connecteur
+personnalise, meme quand le serveur cible n'a besoin d'aucune auth.
+
+Modele repris de bolbol/nas-runbooks (fix mcp-vps, juillet 2026).
+"""
+import hashlib
+import base64
+import secrets
+import time
+from urllib.parse import urlencode
+
+from starlette.applications import Starlette
+from starlette.responses import JSONResponse, RedirectResponse
+from starlette.routing import Route
+from starlette.requests import Request
+
+CLIENT_ID = "claude-baserow-mcp"
+PUBLIC_BASE = "https://baserow.bolbol.tn"
+AUTH_TTL = 300
+TOKEN_TTL = 3600 * 24 * 30
+
+_auth_codes = {}
+_access_tokens = {}
+
+
+def _pkce_ok(verifier, challenge, method):
+    if not challenge:
+        return True
+    if method == "S256":
+        digest = hashlib.sha256(verifier.encode()).digest()
+        computed = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
+        return computed == challenge
+    return verifier == challenge
+
+
+async def client_register(request: Request) -> JSONResponse:
+    return JSONResponse({
+        "client_id": CLIENT_ID,
+        "client_id_issued_at": int(time.time()),
+        "redirect_uris": [],
+        "grant_types": ["authorization_code", "refresh_token"],
+        "response_types": ["code"],
+        "token_endpoint_auth_method": "none",
+    }, status_code=201)
+
+
+async def authorize(request: Request):
+    q = request.query_params
+    redirect_uri = q.get("redirect_uri", "")
+    state = q.get("state", "")
+    code_challenge = q.get("code_challenge", "")
+    code_challenge_method = q.get("code_challenge_method", "S256")
+    if not redirect_uri:
+        return JSONResponse({"error": "invalid_request", "error_description": "redirect_uri manquant"}, status_code=400)
+    code = secrets.token_urlsafe(32)
+    _auth_codes[code] = {
+        "redirect_uri": redirect_uri,
+        "code_challenge": code_challenge,
+        "code_challenge_method": code_challenge_method,
+        "expires": time.time() + AUTH_TTL,
+    }
+    params = {"code": code}
+    if state:
+        params["state"] = state
+    return RedirectResponse(f"{redirect_uri}?{urlencode(params)}", status_code=302)
+
+
+async def token(request: Request) -> JSONResponse:
+    form = await request.form()
+    grant_type = form.get("grant_type", "")
+    if grant_type == "authorization_code":
+        code = form.get("code", "")
+        entry = _auth_codes.pop(code, None)
+        if not entry or entry["expires"] < time.time():
+            return JSONResponse({"error": "invalid_grant"}, status_code=400)
+        verifier = form.get("code_verifier", "")
+        if not _pkce_ok(verifier, entry["code_challenge"], entry["code_challenge_method"]):
+            return JSONResponse({"error": "invalid_grant", "error_description": "PKCE invalide"}, status_code=400)
+        access_token = secrets.token_urlsafe(32)
+        refresh_token = secrets.token_urlsafe(32)
+        _access_tokens[access_token] = time.time() + TOKEN_TTL
+        return JSONResponse({
+            "access_token": access_token,
+            "token_type": "bearer",
+            "expires_in": TOKEN_TTL,
+            "refresh_token": refresh_token,
+        })
+    elif grant_type == "refresh_token":
+        access_token = secrets.token_urlsafe(32)
+        _access_tokens[access_token] = time.time() + TOKEN_TTL
+        return JSONResponse({
+            "access_token": access_token,
+            "token_type": "bearer",
+            "expires_in": TOKEN_TTL,
+        })
+    return JSONResponse({"error": "unsupported_grant_type"}, status_code=400)
+
+
+async def health(request: Request) -> JSONResponse:
+    return JSONResponse({"status": "ok"})
+
+
+app = Starlette(routes=[
+    Route("/health", health, methods=["GET"]),
+    Route("/register", client_register, methods=["POST"]),
+    Route("/authorize", authorize, methods=["GET"]),
+    Route("/token", token, methods=["POST"]),
+])