270 lines
11 KiB
Python
270 lines
11 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
mcp_oauth.py — OAuth 2.1 partagé pour les serveurs MCP · Chantier C8
|
||
|
|
===================================================================
|
||
|
|
Extrait fidèle du bloc OAuth de mcp_server.py (Context Continuity),
|
||
|
|
rendu réutilisable : chaque serveur MCP l'instancie avec sa propre URL
|
||
|
|
de base et son propre fichier d'état. Aucune logique métier ici.
|
||
|
|
|
||
|
|
Modèle : OAuth 2.1 mono-utilisateur, protégé par un « mot de sécurité »
|
||
|
|
(SECURITY_WORD). Dynamic Client Registration (RFC 7591), PKCE S256,
|
||
|
|
authorization_code + refresh_token. Tokens émis préfixés « at_ ».
|
||
|
|
État persistant dans un JSON (clients / auth_codes / tokens).
|
||
|
|
|
||
|
|
Compatible Python 3.9, Starlette pur.
|
||
|
|
|
||
|
|
Usage dans un serveur MCP :
|
||
|
|
from mcp_oauth import OAuthProvider
|
||
|
|
oauth = OAuthProvider(
|
||
|
|
base_url=os.getenv("OAUTH_BASE_URL", "https://mcp-x.exemple.fr"),
|
||
|
|
state_file=os.path.expanduser("~/App/X/oauth_state.json"),
|
||
|
|
security_word=os.getenv("SECURITY_WORD", ""),
|
||
|
|
realm="X MCP")
|
||
|
|
# routes : oauth.routes() → à concaténer aux routes du serveur
|
||
|
|
# auth : oauth.check_token(token) → True si token at_ valide
|
||
|
|
"""
|
||
|
|
import base64
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import secrets
|
||
|
|
import time
|
||
|
|
|
||
|
|
from starlette.responses import (HTMLResponse, JSONResponse,
|
||
|
|
RedirectResponse)
|
||
|
|
from starlette.routing import Route
|
||
|
|
|
||
|
|
AUTHORIZE_PAGE = """<!DOCTYPE html>
|
||
|
|
<html lang="fr"><head><meta charset="utf-8"><title>Autorisation {realm}</title>
|
||
|
|
<style>body{{font-family:sans-serif;max-width:400px;margin:80px auto;padding:20px}}
|
||
|
|
input{{width:100%;padding:10px;margin:10px 0;box-sizing:border-box}}
|
||
|
|
button{{padding:10px 20px;background:#061033;color:#fff;border:none;cursor:pointer}}
|
||
|
|
.err{{color:#c00}}</style></head>
|
||
|
|
<body><h2>Autorisation {realm}</h2>
|
||
|
|
<p>Saisissez le mot de securite pour autoriser l acces.</p>
|
||
|
|
{error}
|
||
|
|
<form method="POST" action="/authorize">
|
||
|
|
<input type="hidden" name="client_id" value="{client_id}">
|
||
|
|
<input type="hidden" name="redirect_uri" value="{redirect_uri}">
|
||
|
|
<input type="hidden" name="state" value="{state}">
|
||
|
|
<input type="hidden" name="code_challenge" value="{code_challenge}">
|
||
|
|
<input type="hidden" name="code_challenge_method" value="{code_challenge_method}">
|
||
|
|
<input type="password" name="security_word" placeholder="Mot de securite" autofocus>
|
||
|
|
<button type="submit">Autoriser</button>
|
||
|
|
</form></body></html>"""
|
||
|
|
|
||
|
|
|
||
|
|
def _verify_pkce(code_verifier, code_challenge):
|
||
|
|
"""PKCE S256 : base64url(sha256(verifier)) == challenge."""
|
||
|
|
digest = hashlib.sha256(code_verifier.encode("ascii")).digest()
|
||
|
|
computed = base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
|
||
|
|
return computed == code_challenge
|
||
|
|
|
||
|
|
|
||
|
|
class OAuthProvider:
|
||
|
|
def __init__(self, base_url, state_file, security_word, realm="MCP"):
|
||
|
|
self.base_url = base_url.rstrip("/")
|
||
|
|
self.state_file = state_file
|
||
|
|
self.security_word = security_word
|
||
|
|
self.realm = realm
|
||
|
|
self._ensure_state()
|
||
|
|
|
||
|
|
# ── état ────────────────────────────────────────────────────────────
|
||
|
|
def _ensure_state(self):
|
||
|
|
if not os.path.exists(self.state_file):
|
||
|
|
os.makedirs(os.path.dirname(self.state_file), exist_ok=True)
|
||
|
|
self._save({"clients": {}, "auth_codes": {}, "tokens": {}})
|
||
|
|
|
||
|
|
def _load(self):
|
||
|
|
with open(self.state_file) as f:
|
||
|
|
state = json.load(f)
|
||
|
|
for k in ("clients", "auth_codes", "tokens"):
|
||
|
|
state.setdefault(k, {})
|
||
|
|
return state
|
||
|
|
|
||
|
|
def _save(self, state):
|
||
|
|
with open(self.state_file, "w") as f:
|
||
|
|
json.dump(state, f, indent=2)
|
||
|
|
|
||
|
|
# ── vérification (appelée par check_auth du serveur) ────────────────
|
||
|
|
def check_token(self, token):
|
||
|
|
"""True si le token est un access_token OAuth émis et connu."""
|
||
|
|
if not token or not token.startswith("at_"):
|
||
|
|
return False
|
||
|
|
return bool(self._load()["tokens"].get(token))
|
||
|
|
|
||
|
|
# ── endpoints ───────────────────────────────────────────────────────
|
||
|
|
async def well_known_protected_resource(self, request):
|
||
|
|
return JSONResponse({
|
||
|
|
"resource": self.base_url + "/mcp",
|
||
|
|
"authorization_servers": [self.base_url],
|
||
|
|
})
|
||
|
|
|
||
|
|
async def well_known_authorization_server(self, request):
|
||
|
|
return JSONResponse({
|
||
|
|
"issuer": self.base_url,
|
||
|
|
"authorization_endpoint": self.base_url + "/authorize",
|
||
|
|
"token_endpoint": self.base_url + "/token",
|
||
|
|
"registration_endpoint": self.base_url + "/register",
|
||
|
|
"response_types_supported": ["code"],
|
||
|
|
"grant_types_supported": ["authorization_code",
|
||
|
|
"refresh_token"],
|
||
|
|
"code_challenge_methods_supported": ["S256"],
|
||
|
|
"token_endpoint_auth_methods_supported": ["none"],
|
||
|
|
})
|
||
|
|
|
||
|
|
async def register(self, request):
|
||
|
|
"""Dynamic Client Registration (RFC 7591)."""
|
||
|
|
try:
|
||
|
|
body = await request.json()
|
||
|
|
except Exception:
|
||
|
|
body = {}
|
||
|
|
client_id = "client_" + secrets.token_urlsafe(16)
|
||
|
|
state = self._load()
|
||
|
|
state["clients"][client_id] = {
|
||
|
|
"client_id": client_id,
|
||
|
|
"redirect_uris": body.get("redirect_uris", []),
|
||
|
|
"client_name": body.get("client_name", "unknown"),
|
||
|
|
"created_at": int(time.time()),
|
||
|
|
}
|
||
|
|
self._save(state)
|
||
|
|
return JSONResponse({
|
||
|
|
"client_id": client_id,
|
||
|
|
"redirect_uris": body.get("redirect_uris", []),
|
||
|
|
"client_name": body.get("client_name", "unknown"),
|
||
|
|
"token_endpoint_auth_method": "none",
|
||
|
|
"grant_types": ["authorization_code", "refresh_token"],
|
||
|
|
"response_types": ["code"],
|
||
|
|
}, status_code=201)
|
||
|
|
|
||
|
|
async def authorize_get(self, request):
|
||
|
|
q = request.query_params
|
||
|
|
page = AUTHORIZE_PAGE.format(
|
||
|
|
realm=self.realm, error="",
|
||
|
|
client_id=q.get("client_id", ""),
|
||
|
|
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"),
|
||
|
|
)
|
||
|
|
return HTMLResponse(page)
|
||
|
|
|
||
|
|
async def authorize_post(self, request):
|
||
|
|
form = await request.form()
|
||
|
|
word = form.get("security_word", "")
|
||
|
|
client_id = form.get("client_id", "")
|
||
|
|
redirect_uri = form.get("redirect_uri", "")
|
||
|
|
state_param = form.get("state", "")
|
||
|
|
code_challenge = form.get("code_challenge", "")
|
||
|
|
code_challenge_method = form.get("code_challenge_method", "S256")
|
||
|
|
|
||
|
|
if word != self.security_word:
|
||
|
|
page = AUTHORIZE_PAGE.format(
|
||
|
|
realm=self.realm,
|
||
|
|
error='<p class="err">Mot de securite incorrect.</p>',
|
||
|
|
client_id=client_id, redirect_uri=redirect_uri,
|
||
|
|
state=state_param, code_challenge=code_challenge,
|
||
|
|
code_challenge_method=code_challenge_method,
|
||
|
|
)
|
||
|
|
return HTMLResponse(page, status_code=401)
|
||
|
|
|
||
|
|
oauth = self._load()
|
||
|
|
if client_id not in oauth["clients"]:
|
||
|
|
return JSONResponse({"error": "client inconnu"},
|
||
|
|
status_code=400)
|
||
|
|
code = "code_" + secrets.token_urlsafe(24)
|
||
|
|
oauth["auth_codes"][code] = {
|
||
|
|
"client_id": client_id,
|
||
|
|
"redirect_uri": redirect_uri,
|
||
|
|
"code_challenge": code_challenge,
|
||
|
|
"code_challenge_method": code_challenge_method,
|
||
|
|
"created_at": int(time.time()),
|
||
|
|
"used": False,
|
||
|
|
}
|
||
|
|
self._save(oauth)
|
||
|
|
sep = "&" if "?" in redirect_uri else "?"
|
||
|
|
location = redirect_uri + sep + "code=" + code
|
||
|
|
if state_param:
|
||
|
|
location += "&state=" + state_param
|
||
|
|
return RedirectResponse(location, status_code=302)
|
||
|
|
|
||
|
|
async def token(self, request):
|
||
|
|
form = await request.form()
|
||
|
|
grant_type = form.get("grant_type", "")
|
||
|
|
oauth = self._load()
|
||
|
|
|
||
|
|
if grant_type == "authorization_code":
|
||
|
|
code = form.get("code", "")
|
||
|
|
code_verifier = form.get("code_verifier", "")
|
||
|
|
entry = oauth["auth_codes"].get(code)
|
||
|
|
if not entry:
|
||
|
|
return JSONResponse({"error": "invalid_grant"},
|
||
|
|
status_code=400)
|
||
|
|
if entry.get("used"):
|
||
|
|
return JSONResponse(
|
||
|
|
{"error": "invalid_grant",
|
||
|
|
"error_description": "code deja utilise"},
|
||
|
|
status_code=400)
|
||
|
|
challenge = entry.get("code_challenge", "")
|
||
|
|
if challenge:
|
||
|
|
if not code_verifier or not _verify_pkce(code_verifier,
|
||
|
|
challenge):
|
||
|
|
return JSONResponse(
|
||
|
|
{"error": "invalid_grant",
|
||
|
|
"error_description": "PKCE echec"},
|
||
|
|
status_code=400)
|
||
|
|
access_token = "at_" + secrets.token_urlsafe(32)
|
||
|
|
refresh_token = "rt_" + secrets.token_urlsafe(32)
|
||
|
|
oauth["tokens"][access_token] = {
|
||
|
|
"client_id": entry["client_id"],
|
||
|
|
"refresh_token": refresh_token,
|
||
|
|
"created_at": int(time.time()),
|
||
|
|
"type": "access",
|
||
|
|
}
|
||
|
|
oauth["auth_codes"][code]["used"] = True
|
||
|
|
self._save(oauth)
|
||
|
|
return JSONResponse({
|
||
|
|
"access_token": access_token,
|
||
|
|
"token_type": "Bearer",
|
||
|
|
"expires_in": 31536000,
|
||
|
|
"refresh_token": refresh_token,
|
||
|
|
})
|
||
|
|
|
||
|
|
elif grant_type == "refresh_token":
|
||
|
|
rt = form.get("refresh_token", "")
|
||
|
|
for at, meta in list(oauth["tokens"].items()):
|
||
|
|
if meta.get("refresh_token") == rt:
|
||
|
|
new_at = "at_" + secrets.token_urlsafe(32)
|
||
|
|
oauth["tokens"][new_at] = {
|
||
|
|
"client_id": meta["client_id"],
|
||
|
|
"refresh_token": rt,
|
||
|
|
"created_at": int(time.time()),
|
||
|
|
"type": "access",
|
||
|
|
}
|
||
|
|
self._save(oauth)
|
||
|
|
return JSONResponse({
|
||
|
|
"access_token": new_at,
|
||
|
|
"token_type": "Bearer",
|
||
|
|
"expires_in": 31536000,
|
||
|
|
"refresh_token": rt,
|
||
|
|
})
|
||
|
|
return JSONResponse({"error": "invalid_grant"},
|
||
|
|
status_code=400)
|
||
|
|
|
||
|
|
return JSONResponse({"error": "unsupported_grant_type"},
|
||
|
|
status_code=400)
|
||
|
|
|
||
|
|
def routes(self):
|
||
|
|
"""Routes OAuth à concaténer à celles du serveur MCP."""
|
||
|
|
return [
|
||
|
|
Route("/.well-known/oauth-protected-resource",
|
||
|
|
self.well_known_protected_resource, methods=["GET"]),
|
||
|
|
Route("/.well-known/oauth-authorization-server",
|
||
|
|
self.well_known_authorization_server, methods=["GET"]),
|
||
|
|
Route("/register", self.register, methods=["POST"]),
|
||
|
|
Route("/authorize", self.authorize_get, methods=["GET"]),
|
||
|
|
Route("/authorize", self.authorize_post, methods=["POST"]),
|
||
|
|
Route("/token", self.token, methods=["POST"]),
|
||
|
|
]
|