Bastien Gourdon
659 lines
25 KiB
Python
659 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
mcp_forgejo_server.py — Serveur MCP pur Starlette (compatible Python 3.9)
|
|
Acces en lecture aux depots Forgejo via le protocole MCP (JSON-RPC 2.0).
|
|
|
|
Calque sur ~/App/Context_continuity/mcp_server.py :
|
|
transport Streamable HTTP sur /mcp, auth Bearer (tokens statiques + OAuth 2.1
|
|
avec PKCE), etat OAuth isole dans oauth_state_forgejo.json.
|
|
|
|
Usage : python3 mcp_forgejo_server.py -> ecoute sur 127.0.0.1:8770
|
|
|
|
Variables .env requises :
|
|
FORGEJO_URL (defaut http://localhost:3000)
|
|
FORGEJO_TOKEN token d'API Forgejo, permissions lecture repo
|
|
API_KEY_CLAUDE token statique Claude
|
|
API_KEY_LECHAT token statique Le Chat
|
|
OAUTH_BASE_URL defaut https://mcp-forgejo.bertha-cloud.fr
|
|
SECURITY_WORD mot de securite de la page /authorize
|
|
"""
|
|
import os
|
|
import json
|
|
import secrets
|
|
import hashlib
|
|
import base64
|
|
import time
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Route
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse, Response, HTMLResponse, RedirectResponse
|
|
|
|
load_dotenv()
|
|
|
|
# --- Configuration Forgejo ---
|
|
FORGEJO_URL = os.getenv("FORGEJO_URL", "http://localhost:3000")
|
|
FORGEJO_TOKEN = os.getenv("FORGEJO_TOKEN", "")
|
|
|
|
if not FORGEJO_TOKEN:
|
|
raise ValueError("FORGEJO_TOKEN doit etre defini (token d'API Forgejo, acces lecture).")
|
|
|
|
# --- Auth MCP ---
|
|
API_KEY_CLAUDE = os.getenv("API_KEY_CLAUDE", "")
|
|
API_KEY_LECHAT = os.getenv("API_KEY_LECHAT", "")
|
|
VALID_TOKENS = {}
|
|
if API_KEY_CLAUDE:
|
|
VALID_TOKENS[API_KEY_CLAUDE] = "Claude"
|
|
if API_KEY_LECHAT:
|
|
VALID_TOKENS[API_KEY_LECHAT] = "LeChat"
|
|
|
|
PROTOCOL_VERSION = "2025-06-18"
|
|
SERVER_INFO = {"name": "Forgejo Code Access", "version": "1.0.0"}
|
|
|
|
# --- Config OAuth 2.1 (etat isole du serveur Trilium) ---
|
|
OAUTH_BASE_URL = os.getenv("OAUTH_BASE_URL", "https://mcp-forgejo.bertha-cloud.fr")
|
|
SECURITY_WORD = os.getenv("SECURITY_WORD", "")
|
|
OAUTH_STATE_FILE = os.path.expanduser("~/App/Code_versioning/oauth_state_forgejo.json")
|
|
|
|
|
|
def load_oauth_state():
|
|
if not os.path.exists(OAUTH_STATE_FILE):
|
|
return {"clients": {}, "auth_codes": {}, "tokens": {}}
|
|
with open(OAUTH_STATE_FILE) as f:
|
|
state = json.load(f)
|
|
state.setdefault("clients", {})
|
|
state.setdefault("auth_codes", {})
|
|
state.setdefault("tokens", {})
|
|
return state
|
|
|
|
|
|
def save_oauth_state(state):
|
|
with open(OAUTH_STATE_FILE, "w") as f:
|
|
json.dump(state, f, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper API Forgejo
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def forgejo_api(method, endpoint, **kwargs):
|
|
"""Appel generique a l'API Forgejo."""
|
|
url = "%s/%s" % (FORGEJO_URL.rstrip("/"), endpoint.lstrip("/"))
|
|
headers = {"Authorization": "token %s" % FORGEJO_TOKEN}
|
|
response = requests.request(method, url, headers=headers, timeout=30, **kwargs)
|
|
response.raise_for_status()
|
|
return response.json() if response.content else None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Definition des tools (schema JSON)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TOOLS = [
|
|
{
|
|
"name": "list_repos",
|
|
"description": "Liste tous les depots Forgejo accessibles avec le token (nom complet, description, branche par defaut, visibilite).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {},
|
|
"required": [],
|
|
},
|
|
},
|
|
{
|
|
"name": "list_files",
|
|
"description": "Liste les fichiers et dossiers d un depot a un chemin donne. repo au format proprietaire/depot (ex: Master/sliding-automation). path vide = racine.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "proprietaire/depot"},
|
|
"path": {"type": "string", "description": "Chemin dans le depot (defaut: racine)"},
|
|
"ref": {"type": "string", "description": "Branche, tag ou commit (defaut: branche par defaut)"},
|
|
},
|
|
"required": ["repo"],
|
|
},
|
|
},
|
|
{
|
|
"name": "read_file",
|
|
"description": "Lit le contenu texte d un fichier d un depot. repo au format proprietaire/depot. Refuse les fichiers binaires et ceux depassant la taille limite.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "proprietaire/depot"},
|
|
"path": {"type": "string", "description": "Chemin du fichier dans le depot"},
|
|
"ref": {"type": "string", "description": "Branche, tag ou commit (defaut: branche par defaut)"},
|
|
},
|
|
"required": ["repo", "path"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_metadata",
|
|
"description": "Metadonnees d un fichier (taille, sha, type, urls) sans telecharger le contenu. Utile avant de lire un gros fichier.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "proprietaire/depot"},
|
|
"path": {"type": "string", "description": "Chemin du fichier dans le depot"},
|
|
"ref": {"type": "string", "description": "Branche, tag ou commit (defaut: branche par defaut)"},
|
|
},
|
|
"required": ["repo", "path"],
|
|
},
|
|
},
|
|
|
|
{
|
|
"name": "list_commits",
|
|
"description": "Historique des commits d un depot (sha, auteur, date, message). path optionnel : limite l historique a un fichier ou dossier precis. Ne retourne pas les diffs.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "proprietaire/depot"},
|
|
"path": {"type": "string", "description": "Optionnel - limite l historique a ce fichier ou dossier"},
|
|
"ref": {"type": "string", "description": "Branche, tag ou commit (defaut: branche par defaut)"},
|
|
"limit": {"type": "integer", "description": "Nombre de commits (defaut 20, max 50)"},
|
|
},
|
|
"required": ["repo"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_commit",
|
|
"description": "Detail d un commit : metadonnees, fichiers touches et statistiques. include_diff=true ajoute le diff complet (refuse au-dela de la limite de taille).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "proprietaire/depot"},
|
|
"sha": {"type": "string", "description": "SHA du commit (complet ou abrege)"},
|
|
"include_diff": {"type": "boolean", "description": "Inclure le diff complet (defaut false)"},
|
|
},
|
|
"required": ["repo", "sha"],
|
|
},
|
|
},
|
|
]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Implementation des tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
MAX_FILE_BYTES = 200000 # garde-fou : au-dela, on renvoie les metadonnees
|
|
|
|
|
|
def _valider_repo(repo):
|
|
"""Le format attendu est proprietaire/depot. Evite les traversees de chemin."""
|
|
if not repo or repo.count("/") != 1:
|
|
return {"error": "repo doit etre au format proprietaire/depot (ex: Master/sliding-automation)"}
|
|
if ".." in repo:
|
|
return {"error": "repo invalide"}
|
|
return None
|
|
|
|
|
|
def _valider_path(path):
|
|
if path and ".." in path:
|
|
return {"error": "path invalide (traversee de repertoire)"}
|
|
return None
|
|
|
|
|
|
def tool_list_repos():
|
|
repos = forgejo_api("GET", "/api/v1/user/repos") or []
|
|
return [{
|
|
"full_name": r.get("full_name"),
|
|
"description": r.get("description") or "",
|
|
"default_branch": r.get("default_branch"),
|
|
"private": r.get("private"),
|
|
"updated_at": r.get("updated_at"),
|
|
"size_kb": r.get("size"),
|
|
} for r in repos]
|
|
|
|
|
|
def tool_list_files(repo, path="", ref=None):
|
|
err = _valider_repo(repo) or _valider_path(path)
|
|
if err:
|
|
return err
|
|
params = {"ref": ref} if ref else None
|
|
data = forgejo_api("GET", "/api/v1/repos/%s/contents/%s" % (repo, path.lstrip("/")), params=params)
|
|
if isinstance(data, dict):
|
|
data = [data]
|
|
return [{
|
|
"name": e.get("name"),
|
|
"path": e.get("path"),
|
|
"type": e.get("type"),
|
|
"size": e.get("size"),
|
|
} for e in (data or [])]
|
|
|
|
|
|
def tool_read_file(repo, path, ref=None):
|
|
err = _valider_repo(repo) or _valider_path(path)
|
|
if err:
|
|
return err
|
|
params = {"ref": ref} if ref else None
|
|
data = forgejo_api("GET", "/api/v1/repos/%s/contents/%s" % (repo, path.lstrip("/")), params=params)
|
|
if not data or data.get("type") != "file":
|
|
return {"error": "Fichier introuvable ou chemin non fichier : %s" % path}
|
|
size = data.get("size") or 0
|
|
if size > MAX_FILE_BYTES:
|
|
return {"error": "Fichier trop volumineux (%d octets, limite %d). Utiliser get_metadata." % (size, MAX_FILE_BYTES),
|
|
"path": path, "repo": repo, "size": size}
|
|
encoded = data.get("content")
|
|
if encoded and data.get("encoding") == "base64":
|
|
raw = base64.b64decode(encoded)
|
|
else:
|
|
dl = requests.get(data["download_url"],
|
|
headers={"Authorization": "token %s" % FORGEJO_TOKEN},
|
|
timeout=30)
|
|
dl.raise_for_status()
|
|
raw = dl.content
|
|
try:
|
|
content = raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return {"error": "Fichier binaire, lecture texte impossible", "path": path, "repo": repo}
|
|
return {"repo": repo, "path": path, "ref": ref or data.get("ref"),
|
|
"size": size, "sha": data.get("sha"), "content": content}
|
|
|
|
|
|
def tool_get_metadata(repo, path, ref=None):
|
|
err = _valider_repo(repo) or _valider_path(path)
|
|
if err:
|
|
return err
|
|
params = {"ref": ref} if ref else None
|
|
data = forgejo_api("GET", "/api/v1/repos/%s/contents/%s" % (repo, path.lstrip("/")), params=params)
|
|
if not data:
|
|
return {"error": "Introuvable : %s" % path}
|
|
if isinstance(data, list):
|
|
return {"repo": repo, "path": path, "type": "dir", "entries": len(data)}
|
|
return {
|
|
"repo": repo,
|
|
"path": data.get("path"),
|
|
"type": data.get("type"),
|
|
"size": data.get("size"),
|
|
"sha": data.get("sha"),
|
|
"encoding": data.get("encoding"),
|
|
"html_url": data.get("html_url"),
|
|
"download_url": data.get("download_url"),
|
|
}
|
|
|
|
MAX_DIFF_BYTES = 100000 # garde-fou diff : au-dela, on renvoie seulement les stats
|
|
|
|
|
|
def tool_list_commits(repo, path=None, ref=None, limit=20):
|
|
err = _valider_repo(repo) or _valider_path(path)
|
|
if err:
|
|
return err
|
|
try:
|
|
limit = int(limit)
|
|
except (TypeError, ValueError):
|
|
return {"error": "limit doit etre un entier"}
|
|
limit = max(1, min(limit, 50))
|
|
params = {"limit": limit, "stat": "false", "verification": "false", "files": "false"}
|
|
if path:
|
|
params["path"] = path.lstrip("/")
|
|
if ref:
|
|
params["sha"] = ref
|
|
data = forgejo_api("GET", "/api/v1/repos/%s/commits" % repo, params=params) or []
|
|
out = []
|
|
for c in data:
|
|
commit = c.get("commit") or {}
|
|
author = commit.get("author") or {}
|
|
message = (commit.get("message") or "").strip()
|
|
parts = message.split("\n", 1)
|
|
out.append({
|
|
"sha": (c.get("sha") or "")[:10],
|
|
"sha_full": c.get("sha"),
|
|
"date": author.get("date"),
|
|
"author": author.get("name"),
|
|
"subject": parts[0],
|
|
"body": parts[1].strip() if len(parts) > 1 else "",
|
|
})
|
|
return {"repo": repo, "path": path or "", "count": len(out), "commits": out}
|
|
|
|
|
|
def tool_get_commit(repo, sha, include_diff=False):
|
|
err = _valider_repo(repo)
|
|
if err:
|
|
return err
|
|
if not sha or "/" in sha or ".." in sha:
|
|
return {"error": "sha invalide"}
|
|
data = forgejo_api("GET", "/api/v1/repos/%s/git/commits/%s" % (repo, sha))
|
|
if not data:
|
|
return {"error": "Commit introuvable : %s" % sha}
|
|
commit = data.get("commit") or {}
|
|
author = commit.get("author") or {}
|
|
stats = data.get("stats") or {}
|
|
message = (commit.get("message") or "").strip()
|
|
parts = message.split("\n", 1)
|
|
result = {
|
|
"repo": repo,
|
|
"sha": data.get("sha"),
|
|
"date": author.get("date"),
|
|
"author": author.get("name"),
|
|
"subject": parts[0],
|
|
"body": parts[1].strip() if len(parts) > 1 else "",
|
|
"stats": {"additions": stats.get("additions"),
|
|
"deletions": stats.get("deletions"),
|
|
"total": stats.get("total")},
|
|
"files": [{"filename": f.get("filename"),
|
|
"status": f.get("status"),
|
|
"additions": f.get("additions"),
|
|
"deletions": f.get("deletions")}
|
|
for f in (data.get("files") or [])],
|
|
}
|
|
if include_diff:
|
|
url = "%s/api/v1/repos/%s/git/commits/%s.diff" % (FORGEJO_URL.rstrip("/"), repo, sha)
|
|
resp = requests.get(url, headers={"Authorization": "token %s" % FORGEJO_TOKEN}, timeout=30)
|
|
resp.raise_for_status()
|
|
raw = resp.content
|
|
if len(raw) > MAX_DIFF_BYTES:
|
|
result["diff_error"] = ("Diff trop volumineux (%d octets, limite %d). "
|
|
"Utiliser read_file sur les fichiers concernes."
|
|
% (len(raw), MAX_DIFF_BYTES))
|
|
else:
|
|
try:
|
|
result["diff"] = raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
result["diff_error"] = "Diff non decodable en UTF-8 (fichiers binaires)"
|
|
return result
|
|
|
|
|
|
DISPATCH = {
|
|
"list_repos": tool_list_repos,
|
|
"list_files": tool_list_files,
|
|
"read_file": tool_read_file,
|
|
"get_metadata": tool_get_metadata,
|
|
"list_commits": tool_list_commits,
|
|
"get_commit": tool_get_commit,
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Handler JSON-RPC MCP
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def check_auth(request):
|
|
auth = request.headers.get("authorization", "")
|
|
token = auth.replace("Bearer ", "").strip()
|
|
# 1) Tokens statiques (Le Chat, cle Claude historique)
|
|
if token and token in VALID_TOKENS:
|
|
return VALID_TOKENS[token]
|
|
# 2) Tokens OAuth emis (Claude Desktop)
|
|
if token.startswith("at_"):
|
|
oauth = load_oauth_state()
|
|
meta = oauth["tokens"].get(token)
|
|
if meta:
|
|
return "Claude-OAuth"
|
|
return None
|
|
|
|
|
|
async def mcp_endpoint(request: Request):
|
|
# Auth
|
|
llm = check_auth(request)
|
|
if not llm:
|
|
return JSONResponse(
|
|
{"error": "invalid_token", "error_description": "Authentication required"},
|
|
status_code=401,
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
body = await request.json()
|
|
method = body.get("method")
|
|
req_id = body.get("id")
|
|
params = body.get("params", {})
|
|
|
|
# initialize
|
|
if method == "initialize":
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"result": {
|
|
"protocolVersion": PROTOCOL_VERSION,
|
|
"capabilities": {"tools": {}},
|
|
"serverInfo": SERVER_INFO,
|
|
},
|
|
})
|
|
|
|
# notifications/initialized (pas de reponse attendue)
|
|
if method == "notifications/initialized":
|
|
return Response(status_code=202)
|
|
|
|
# tools/list
|
|
if method == "tools/list":
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"result": {"tools": TOOLS},
|
|
})
|
|
|
|
# tools/call
|
|
if method == "tools/call":
|
|
tool_name = params.get("name")
|
|
args = params.get("arguments", {})
|
|
fn = DISPATCH.get(tool_name)
|
|
if not fn:
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"error": {"code": -32601, "message": "Tool inconnu : %s" % tool_name},
|
|
})
|
|
try:
|
|
result = fn(**args)
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"result": {
|
|
"content": [{"type": "text",
|
|
"text": json.dumps(result, ensure_ascii=False, indent=2)}],
|
|
},
|
|
})
|
|
except requests.HTTPError as e:
|
|
status = e.response.status_code if e.response is not None else "?"
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"error": {"code": -32603, "message": "Forgejo HTTP %s : %s" % (status, str(e))},
|
|
})
|
|
except Exception as e:
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"error": {"code": -32603, "message": str(e)},
|
|
})
|
|
|
|
# methode inconnue
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"error": {"code": -32601, "message": "Methode inconnue : %s" % method},
|
|
})
|
|
|
|
|
|
async def health(request: Request):
|
|
return JSONResponse({"status": "ok", "server": "mcp-forgejo", "version": "1.0.0"})
|
|
|
|
|
|
async def well_known_protected_resource(request: Request):
|
|
return JSONResponse({
|
|
"resource": OAUTH_BASE_URL + "/mcp",
|
|
"authorization_servers": [OAUTH_BASE_URL],
|
|
})
|
|
|
|
|
|
async def well_known_authorization_server(request: Request):
|
|
return JSONResponse({
|
|
"issuer": OAUTH_BASE_URL,
|
|
"authorization_endpoint": OAUTH_BASE_URL + "/authorize",
|
|
"token_endpoint": OAUTH_BASE_URL + "/token",
|
|
"registration_endpoint": OAUTH_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(request: Request):
|
|
"""Dynamic Client Registration (RFC 7591)."""
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
body = {}
|
|
client_id = "client_" + secrets.token_urlsafe(16)
|
|
state = load_oauth_state()
|
|
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()),
|
|
}
|
|
save_oauth_state(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)
|
|
|
|
|
|
AUTHORIZE_PAGE = """<!DOCTYPE html>
|
|
<html lang="fr"><head><meta charset="utf-8"><title>Autorisation Forgejo MCP</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:#000;color:#fff;border:none;cursor:pointer}}
|
|
.err{{color:#c00}}</style></head>
|
|
<body><h2>Autorisation Forgejo MCP</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>"""
|
|
|
|
|
|
async def authorize_get(request: Request):
|
|
q = request.query_params
|
|
page = AUTHORIZE_PAGE.format(
|
|
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(request: 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 not SECURITY_WORD or word != SECURITY_WORD:
|
|
page = AUTHORIZE_PAGE.format(
|
|
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 = load_oauth_state()
|
|
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,
|
|
}
|
|
save_oauth_state(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)
|
|
|
|
|
|
def verify_pkce(code_verifier, code_challenge):
|
|
"""Verifie 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
|
|
|
|
|
|
async def token(request: Request):
|
|
form = await request.form()
|
|
grant_type = form.get("grant_type", "")
|
|
oauth = load_oauth_state()
|
|
|
|
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
|
|
save_oauth_state(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",
|
|
}
|
|
save_oauth_state(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)
|
|
|
|
|
|
app = Starlette(routes=[
|
|
Route("/mcp", mcp_endpoint, methods=["POST"]),
|
|
Route("/", mcp_endpoint, methods=["POST"]),
|
|
Route("/health", health, methods=["GET"]),
|
|
Route("/.well-known/oauth-protected-resource", well_known_protected_resource, methods=["GET"]),
|
|
Route("/.well-known/oauth-authorization-server", well_known_authorization_server, methods=["GET"]),
|
|
Route("/register", register, methods=["POST"]),
|
|
Route("/authorize", authorize_get, methods=["GET"]),
|
|
Route("/authorize", authorize_post, methods=["POST"]),
|
|
Route("/token", token, methods=["POST"]),
|
|
])
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="127.0.0.1", port=8770)
|