mirror of
https://github.com/suitenumerique/messages.git
synced 2026-08-17 21:25:41 +02:00
✨(backend) add mobile OIDC session handoff
Capacitor apps must run the OIDC flow in the system browser (the IdP cookie has to live there to provide cross-app SSO), but the browser's cookies never reach the app's native HTTP layer, so the Django session created by the callback would be stranded. The callback now redirects to an allowlisted app deep link with a one-time token that the app exchanges for its session cookie and CSRF token. The token is bound to the initiating app instance with a PKCE S256 verifier, single-use, short-lived (MOBILE_AUTH_TOKEN_TTL) and the anonymous exchange endpoint is throttled per IP to cap brute-force guessing. An empty MOBILE_AUTH_CALLBACK_SCHEMES (the default) keeps the whole handoff disabled.
This commit is contained in:
+10
@@ -236,6 +236,16 @@ _Those settings are deprecated and will be removed in the future._
|
||||
| `LOGOUT_REDIRECT_URL` | `http://localhost:8900` | Post-logout redirect URL | Optional |
|
||||
| `ALLOW_LOGOUT_GET_METHOD` | `True` | Allow GET method for logout | Optional |
|
||||
|
||||
### Mobile App Authentication (Capacitor)
|
||||
|
||||
| Variable | Default | Description | Required |
|
||||
|----------|---------|-------------|----------|
|
||||
| `MOBILE_AUTH_CALLBACK_SCHEMES` | `[]` | JSON list of deep-link schemes the OIDC callback may redirect to after a mobile-initiated login (e.g. `["stmessages"]`). An empty list disables the mobile session handoff. | Optional |
|
||||
| `MOBILE_AUTH_TOKEN_TTL` | `60` | Lifetime (seconds) of the one-time token a mobile app exchanges for its session cookie on `/api/v1.0/mobile/auth/exchange/` | Optional |
|
||||
| `API_MOBILE_AUTH_EXCHANGE_THROTTLE_RATE` | `10/minute` | Per-IP rate limit on the anonymous `/api/v1.0/mobile/auth/exchange/` endpoint. A legitimate login exchanges once; the cap only slows down brute-force guessing of the one-time token | Optional |
|
||||
|
||||
> **Note**: mobile builds of the frontend must set `NEXT_PUBLIC_API_ORIGIN` explicitly — inside the Capacitor WebView there is no meaningful `window.location.origin` fallback.
|
||||
|
||||
## Security & CORS
|
||||
|
||||
| Variable | Default | Description | Required |
|
||||
|
||||
@@ -74,6 +74,9 @@ LOGOUT_REDIRECT_URL=http://localhost:8900
|
||||
OIDC_REDIRECT_ALLOWED_HOSTS=["http://localhost:8902", "http://localhost:8900"]
|
||||
OIDC_AUTH_REQUEST_EXTRA_PARAMS={"acr_values": "eidas1"}
|
||||
|
||||
# Mobile apps (Capacitor) session handoff
|
||||
MOBILE_AUTH_CALLBACK_SCHEMES=["stmessages"]
|
||||
|
||||
# keycloak
|
||||
IDENTITY_PROVIDER=keycloak
|
||||
KEYCLOAK_REALM=messages
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
"""API views managing the mobile (Capacitor) Django session lifecycle.
|
||||
|
||||
Counterpart of `core.authentication.views`: the OIDC callback hands a one-time
|
||||
token to the mobile app via deep link, and the app calls the exchange endpoint
|
||||
from its native HTTP layer to obtain the session cookie used by all subsequent
|
||||
API calls. The logout endpoint ends that session without touching the IdP one.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import secrets
|
||||
from importlib import import_module
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth import SESSION_KEY
|
||||
from django.contrib.auth import logout as auth_logout
|
||||
from django.core.cache import cache
|
||||
from django.middleware.csrf import get_token
|
||||
|
||||
from drf_spectacular.utils import extend_schema
|
||||
from rest_framework import status
|
||||
from rest_framework.permissions import AllowAny
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.throttling import ScopedRateThrottle
|
||||
from rest_framework.views import APIView
|
||||
|
||||
from core.authentication.views import MOBILE_AUTH_TOKEN_CACHE_PREFIX
|
||||
|
||||
|
||||
def _s256(code_verifier):
|
||||
"""Compute the RFC 7636 S256 transform of a PKCE code verifier."""
|
||||
digest = hashlib.sha256(code_verifier.encode("utf-8")).digest()
|
||||
return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
|
||||
|
||||
|
||||
class MobileSessionExchangeView(APIView):
|
||||
"""Exchange the one-time token minted by the OIDC callback for a session cookie.
|
||||
|
||||
Anonymous by design: the request is authenticated by the single-use token
|
||||
and the PKCE verifier generated by the app before opening the system browser.
|
||||
Throttled per IP — a legitimate login exchanges exactly once, so the cap
|
||||
only slows down brute-force guessing of the token.
|
||||
"""
|
||||
|
||||
authentication_classes = []
|
||||
permission_classes = [AllowAny]
|
||||
throttle_classes = [ScopedRateThrottle]
|
||||
throttle_scope = "mobile_auth_exchange"
|
||||
|
||||
# Excluded from the OpenAPI schema: this endpoint is part of the mobile
|
||||
# auth bootstrap (like /authenticate/), not of the Orval-generated client.
|
||||
@extend_schema(exclude=True)
|
||||
def post(self, request):
|
||||
"""Verify the token and PKCE verifier, then attach the session to the response."""
|
||||
# JSON bodies may be a non-object or carry non-string values, which
|
||||
# would crash .get() or _s256() with a 500 instead of a controlled
|
||||
# rejection.
|
||||
data = request.data if isinstance(request.data, dict) else {}
|
||||
token = data.get("token", "")
|
||||
code_verifier = data.get("code_verifier", "")
|
||||
if (
|
||||
not isinstance(token, str)
|
||||
or not isinstance(code_verifier, str)
|
||||
or not token
|
||||
or not code_verifier
|
||||
):
|
||||
return Response(
|
||||
{"detail": "Missing token or code_verifier."},
|
||||
status=status.HTTP_400_BAD_REQUEST,
|
||||
)
|
||||
|
||||
cache_key = f"{MOBILE_AUTH_TOKEN_CACHE_PREFIX}:{token}"
|
||||
payload = cache.get(cache_key)
|
||||
# Single use: consume the token before verifying it so a failed attempt
|
||||
# cannot be retried with another verifier. get() then delete() is not
|
||||
# atomic, so two concurrent exchanges may both read the payload — but
|
||||
# the backend deletes the key exactly once, so only the request whose
|
||||
# delete() returns True is allowed to proceed.
|
||||
consumed = cache.delete(cache_key)
|
||||
if (
|
||||
payload is None
|
||||
or not consumed
|
||||
or not secrets.compare_digest(
|
||||
_s256(code_verifier), payload["code_challenge"]
|
||||
)
|
||||
):
|
||||
return Response(
|
||||
{"detail": "Invalid or expired token."},
|
||||
status=status.HTTP_403_FORBIDDEN,
|
||||
)
|
||||
|
||||
engine = import_module(settings.SESSION_ENGINE)
|
||||
session = engine.SessionStore(payload["session_key"])
|
||||
if session.get(SESSION_KEY) is None:
|
||||
return Response(
|
||||
{"detail": "Session expired."},
|
||||
status=status.HTTP_403_FORBIDDEN,
|
||||
)
|
||||
|
||||
# Attach the session to the underlying HttpRequest (not the DRF proxy)
|
||||
# so SessionMiddleware emits the Set-Cookie header on this response.
|
||||
http_request = request._request # noqa: SLF001 pylint: disable=protected-access
|
||||
http_request.session = session
|
||||
http_request.session.modified = True
|
||||
# With CSRF_USE_SESSIONS this stores the CSRF secret in the session and
|
||||
# returns the token the app echoes in the X-CSRFToken header; the native
|
||||
# HTTP jar replays the session cookie, so the check validates server-side
|
||||
# without depending on a csrf cookie the jar would not resend.
|
||||
csrf_token = get_token(http_request)
|
||||
|
||||
return Response({"csrf_token": csrf_token})
|
||||
|
||||
|
||||
class MobileLogoutView(APIView):
|
||||
"""Invalidate the Django session without the RP-initiated IdP logout.
|
||||
|
||||
The web logout (`/logout/`) redirects to the identity provider with an
|
||||
`id_token_hint`, which terminates the IdP session and therefore cross-app
|
||||
SSO. Mobile logout must only flush the server-side Django session: the app
|
||||
clears its local cookie jar, and the IdP session stays alive for the other
|
||||
apps of the suite.
|
||||
|
||||
Anonymous requests are a no-op (204): the app calls this best-effort, and
|
||||
logging out an already expired session must not fail.
|
||||
"""
|
||||
|
||||
# Excluded from the OpenAPI schema: this endpoint is part of the mobile
|
||||
# auth bootstrap (like /authenticate/), not of the Orval-generated client.
|
||||
@extend_schema(exclude=True)
|
||||
def post(self, request):
|
||||
"""Flush the server-side session bound to the request cookie."""
|
||||
auth_logout(request)
|
||||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||||
@@ -0,0 +1,115 @@
|
||||
"""Mobile-aware OIDC views handing the Django session over to the Capacitor apps.
|
||||
|
||||
The mobile apps run the OIDC flow in the system browser (ASWebAuthenticationSession
|
||||
on iOS, Chrome Custom Tabs on Android) so the identity provider session cookie is
|
||||
shared across apps and provides cross-app SSO. The backend remains the confidential
|
||||
OIDC client: once the callback has created the Django session, it redirects the
|
||||
system browser to the app deep link with a one-time token that the app exchanges
|
||||
for the session cookie (see `core.api.viewsets.mobile_auth`).
|
||||
"""
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
import time
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.cache import cache
|
||||
from django.core.exceptions import SuspiciousOperation
|
||||
from django.core.handlers.wsgi import WSGIRequest
|
||||
from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseRedirect
|
||||
|
||||
from lasuite.oidc_login.views import (
|
||||
OIDCAuthenticationCallbackView as LaSuiteOIDCAuthenticationCallbackView,
|
||||
)
|
||||
from lasuite.oidc_login.views import (
|
||||
OIDCAuthenticationRequestView as LaSuiteOIDCAuthenticationRequestView,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MOBILE_AUTH_SESSION_KEY = "mobile_auth"
|
||||
MOBILE_AUTH_TOKEN_CACHE_PREFIX = "mobile-auth-token" # noqa: S105 (cache key prefix, not a secret)
|
||||
# A pending mobile login older than this is ignored by the callback, so an
|
||||
# abandoned mobile attempt cannot turn a later web login performed in the same
|
||||
# browser session into a deep-link redirect.
|
||||
MOBILE_AUTH_FLAG_MAX_AGE = 60 * 10
|
||||
|
||||
|
||||
class AppSchemeRedirect(HttpResponseRedirect):
|
||||
"""Redirect response whose allowed schemes are the configured mobile app schemes."""
|
||||
|
||||
def __init__(self, redirect_to: str, *args, **kwargs) -> None:
|
||||
"""Allow the configured mobile schemes before validating the redirect URL."""
|
||||
self.allowed_schemes = list(settings.MOBILE_AUTH_CALLBACK_SCHEMES)
|
||||
super().__init__(redirect_to, *args, **kwargs)
|
||||
|
||||
|
||||
class OIDCAuthenticationRequestView(LaSuiteOIDCAuthenticationRequestView):
|
||||
"""Authentication request view supporting a mobile session handoff.
|
||||
|
||||
Mobile apps add `mobile_scheme` (an allowlisted deep-link scheme) and
|
||||
`code_challenge` (PKCE S256) to the authenticate URL. The pair is stored in
|
||||
the session so the callback view knows where to hand the session over.
|
||||
"""
|
||||
|
||||
def get(self, request: WSGIRequest) -> HttpResponse:
|
||||
"""Flag the session when the login is initiated by a mobile app."""
|
||||
mobile_scheme = request.GET.get("mobile_scheme", "")
|
||||
if mobile_scheme:
|
||||
if mobile_scheme not in settings.MOBILE_AUTH_CALLBACK_SCHEMES:
|
||||
raise SuspiciousOperation("Unknown mobile callback scheme.")
|
||||
code_challenge = request.GET.get("code_challenge", "")
|
||||
if not code_challenge:
|
||||
return HttpResponseBadRequest("Missing code_challenge.")
|
||||
request.session[MOBILE_AUTH_SESSION_KEY] = {
|
||||
"scheme": mobile_scheme,
|
||||
"code_challenge": code_challenge,
|
||||
"created_at": time.time(),
|
||||
}
|
||||
# The parent view forces a session save before redirecting to the
|
||||
# identity provider, so the flag survives the OIDC round-trip.
|
||||
return super().get(request)
|
||||
|
||||
|
||||
class OIDCAuthenticationCallbackView(LaSuiteOIDCAuthenticationCallbackView):
|
||||
"""Callback view handing the session over to the mobile app via deep link."""
|
||||
|
||||
def _pop_mobile_auth(self) -> dict | None:
|
||||
"""Pop and return the pending mobile login data, if any and still fresh."""
|
||||
mobile_auth = self.request.session.pop(MOBILE_AUTH_SESSION_KEY, None)
|
||||
if mobile_auth is None:
|
||||
return None
|
||||
self.request.session.save()
|
||||
if time.time() - mobile_auth["created_at"] > MOBILE_AUTH_FLAG_MAX_AGE:
|
||||
logger.warning("Ignoring stale mobile login flag.")
|
||||
return None
|
||||
return mobile_auth
|
||||
|
||||
def login_success(self) -> HttpResponse:
|
||||
"""Redirect to the mobile app with a one-time token after a mobile login."""
|
||||
response = super().login_success()
|
||||
mobile_auth = self._pop_mobile_auth()
|
||||
if mobile_auth is None:
|
||||
return response
|
||||
|
||||
token = secrets.token_urlsafe(32)
|
||||
cache.set(
|
||||
f"{MOBILE_AUTH_TOKEN_CACHE_PREFIX}:{token}",
|
||||
{
|
||||
# auth.login() cycled the session key, hence read after super().
|
||||
"session_key": self.request.session.session_key,
|
||||
"code_challenge": mobile_auth["code_challenge"],
|
||||
},
|
||||
timeout=settings.MOBILE_AUTH_TOKEN_TTL,
|
||||
)
|
||||
query = urlencode({"token": token})
|
||||
return AppSchemeRedirect(f"{mobile_auth['scheme']}://auth?{query}")
|
||||
|
||||
def login_failure(self) -> HttpResponse:
|
||||
"""Redirect to the mobile app with an error after a failed mobile login."""
|
||||
response = super().login_failure()
|
||||
mobile_auth = self._pop_mobile_auth()
|
||||
if mobile_auth is None:
|
||||
return response
|
||||
return AppSchemeRedirect(f"{mobile_auth['scheme']}://auth?error=login_failed")
|
||||
@@ -0,0 +1,389 @@
|
||||
"""Tests for the mobile (Capacitor) OIDC session handoff."""
|
||||
|
||||
import time
|
||||
from importlib import import_module
|
||||
from unittest.mock import patch
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth import BACKEND_SESSION_KEY, HASH_SESSION_KEY, SESSION_KEY
|
||||
from django.contrib.sessions.middleware import SessionMiddleware
|
||||
from django.core.cache import cache
|
||||
from django.test import RequestFactory
|
||||
from django.test.utils import override_settings
|
||||
from django.urls import reverse
|
||||
|
||||
import pytest
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
from rest_framework.throttling import ScopedRateThrottle
|
||||
|
||||
from core import factories
|
||||
from core.api.viewsets.mobile_auth import _s256
|
||||
from core.authentication.views import (
|
||||
MOBILE_AUTH_SESSION_KEY,
|
||||
MOBILE_AUTH_TOKEN_CACHE_PREFIX,
|
||||
OIDCAuthenticationCallbackView,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated_cache():
|
||||
"""Reset the shared locmem cache before each test.
|
||||
|
||||
The exchange endpoint throttles per IP and every test client posts from
|
||||
127.0.0.1, so a shared counter would leak requests across tests and make
|
||||
outcomes order-dependent.
|
||||
"""
|
||||
cache.clear()
|
||||
|
||||
|
||||
AUTHENTICATE_SETTINGS = {
|
||||
"MOBILE_AUTH_CALLBACK_SCHEMES": ["stmessagesa", "stmessagesb"],
|
||||
"OIDC_OP_AUTHORIZATION_ENDPOINT": "https://oidc.test/authorize",
|
||||
}
|
||||
|
||||
|
||||
def _make_authenticated_session(user):
|
||||
"""Create a server-side session authenticated as the given user."""
|
||||
engine = import_module(settings.SESSION_ENGINE)
|
||||
session = engine.SessionStore()
|
||||
session[SESSION_KEY] = str(user.pk)
|
||||
session[BACKEND_SESSION_KEY] = (
|
||||
"core.authentication.backends.OIDCAuthenticationBackend"
|
||||
)
|
||||
session[HASH_SESSION_KEY] = user.get_session_auth_hash()
|
||||
session.save()
|
||||
return session
|
||||
|
||||
|
||||
class TestMobileAuthenticationRequest:
|
||||
"""Tests for the mobile parameters of the authenticate view."""
|
||||
|
||||
@override_settings(**AUTHENTICATE_SETTINGS)
|
||||
def test_unknown_scheme_is_rejected(self):
|
||||
"""A scheme not in the allowlist must be rejected as suspicious."""
|
||||
response = APIClient().get(
|
||||
reverse("oidc_authentication_init"),
|
||||
{"mobile_scheme": "evilapp", "code_challenge": "challenge"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
@override_settings(**AUTHENTICATE_SETTINGS)
|
||||
def test_missing_code_challenge_is_rejected(self):
|
||||
"""A mobile login without PKCE challenge must be rejected."""
|
||||
response = APIClient().get(
|
||||
reverse("oidc_authentication_init"), {"mobile_scheme": "stmessagesa"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
@override_settings(**AUTHENTICATE_SETTINGS)
|
||||
def test_mobile_login_flags_the_session(self):
|
||||
"""A valid mobile login redirects to the IdP and flags the session."""
|
||||
client = APIClient()
|
||||
response = client.get(
|
||||
reverse("oidc_authentication_init"),
|
||||
{"mobile_scheme": "stmessagesa", "code_challenge": "challenge"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert response["Location"].startswith("https://oidc.test/authorize")
|
||||
mobile_auth = client.session[MOBILE_AUTH_SESSION_KEY]
|
||||
assert mobile_auth["scheme"] == "stmessagesa"
|
||||
assert mobile_auth["code_challenge"] == "challenge"
|
||||
|
||||
@override_settings(**AUTHENTICATE_SETTINGS)
|
||||
def test_web_login_does_not_flag_the_session(self):
|
||||
"""The web flow must not be affected by the mobile handoff support."""
|
||||
client = APIClient()
|
||||
response = client.get(reverse("oidc_authentication_init"))
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert MOBILE_AUTH_SESSION_KEY not in client.session
|
||||
|
||||
|
||||
# `override_settings` only works as a class decorator on Django TestCase
|
||||
# subclasses, so plain pytest classes apply it per method.
|
||||
CALLBACK_SETTINGS = {
|
||||
"MOBILE_AUTH_CALLBACK_SCHEMES": ["stmessagesa", "stmessagesb"],
|
||||
"LOGIN_REDIRECT_URL": "/",
|
||||
"LOGIN_REDIRECT_URL_FAILURE": "/auth-failure",
|
||||
}
|
||||
|
||||
|
||||
class TestMobileAuthenticationCallback:
|
||||
"""Tests for the mobile handoff performed by the callback view."""
|
||||
|
||||
def _build_view(self, session_data=None):
|
||||
"""Return a callback view bound to a request with a real session."""
|
||||
request = RequestFactory().get("/api/v1.0/callback/")
|
||||
SessionMiddleware(lambda _request: None).process_request(request)
|
||||
for key, value in (session_data or {}).items():
|
||||
request.session[key] = value
|
||||
request.session.save()
|
||||
|
||||
user = factories.UserFactory()
|
||||
user.backend = "core.authentication.backends.OIDCAuthenticationBackend"
|
||||
|
||||
view = OIDCAuthenticationCallbackView()
|
||||
view.request = request
|
||||
view.user = user
|
||||
return view
|
||||
|
||||
@override_settings(**CALLBACK_SETTINGS)
|
||||
def test_mobile_login_success_redirects_to_the_app(self):
|
||||
"""A mobile login ends with a deep link carrying a one-time token."""
|
||||
view = self._build_view(
|
||||
{
|
||||
MOBILE_AUTH_SESSION_KEY: {
|
||||
"scheme": "stmessagesa",
|
||||
"code_challenge": "challenge",
|
||||
"created_at": time.time(),
|
||||
}
|
||||
}
|
||||
)
|
||||
response = view.login_success()
|
||||
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
location = urlparse(response["Location"])
|
||||
assert location.scheme == "stmessagesa"
|
||||
assert location.netloc == "auth"
|
||||
|
||||
token = parse_qs(location.query)["token"][0]
|
||||
payload = cache.get(f"{MOBILE_AUTH_TOKEN_CACHE_PREFIX}:{token}")
|
||||
assert payload["code_challenge"] == "challenge"
|
||||
# The session key is cycled by auth.login, the cache entry must hold
|
||||
# the post-login key so the exchanged cookie authenticates requests.
|
||||
assert payload["session_key"] == view.request.session.session_key
|
||||
assert MOBILE_AUTH_SESSION_KEY not in view.request.session
|
||||
|
||||
@override_settings(**CALLBACK_SETTINGS)
|
||||
def test_web_login_success_is_unchanged(self):
|
||||
"""Without the mobile flag, the callback keeps its web behavior."""
|
||||
view = self._build_view()
|
||||
response = view.login_success()
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert response["Location"] == "/"
|
||||
|
||||
@override_settings(**CALLBACK_SETTINGS)
|
||||
def test_stale_mobile_flag_is_ignored(self):
|
||||
"""An abandoned mobile attempt must not hijack a later web login."""
|
||||
view = self._build_view(
|
||||
{
|
||||
MOBILE_AUTH_SESSION_KEY: {
|
||||
"scheme": "stmessagesa",
|
||||
"code_challenge": "challenge",
|
||||
"created_at": time.time() - 3600,
|
||||
}
|
||||
}
|
||||
)
|
||||
response = view.login_success()
|
||||
assert response["Location"] == "/"
|
||||
assert MOBILE_AUTH_SESSION_KEY not in view.request.session
|
||||
|
||||
@override_settings(**CALLBACK_SETTINGS)
|
||||
def test_mobile_login_failure_redirects_to_the_app(self):
|
||||
"""A failed mobile login notifies the app through the deep link."""
|
||||
view = self._build_view(
|
||||
{
|
||||
MOBILE_AUTH_SESSION_KEY: {
|
||||
"scheme": "stmessagesa",
|
||||
"code_challenge": "challenge",
|
||||
"created_at": time.time(),
|
||||
}
|
||||
}
|
||||
)
|
||||
response = view.login_failure()
|
||||
assert response.status_code == status.HTTP_302_FOUND
|
||||
assert response["Location"] == "stmessagesa://auth?error=login_failed"
|
||||
|
||||
@override_settings(**CALLBACK_SETTINGS)
|
||||
def test_web_login_failure_is_unchanged(self):
|
||||
"""Without the mobile flag, a failed login keeps its web behavior."""
|
||||
view = self._build_view()
|
||||
response = view.login_failure()
|
||||
assert response["Location"] == "/auth-failure"
|
||||
|
||||
|
||||
class TestMobileSessionExchange:
|
||||
"""Tests for the one-time token → session cookie exchange endpoint."""
|
||||
|
||||
VERIFIER = "mobile-app-code-verifier"
|
||||
TOKEN = "one-time-token"
|
||||
|
||||
def _mint_token(self, user):
|
||||
"""Create an authenticated session and the matching one-time token."""
|
||||
session = _make_authenticated_session(user)
|
||||
|
||||
cache.set(
|
||||
f"{MOBILE_AUTH_TOKEN_CACHE_PREFIX}:{self.TOKEN}",
|
||||
{
|
||||
"session_key": session.session_key,
|
||||
"code_challenge": _s256(self.VERIFIER),
|
||||
},
|
||||
timeout=60,
|
||||
)
|
||||
return session
|
||||
|
||||
def _exchange(self, client, **overrides):
|
||||
"""POST the exchange payload, allowing per-test overrides."""
|
||||
payload = {"token": self.TOKEN, "code_verifier": self.VERIFIER, **overrides}
|
||||
return client.post(
|
||||
reverse("mobile-auth-exchange"),
|
||||
{key: value for key, value in payload.items() if value is not None},
|
||||
format="json",
|
||||
)
|
||||
|
||||
def test_exchange_success_sets_the_session_cookie(self):
|
||||
"""A valid exchange returns the session cookie and a CSRF token."""
|
||||
user = factories.UserFactory()
|
||||
session = self._mint_token(user)
|
||||
client = APIClient()
|
||||
|
||||
response = self._exchange(client)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data["csrf_token"]
|
||||
assert (
|
||||
response.cookies[settings.SESSION_COOKIE_NAME].value == session.session_key
|
||||
)
|
||||
# With CSRF_USE_SESSIONS the secret lives in the session: no csrftoken
|
||||
# cookie is ever emitted, the app only relies on the body token above.
|
||||
assert settings.CSRF_COOKIE_NAME not in response.cookies
|
||||
# The cookie carried over by the client now authenticates API calls.
|
||||
me_response = client.get("/api/v1.0/users/me/")
|
||||
assert me_response.status_code == status.HTTP_200_OK
|
||||
assert me_response.data["email"] == user.email
|
||||
|
||||
def test_exchange_token_is_single_use(self):
|
||||
"""Replaying a consumed token must be rejected."""
|
||||
self._mint_token(factories.UserFactory())
|
||||
client = APIClient()
|
||||
|
||||
assert self._exchange(client).status_code == status.HTTP_200_OK
|
||||
assert self._exchange(client).status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
def test_exchange_lost_consume_race_is_rejected(self):
|
||||
"""An exchange that loses the delete race must fail even with the payload.
|
||||
|
||||
get() then delete() is not atomic: two concurrent exchanges can both
|
||||
read the payload, but the cache deletes the key exactly once and only
|
||||
the request whose delete() returns True may proceed.
|
||||
"""
|
||||
self._mint_token(factories.UserFactory())
|
||||
client = APIClient()
|
||||
|
||||
with patch("core.api.viewsets.mobile_auth.cache.delete", return_value=False):
|
||||
assert self._exchange(client).status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
def test_exchange_wrong_verifier_consumes_the_token(self):
|
||||
"""A wrong PKCE verifier is rejected and burns the token."""
|
||||
self._mint_token(factories.UserFactory())
|
||||
client = APIClient()
|
||||
|
||||
response = self._exchange(client, code_verifier="wrong-verifier")
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
# The token was consumed by the failed attempt.
|
||||
assert self._exchange(client).status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
def test_exchange_missing_parameters(self):
|
||||
"""Both the token and the verifier are required."""
|
||||
client = APIClient()
|
||||
assert (
|
||||
self._exchange(client, code_verifier=None).status_code
|
||||
== status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
assert (
|
||||
self._exchange(client, token=None).status_code
|
||||
== status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
|
||||
def test_exchange_non_string_parameters(self):
|
||||
"""Non-string JSON values must be rejected instead of crashing _s256()."""
|
||||
self._mint_token(factories.UserFactory())
|
||||
client = APIClient()
|
||||
assert (
|
||||
self._exchange(client, code_verifier=1).status_code
|
||||
== status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
assert (
|
||||
self._exchange(client, token=["a"]).status_code
|
||||
== status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
# A body that is not a JSON object must not crash .get() either.
|
||||
list_body = APIClient().post(
|
||||
reverse("mobile-auth-exchange"), [1, 2], format="json"
|
||||
)
|
||||
assert list_body.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
def test_exchange_is_throttled_per_ip(self):
|
||||
"""Repeated exchange attempts from one IP must be rate limited.
|
||||
|
||||
The endpoint is anonymous and the one-time token is its only secret:
|
||||
without a cap, nothing slows down brute-force guessing.
|
||||
"""
|
||||
client = APIClient()
|
||||
with patch.object(ScopedRateThrottle, "get_rate", return_value="2/minute"):
|
||||
for _ in range(2):
|
||||
assert self._exchange(client).status_code == status.HTTP_403_FORBIDDEN
|
||||
response = self._exchange(client)
|
||||
|
||||
assert response.status_code == status.HTTP_429_TOO_MANY_REQUESTS
|
||||
|
||||
def test_exchange_expired_session(self):
|
||||
"""A token referencing a vanished session must be rejected."""
|
||||
cache.set(
|
||||
f"{MOBILE_AUTH_TOKEN_CACHE_PREFIX}:{self.TOKEN}",
|
||||
{
|
||||
"session_key": "vanished-session-key",
|
||||
"code_challenge": _s256(self.VERIFIER),
|
||||
},
|
||||
timeout=60,
|
||||
)
|
||||
response = self._exchange(APIClient())
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
class TestMobileLogout:
|
||||
"""Tests for the mobile logout endpoint.
|
||||
|
||||
Unlike `/logout/` it must end the Django session without the RP-initiated
|
||||
IdP logout, so the cross-app SSO session survives.
|
||||
"""
|
||||
|
||||
def _login(self, client, user):
|
||||
"""Attach an authenticated session cookie to the client."""
|
||||
session = _make_authenticated_session(user)
|
||||
client.cookies[settings.SESSION_COOKIE_NAME] = session.session_key
|
||||
return session
|
||||
|
||||
def test_logout_flushes_the_server_side_session(self):
|
||||
"""Logout must invalidate the session itself, not only the cookie."""
|
||||
user = factories.UserFactory()
|
||||
client = APIClient()
|
||||
session = self._login(client, user)
|
||||
|
||||
response = client.post(reverse("mobile-auth-logout"))
|
||||
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
engine = import_module(settings.SESSION_ENGINE)
|
||||
assert not engine.SessionStore().exists(session.session_key)
|
||||
# Replaying the old cookie (e.g. a cookie jar the app failed to clear)
|
||||
# must not authenticate anymore.
|
||||
client.cookies[settings.SESSION_COOKIE_NAME] = session.session_key
|
||||
me_response = client.get("/api/v1.0/users/me/")
|
||||
assert me_response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_logout_is_a_noop_for_anonymous_requests(self):
|
||||
"""A logout without a live session must succeed silently."""
|
||||
response = APIClient().post(reverse("mobile-auth-logout"))
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
def test_logout_enforces_csrf(self):
|
||||
"""The session-authenticated POST must echo the CSRF token."""
|
||||
client = APIClient(enforce_csrf_checks=True)
|
||||
self._login(client, factories.UserFactory())
|
||||
|
||||
response = client.post(reverse("mobile-auth-logout"))
|
||||
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
@@ -42,6 +42,7 @@ from core.api.viewsets.metrics import (
|
||||
MailboxUsageMetricsApiView,
|
||||
MailDomainUsersMetricsApiView,
|
||||
)
|
||||
from core.api.viewsets.mobile_auth import MobileLogoutView, MobileSessionExchangeView
|
||||
from core.api.viewsets.placeholder import DraftPlaceholderView, PlaceholderView
|
||||
from core.api.viewsets.provisioning import (
|
||||
ProvisioningMailboxView,
|
||||
@@ -228,6 +229,16 @@ urlpatterns = [
|
||||
),
|
||||
),
|
||||
path(f"api/{settings.API_VERSION}/config/", ConfigView.as_view()),
|
||||
path(
|
||||
f"api/{settings.API_VERSION}/mobile/auth/exchange/",
|
||||
MobileSessionExchangeView.as_view(),
|
||||
name="mobile-auth-exchange",
|
||||
),
|
||||
path(
|
||||
f"api/{settings.API_VERSION}/mobile/auth/logout/",
|
||||
MobileLogoutView.as_view(),
|
||||
name="mobile-auth-logout",
|
||||
),
|
||||
path(
|
||||
f"api/{settings.API_VERSION}/flag/",
|
||||
ChangeFlagView.as_view(),
|
||||
|
||||
@@ -886,6 +886,15 @@ class Base(Configuration):
|
||||
environ_name="API_WIDGET_INBOUND_IP_THROTTLE_RATE",
|
||||
environ_prefix=None,
|
||||
),
|
||||
# Anonymous mobile token → session exchange, keyed per IP (there
|
||||
# is no user yet). Legitimate use is one call per login (plus the
|
||||
# odd retry), so this only caps brute-force guessing of the
|
||||
# one-time token without getting in the way of real devices.
|
||||
"mobile_auth_exchange": values.Value(
|
||||
default="10/minute",
|
||||
environ_name="API_MOBILE_AUTH_EXCHANGE_THROTTLE_RATE",
|
||||
environ_prefix=None,
|
||||
),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -1051,8 +1060,20 @@ class Base(Configuration):
|
||||
OIDC_RP_SCOPES = values.Value(
|
||||
"openid email", environ_name="OIDC_RP_SCOPES", environ_prefix=None
|
||||
)
|
||||
OIDC_AUTHENTICATE_CLASS = "lasuite.oidc_login.views.OIDCAuthenticationRequestView"
|
||||
OIDC_CALLBACK_CLASS = "lasuite.oidc_login.views.OIDCAuthenticationCallbackView"
|
||||
OIDC_AUTHENTICATE_CLASS = "core.authentication.views.OIDCAuthenticationRequestView"
|
||||
OIDC_CALLBACK_CLASS = "core.authentication.views.OIDCAuthenticationCallbackView"
|
||||
|
||||
# Mobile apps (Capacitor) session handoff.
|
||||
# Deep-link schemes the OIDC callback is allowed to redirect to after a
|
||||
# mobile-initiated login. An empty list disables the mobile handoff.
|
||||
MOBILE_AUTH_CALLBACK_SCHEMES = JSONValue(
|
||||
default=[], environ_name="MOBILE_AUTH_CALLBACK_SCHEMES", environ_prefix=None
|
||||
)
|
||||
# Lifetime (seconds) of the one-time token a mobile app exchanges for its
|
||||
# Django session cookie.
|
||||
MOBILE_AUTH_TOKEN_TTL = values.PositiveIntegerValue(
|
||||
default=60, environ_name="MOBILE_AUTH_TOKEN_TTL", environ_prefix=None
|
||||
)
|
||||
LOGIN_REDIRECT_URL = values.Value(
|
||||
None, environ_name="LOGIN_REDIRECT_URL", environ_prefix=None
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user