Saisissez le mot de securite pour autoriser l acces.
{error} """ 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='Mot de securite incorrect.
', 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"]), ]