Compare commits

..

2 Commits

Author SHA1 Message Date
lebaudantoine e701a89036 (backend) wip introduce a token exchange endpoint 2026-08-01 15:52:37 +02:00
lebaudantoine 7fbcbc89ed 🧑‍💻(devx) prototype OrbStack as local Kubernetes provider on macOS
Add a prototype setup that uses OrbStack as the local Kubernetes
provider on macOS, aiming to save a few GB of RAM compared to the
current stack.

The script has been tested locally but was mostly built through vibe
coding, so it has not been thoroughly reviewed yet.

Follow-ups to look into:

* Simplify the ingress and CoreDNS setup if possible.
* Wire up a way to run tests and lint against the Tilt stack.
2026-07-31 14:18:37 +02:00
27 changed files with 1374 additions and 7 deletions
+6
View File
@@ -389,6 +389,12 @@ build-k8s-cluster: \
./bin/start-kind.sh
.PHONY: build-k8s-cluster
build-k8s-cluster-orbstack: ## setup the kubernetes environment on OrbStack's built-in cluster (macOS)
build-k8s-cluster-orbstack: \
env.d/development/kube-secret
./bin/start-orbstack.sh
.PHONY: build-k8s-cluster-orbstack
start-tilt-keycloak: ## start the kubernetes cluster using kind, without Pro Connect for authentication, use keycloak
DEV_ENV=dev-keycloak tilt up --namespace=meet -f ./bin/Tiltfile
.PHONY: build-k8s-cluster
+6
View File
@@ -1,5 +1,11 @@
load('ext://uibutton', 'cmd_button', 'bool_input', 'location')
load('ext://namespace', 'namespace_create', 'namespace_inject')
# OrbStack's built-in cluster (macOS) is a supported alternative to kind.
# Recent Tilt versions (>= 0.33) detect it as a local dev cluster; this is
# a no-op for kind and a safety net for older Tilt versions.
allow_k8s_contexts('orbstack')
namespace_create('meet')
DEV_ENV = os.getenv('DEV_ENV', 'dev-keycloak')
+182
View File
@@ -0,0 +1,182 @@
#!/usr/bin/env bash
#
# Bootstrap the local dev environment on OrbStack's built-in Kubernetes
# cluster (macOS) instead of kind.
#
# This replicates what bin/start-kind.sh (numerique-gouv/tools
# kind/create_cluster.sh) provides, minus what OrbStack makes unnecessary:
# - no kind cluster: OrbStack ships a lightweight single-node cluster
# - no local registry (kind-registry): OrbStack's cluster shares the
# Docker image store, so images built by Tilt are directly visible
# to pods. Tilt detects the "orbstack" context as a local cluster
# and skips pushing images entirely.
#
# Requirements: OrbStack (with Kubernetes enabled), kubectl, mkcert, curl.
set -o errexit
APPLICATION=${1:-meet}
CONTEXT="orbstack"
echo "0. Check OrbStack Kubernetes is available"
if ! command -v mkcert >/dev/null 2>&1; then
echo "❌ mkcert is not installed. Install it first: brew install mkcert"
exit 1
fi
if ! kubectl config get-contexts -o name | grep -qx "${CONTEXT}"; then
echo "Context '${CONTEXT}' not found. Trying to start OrbStack Kubernetes..."
if command -v orb >/dev/null 2>&1; then
orb start k8s
else
echo "❌ Enable Kubernetes in OrbStack (Settings > Kubernetes) and retry."
exit 1
fi
fi
kubectl config use-context "${CONTEXT}"
echo "0b. Check ports 80/443 are free on localhost"
# OrbStack forwards LoadBalancer service ports to 127.0.0.1. If the kind
# cluster is still running, its docker proxy already holds 80/443.
# Skip the check if ingress-nginx is already installed here: in that case
# the listener on 80/443 is our own LoadBalancer.
if ! kubectl -n ingress-nginx get deployment ingress-nginx-controller >/dev/null 2>&1; then
for port in 80 443; do
if lsof -nP -iTCP:"${port}" -sTCP:LISTEN >/dev/null 2>&1; then
echo "❌ Port ${port} is already in use on the host."
echo " If the kind cluster is running, delete it first:"
echo " kind delete cluster --name suite"
exit 1
fi
done
fi
echo "1. Create ca"
CURRENT_DIR=$(pwd)
mkcert -install
cd /tmp
mkcert "127.0.0.1.nip.io" "*.127.0.0.1.nip.io"
cd "${CURRENT_DIR}"
echo "2. Install ingress-nginx (cloud provider: LoadBalancer service)"
# OrbStack exposes LoadBalancer services on 127.0.0.1, so the cloud
# manifest replaces kind's hostPort-based deploy. Every sub-step below is
# guarded individually so the script is safe to re-run after a partial
# failure (unlike the upstream kind script, which guards the whole block
# on namespace existence).
# Make sure no stale registry configmap tells Tilt to push to localhost:5001
# (there is no registry on OrbStack).
kubectl -n kube-public delete configmap local-registry-hosting --ignore-not-found
if ! kubectl -n ingress-nginx get deployment ingress-nginx-controller >/dev/null 2>&1; then
kubectl apply -f https://raw.githubusercontent.com/kubernetes/ingress-nginx/main/deploy/static/provider/cloud/deploy.yaml
fi
if ! kubectl -n ingress-nginx get deployment nginx-errors >/dev/null 2>&1; then
kubectl apply -n ingress-nginx -f https://raw.githubusercontent.com/kubernetes/ingress-nginx/refs/heads/main/docs/examples/customization/custom-errors/custom-default-backend.yaml
fi
kubectl -n ingress-nginx create secret tls mkcert --key /tmp/127.0.0.1.nip.io+1-key.pem --cert /tmp/127.0.0.1.nip.io+1.pem || echo ok
# The meet charts render Ingresses without ingressClassName. The kind
# provider manifest handles this via --watch-ingress-without-class=true;
# the cloud manifest does not, so add it here (otherwise: 404 everywhere).
if ! kubectl -n ingress-nginx get deployment ingress-nginx-controller -o jsonpath='{.spec.template.spec.containers[0].args}' | grep -q 'watch-ingress-without-class'; then
kubectl -n ingress-nginx patch deployments.apps ingress-nginx-controller --type 'json' -p '[{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value":"--watch-ingress-without-class=true"},{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value":"--default-ssl-certificate=ingress-nginx/mkcert"},{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value":"--default-backend-service=ingress-nginx/nginx-errors"}
]'
fi
if ! kubectl -n ingress-nginx get deployment nginx-errors -o jsonpath='{.spec.template.spec.containers[0].image}' | grep -q 'error-pages'; then
kubectl -n ingress-nginx patch deployment nginx-errors --type=json -p='[
{"op": "replace", "path": "/spec/template/spec/containers/0/image", "value": "ghcr.io/tarampampam/error-pages:3.3.0"},
{"op": "add", "path": "/spec/template/spec/containers/0/env", "value": [{"name": "TEMPLATE_NAME", "value": "ghost"}, {"name": "SHOW_DETAILS", "value": "false"}, {"name": "SEND_SAME_HTTP_CODE", "value": "true"}]}
]'
fi
cat <<EOF | kubectl apply -n ingress-nginx -f -
apiVersion: v1
data:
allow-snippet-annotations: "true"
annotations-risk-level: Critical
custom-http-errors: 500,501,502,503,504
kind: ConfigMap
metadata:
name: ingress-nginx-controller
namespace: ingress-nginx
EOF
echo "2b. Wait for the ingress controller to be ready"
kubectl -n ingress-nginx rollout status deployment/ingress-nginx-controller --timeout=180s
echo "3. Patch CoreDNS so in-cluster pods resolve *.127.0.0.1.nip.io to the ingress"
# nip.io resolves to 127.0.0.1, which inside a pod is the pod itself.
# Rewrite these names to the ingress-nginx service, like the kind setup does.
# Unlike kind, we amend OrbStack's existing Corefile instead of replacing it.
if ! kubectl -n kube-system get configmap coredns -o jsonpath='{.data.Corefile}' | grep -q '127\.0\.0\.1\.nip\.io'; then
kubectl -n kube-system get configmap coredns -o jsonpath='{.data.Corefile}' \
| awk '/forward \./ && !done { print " rewrite stop {"; print " name regex (.*).127.0.0.1.nip.io ingress-nginx-controller.ingress-nginx.svc.cluster.local answer auto"; print " }"; done=1 } { print }' \
>/tmp/Corefile.orbstack
kubectl -n kube-system create configmap coredns --from-file=Corefile=/tmp/Corefile.orbstack --dry-run=client -o yaml | kubectl apply -f -
kubectl -n kube-system rollout restart deployments/coredns
fi
if ! kubectl get ns "${APPLICATION}" >/dev/null 2>&1; then
echo "4. Setup namespace"
kubectl create ns "${APPLICATION}"
fi
kubectl config set-context --current --namespace="${APPLICATION}"
kubectl -n "${APPLICATION}" create secret generic mkcert --from-file=rootCA.pem="$(mkcert -CAROOT)/rootCA.pem" || echo ok
if ! kubectl get configmap certifi -n "${APPLICATION}" >/dev/null 2>&1; then
echo "5. Inject our custom CA in a configmap for certifi"
curl https://raw.githubusercontent.com/certifi/python-certifi/refs/heads/master/certifi/cacert.pem -o /tmp/cacert.pem
cat "$(mkcert -CAROOT)/rootCA.pem" >>/tmp/cacert.pem
kubectl -n "${APPLICATION}" create configmap certifi --from-file=cacert.pem=/tmp/cacert.pem
kubectl -n "${APPLICATION}" create secret generic certifi --from-file=/tmp/cacert.pem || echo ok
fi
echo "5b. Smoke test: the ingress chain answers on https://127.0.0.1"
# Before Tilt deploys the app this returns the styled 404 from the default
# backend — that still proves LB -> controller works. 000 means the
# LoadBalancer is not bound to localhost.
HTTP_CODE=$(curl -sk -o /dev/null -w '%{http_code}' --max-time 10 https://127.0.0.1/ || true)
if [ "${HTTP_CODE}" = "000" ]; then
echo "⚠️ Nothing answered on https://127.0.0.1 — check the LoadBalancer:"
echo " kubectl -n ingress-nginx get svc ingress-nginx-controller"
else
echo "✅ Ingress reachable (HTTP ${HTTP_CODE})"
fi
echo "6. Check pod readiness across all namespaces..."
sleep_interval=10
echo "Initial wait time: $((sleep_interval * 2)) seconds…"
sleep $((sleep_interval * 2))
check_pods_ready() {
local max_attempts=60 # Maximum number of attempts (10 minutes with 10s intervals)
local attempt=1
while [ $attempt -le $max_attempts ]; do
echo "Attempt $attempt/$max_attempts - Checking pod status..."
not_ready_count=$( kubectl get po -A --no-headers | grep -v -E "Running|Completed"| wc -l | tr -d ' ')
if [ "$not_ready_count" -eq 0 ]; then
echo "✅ All pods are ready!"
return 0
else
echo "$not_ready_count pod(s) still not ready. Waiting $sleep_interval seconds…"
sleep $sleep_interval
((attempt++))
fi
done
echo "❌ Timeout: Some pods are still not ready after 10 minutes"
echo "Final pod status:"
kubectl get po -A
return 1
}
if check_pods_ready; then
echo "🎉 Cluster is fully ready!"
else
echo "⚠️ Some pods may need manual intervention"
exit 1
fi
+21
View File
@@ -143,3 +143,24 @@ $ make start-tilt-keycloak
```
Monitor Tilts progress at [http://localhost:10350/](http://localhost:10350/). After Tilt actions finish, you can access the app at [https://meet.127.0.0.1.nip.io/](https://meet.127.0.0.1.nip.io/).
### Alternative: OrbStack's built-in Kubernetes (macOS)
If you use [OrbStack](https://orbstack.dev/) on macOS, you can run the stack on its built-in Kubernetes cluster instead of kind. It uses noticeably less RAM (no nested kubeadm node container) and no local registry is needed: OrbStack's cluster shares the Docker image store, so Tilt uses images directly without pushing.
Enable Kubernetes in OrbStack (Settings > Kubernetes), then:
```shellscript
$ make build-k8s-cluster-orbstack
```
This installs ingress-nginx (exposed by OrbStack on `127.0.0.1:80/443`), the mkcert TLS certificates, and the CoreDNS rewrite for `*.127.0.0.1.nip.io`, then you start Tilt as usual:
```shellscript
$ make start-tilt-keycloak
```
Notes:
- Ports 80/443 must be free: delete the kind cluster first if you used it (`kind delete cluster --name suite`).
- If you "Reset Kubernetes" in OrbStack, re-run `make build-k8s-cluster-orbstack`.
- kind remains the reference setup (matches CI and lets you pin the Kubernetes version).
+1
View File
@@ -16,6 +16,7 @@ class FeatureFlag:
"file_upload": "FILE_UPLOAD_ENABLED",
"addons": "ADDONS_ENABLED",
"application": "APPLICATION_ENABLED",
"user_access_token": "USER_ACCESS_TOKEN_ENABLED",
}
@classmethod
+22
View File
@@ -580,3 +580,25 @@ class ExternalProcessEventSerializer(BaseValidationOnlySerializer):
# useless bad requests
type = 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."""
# todo if I can pass the max length directly to the char field
code = serializers.CharField(max_length=255, trim_whitespace=True)
def validate_code(self, value):
"""Reject codes whose length cannot match a generated one.
`secrets.token_urlsafe(nbytes)` produces (4 * nbytes + 2) // 3
url-safe characters. Checking the length against the configured
TRANSIT_CODE_NBYTES makes malformed codes fail fast with a 400,
before any cache lookup.
"""
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
@@ -73,3 +73,14 @@ class CreationCallbackAnonRateThrottle(MonitoredAnonRateThrottle):
"""Throttle Anonymous user requesting room generation callback"""
scope = "creation_callback"
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"
+72
View File
@@ -69,6 +69,7 @@ from core.recording.worker.mediator import (
WorkerServiceMediator,
)
from core.services.invitation import InvitationService
from core.services.jwt_token import JwtTokenService
from core.services.livekit_events import (
LiveKitEventsService,
LiveKitWebhookError,
@@ -93,6 +94,7 @@ from core.services.room_roles import (
RoomRoleService,
)
from core.services.subtitle import SubtitleException, SubtitleService
from core.services.transit_code import TransitCodeService
from core.tasks.file import process_file_deletion
from ..authentication.livekit import LiveKitTokenAuthentication
@@ -229,6 +231,76 @@ class UserViewSet(
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.
"""
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
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,
)
# todo - discuss wether it's the relevant scope
data = token_service.generate_jwt(
user,
"user:access",
{
"token_type": "user_access",
"client_id": code_data.get("client_id", "unknown"),
},
)
# 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(
mixins.CreateModelMixin,
@@ -0,0 +1,71 @@
"""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
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.
"""
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.")
# Every token we issue carries the client_id of the application the
# transit code was minted for: its absence means the token does not
# come from the exchange endpoint.
if not payload.get("client_id"):
logger.warning("Missing 'client_id' in user access token payload")
raise exceptions.AuthenticationFailed("Invalid token claims.")
@@ -86,6 +86,14 @@ class HasRequiredRoomScope(BaseScopePermission):
}
class HasRequiredUserScope(BaseScopePermission):
"""Scope-based permissions for the external user endpoints."""
scope_map = {
"generate_transit_code": models.ApplicationScope.USERS_SESSION,
}
class RoomPermissions(permissions.BasePermission):
"""Permissions applying to the room API endpoint."""
+60
View File
@@ -22,6 +22,7 @@ from rest_framework import (
from core import analytics, api, models
from core.api.feature_flag import FeatureFlag
from core.services.jwt_token import JwtTokenService
from core.services.transit_code import TransitCodeService
from ..services.provisional_user_service import (
ProvisionalUserCreationDisabledError,
@@ -218,3 +219,62 @@ class RoomViewSet(
"$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,
ResourceServerAuthentication,
]
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.
"""
auth_method = type(request.successful_authenticator).__name__
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, auth_method=%s",
request.user.id,
client_id,
auth_method,
)
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),
),
]
+1
View File
@@ -769,6 +769,7 @@ class ApplicationScope(models.TextChoices):
ROOMS_RETRIEVE = "rooms:retrieve", _("Retrieve room details")
ROOMS_UPDATE = "rooms:update", _("Update rooms")
ROOMS_DELETE = "rooms:delete", _("Delete rooms")
USERS_SESSION = "users:session", _("Create user session tokens")
class Application(BaseModel):
+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,9 +2,14 @@
Test rooms API endpoints in the Meet core app: create.
"""
from datetime import datetime, timedelta, timezone
from django.conf import settings as django_settings
# pylint: disable=redefined-outer-name,unused-argument
from django.core.cache import cache
import jwt
import pytest
from rest_framework.test import APIClient
@@ -109,3 +114,38 @@ def test_api_rooms_create_authenticated_existing_slug():
assert response.status_code == 400
assert response.json() == {"slug": ["Room with this Slug already exists."]}
def generate_user_access_token(user):
"""Generate a valid user access JWT signed with the token secret."""
now = datetime.now(timezone.utc)
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": "test-app",
"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_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,8 +2,12 @@
Test rooms API endpoints in the Meet core app: list.
"""
from datetime import datetime, timedelta, timezone
from unittest import mock
from django.conf import settings as django_settings
import jwt
import pytest
from rest_framework.pagination import PageNumberPagination
from rest_framework.test import APIClient
@@ -156,3 +160,40 @@ def test_api_rooms_list_pagination_page_size():
assert len(content["results"]) == 3
assert content["next"] == "http://testserver/api/v1.0/rooms/?page=2&page_size=3"
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)
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": "test-app",
"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)
@@ -3,11 +3,14 @@ Test rooms API endpoints in the Meet core app: retrieve.
"""
import random
from datetime import datetime, timedelta, timezone
from unittest import mock
from django.conf import settings as django_settings
from django.contrib.auth.models import AnonymousUser
from django.test.utils import override_settings
import jwt
import pytest
from rest_framework.test import APIClient
@@ -503,3 +506,40 @@ def test_api_rooms_retrieve_administrators(
role=str(user_access.role),
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)
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": "test-app",
"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)
# Authenticated as the owner: privileged fields are included
assert response.data["pin_code"] == room.pin_code
@@ -3,8 +3,12 @@ Test rooms API endpoints in the Meet core app: update.
"""
import random
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from django.conf import settings as django_settings
import jwt
import pytest
from rest_framework.test import APIClient
@@ -437,3 +441,45 @@ def test_api_rooms_update_livekit_sync_failure(mock_update_metadata):
"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)
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": "test-app",
"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,46 @@
"""
Unit tests for the TransitCodeService.
"""
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
@@ -0,0 +1,200 @@
"""
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 RoomFactory, UserFactory
from core.models import RoleChoices
pytestmark = pytest.mark.django_db
def generate_user_access_token(user, **overrides):
"""Generate a valid user access JWT signed with the token secret."""
now = datetime.now(timezone.utc)
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": "test-app",
"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_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_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_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
@@ -0,0 +1,165 @@
"""
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 UserFactory
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_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()
code = TransitCodeService().create_code(user, client_id="my-app")
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["token_type"] == "user_access"
assert payload["user_id"] == str(user.id)
assert payload["client_id"] == "my-app"
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()
code = TransitCodeService().create_code(user)
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()
code = TransitCodeService().create_code(user)
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()
code = TransitCodeService().create_code(user)
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
@@ -0,0 +1,166 @@
"""
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_test_token(user, scopes):
"""Generate a valid application JWT token for testing."""
now = datetime.now(timezone.utc)
scope_string = " ".join(scopes)
application = ApplicationFactory()
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_with_rs_token():
"""A resource-server-authenticated user should be able to mint a code."""
user = UserFactory()
# todo - add a decorator instead
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/")
mock_rs_authenticate.assert_called_once()
assert response.status_code == 200
code_data = TransitCodeService().consume_code(response.data["transit_code"])
assert code_data == {
"user_id": str(user.id),
"client_id": "rs-client",
}
def test_api_users_transit_code_with_rs_token_missing_scope():
"""A resource server token without the scope should be rejected."""
user = UserFactory()
# todo - add a decorator instead
with mock.patch.object(
ResourceServerAuthentication,
"authenticate",
return_value=(user, {"scope": "rooms:list", "client_id": "rs-client"}),
):
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 == 403
assert "users:session" in str(response.data)
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
# todo - add a test to make sure the addon authentification doesn't allow to mint a transit token
+5
View File
@@ -37,6 +37,11 @@ external_router.register(
external_viewsets.RoomViewSet,
basename="external_room",
)
external_router.register(
"users",
external_viewsets.UserViewSet,
basename="external_user",
)
urlpatterns = [
path(
+65
View File
@@ -324,6 +324,7 @@ class Base(Configuration):
REST_FRAMEWORK = {
"DEFAULT_AUTHENTICATION_CLASSES": (
"core.authentication.user_token.UserAccessJWTAuthentication",
"core.authentication.backends.SessionAuthenticationWith401",
),
"DEFAULT_PARSER_CLASSES": [
@@ -344,6 +345,11 @@ class Base(Configuration):
environ_name="REQUEST_ENTRY_THROTTLE_RATES",
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(
default="600/minute",
environ_name="CREATION_CALLBACK_THROTTLE_RATES",
@@ -953,6 +959,61 @@ class Base(Configuration):
environ_name="APPLICATION_BASE_URL",
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
# EXTERNAL_API_DEFAULT_ACCESS_LEVEL=public.
EXTERNAL_API_ALLOW_PUBLIC_ACCESS = values.BooleanValue(
@@ -1250,6 +1311,10 @@ class Test(Base):
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
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
def __init__(self):
# pylint: disable=invalid-name
self.INSTALLED_APPS += ["drf_spectacular_sidecar"]
+1 -1
View File
@@ -40,7 +40,7 @@ dependencies = [
"django-storages[s3]==1.14.6",
"django-timezone-field>=5.1",
"django-pydantic-field==0.5.4",
"django==5.2.16",
"django==5.2.14",
"djangorestframework==3.17.1",
"drf_spectacular==0.29.0",
"dockerflow==2026.3.4",
+4 -4
View File
@@ -586,16 +586,16 @@ wheels = [
[[package]]
name = "django"
version = "5.2.16"
version = "5.2.14"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "asgiref" },
{ name = "sqlparse" },
{ name = "tzdata", marker = "sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a9/26/889449d521ae508b26de715954faecd8bcf3f740affb81b2d146a83b42a5/django-5.2.16.tar.gz", hash = "sha256:59ea02020c3136fce14bef0bbece21a10a4febef5eed1c51c22ae468efa22200", size = 10890894, upload-time = "2026-07-07T13:52:17.005Z" }
sdist = { url = "https://files.pythonhosted.org/packages/65/95/95f7faa0950867afaa0bef2460c6263afd6a2c78cc9434046ed28160b015/django-5.2.14.tar.gz", hash = "sha256:58a63ba841662e5c686b57ba1fec52ddd68c0b93bd96ac3029d55728f00bf8a2", size = 10895118, upload-time = "2026-05-05T13:57:31.104Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/4e/13/1e5e3e4c15dcecb04281b3cb2a46a4670e1cef131068e202f6040df19224/django-5.2.16-py3-none-any.whl", hash = "sha256:04f354bf9d807a86ad1a8392fe3808d362358a8eafc322848e0e43e59b24371d", size = 8311943, upload-time = "2026-07-07T13:52:11.223Z" },
{ url = "https://files.pythonhosted.org/packages/14/44/f172870cf87aa25afef48fb72adba89ee8b77fcab6f3b23d240b923f1528/django-5.2.14-py3-none-any.whl", hash = "sha256:6f712143bd3064310d1f50fac859c3e9a274bdcfc9595339853be7779297fc76", size = 8311320, upload-time = "2026-05-05T13:57:25.795Z" },
]
[[package]]
@@ -1260,7 +1260,7 @@ requires-dist = [
{ name = "brotli", specifier = "==1.2.0" },
{ name = "celery", extras = ["redis"], specifier = "==5.6.3" },
{ name = "dj-database-url", specifier = "==3.1.2" },
{ name = "django", specifier = "==5.2.16" },
{ name = "django", specifier = "==5.2.14" },
{ name = "django-configurations", specifier = "==2.5.1" },
{ name = "django-cors-headers", specifier = "==4.9.0" },
{ name = "django-countries", specifier = "==9.0.0" },
+1 -2
View File
@@ -59,10 +59,9 @@ export const Avatar = React.memo(
<text
x="50"
y="50"
dy="-0.08em"
textAnchor="middle"
dominantBaseline="central"
fontSize="52"
fontSize={initials.length > 1 ? 48 : 52}
fontWeight="500"
fill="currentColor"
>