feat(mcp): routes découverte + DCR + révocation OAuth (#114)
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
def authorize():
|
||||
return "", 501
|
||||
|
||||
|
||||
def token():
|
||||
return "", 501
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user