💄(frontend) adjust centering of Avatar initials

Fine-tune the vertical alignment of the initials in the Avatar so they sit properly centered inside the circle.
2026-08-01 14:42:15 +00:00 · 2026-07-30 15:16:29 +02:00
25 changed files with 2 additions and 1369 deletions
@@ -389,12 +389,6 @@ 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
@@ -1,11 +1,5 @@
 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')
@@ -1,182 +0,0 @@
 #!/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
@@ -143,24 +143,3 @@ $ make start-tilt-keycloak
 ```
 Monitor Tilt’s 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).
@@ -16,7 +16,6 @@ class FeatureFlag:
        "file_upload": "FILE_UPLOAD_ENABLED",
        "addons": "ADDONS_ENABLED",
        "application": "APPLICATION_ENABLED",
        "user_access_token": "USER_ACCESS_TOKEN_ENABLED",
    }
    @classmethod
@@ -580,25 +580,3 @@ 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
@@ -73,14 +73,3 @@ 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"
@@ -69,7 +69,6 @@ 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,
@@ -94,7 +93,6 @@ 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
@@ -231,76 +229,6 @@ 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,
@@ -1,71 +0,0 @@
 """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,14 +86,6 @@ 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."""
@@ -22,7 +22,6 @@ 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,
@@ -219,62 +218,3 @@ 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,
        )
@@ -1,19 +0,0 @@
 # 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),
        ),
    ]
@@ -769,7 +769,6 @@ 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):
@@ -1,74 +0,0 @@
 """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,14 +2,9 @@
 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
@@ -114,38 +109,3 @@ 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,12 +2,8 @@
 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
@@ -160,40 +156,3 @@ 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,14 +3,11 @@ 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
@@ -506,40 +503,3 @@ 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,12 +3,8 @@ 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
@@ -441,45 +437,3 @@ 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"
@@ -1,46 +0,0 @@
 """
 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
@@ -1,200 +0,0 @@
 """
 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
@@ -1,165 +0,0 @@
 """
 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
@@ -1,166 +0,0 @@
 """
 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
@@ -37,11 +37,6 @@ external_router.register(
    external_viewsets.RoomViewSet,
    basename="external_room",
 )
 external_router.register(
    "users",
    external_viewsets.UserViewSet,
    basename="external_user",
 )
 urlpatterns = [
    path(
@@ -324,7 +324,6 @@ class Base(Configuration):
    REST_FRAMEWORK = {
        "DEFAULT_AUTHENTICATION_CLASSES": (
            "core.authentication.user_token.UserAccessJWTAuthentication",
            "core.authentication.backends.SessionAuthenticationWith401",
        ),
        "DEFAULT_PARSER_CLASSES": [
@@ -345,11 +344,6 @@ 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",
@@ -959,61 +953,6 @@ 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(
@@ -1311,10 +1250,6 @@ 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"]
@@ -59,9 +59,10 @@ export const Avatar = React.memo(
          <text
            x="50"
            y="50"
            dy="-0.08em"
            textAnchor="middle"
            dominantBaseline="central"
-            fontSize={initials.length > 1 ? 48 : 52}
+            fontSize="52"
            fontWeight="500"
            fill="currentColor"
          >