diff --git a/src/mcp/account.py b/src/mcp/account.py new file mode 100644 index 0000000..d96d127 --- /dev/null +++ b/src/mcp/account.py @@ -0,0 +1,48 @@ +import os + +from flask import Blueprint, redirect, request, session +from flask_login import current_user + +from src.api import tokens_db + +mcp_account_bp = Blueprint("mcp_account", __name__) + +_LABEL_MAX = 100 + + +def _has_active_subscription() -> bool: + # Réutilise la vérification canonique (gère TOUS_ABONNES). + from src.pages._compte_shell import current_user_has_subscription + + return current_user_has_subscription() + + +def _guard(): + """Renvoie une redirection si l'accès est refusé, sinon None.""" + if not current_user.is_authenticated: + return redirect("/connexion?next=/compte/mcp") + if not _has_active_subscription(): + return redirect("/compte/abonnement") + return None + + +@mcp_account_bp.route("/compte/mcp/creer", methods=["POST"]) +def creer(): + denied = _guard() + if denied is not None: + return denied + label = (request.form.get("label") or "").strip()[:_LABEL_MAX] or "Sans nom" + token, _ = tokens_db.create_token( + os.environ["USERS_DB_PATH"], label, user_id=current_user.id, kind="mcp" + ) + session["mcp_new_token"] = token + return redirect("/compte/mcp") + + +@mcp_account_bp.route("/compte/mcp/revoquer/", methods=["POST"]) +def revoquer(token_id): + denied = _guard() + if denied is not None: + return denied + tokens_db.revoke_user_token(os.environ["USERS_DB_PATH"], token_id, current_user.id) + return redirect("/compte/mcp") diff --git a/tests/mcp/test_account_routes.py b/tests/mcp/test_account_routes.py new file mode 100644 index 0000000..84a60c0 --- /dev/null +++ b/tests/mcp/test_account_routes.py @@ -0,0 +1,114 @@ +import pytest +from flask import Flask + + +@pytest.fixture +def account_client(monkeypatch, tmp_path): + from src.api import tokens_db + from src.auth import db as auth_db + from src.auth.setup import init_auth + from src.mcp.account import mcp_account_bp + from src.subscriptions import db as sub_db + + monkeypatch.setenv("USERS_DB_PATH", str(tmp_path / "users.test.sqlite")) + monkeypatch.setenv("SECRET_KEY", "test-secret-key") + monkeypatch.setenv("APP_BASE_URL", "http://localhost:8050") + auth_db.reset_conn_for_tests() + + app = Flask(__name__) + app.config["WTF_CSRF_ENABLED"] = False + init_auth(app) # login manager + CSRF (désactivé) + schéma auth + sub_db.init_schema() + db_path = tmp_path / "users.test.sqlite" + tokens_db.init_schema(db_path) + app.register_blueprint(mcp_account_bp) + + yield app, db_path + auth_db.reset_conn_for_tests() + + +def _login(app, uid): + client = app.test_client() + with client.session_transaction() as sess: + sess["_user_id"] = str(uid) + sess["_fresh"] = True + return client + + +def _subscribed_uid(): + from src.auth import db as auth_db + from src.subscriptions import db as sub_db + + uid = auth_db.create_user("sub@ex.fr", "hash") + _, sub_id = sub_db.create_pending(uid, "colibre-1", "simple") + sub_db.set_status(sub_id, "active") + return uid + + +def test_creer_generates_mcp_token_for_subscriber(account_client): + from src.api import tokens_db + + app, db_path = account_client + uid = _subscribed_uid() + client = _login(app, uid) + + resp = client.post("/compte/mcp/creer", data={"label": "Claude portable"}) + assert resp.status_code == 302 + assert resp.headers["Location"].endswith("/compte/mcp") + + with client.session_transaction() as sess: + assert sess.get("mcp_new_token", "").startswith("colibre_") + + rows = tokens_db.list_user_tokens(db_path, uid, "mcp") + assert [r["label"] for r in rows] == ["Claude portable"] + assert rows[0]["kind"] == "mcp" + assert rows[0]["user_id"] == uid + + +def test_revoquer_own_token(account_client): + from src.api import tokens_db + + app, db_path = account_client + uid = _subscribed_uid() + client = _login(app, uid) + token, tid = tokens_db.create_token(db_path, "x", user_id=uid, kind="mcp") + + resp = client.post(f"/compte/mcp/revoquer/{tid}") + assert resp.status_code == 302 + assert tokens_db.get_token_by_plaintext(db_path, token)["revoked_at"] is not None + + +def test_revoquer_other_users_token_is_noop(account_client): + from src.api import tokens_db + + app, db_path = account_client + uid = _subscribed_uid() + client = _login(app, uid) + other_token, other_id = tokens_db.create_token( + db_path, "y", user_id=99999, kind="mcp" + ) + + resp = client.post(f"/compte/mcp/revoquer/{other_id}") + assert resp.status_code == 302 + assert tokens_db.get_token_by_plaintext(db_path, other_token)["revoked_at"] is None + + +def test_creer_blocked_without_subscription(account_client): + from src.api import tokens_db + from src.auth import db as auth_db + + app, db_path = account_client + uid = auth_db.create_user("nosub@ex.fr", "hash") # pas d'abonnement + client = _login(app, uid) + + resp = client.post("/compte/mcp/creer", data={"label": "x"}) + assert resp.status_code == 302 + assert resp.headers["Location"].endswith("/compte/abonnement") + assert tokens_db.list_user_tokens(db_path, uid, "mcp") == [] + + +def test_creer_blocked_when_anonymous(account_client): + app, db_path = account_client + resp = app.test_client().post("/compte/mcp/creer", data={"label": "x"}) + assert resp.status_code == 302 + assert "/connexion" in resp.headers["Location"]