code-versioning/mcp_forgejo_server.py

#!/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)