feat(auth): routes /auth/linkedin et callback OIDC (#88)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+39
-1
@@ -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)
|
||||
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user