From 697711f6761c5787c603ece7e051c09bb407510f Mon Sep 17 00:00:00 2001 From: Colin Maudry Date: Fri, 10 Jul 2026 14:46:50 +0200 Subject: [PATCH] feat(mcp): garde d'abonnement sur /_mcp (scope B #111) --- src/mcp/auth.py | 60 ++++++++++++++++++++ tests/mcp/test_auth.py | 122 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 182 insertions(+) create mode 100644 src/mcp/auth.py create mode 100644 tests/mcp/test_auth.py diff --git a/src/mcp/auth.py b/src/mcp/auth.py new file mode 100644 index 0000000..c9f8a0b --- /dev/null +++ b/src/mcp/auth.py @@ -0,0 +1,60 @@ +import os + +from flask import Flask, jsonify, request + +from src.api import tokens_db +from src.subscriptions.db import has_active_subscription +from src.utils import TOUS_ABONNES + + +def _unauthorized(): + resp = jsonify( + {"error": "unauthorized", "message": "Jeton MCP absent ou invalide."} + ) + resp.status_code = 401 + resp.headers["WWW-Authenticate"] = 'Bearer realm="colibre-mcp"' + return resp + + +def _forbidden(): + resp = jsonify( + { + "error": "no_active_subscription", + "message": "Un abonnement colibre actif est requis pour le connecteur MCP.", + } + ) + resp.status_code = 403 + return resp + + +def _authenticate_mcp(): + header = request.headers.get("Authorization", "") + if not header.startswith("Bearer "): + return _unauthorized() + token = header[len("Bearer ") :].strip() + if not token: + return _unauthorized() + + db_path = os.environ["USERS_DB_PATH"] + row = tokens_db.get_token_by_plaintext(db_path, token) + if row is None or row["revoked_at"] is not None or row["kind"] != "mcp": + return _unauthorized() + + user_id = row["user_id"] + if user_id is None: + return _forbidden() + if not (TOUS_ABONNES or has_active_subscription(user_id)): + return _forbidden() + + tokens_db.increment_usage(db_path, row["id"]) + return None + + +def init_mcp_auth(server: Flask) -> None: + """Enregistre le garde d'authentification du serveur MCP (/_mcp).""" + + @server.before_request + def _guard_mcp(): + if request.path == "/_mcp" or request.path.startswith("/_mcp/"): + return _authenticate_mcp() + return None diff --git a/tests/mcp/test_auth.py b/tests/mcp/test_auth.py new file mode 100644 index 0000000..9e3b406 --- /dev/null +++ b/tests/mcp/test_auth.py @@ -0,0 +1,122 @@ +import pytest +from flask import Flask + + +@pytest.fixture +def mcp_app(monkeypatch, tmp_path): + from src.api import tokens_db + from src.auth import db as auth_db + from src.mcp.auth import init_mcp_auth + from src.subscriptions import db as sub_db + + db_path = tmp_path / "users.test.sqlite" + monkeypatch.setenv("USERS_DB_PATH", str(db_path)) + auth_db.reset_conn_for_tests() + auth_db.init_schema() + sub_db.init_schema() + tokens_db.init_schema(db_path) + + app = Flask(__name__) + + @app.route("/_mcp", methods=["GET", "POST"]) + def _mcp(): + return "ok", 200 + + init_mcp_auth(app) + yield app, db_path + auth_db.reset_conn_for_tests() + + +def _subscribed_uid(): + from src.auth import db as auth_db + from src.subscriptions import db as sub_db + + uid = auth_db.create_user("mcp@ex.fr", "hash") + _, sub_id = sub_db.create_pending(uid, "colibre-1", "simple") + sub_db.set_status(sub_id, "active") + return uid + + +def test_missing_header_401_with_challenge(mcp_app): + app, _ = mcp_app + resp = app.test_client().post("/_mcp") + assert resp.status_code == 401 + assert resp.headers["WWW-Authenticate"] == 'Bearer realm="colibre-mcp"' + + +def test_empty_bearer_401(mcp_app): + app, _ = mcp_app + resp = app.test_client().post("/_mcp", headers={"Authorization": "Bearer "}) + assert resp.status_code == 401 + + +def test_unknown_token_401(mcp_app): + app, _ = mcp_app + resp = app.test_client().post( + "/_mcp", headers={"Authorization": "Bearer colibre_unknown"} + ) + assert resp.status_code == 401 + + +def test_revoked_mcp_token_401(mcp_app): + from src.api import tokens_db + + app, db_path = mcp_app + uid = _subscribed_uid() + token, tid = tokens_db.create_token(db_path, "x", user_id=uid, kind="mcp") + tokens_db.revoke_user_token(db_path, tid, uid) + resp = app.test_client().post("/_mcp", headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 401 + + +def test_api_kind_token_rejected_401(mcp_app): + from src.api import tokens_db + + app, db_path = mcp_app + uid = _subscribed_uid() + token, _ = tokens_db.create_token(db_path, "x", user_id=uid, kind="api") + resp = app.test_client().post("/_mcp", headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 401 + + +def test_mcp_token_without_user_403(mcp_app): + from src.api import tokens_db + + app, db_path = mcp_app + token, _ = tokens_db.create_token(db_path, "x", user_id=None, kind="mcp") + resp = app.test_client().post("/_mcp", headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 403 + + +def test_mcp_token_no_active_subscription_403(mcp_app): + from src.api import tokens_db + from src.auth import db as auth_db + + app, db_path = mcp_app + uid = auth_db.create_user("nosub@ex.fr", "hash") # aucun abonnement + token, _ = tokens_db.create_token(db_path, "x", user_id=uid, kind="mcp") + resp = app.test_client().post("/_mcp", headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 403 + + +def test_mcp_token_active_subscription_passes_and_increments(mcp_app): + from src.api import tokens_db + + app, db_path = mcp_app + uid = _subscribed_uid() + token, _ = tokens_db.create_token(db_path, "x", user_id=uid, kind="mcp") + resp = app.test_client().post("/_mcp", headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 200 + assert tokens_db.get_token_by_plaintext(db_path, token)["count_total"] == 1 + + +def test_tous_abonnes_bypasses_subscription(mcp_app, monkeypatch): + from src.api import tokens_db + from src.auth import db as auth_db + + monkeypatch.setattr("src.mcp.auth.TOUS_ABONNES", True) + app, db_path = mcp_app + uid = auth_db.create_user("nosub2@ex.fr", "hash") + token, _ = tokens_db.create_token(db_path, "x", user_id=uid, kind="mcp") + resp = app.test_client().post("/_mcp", headers={"Authorization": f"Bearer {token}"}) + assert resp.status_code == 200