diff --git a/common/baserow-oauth-stub/oauth_app.py b/common/baserow-oauth-stub/oauth_app.py new file mode 100644 index 0000000..30fc0dc --- /dev/null +++ b/common/baserow-oauth-stub/oauth_app.py @@ -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//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"]), +])