From 633462f18b019c0dcfddc7c08b61054551acdb45 Mon Sep 17 00:00:00 2001 From: Colin Maudry Date: Mon, 13 Jul 2026 14:19:49 +0200 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20routes=20d=C3=A9couverte=20+=20DCR?= =?UTF-8?q?=20+=20r=C3=A9vocation=20OAuth=20(#114)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mcp/oauth/authorize.py | 6 +++ src/mcp/oauth/routes.py | 50 +++++++++++++++++++++++++ tests/mcp/test_oauth_routes_metadata.py | 50 +++++++++++++++++++++++++ 3 files changed, 106 insertions(+) create mode 100644 src/mcp/oauth/authorize.py create mode 100644 src/mcp/oauth/routes.py create mode 100644 tests/mcp/test_oauth_routes_metadata.py diff --git a/src/mcp/oauth/authorize.py b/src/mcp/oauth/authorize.py new file mode 100644 index 0000000..3f7714e --- /dev/null +++ b/src/mcp/oauth/authorize.py @@ -0,0 +1,6 @@ +def authorize(): + return "", 501 + + +def token(): + return "", 501 diff --git a/src/mcp/oauth/routes.py b/src/mcp/oauth/routes.py new file mode 100644 index 0000000..2d68bd1 --- /dev/null +++ b/src/mcp/oauth/routes.py @@ -0,0 +1,50 @@ +import os + +from flask import Blueprint, jsonify + +from src.mcp.oauth import metadata, server + +oauth_bp = Blueprint("mcp_oauth", __name__) + +_server = None # AuthorizationServer, initialisé par init_oauth + + +def _base() -> str: + return os.getenv("APP_BASE_URL", "").rstrip("/") + + +@oauth_bp.route("/.well-known/oauth-protected-resource") +@oauth_bp.route("/.well-known/oauth-protected-resource/_mcp") +def protected_resource(): + return jsonify(metadata.protected_resource_metadata(_base())) + + +@oauth_bp.route("/.well-known/oauth-authorization-server") +@oauth_bp.route("/.well-known/oauth-authorization-server/_mcp") +@oauth_bp.route("/.well-known/openid-configuration") +def as_metadata(): + return jsonify(metadata.authorization_server_metadata(_base())) + + +@oauth_bp.route("/oauth/register", methods=["POST"]) +def register(): + return _server.create_endpoint_response("client_registration") + + +@oauth_bp.route("/oauth/revoke", methods=["POST"]) +def revoke(): + return _server.create_endpoint_response("revocation") + + +def init_oauth(app) -> None: + global _server + _server = server.create_authorization_server(app) + + from src.mcp.oauth.authorize import authorize as _authorize # Task 9 + from src.mcp.oauth.authorize import token as _token # Task 9 + + app.add_url_rule( + "/oauth/authorize", "oauth_authorize", _authorize, methods=["GET", "POST"] + ) + app.add_url_rule("/oauth/token", "oauth_token", _token, methods=["POST"]) + app.register_blueprint(oauth_bp) diff --git a/tests/mcp/test_oauth_routes_metadata.py b/tests/mcp/test_oauth_routes_metadata.py new file mode 100644 index 0000000..051d041 --- /dev/null +++ b/tests/mcp/test_oauth_routes_metadata.py @@ -0,0 +1,50 @@ +import pytest +from flask import Flask + + +@pytest.fixture +def oauth_app(monkeypatch, tmp_path): + from src.mcp.oauth import routes, store + + monkeypatch.setenv("USERS_DB_PATH", str(tmp_path / "u.sqlite")) + monkeypatch.setenv("APP_BASE_URL", "https://colibre.fr") + store.init_schema(tmp_path / "u.sqlite") + app = Flask(__name__) + app.config["SECRET_KEY"] = "x" + routes.init_oauth(app) + return app + + +def test_protected_resource_wellknown(oauth_app): + resp = oauth_app.test_client().get("/.well-known/oauth-protected-resource") + assert resp.status_code == 200 + assert resp.get_json()["resource"] == "https://colibre.fr/_mcp" + + +def test_protected_resource_wellknown_mcp_suffix(oauth_app): + resp = oauth_app.test_client().get("/.well-known/oauth-protected-resource/_mcp") + assert resp.status_code == 200 + assert resp.get_json()["resource"] == "https://colibre.fr/_mcp" + + +def test_as_metadata_and_openid(oauth_app): + for path in ( + "/.well-known/oauth-authorization-server", + "/.well-known/openid-configuration", + ): + resp = oauth_app.test_client().get(path) + assert resp.status_code == 200 + assert resp.get_json()["code_challenge_methods_supported"] == ["S256"] + + +def test_dynamic_client_registration(oauth_app): + resp = oauth_app.test_client().post( + "/oauth/register", + json={ + "redirect_uris": ["https://claude.ai/api/mcp/auth_callback"], + "client_name": "Claude", + "token_endpoint_auth_method": "none", + }, + ) + assert resp.status_code == 201 + assert "client_id" in resp.get_json()