diff --git a/src/auth/routes.py b/src/auth/routes.py index 6c23744..6bc5d5c 100644 --- a/src/auth/routes.py +++ b/src/auth/routes.py @@ -1,10 +1,13 @@ +import os + from email_validator import EmailNotValidError, validate_email -from flask import Blueprint, redirect, request +from flask import Blueprint, redirect, request, session from flask_login import current_user, login_required, login_user, logout_user from werkzeug.security import check_password_hash, generate_password_hash from src.auth import db, mailer, tokens from src.auth.models import User +from src.auth.oauth import oauth from src.auth.setup import safe_next from src.utils import logger @@ -237,3 +240,38 @@ def delete_account(): db.delete_user(user_id) logout_user() return redirect("/?account_deleted=1") + + +@auth_bp.route("/linkedin", methods=["GET"]) +def linkedin_login(): + session["oauth_next"] = safe_next( + request.args.get("next"), fallback="/compte/admin" + ) + redirect_uri = f"{os.getenv('APP_BASE_URL', '')}/auth/linkedin/callback" + return oauth.linkedin.authorize_redirect(redirect_uri) + + +@auth_bp.route("/linkedin/callback", methods=["GET"]) +def linkedin_callback(): + next_url = safe_next(session.pop("oauth_next", None), fallback="/compte/admin") + if request.args.get("error"): + # L'utilisateur a refusé / annulé l'autorisation côté LinkedIn. + return _redirect_with_error("/connexion", "oauth_cancelled") + try: + token = oauth.linkedin.authorize_access_token() + except Exception: + logger.exception("Échec de l'échange de token LinkedIn") + return _redirect_with_error("/connexion", "oauth_failed") + + userinfo = token.get("userinfo") or {} + subject = userinfo.get("sub") + email = (userinfo.get("email") or "").strip().lower() + if not subject or not email: + logger.error("Réponse LinkedIn sans sub/email : %s", userinfo) + return _redirect_with_error("/connexion", "oauth_failed") + + user = resolve_oauth_user( + "linkedin", subject, email, bool(userinfo.get("email_verified")) + ) + login_user(user, remember=True) + return redirect(next_url) diff --git a/tests/auth/test_oauth_routes.py b/tests/auth/test_oauth_routes.py new file mode 100644 index 0000000..7437051 --- /dev/null +++ b/tests/auth/test_oauth_routes.py @@ -0,0 +1,74 @@ +import pytest + +from src.auth import db +from src.auth import oauth as oauth_module + + +@pytest.fixture +def fake_userinfo(monkeypatch): + """Patche authorize_access_token pour éviter tout appel réseau.""" + + state = {"userinfo": None, "raise": False} + + def _authorize_access_token(): + if state["raise"]: + raise RuntimeError("token exchange failed") + return {"userinfo": state["userinfo"]} + + monkeypatch.setattr( + oauth_module.oauth.linkedin, + "authorize_access_token", + _authorize_access_token, + raising=False, + ) + return state + + +def test_login_route_redirects_to_linkedin(client, monkeypatch): + from flask import redirect + + monkeypatch.setattr( + oauth_module.oauth.linkedin, + "authorize_redirect", + lambda redirect_uri, **kw: redirect("https://www.linkedin.com/oauth/authorize"), + raising=False, + ) + resp = client.get("/auth/linkedin") + assert resp.status_code == 302 + assert "linkedin.com" in resp.headers["Location"] + + +def test_callback_creates_user_and_logs_in(client, fake_userinfo, users_db_path): + db.init_schema() + fake_userinfo["userinfo"] = { + "sub": "sub-xyz", + "email": "newbie@example.com", + "email_verified": True, + } + resp = client.get("/auth/linkedin/callback") + assert resp.status_code == 302 + assert resp.headers["Location"].endswith("/compte/admin") + assert db.get_user_by_email("newbie@example.com") is not None + + +def test_callback_without_email_fails(client, fake_userinfo, users_db_path): + db.init_schema() + fake_userinfo["userinfo"] = {"sub": "sub-noemail", "email_verified": True} + resp = client.get("/auth/linkedin/callback") + assert resp.status_code == 302 + assert "error=oauth_failed" in resp.headers["Location"] + + +def test_callback_token_error_redirects(client, fake_userinfo, users_db_path): + db.init_schema() + fake_userinfo["raise"] = True + resp = client.get("/auth/linkedin/callback") + assert resp.status_code == 302 + assert "error=oauth_failed" in resp.headers["Location"] + + +def test_callback_user_cancelled_redirects(client, users_db_path): + db.init_schema() + resp = client.get("/auth/linkedin/callback?error=user_cancelled_login") + assert resp.status_code == 302 + assert "error=oauth_cancelled" in resp.headers["Location"]