(backend) introduce a token exchange endpoint for iframe embeds

Some integrators render our videoconference inside an iframe, where
our cookie-based authentication does not work: our cookies are
SameSite=Lax/Strict, so the iframe drops them.

We looked at what Jitsi offers: a shared secret used to sign JWTs
that authenticate users coming from external services. Since we
already expose an external API where third parties authenticate as
a given user, it was simpler for us to add an exchange mechanism on
top of that.

Flow:

* Through the external API, mint a short-lived, single-use exchange
  code for a user.
* The third party hands that code to the frontend as a URL fragment.
* The frontend exchanges the code for a longer-lived JWT that can be
  used to query the regular API viewsets.

Known limitations and follow-ups:

* At some point it would be nice to shorten the JWT lifetime and
  add a refresh mechanism. This will be handled in a follow-up PR
  when actually needed.
* CSP rules to control which origins are allowed to embed the app
  in an iframe still need to be added.
* This alternative authentication cannot easily be scoped to a
  subset of endpoints without adding a lot of complexity, so it is
  accepted globally on the API for now.
This commit is contained in:
lebaudantoine
2026-08-01 15:52:37 +02:00
parent f1d3799434
commit fcb7087f4d
24 changed files with 1537 additions and 38 deletions
+1
View File
@@ -18,6 +18,7 @@ class FeatureFlag:
"application": "APPLICATION_ENABLED", "application": "APPLICATION_ENABLED",
"roomkit": "ROOMKIT_ENABLED", "roomkit": "ROOMKIT_ENABLED",
"connection_test": "CONNECTION_TEST_ENABLED", "connection_test": "CONNECTION_TEST_ENABLED",
"user_access_token": "USER_ACCESS_TOKEN_ENABLED",
} }
@classmethod @classmethod
+17
View File
@@ -599,3 +599,20 @@ class ExternalProcessEventSerializer(BaseValidationOnlySerializer):
# useless bad requests # useless bad requests
type = serializers.CharField(required=False, allow_null=True, allow_blank=True) type = serializers.CharField(required=False, allow_null=True, allow_blank=True)
status = serializers.CharField(required=False, allow_null=True, allow_blank=True) status = serializers.CharField(required=False, allow_null=True, allow_blank=True)
class TransitCodeSerializer(BaseValidationOnlySerializer):
"""Validate the single-use transit code sent to the exchange endpoint."""
code = serializers.CharField(trim_whitespace=True)
def validate_code(self, value):
"""Reject codes whose length cannot match a generated one."""
# Calculates urlsafe_b64encode length without padding
expected_length = (4 * settings.TRANSIT_CODE_NBYTES + 2) // 3
if len(value) != expected_length:
raise serializers.ValidationError("Invalid transit code format.")
return value
+11
View File
@@ -97,3 +97,14 @@ class ConnectionTestAnonRateThrottle(MonitoredAnonRateThrottle):
"""Throttle anonymous users requesting connection test tokens.""" """Throttle anonymous users requesting connection test tokens."""
scope = "connection_test" scope = "connection_test"
class ExchangeAccessTokenAnonRateThrottle(MonitoredAnonRateThrottle):
"""Throttle anonymous transit code exchange attempts.
Abuse mitigation only, not a security boundary: DRF throttling is
best-effort. The security of the exchange rests on the codes'
entropy and single use.
"""
scope = "exchange_access_token"
+93
View File
@@ -43,6 +43,7 @@ from rest_framework.settings import api_settings
from core import analytics, enums, models, utils from core import analytics, enums, models, utils
from core.api import throttling from core.api import throttling
from core.api.filters import ListFileFilter from core.api.filters import ListFileFilter
from core.authentication.user_token import USER_ACCESS_TOKEN_TYPE_CLAIM
from core.enums import MEDIA_STORAGE_URL_PATTERN from core.enums import MEDIA_STORAGE_URL_PATTERN
from core.recording.enums import FileExtension from core.recording.enums import FileExtension
from core.recording.event.authentication import ( from core.recording.event.authentication import (
@@ -75,6 +76,7 @@ from core.recording.worker.mediator import (
WorkerServiceMediator, WorkerServiceMediator,
) )
from core.services.invitation import InvitationService from core.services.invitation import InvitationService
from core.services.jwt_token import JwtTokenService
from core.services.livekit_events import ( from core.services.livekit_events import (
LiveKitEventsService, LiveKitEventsService,
LiveKitWebhookError, LiveKitWebhookError,
@@ -99,6 +101,7 @@ from core.services.room_roles import (
RoomRoleService, RoomRoleService,
) )
from core.services.subtitle import SubtitleException, SubtitleService from core.services.subtitle import SubtitleException, SubtitleService
from core.services.transit_code import TransitCodeService
from core.tasks.connection_test import delete_connection_test_room from core.tasks.connection_test import delete_connection_test_room
from core.tasks.file import process_file_deletion from core.tasks.file import process_file_deletion
from core.utils import generate_token from core.utils import generate_token
@@ -237,6 +240,96 @@ class UserViewSet(
self.serializer_class(request.user, context=context).data self.serializer_class(request.user, context=context).data
) )
@decorators.action(
detail=False,
methods=["post"],
url_path="exchange-access-token",
permission_classes=[],
throttle_classes=[throttling.ExchangeAccessTokenAnonRateThrottle],
)
@FeatureFlag.require("user_access_token")
def exchange_access_token(self, request):
"""Exchange a single-use transit code for a user access token.
The endpoint is unauthenticated: the transit code itself, an opaque
random string obtained through the external API and delivered to
the embedded frontend via a URL fragment, is the credential. Each
code can be exchanged exactly once (consuming it deletes it from
the cache); replaying a consumed code is denied and logged.
The issued JWT authenticates the user the code was minted for on
the whole core API, exactly like a session cookie would (similar
to lib-jitsi-meet's token authentication), and never appears in
any URL. Role-based permissions apply unchanged.
"""
if request.user and request.user.is_authenticated:
logger.warning(
"Transit code exchange refused: request is already "
"session-authenticated (user_id=%s)",
request.user.id,
)
raise drf_exceptions.PermissionDenied("Already authenticated.")
serializer = serializers.TransitCodeSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
code_data = TransitCodeService().consume_code(serializer.validated_data["code"])
if code_data is None:
logger.warning("Invalid, expired or already used transit code")
raise drf_exceptions.PermissionDenied(
"Invalid, expired or already used transit code."
)
# Re-check the user at exchange time so that a deactivation after
# the transit code was minted is taken into account.
try:
user = models.User.objects.get(id=code_data["user_id"], is_active=True)
except models.User.DoesNotExist as excpt:
raise drf_exceptions.PermissionDenied(
"This account can no longer access the application."
) from excpt
if not models.Application.has_active_scope(
code_data.get("client_id"), models.ApplicationScope.USERS_SESSION
):
logger.warning(
"Transit code exchange refused: application '%s' no longer "
"holds the '%s' grant",
code_data.get("client_id"),
models.ApplicationScope.USERS_SESSION,
)
raise drf_exceptions.PermissionDenied(
"This application can no longer create user sessions."
)
token_service = JwtTokenService(
secret_key=settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithm=settings.USER_ACCESS_TOKEN_ALG,
issuer=settings.USER_ACCESS_TOKEN_ISSUER,
audience=settings.USER_ACCESS_TOKEN_AUDIENCE,
expiration_seconds=settings.USER_ACCESS_TOKEN_TTL,
token_type=settings.USER_ACCESS_TOKEN_TYPE,
)
data = token_service.generate_jwt(
user,
"user:access",
{
"client_id": code_data.get("client_id", "unknown"),
"token_type": USER_ACCESS_TOKEN_TYPE_CLAIM,
},
)
# Log for auditing
logger.info(
"User access token issued from transit code: user_id=%s, client_id=%s",
user.id,
code_data.get("client_id", "unknown"),
)
return drf_response.Response(data)
class RoomViewSet( class RoomViewSet(
mixins.CreateModelMixin, mixins.CreateModelMixin,
@@ -0,0 +1,82 @@
"""User access JWT authentication for the Meet core API.
Allows an embedded frontend (e.g. rendered in an iframe, where third-party
session cookies are blocked) to authenticate requests on the core API with
a JWT, obtained by exchanging a single-use transit code (see
core.services.transit_code and the users exchange-access-token endpoint)
and passed as a Bearer header. The JWT itself never appears in any URL.
Similar to lib-jitsi-meet's token authentication, the token is bound to a
user, not to a resource: once authenticated, the request is treated
exactly like a session-authenticated one, and the existing role-based
permissions apply unchanged.
"""
import logging
from django.conf import settings
from rest_framework import exceptions
from core.external_api.authentication import BaseJWTAuthentication
from core.models import Application, ApplicationScope
logger = logging.getLogger(__name__)
USER_ACCESS_TOKEN_TYPE_CLAIM = "user_access" # noqa: S105
class UserAccessJWTAuthentication(BaseJWTAuthentication):
"""JWT authentication for user access tokens.
Validates user access tokens issued by the users exchange-access-token
endpoint and authenticates the user they were issued for. A bearer
token that does not verify against the user access token secret is
deferred to the next authentication backend; a token that does verify
but carries wrong claims is rejected.
When the feature is disabled (USER_ACCESS_TOKEN_ENABLED=False), the
backend is entirely inert: `BaseJWTAuthentication.authenticate`
returns None before reading the Authorization header, deferring every
request to the next authentication backend.
"""
def __init__(self):
"""Initialize the backend with user access token settings."""
super().__init__(
secret_key=settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithm=settings.USER_ACCESS_TOKEN_ALG,
issuer=settings.USER_ACCESS_TOKEN_ISSUER,
audience=settings.USER_ACCESS_TOKEN_AUDIENCE,
expiration_seconds=settings.USER_ACCESS_TOKEN_TTL,
token_type=settings.USER_ACCESS_TOKEN_TYPE,
is_enabled=settings.USER_ACCESS_TOKEN_ENABLED,
)
def validate_payload(self, payload):
"""Validate the token type and the issuance-audit claim.
Raises:
AuthenticationFailed: If the token verified against the user
access token secret but does not carry the expected
claims, or if the issuing application lost its grant.
"""
if payload.get("token_type") != USER_ACCESS_TOKEN_TYPE_CLAIM:
logger.warning("Wrong 'token_type' in user access token payload")
raise exceptions.AuthenticationFailed("Invalid token type.")
if not payload.get("client_id"):
logger.warning("Missing 'client_id' in user access token payload")
raise exceptions.AuthenticationFailed("Invalid token claims.")
if not Application.has_active_scope(
payload["client_id"], ApplicationScope.USERS_SESSION
):
logger.warning(
"User access token refused: application '%s' no longer "
"holds the '%s' grant",
payload["client_id"],
ApplicationScope.USERS_SESSION,
)
raise exceptions.AuthenticationFailed("Application access revoked.")
+95 -27
View File
@@ -20,8 +20,60 @@ class BaseScopePermission(permissions.BasePermission):
scope_map: Dict[str, str] = {} scope_map: Dict[str, str] = {}
def get_required_scope(self, view):
"""Return the scope required by the view's current action.
Returns:
The required scope, or None for an unsupported method so
DRF's router can answer 405.
Raises:
PermissionDenied: If the action is not in scope_map (deny by
default).
"""
# Get the current action (e.g., 'list', 'create'), if None let DRF handle it
action = getattr(view, "action", None)
if not action:
# DRF routers return a 405 for unsupported methods
return None
required_scope = self.scope_map.get(action)
if not required_scope:
# Action not in scope_map, deny by default
raise exceptions.PermissionDenied(
f"Insufficient permissions. Required scope: {required_scope}"
)
return required_scope
def get_token_scopes(self, request):
"""Extract and normalize the scopes claimed by the token."""
token_scopes = (request.auth or {}).get("scope")
if not token_scopes:
return []
# Ensure scopes is a list (handle both list and space-separated string)
if isinstance(token_scopes, str):
token_scopes = token_scopes.split()
# Ensure scopes is a deduplicated list (preserving order) and lowercase all scopes
token_scopes = list(dict.fromkeys(scope.lower() for scope in token_scopes))
return self.strip_scope_prefix(token_scopes)
@staticmethod
def strip_scope_prefix(token_scopes):
"""Strip the OIDC resource server prefix, when configured."""
if settings.OIDC_RS_SCOPES_PREFIX:
return [
scope.removeprefix(f"{settings.OIDC_RS_SCOPES_PREFIX}:")
for scope in token_scopes
]
return token_scopes
def has_permission(self, request, view): def has_permission(self, request, view):
"""Check if the JWT token contains the required scope for this action. """Check if the token claims the scope required by this action.
Args: Args:
request: DRF request object with authenticated user request: DRF request object with authenticated user
@@ -33,38 +85,15 @@ class BaseScopePermission(permissions.BasePermission):
Raises: Raises:
PermissionDenied: If required scope is missing from token PermissionDenied: If required scope is missing from token
""" """
# Get the current action (e.g., 'list', 'create'), if None let DRF handle it required_scope = self.get_required_scope(view)
action = getattr(view, "action", None) if required_scope is None:
if not action:
# DRF routers return a 405 for unsupported methods
return True return True
required_scope = self.scope_map.get(action) token_scopes = self.get_token_scopes(request)
if not required_scope:
# Action not in scope_map, deny by default
raise exceptions.PermissionDenied(
f"Insufficient permissions. Required scope: {required_scope}"
)
token_payload = request.auth
token_scopes = token_payload.get("scope")
if not token_scopes: if not token_scopes:
raise exceptions.PermissionDenied("Insufficient permissions.") raise exceptions.PermissionDenied("Insufficient permissions.")
# Ensure scopes is a list (handle both list and space-separated string)
if isinstance(token_scopes, str):
token_scopes = token_scopes.split()
# Ensure scopes is a deduplicated list (preserving order) and lowercase all scopes
token_scopes = list(dict.fromkeys(scope.lower() for scope in token_scopes))
if settings.OIDC_RS_SCOPES_PREFIX:
token_scopes = [
scope.removeprefix(f"{settings.OIDC_RS_SCOPES_PREFIX}:")
for scope in token_scopes
]
if required_scope not in token_scopes: if required_scope not in token_scopes:
raise exceptions.PermissionDenied( raise exceptions.PermissionDenied(
f"Insufficient permissions. Required scope: {required_scope}" f"Insufficient permissions. Required scope: {required_scope}"
@@ -73,6 +102,37 @@ class BaseScopePermission(permissions.BasePermission):
return True return True
class ApplicationScopePermission(BaseScopePermission):
"""Scope-based permission for application-authenticated endpoints."""
@staticmethod
def strip_scope_prefix(token_scopes):
"""Compare application scopes verbatim."""
return token_scopes
def has_permission(self, request, view):
"""Check the scope claim, then the grant recorded in the database."""
granted = super().has_permission(request, view)
required_scope = self.get_required_scope(view)
if granted and required_scope:
client_id = (request.auth or {}).get("client_id")
if not models.Application.has_active_scope(client_id, required_scope):
logger.warning(
"Application '%s' presented scope '%s' without a matching "
"grant in database",
client_id,
required_scope,
)
raise exceptions.PermissionDenied(
f"Application is not granted the required scope: {required_scope}"
)
return granted
class HasRequiredRoomScope(BaseScopePermission): class HasRequiredRoomScope(BaseScopePermission):
"""Permission class for Room-related operations.""" """Permission class for Room-related operations."""
@@ -86,6 +146,14 @@ class HasRequiredRoomScope(BaseScopePermission):
} }
class HasRequiredUserScope(ApplicationScopePermission):
"""Scope-based permissions for the external user endpoints."""
scope_map = {
"generate_transit_code": models.ApplicationScope.USERS_SESSION,
}
class RoomPermissions(permissions.BasePermission): class RoomPermissions(permissions.BasePermission):
"""Permissions applying to the room API endpoint.""" """Permissions applying to the room API endpoint."""
+57
View File
@@ -25,6 +25,7 @@ from rest_framework import (
from core import analytics, api, models from core import analytics, api, models
from core.api.feature_flag import FeatureFlag from core.api.feature_flag import FeatureFlag
from core.services.jwt_token import JwtTokenService from core.services.jwt_token import JwtTokenService
from core.services.transit_code import TransitCodeService
from ..services.provisional_user_service import ( from ..services.provisional_user_service import (
ProvisionalUserCreationDisabledError, ProvisionalUserCreationDisabledError,
@@ -222,3 +223,59 @@ class RoomViewSet(
"$set": {"email": self.request.user.email}, "$set": {"email": self.request.user.email},
}, },
) )
class UserViewSet(viewsets.GenericViewSet):
"""Application-delegated API for user operations.
Provides JWT-authenticated access to user operations for external
applications acting on behalf of users. All operations are
scope-based. Meant to grow with the other user actions exposed to
third parties.
Supported operations:
- transit-code: Mint a single-use transit code for the delegated user
(requires 'users:session' scope)
"""
authentication_classes = [
authentication.ApplicationJWTAuthentication,
]
permission_classes = [
api.permissions.IsAuthenticated & permissions.HasRequiredUserScope
]
@decorators.action(
detail=False,
methods=["post"],
url_path="transit-code",
url_name="transit-code",
)
@FeatureFlag.require("user_access_token")
def generate_transit_code(self, request):
"""Mint a transit code for the delegated user.
Returns a short-lived, single-use opaque code to pass to an embedded
frontend (e.g. via a URL fragment when cookies are unavailable). The
frontend exchanges it once on
POST /api/v1.0/users/exchange-access-token/ for a JWT access token,
equivalent to session-cookie authentication and never exposed in a URL.
"""
client_id = (request.auth or {}).get("client_id", "unknown")
code = TransitCodeService().create_code(request.user, client_id=client_id)
# Log for auditing
logger.info(
"Transit code issued: user_id=%s, client_id=%s",
request.user.id,
client_id,
)
return drf_response.Response(
{
"transit_code": code,
"expires_in": settings.TRANSIT_CODE_TTL,
},
status=drf_status.HTTP_200_OK,
)
@@ -0,0 +1,19 @@
# Generated by Django 5.2.14 on 2026-07-31 18:27
import django.contrib.postgres.fields
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0021_recording_external_process_id_alter_recording_status'),
]
operations = [
migrations.AlterField(
model_name='application',
name='scopes',
field=django.contrib.postgres.fields.ArrayField(base_field=models.CharField(choices=[('rooms:create', 'Create rooms'), ('rooms:list', 'List rooms'), ('rooms:retrieve', 'Retrieve room details'), ('rooms:update', 'Update rooms'), ('rooms:delete', 'Delete rooms'), ('users:session', 'Create user session tokens')], max_length=50), blank=True, default=list, size=None),
),
]
@@ -6,7 +6,7 @@ from django.db import migrations, models
class Migration(migrations.Migration): class Migration(migrations.Migration):
dependencies = [ dependencies = [
('core', '0021_recording_external_process_id_alter_recording_status'), ('core', '0022_alter_application_scopes'),
] ]
operations = [ operations = [
+13
View File
@@ -795,6 +795,7 @@ class ApplicationScope(models.TextChoices):
ROOMS_RETRIEVE = "rooms:retrieve", _("Retrieve room details") ROOMS_RETRIEVE = "rooms:retrieve", _("Retrieve room details")
ROOMS_UPDATE = "rooms:update", _("Update rooms") ROOMS_UPDATE = "rooms:update", _("Update rooms")
ROOMS_DELETE = "rooms:delete", _("Delete rooms") ROOMS_DELETE = "rooms:delete", _("Delete rooms")
USERS_SESSION = "users:session", _("Create user session tokens")
class Application(BaseModel): class Application(BaseModel):
@@ -844,6 +845,18 @@ class Application(BaseModel):
domain = get_domain_from_email(email) domain = get_domain_from_email(email)
return self.allowed_domains.filter(domain__iexact=domain).exists() return self.allowed_domains.filter(domain__iexact=domain).exists()
@classmethod
def has_active_scope(cls, client_id, scope) -> bool:
"""Check that an active application holds a scope."""
if not client_id or not scope:
return False
return cls.objects.filter(
client_id=client_id,
is_active=True,
scopes__contains=[scope],
).exists()
class ApplicationDomain(BaseModel): class ApplicationDomain(BaseModel):
"""Domain authorized for application delegation.""" """Domain authorized for application delegation."""
+74
View File
@@ -0,0 +1,74 @@
"""Service handling the lifecycle of transit codes.
A transit code is an opaque, cryptographically random, single-use code
handed to an embedded frontend (through a URL fragment) so it can obtain a
user access token on the core API without a session cookie. The code
carries no information by itself: everything it references (user, client)
is stored server-side in the cache, and consumed atomically on exchange.
"""
import hashlib
import secrets
from django.conf import settings
from django.core.cache import cache
class TransitCodeService:
"""Create and consume single-use transit codes."""
@staticmethod
def _cache_key(code):
"""Build the cache key for a code.
The code is hashed so that a dump of the cache never reveals
directly usable codes.
"""
digest = hashlib.sha256(code.encode("utf-8")).hexdigest()
return f"{settings.TRANSIT_CODE_CACHE_PREFIX}:{digest}"
def create_code(self, user, client_id="unknown"):
"""Generate a transit code for a user, and store it.
The code expires after TRANSIT_CODE_TTL seconds.
Returns:
str: The opaque code to hand to the client.
"""
# Default 48 random bytes -> 64 url-safe characters, 384 bits of
# entropy: unguessable and safe to transit through a URL fragment.
code = secrets.token_urlsafe(settings.TRANSIT_CODE_NBYTES)
cache.set(
self._cache_key(code),
{
"user_id": str(user.id),
"client_id": client_id,
},
timeout=settings.TRANSIT_CODE_TTL,
)
return code
def consume_code(self, code):
"""Consume a transit code, enforcing single use.
The code is deleted from the cache upon consumption. `cache.delete`
returns whether a key was actually deleted, so if two requests race
on the same code, only one of them wins.
Returns:
dict | None: The data stored at creation time ('user_id',
'client_id'), or None if the code is unknown, expired or
already consumed.
"""
if not code:
return None
key = self._cache_key(code)
data = cache.get(key)
if data is None or not cache.delete(key):
return None
return data
@@ -2,15 +2,18 @@
Test rooms API endpoints in the Meet core app: create. Test rooms API endpoints in the Meet core app: create.
""" """
from datetime import datetime, timedelta, timezone
# pylint: disable=redefined-outer-name,unused-argument # pylint: disable=redefined-outer-name,unused-argument
from django.conf import settings from django.conf import settings
from django.core.cache import cache from django.core.cache import cache
import jwt
import pytest import pytest
from rest_framework.test import APIClient from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory from ...factories import ApplicationFactory, RoomFactory, UserFactory
from ...models import Room, RoomAccessLevel from ...models import ApplicationScope, Room, RoomAccessLevel
pytestmark = pytest.mark.django_db pytestmark = pytest.mark.django_db
@@ -312,3 +315,39 @@ def test_api_rooms_create_authenticated_blank_user_default_access_level():
assert response.status_code == 201 assert response.status_code == 201
room = Room.objects.get() room = Room.objects.get()
assert room.access_level == settings.RESOURCE_DEFAULT_ACCESS_LEVEL assert room.access_level == settings.RESOURCE_DEFAULT_ACCESS_LEVEL
def generate_user_access_token(user):
"""Generate a valid user access JWT signed with the token secret."""
now = datetime.now(timezone.utc)
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
payload = {
"iss": settings.USER_ACCESS_TOKEN_ISSUER,
"aud": settings.USER_ACCESS_TOKEN_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=settings.USER_ACCESS_TOKEN_TTL),
"user_id": str(user.id),
"token_type": "user_access",
"client_id": application.client_id,
"scope": "user:access",
}
return jwt.encode(
payload,
settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithm=settings.USER_ACCESS_TOKEN_ALG,
)
def test_api_rooms_create_authenticated_with_user_access_token():
"""A user access token should create a room exactly like a session would."""
user = UserFactory()
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(user)}")
response = client.post("/api/v1.0/rooms/", {"name": "my room"})
assert response.status_code == 201
room = Room.objects.get()
assert room.accesses.filter(role="owner", user=user).exists()
@@ -2,14 +2,18 @@
Test rooms API endpoints in the Meet core app: list. Test rooms API endpoints in the Meet core app: list.
""" """
from datetime import datetime, timedelta, timezone
from unittest import mock from unittest import mock
from django.conf import settings as django_settings
import jwt
import pytest import pytest
from rest_framework.pagination import PageNumberPagination from rest_framework.pagination import PageNumberPagination
from rest_framework.test import APIClient from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory from ...factories import ApplicationFactory, RoomFactory, UserFactory
from ...models import RoomAccessLevel from ...models import ApplicationScope, RoomAccessLevel
pytestmark = pytest.mark.django_db pytestmark = pytest.mark.django_db
@@ -156,3 +160,41 @@ def test_api_rooms_list_pagination_page_size():
assert len(content["results"]) == 3 assert len(content["results"]) == 3
assert content["next"] == "http://testserver/api/v1.0/rooms/?page=2&page_size=3" assert content["next"] == "http://testserver/api/v1.0/rooms/?page=2&page_size=3"
assert content["previous"] is None assert content["previous"] is None
def generate_user_access_token(user):
"""Generate a valid user access JWT signed with the token secret."""
now = datetime.now(timezone.utc)
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
payload = {
"iss": django_settings.USER_ACCESS_TOKEN_ISSUER,
"aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL),
"user_id": str(user.id),
"token_type": "user_access",
"client_id": application.client_id,
"scope": "user:access",
}
return jwt.encode(
payload,
django_settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithm=django_settings.USER_ACCESS_TOKEN_ALG,
)
def test_api_rooms_list_authenticated_with_user_access_token():
"""A user access token should list rooms exactly like a session would."""
user = UserFactory()
room = RoomFactory(users=[(user, "owner")])
RoomFactory() # another user's room, not listed
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(user)}")
response = client.get("/api/v1.0/rooms/")
assert response.status_code == 200
assert response.data["count"] == 1
assert response.data["results"][0]["id"] == str(room.id)
@@ -19,8 +19,23 @@ from rest_framework import status
from rest_framework.test import APIClient from rest_framework.test import APIClient
from core import utils from core import utils
<<<<<<< HEAD
from core.factories import RoomFactory, UserFactory, UserResourceAccessFactory from core.factories import RoomFactory, UserFactory, UserResourceAccessFactory
from core.services.lobby import LobbyService from core.services.lobby import LobbyService
=======
from core.factories import (
ApplicationFactory,
RoomFactory,
UserFactory,
UserResourceAccessFactory,
)
from core.models import ApplicationScope
from core.services.lobby import (
LobbyParticipant,
LobbyParticipantStatus,
LobbyService,
)
>>>>>>> 32a253e08 (fixup! (backend) introduce a token exchange endpoint for iframe embeds)
pytestmark = pytest.mark.django_db pytestmark = pytest.mark.django_db
@@ -17,7 +17,13 @@ from rest_framework import status
from rest_framework.test import APIClient from rest_framework.test import APIClient
from core import utils from core import utils
from core.factories import RoomFactory, UserFactory from core.factories import (
ApplicationFactory,
RoomFactory,
UserFactory,
UserResourceAccessFactory,
)
from core.models import ApplicationScope
pytestmark = pytest.mark.django_db pytestmark = pytest.mark.django_db
@@ -3,16 +3,24 @@ Test rooms API endpoints in the Meet core app: retrieve.
""" """
import random import random
from datetime import datetime, timedelta, timezone
from unittest import mock from unittest import mock
from django.conf import settings as django_settings
from django.contrib.auth.models import AnonymousUser from django.contrib.auth.models import AnonymousUser
from django.test.utils import override_settings from django.test.utils import override_settings
import jwt
import pytest import pytest
from rest_framework.test import APIClient from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory, UserResourceAccessFactory from ...factories import (
from ...models import RoleChoices, RoomAccessLevel ApplicationFactory,
RoomFactory,
UserFactory,
UserResourceAccessFactory,
)
from ...models import ApplicationScope, RoleChoices, RoomAccessLevel
pytestmark = pytest.mark.django_db pytestmark = pytest.mark.django_db
@@ -507,3 +515,41 @@ def test_api_rooms_retrieve_administrators(
role=str(user_access.role), role=str(user_access.role),
participant_id=None, participant_id=None,
) )
def generate_user_access_token(user):
"""Generate a valid user access JWT signed with the token secret."""
now = datetime.now(timezone.utc)
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
payload = {
"iss": django_settings.USER_ACCESS_TOKEN_ISSUER,
"aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL),
"user_id": str(user.id),
"token_type": "user_access",
"client_id": application.client_id,
"scope": "user:access",
}
return jwt.encode(
payload,
django_settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithm=django_settings.USER_ACCESS_TOKEN_ALG,
)
def test_api_rooms_retrieve_authenticated_with_user_access_token():
"""A user access token should retrieve a room exactly like a session would."""
user = UserFactory()
room = RoomFactory(users=[(user, "owner")])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(user)}")
response = client.get(f"/api/v1.0/rooms/{room.id!s}/")
assert response.status_code == 200
assert response.data["id"] == str(room.id)
assert response.data["pin_code"] == room.pin_code
assert "accesses" in response.data
@@ -12,7 +12,8 @@ import pytest
from livekit.api import AccessToken, TwirpError, VideoGrants from livekit.api import AccessToken, TwirpError, VideoGrants
from rest_framework.test import APIClient from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory from ...factories import ApplicationFactory, RoomFactory, UserFactory
from ...models import ApplicationScope
pytestmark = pytest.mark.django_db pytestmark = pytest.mark.django_db
@@ -3,13 +3,17 @@ Test rooms API endpoints in the Meet core app: update.
""" """
import random import random
from datetime import datetime, timedelta, timezone
from unittest.mock import patch from unittest.mock import patch
from django.conf import settings as django_settings
import jwt
import pytest import pytest
from rest_framework.test import APIClient from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory from ...factories import ApplicationFactory, RoomFactory, UserFactory
from ...models import RoomAccessLevel from ...models import ApplicationScope, RoomAccessLevel
from ...services.room_management import ( from ...services.room_management import (
RoomManagement, RoomManagement,
RoomManagementException, RoomManagementException,
@@ -437,3 +441,46 @@ def test_api_rooms_update_livekit_sync_failure(mock_update_metadata):
"configuration": {"can_publish_sources": ["camera"]}, "configuration": {"can_publish_sources": ["camera"]},
}, },
) )
def generate_user_access_token(user):
"""Generate a valid user access JWT signed with the token secret."""
now = datetime.now(timezone.utc)
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
payload = {
"iss": django_settings.USER_ACCESS_TOKEN_ISSUER,
"aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL),
"user_id": str(user.id),
"token_type": "user_access",
"client_id": application.client_id,
"scope": "user:access",
}
return jwt.encode(
payload,
django_settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithm=django_settings.USER_ACCESS_TOKEN_ALG,
)
def test_api_rooms_update_authenticated_with_user_access_token():
"""Role-based permissions apply unchanged with a user access token."""
user = UserFactory()
room = RoomFactory(users=[(user, "member")])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(user)}")
# A simple member cannot update the room
response = client.patch(f"/api/v1.0/rooms/{room.id!s}/", {"name": "new name"})
assert response.status_code == 403
# An administrator can
room.accesses.filter(user=user).update(role="administrator")
response = client.patch(f"/api/v1.0/rooms/{room.id!s}/", {"name": "new name"})
assert response.status_code == 200
room.refresh_from_db()
assert room.name == "new name"
@@ -0,0 +1,58 @@
"""
Unit tests for the TransitCodeService.
"""
from unittest.mock import patch
import pytest
from core.factories import UserFactory
from core.services.transit_code import TransitCodeService
pytestmark = pytest.mark.django_db
def test_create_code_returns_unique_opaque_codes():
"""Each created code should be a distinct high-entropy string."""
user = UserFactory()
service = TransitCodeService()
codes = {service.create_code(user) for _ in range(5)}
assert len(codes) == 5
for code in codes:
assert len(code) >= 43
def test_consume_code_returns_stored_data_once():
"""Consuming a code should return its data exactly once."""
user = UserFactory()
service = TransitCodeService()
code = service.create_code(user, client_id="my-app")
assert service.consume_code(code) == {
"user_id": str(user.id),
"client_id": "my-app",
}
# Single use: a second consumption fails
assert service.consume_code(code) is None
def test_consume_code_unknown_or_empty():
"""Unknown or empty codes should not be consumable."""
service = TransitCodeService()
assert service.consume_code("unknown-code") is None
assert service.consume_code("") is None
assert service.consume_code(None) is None
@patch("core.services.transit_code.cache.delete", return_value=False)
def test_consume_code_returns_none_when_delete_loses_the_race(mock_delete):
"""If the code was already deleted by a concurrent request, consumption fails."""
user = UserFactory()
service = TransitCodeService()
code = service.create_code(user, client_id="my-app")
assert service.consume_code(code) is None
mock_delete.assert_called_once()
@@ -0,0 +1,270 @@
"""
Tests for user access JWT authentication on the core API.
The token authenticates the user on the whole API, exactly like a session
cookie would (similar to lib-jitsi-meet's token authentication): the
existing role-based permissions apply unchanged. Room endpoint coverage
with a user access token lives in the room test files.
"""
from datetime import datetime, timedelta, timezone
from django.conf import settings as django_settings
import jwt
import pytest
from rest_framework.test import APIClient
from core.factories import ApplicationFactory, RoomFactory, UserFactory
from core.models import ApplicationScope, RoleChoices
pytestmark = pytest.mark.django_db
def generate_user_access_token(user, application=None, **overrides):
"""Generate a valid user access JWT signed with the token secret."""
now = datetime.now(timezone.utc)
if application is None:
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
payload = {
"iss": django_settings.USER_ACCESS_TOKEN_ISSUER,
"aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL),
"user_id": str(user.id),
"token_type": "user_access",
"client_id": application.client_id,
"scope": "user:access",
}
payload.update(overrides)
payload = {key: value for key, value in payload.items() if value is not None}
return jwt.encode(
payload,
django_settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithm=django_settings.USER_ACCESS_TOKEN_ALG,
)
def test_user_access_token_users_me():
"""A user access token should authenticate the user on /users/me/."""
user = UserFactory()
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(user)}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 200
assert response.data["email"] == user.email
def test_user_access_token_expired():
"""An expired user access token should be rejected."""
user = UserFactory()
now = datetime.now(timezone.utc)
token = generate_user_access_token(
user,
iat=now - timedelta(hours=3),
exp=now - timedelta(hours=1),
)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
assert "token expired" in str(response.data).lower()
def test_user_access_token_wrong_token_type():
"""A verified token with the wrong 'token_type' claim should be rejected."""
user = UserFactory()
token = generate_user_access_token(user, token_type="addons")
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
assert "invalid token type" in str(response.data).lower()
def test_user_access_token_invalid_signature():
"""A token signed with the wrong key should defer and end unauthenticated."""
user = UserFactory()
now = datetime.now(timezone.utc)
token = jwt.encode(
{
"iss": django_settings.USER_ACCESS_TOKEN_ISSUER,
"aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=600),
"user_id": str(user.id),
"token_type": "user_access",
"client_id": "test-app",
},
"wrong-secret-key-padded-for-minimum-len!",
algorithm=django_settings.USER_ACCESS_TOKEN_ALG,
)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# UserAccessJWTAuthentication defers, session auth finds no session
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
def test_user_access_token_missing_client_id_claim():
"""A token without the issuance-audit claim should be rejected."""
user = UserFactory()
token = generate_user_access_token(user, client_id=None)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
assert "invalid token claims" in str(response.data).lower()
def test_user_access_token_inactive_user():
"""A user access token for an inactive user should be rejected."""
user = UserFactory(is_active=False)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(user)}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
def test_user_access_token_feature_disabled(settings):
"""When the feature is disabled, user access tokens should be ignored."""
settings.USER_ACCESS_TOKEN_ENABLED = False
user = UserFactory()
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(user)}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
def test_user_access_token_does_not_break_session_authentication():
"""A session-authenticated user should keep full access to the API."""
user = UserFactory()
RoomFactory(users=[(user, RoleChoices.OWNER)])
client = APIClient()
client.force_login(user)
response = client.get("/api/v1.0/rooms/")
assert response.status_code == 200
assert response.data["count"] == 1
def test_user_access_token_application_jwt_not_accepted_on_core_api():
"""An application-delegation JWT must not authenticate on the core API."""
user = UserFactory()
now = datetime.now(timezone.utc)
token = jwt.encode(
{
"iss": django_settings.APPLICATION_JWT_ISSUER,
"aud": django_settings.APPLICATION_JWT_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=600),
"user_id": str(user.id),
"client_id": "some-client",
"delegated": True,
"scope": "rooms:retrieve",
},
django_settings.APPLICATION_JWT_SECRET_KEY,
algorithm=django_settings.APPLICATION_JWT_ALG,
)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
# The user token backend must defer (wrong signature) and the request
# must end up unauthenticated.
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
def test_user_access_token_application_scope_revoked():
"""Revoking the application's grant invalidates its outstanding tokens."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
token = generate_user_access_token(user, application=application)
application.scopes = []
application.save()
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
assert "application access revoked" in str(response.data).lower()
def test_user_access_token_application_deactivated():
"""Deactivating the application invalidates its outstanding tokens."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
token = generate_user_access_token(user, application=application)
application.is_active = False
application.save()
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
assert "application access revoked" in str(response.data).lower()
def test_user_access_token_unknown_application():
"""A token whose client_id matches no application is refused."""
user = UserFactory()
token = generate_user_access_token(user, client_id="not-an-application")
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 401
assert "application access revoked" in str(response.data).lower()
def test_user_access_token_does_not_override_existing_session():
"""A Bearer token must not override the identity of a live session."""
session_user = UserFactory()
token_user = UserFactory()
client = APIClient()
client.force_login(session_user)
client.credentials(
HTTP_AUTHORIZATION=f"Bearer {generate_user_access_token(token_user)}"
)
response = client.get("/api/v1.0/users/me/")
assert response.status_code == 200
assert response.data["email"] == session_user.email
@@ -0,0 +1,262 @@
"""
Test users API endpoints in the Meet core app: exchange transit code.
"""
# pylint: disable=W0621
import secrets
import jwt
import pytest
from rest_framework.test import APIClient
from core.factories import ApplicationFactory, UserFactory
from core.models import ApplicationScope
from core.services.transit_code import TransitCodeService
pytestmark = pytest.mark.django_db
def decode_user_access_token(token, settings):
"""Decode a user access token with the token secret."""
return jwt.decode(
token,
settings.USER_ACCESS_TOKEN_SECRET_KEY,
algorithms=[settings.USER_ACCESS_TOKEN_ALG],
issuer=settings.USER_ACCESS_TOKEN_ISSUER,
audience=settings.USER_ACCESS_TOKEN_AUDIENCE,
)
def generate_unknown_code(settings):
"""Generate a well-formed code that was never stored."""
return secrets.token_urlsafe(settings.TRANSIT_CODE_NBYTES)
@pytest.fixture
def client():
"""Return an anonymous API client with a random source IP.
A fresh IP per test isolates the anonymous throttle history, both
between the tests of this module and between test runs.
"""
# `secrets` rather than `random`: the global random module is seeded
# deterministically by the factories, its sequence repeats across runs.
remote_addr = (
f"10.{secrets.randbelow(256)}.{secrets.randbelow(256)}"
f".{secrets.randbelow(254) + 1}"
)
return APIClient(REMOTE_ADDR=remote_addr)
def test_exchange_access_token_missing_code(client):
"""The exchange endpoint should validate its input."""
response = client.post("/api/v1.0/users/exchange-access-token/")
assert response.status_code == 400
assert "code" in response.data
def test_exchange_access_token_get_method(client):
"""The exchange endpoint should not accept GET."""
response = client.get("/api/v1.0/users/exchange-access-token/")
assert response.status_code == 405
def test_exchange_access_token_malformed_code(client):
"""A code whose length cannot match a generated one should be a 400."""
response = client.post(
"/api/v1.0/users/exchange-access-token/",
{"code": "not-a-valid-code"},
)
assert response.status_code == 400
assert "invalid transit code format" in str(response.data).lower()
def test_exchange_access_token_unknown_code(client, settings):
"""A well-formed but unknown code should be denied."""
response = client.post(
"/api/v1.0/users/exchange-access-token/",
{"code": generate_unknown_code(settings)},
)
assert response.status_code == 403
assert "invalid, expired or already used" in str(response.data).lower()
def test_exchange_access_token_success(client, settings):
"""A valid transit code should be exchangeable for an access token."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 200
assert response.data["token_type"] == settings.USER_ACCESS_TOKEN_TYPE
assert response.data["expires_in"] == settings.USER_ACCESS_TOKEN_TTL
assert response.data["scope"] == "user:access"
payload = decode_user_access_token(response.data["access_token"], settings)
assert payload["user_id"] == str(user.id)
assert payload["client_id"] == application.client_id
assert payload["exp"] - payload["iat"] == settings.USER_ACCESS_TOKEN_TTL
def test_exchange_access_token_single_use(client):
"""A transit code should be exchangeable exactly once."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 200
# Replaying the same code must be denied
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 403
assert "invalid, expired or already used" in str(response.data).lower()
def test_exchange_access_token_inactive_user(client):
"""A code minted for a now-inactive user should be denied."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
user.is_active = False
user.save()
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 403
assert "no longer access" in str(response.data).lower()
def test_exchange_access_token_feature_disabled(client, settings):
"""The exchange endpoint should return 404 when the feature is disabled."""
settings.USER_ACCESS_TOKEN_ENABLED = False
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 404
def test_exchange_access_token_throttled(client, settings):
"""Anonymous exchange attempts should be rate limited."""
throttle_rates = settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"]
initial_rate = throttle_rates["exchange_access_token"]
# The rates dict is mutated in place: restore it explicitly, the
# `settings` fixture only rolls back attribute assignments.
throttle_rates["exchange_access_token"] = "2/minute"
try:
for _ in range(2):
response = client.post(
"/api/v1.0/users/exchange-access-token/",
{"code": generate_unknown_code(settings)},
)
assert response.status_code == 403
response = client.post(
"/api/v1.0/users/exchange-access-token/",
{"code": generate_unknown_code(settings)},
)
assert response.status_code == 429
finally:
throttle_rates["exchange_access_token"] = initial_rate
def test_exchange_access_token_refused_when_already_authenticated(client):
"""A session-authenticated browser must not exchange a transit code."""
user = UserFactory()
session_user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
client.force_login(session_user)
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 403
assert "already authenticated" in str(response.data).lower()
# The code was not consumed: it stays valid for its intended,
# cookieless embedded context.
client.logout()
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 200
def test_exchange_access_token_application_scope_revoked(client):
"""A code is refused once the application's grant is revoked."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
application.scopes = []
application.save()
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 403
assert "no longer create user sessions" in str(response.data).lower()
def test_exchange_access_token_application_deactivated(client):
"""A code is refused once the application is disabled."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
application.is_active = False
application.save()
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 403
assert "no longer create user sessions" in str(response.data).lower()
def test_exchange_access_token_unknown_application(client):
"""A code whose client_id matches no application is refused."""
user = UserFactory()
code = TransitCodeService().create_code(user, client_id="not-an-application")
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 403
assert "no longer create user sessions" in str(response.data).lower()
def test_exchange_access_token_end_to_end(client):
"""A token obtained from the exchange must authenticate on the core API.
Regression test: token issuance and token validation must stay in
sync on the claims they set and require (e.g. 'token_type').
"""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION])
code = TransitCodeService().create_code(user, client_id=application.client_id)
response = client.post("/api/v1.0/users/exchange-access-token/", {"code": code})
assert response.status_code == 200
api_client = APIClient()
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {response.data['access_token']}")
me = api_client.get("/api/v1.0/users/me/")
assert me.status_code == 200
assert me.data["email"] == user.email
@@ -0,0 +1,209 @@
"""
Tests for external API /users endpoints (transit codes)
"""
# pylint: disable=W0621
from datetime import datetime, timedelta, timezone
from unittest import mock
from django.conf import settings as django_settings
import jwt
import pytest
from lasuite.oidc_resource_server.authentication import ResourceServerAuthentication
from rest_framework.test import APIClient
from core.factories import ApplicationFactory, UserFactory
from core.models import ApplicationScope
from core.services.transit_code import TransitCodeService
pytestmark = pytest.mark.django_db
def generate_addons_test_token(user, scopes):
"""Generate a valid JWT token signed with the addons secret for testing."""
now = datetime.now(timezone.utc)
payload = {
"iss": django_settings.ADDONS_TOKEN_ISSUER,
"aud": django_settings.ADDONS_TOKEN_AUDIENCE,
"iat": now,
"exp": now + timedelta(seconds=django_settings.ADDONS_TOKEN_TTL),
"scope": " ".join(scopes),
"user_id": str(user.id),
}
return jwt.encode(
payload,
django_settings.ADDONS_TOKEN_SECRET_KEY,
algorithm=django_settings.ADDONS_TOKEN_ALG,
)
def generate_test_token(user, scopes, application=None):
"""Generate a valid application JWT token for testing."""
now = datetime.now(timezone.utc)
scope_string = " ".join(scopes)
if application is None:
application = ApplicationFactory(scopes=scopes)
payload = {
"iss": django_settings.APPLICATION_JWT_ISSUER,
"aud": django_settings.APPLICATION_JWT_AUDIENCE,
"iat": now,
"exp": now
+ timedelta(seconds=django_settings.APPLICATION_JWT_EXPIRATION_SECONDS),
"client_id": str(application.client_id),
"scope": scope_string,
"user_id": str(user.id),
"delegated": True,
}
return jwt.encode(
payload,
django_settings.APPLICATION_JWT_SECRET_KEY,
algorithm=django_settings.APPLICATION_JWT_ALG,
)
def test_api_users_transit_code_requires_authentication():
"""Minting a transit code without authentication should return 401."""
client = APIClient()
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 401
def test_api_users_transit_code_missing_scope():
"""A token without the 'users:session' scope should be rejected."""
user = UserFactory()
token = generate_test_token(user, [ApplicationScope.ROOMS_RETRIEVE])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 403
assert "users:session" in str(response.data)
def test_api_users_transit_code_success(settings):
"""A delegated user with the scope should be able to mint a transit code."""
user = UserFactory()
token = generate_test_token(user, [ApplicationScope.USERS_SESSION])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 200
assert response.data["expires_in"] == settings.TRANSIT_CODE_TTL
code = response.data["transit_code"]
# Opaque, high-entropy random string
assert len(code) == (4 * settings.TRANSIT_CODE_NBYTES + 2) // 3
# The code is stored server-side and references the delegated user
code_data = TransitCodeService().consume_code(code)
assert code_data == {
"user_id": str(user.id),
"client_id": mock.ANY,
}
def test_api_users_transit_code_scope_claim_exceeding_db_grant():
"""A 'users:session' claim beyond the grant recorded in database is refused."""
user = UserFactory()
application = ApplicationFactory(scopes=[ApplicationScope.ROOMS_RETRIEVE])
token = generate_test_token(
user, [ApplicationScope.USERS_SESSION], application=application
)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 403
assert "not granted" in str(response.data)
def test_api_users_transit_code_get_forbidden():
"""Minting a transit code with a GET should not be allowed."""
user = UserFactory()
token = generate_test_token(user, [ApplicationScope.USERS_SESSION])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.get("/external-api/v1.0/users/transit-code/")
assert response.status_code == 405
def test_api_users_transit_code_resource_server_not_supported():
"""A resource server token must not be able to mint a transit code."""
user = UserFactory()
with mock.patch.object(
ResourceServerAuthentication,
"authenticate",
return_value=(user, {"scope": "users:session", "client_id": "rs-client"}),
) as mock_rs_authenticate:
client = APIClient()
client.credentials(HTTP_AUTHORIZATION="Bearer some-opaque-rs-token")
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 401
mock_rs_authenticate.assert_not_called()
def test_api_users_transit_code_feature_disabled(settings):
"""Minting a transit code should return 404 when the feature is disabled."""
settings.USER_ACCESS_TOKEN_ENABLED = False
user = UserFactory()
token = generate_test_token(user, [ApplicationScope.USERS_SESSION])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 404
def test_api_users_transit_code_inactive_user():
"""An inactive user should not be able to mint a transit code."""
user = UserFactory(is_active=False)
token = generate_test_token(user, [ApplicationScope.USERS_SESSION])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 401
def test_api_users_transit_code_rejects_addons_token():
"""An addons token must not be able to mint a transit code.
The token carries the 'users:session' scope and is signed with the addons
secret, so only the missing backend stands between it and a transit code.
"""
user = UserFactory()
token = generate_addons_test_token(user, [ApplicationScope.USERS_SESSION])
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
with mock.patch.object(
ResourceServerAuthentication, "authenticate", return_value=None
):
response = client.post("/external-api/v1.0/users/transit-code/")
assert response.status_code == 401
+5
View File
@@ -48,6 +48,11 @@ external_router.register(
external_viewsets.RoomViewSet, external_viewsets.RoomViewSet,
basename="external_room", basename="external_room",
) )
external_router.register(
"users",
external_viewsets.UserViewSet,
basename="external_user",
)
urlpatterns = [ urlpatterns = [
path( path(
+64
View File
@@ -325,6 +325,7 @@ class Base(Configuration):
REST_FRAMEWORK = { REST_FRAMEWORK = {
"DEFAULT_AUTHENTICATION_CLASSES": ( "DEFAULT_AUTHENTICATION_CLASSES": (
"core.authentication.backends.SessionAuthenticationWith401", "core.authentication.backends.SessionAuthenticationWith401",
"core.authentication.user_token.UserAccessJWTAuthentication",
), ),
"DEFAULT_PARSER_CLASSES": [ "DEFAULT_PARSER_CLASSES": [
"rest_framework.parsers.JSONParser", "rest_framework.parsers.JSONParser",
@@ -344,6 +345,11 @@ class Base(Configuration):
environ_name="REQUEST_ENTRY_THROTTLE_RATES", environ_name="REQUEST_ENTRY_THROTTLE_RATES",
environ_prefix=None, environ_prefix=None,
), ),
"exchange_access_token": values.Value(
default="30/minute",
environ_name="EXCHANGE_ACCESS_TOKEN_THROTTLE_RATES",
environ_prefix=None,
),
"creation_callback": values.Value( "creation_callback": values.Value(
default="600/minute", default="600/minute",
environ_name="CREATION_CALLBACK_THROTTLE_RATES", environ_name="CREATION_CALLBACK_THROTTLE_RATES",
@@ -1011,6 +1017,61 @@ class Base(Configuration):
environ_name="APPLICATION_BASE_URL", environ_name="APPLICATION_BASE_URL",
environ_prefix=None, environ_prefix=None,
) )
# User access tokens (embedded frontend / iframe support)
USER_ACCESS_TOKEN_ENABLED = values.BooleanValue(
False, environ_name="USER_ACCESS_TOKEN_ENABLED", environ_prefix=None
)
USER_ACCESS_TOKEN_SECRET_KEY = SecretFileValue(
None, environ_name="USER_ACCESS_TOKEN_SECRET_KEY", environ_prefix=None
)
USER_ACCESS_TOKEN_ALG = values.Value(
"HS256",
environ_name="USER_ACCESS_TOKEN_ALG",
environ_prefix=None,
)
USER_ACCESS_TOKEN_ISSUER = values.Value(
"lasuite-meet",
environ_name="USER_ACCESS_TOKEN_ISSUER",
environ_prefix=None,
)
USER_ACCESS_TOKEN_AUDIENCE = values.Value(
None,
environ_name="USER_ACCESS_TOKEN_AUDIENCE",
environ_prefix=None,
)
# Lifetime of the user access token obtained through the exchange
# endpoint. It never transits through a URL, so it can cover a full
# meeting (default: 2 hours).
USER_ACCESS_TOKEN_TTL = values.PositiveIntegerValue(
7200,
environ_name="USER_ACCESS_TOKEN_TTL",
environ_prefix=None,
)
# Lifetime of the single-use transit code handed to the frontend
# through a URL fragment. Kept very short by design: it must only
# survive the redirect and the exchange call.
TRANSIT_CODE_TTL = values.PositiveIntegerValue(
60,
environ_name="TRANSIT_CODE_TTL",
environ_prefix=None,
)
TRANSIT_CODE_CACHE_PREFIX = values.Value(
"transit-code",
environ_name="TRANSIT_CODE_CACHE_PREFIX",
environ_prefix=None,
)
# Number of random bytes per code (48 bytes -> 64 url-safe characters)
TRANSIT_CODE_NBYTES = values.PositiveIntegerValue(
48,
environ_name="TRANSIT_CODE_NBYTES",
environ_prefix=None,
)
USER_ACCESS_TOKEN_TYPE = values.Value(
"Bearer",
environ_name="USER_ACCESS_TOKEN_TYPE",
environ_prefix=None,
)
# Warning: EXTERNAL_API_ALLOW_PUBLIC_ACCESS is ignored when # Warning: EXTERNAL_API_ALLOW_PUBLIC_ACCESS is ignored when
# EXTERNAL_API_DEFAULT_ACCESS_LEVEL=public. # EXTERNAL_API_DEFAULT_ACCESS_LEVEL=public.
EXTERNAL_API_ALLOW_PUBLIC_ACCESS = values.BooleanValue( EXTERNAL_API_ALLOW_PUBLIC_ACCESS = values.BooleanValue(
@@ -1307,6 +1368,9 @@ class Test(Base):
ADDONS_ENABLED = True ADDONS_ENABLED = True
ADDONS_CSRF_SECRET = "secret-key-padded-for-minimum-len!-addons" # noqa:S105 ADDONS_CSRF_SECRET = "secret-key-padded-for-minimum-len!-addons" # noqa:S105
ADDONS_TOKEN_SECRET_KEY = "secret-key-padded-for-minimum-len!-addons" # noqa:S105 ADDONS_TOKEN_SECRET_KEY = "secret-key-padded-for-minimum-len!-addons" # noqa:S105
USER_ACCESS_TOKEN_ENABLED = True
USER_ACCESS_TOKEN_SECRET_KEY = "secret-key-padded-for-minimum-len!-room" # noqa:S105
USER_ACCESS_TOKEN_AUDIENCE = "Test inc." # noqa:S105
CONNECTION_TEST_ENABLED = True CONNECTION_TEST_ENABLED = True