116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
"""
|
|
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"]),
|
|
])
|