✨(backend) draft initial Room viewset for external applications
From a security perspective, the list endpoint should be limited to return only rooms created by the external application. Currently, there is a risk of exposing public rooms through this endpoint. I will address this in upcoming commits by updating the room model to track the source of generation. This will also provide useful information for analytics. The API viewset was largely copied and adapted. The serializer was heavily restricted to return a response more appropriate for external applications, providing ready-to-use information for their users (for example, a clickable link). I plan to extend the room information further, potentially aligning it with the Google Meet API format. This first draft serves as a solid foundation. Although scopes for delete and update exist, these methods have not yet been implemented in the viewset. They will be added in future commits.
This commit is contained in:
lebaudantoine
committed by
aleb_the_flash
aleb_the_flash
parent
b8c3c3df3a
commit
c9fcc2ed60
@@ -69,3 +69,4 @@ FRONTEND_USE_PROCONNECT_BUTTON=False
|
||||
# External Applications
|
||||
APPLICATION_JWT_AUDIENCE=http://localhost:8071/external-api/v1.0/
|
||||
APPLICATION_JWT_SECRET_KEY=devKey
|
||||
APPLICATION_BASE_URL=http://localhost:3000
|
||||
|
||||
@@ -2,8 +2,11 @@
|
||||
|
||||
# pylint: disable=abstract-method
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
from rest_framework import serializers
|
||||
|
||||
from core import models, utils
|
||||
from core.api.serializers import BaseValidationOnlySerializer
|
||||
|
||||
OAUTH2_GRANT_TYPE_CLIENT_CREDENTIALS = "client_credentials"
|
||||
@@ -16,3 +19,55 @@ class ApplicationJwtSerializer(BaseValidationOnlySerializer):
|
||||
client_secret = serializers.CharField(write_only=True)
|
||||
grant_type = serializers.ChoiceField(choices=[OAUTH2_GRANT_TYPE_CLIENT_CREDENTIALS])
|
||||
scope = serializers.CharField(write_only=True)
|
||||
|
||||
|
||||
class RoomSerializer(serializers.ModelSerializer):
|
||||
"""External API serializer for room data exposed to applications.
|
||||
|
||||
Provides limited, safe room information for third-party integrations:
|
||||
- Secure defaults for room creation (trusted access level)
|
||||
- Computed fields (url, telephony) for external consumption
|
||||
- Filtered data appropriate for delegation scenarios
|
||||
- Tracks creation source for auditing
|
||||
|
||||
Intentionally exposes minimal information to external applications,
|
||||
following the principle of least privilege.
|
||||
"""
|
||||
|
||||
class Meta:
|
||||
model = models.Room
|
||||
fields = ["id", "name", "slug", "pin_code", "access_level"]
|
||||
read_only_fields = ["id", "name", "slug", "pin_code", "access_level"]
|
||||
|
||||
def to_representation(self, instance):
|
||||
"""Enrich response with application-specific computed fields."""
|
||||
output = super().to_representation(instance)
|
||||
request = self.context.get("request")
|
||||
pin_code = output.pop("pin_code", None)
|
||||
|
||||
if not request:
|
||||
return output
|
||||
|
||||
# Add room URL for direct access
|
||||
if settings.APPLICATION_BASE_URL:
|
||||
output["url"] = f"{settings.APPLICATION_BASE_URL}/{instance.slug}"
|
||||
|
||||
# Add telephony information if enabled
|
||||
if settings.ROOM_TELEPHONY_ENABLED:
|
||||
output["telephony"] = {
|
||||
"enabled": True,
|
||||
"phone_number": settings.ROOM_TELEPHONY_PHONE_NUMBER,
|
||||
"pin_code": pin_code,
|
||||
"default_country": settings.ROOM_TELEPHONY_DEFAULT_COUNTRY,
|
||||
}
|
||||
|
||||
return output
|
||||
|
||||
def create(self, validated_data):
|
||||
"""Create room with secure defaults for application delegation."""
|
||||
|
||||
# Set secure defaults
|
||||
validated_data["name"] = utils.generate_room_slug()
|
||||
validated_data["access_level"] = models.RoomAccessLevel.TRUSTED
|
||||
|
||||
return super().create(validated_data)
|
||||
|
||||
@@ -9,7 +9,7 @@ from django.core.exceptions import ValidationError
|
||||
from django.core.validators import validate_email
|
||||
|
||||
import jwt
|
||||
from rest_framework import decorators, viewsets
|
||||
from rest_framework import decorators, mixins, viewsets
|
||||
from rest_framework import (
|
||||
exceptions as drf_exceptions,
|
||||
)
|
||||
@@ -20,9 +20,9 @@ from rest_framework import (
|
||||
status as drf_status,
|
||||
)
|
||||
|
||||
from core import models
|
||||
from core import api, models
|
||||
|
||||
from . import serializers
|
||||
from . import authentication, permissions, serializers
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
@@ -129,3 +129,66 @@ class ApplicationViewSet(viewsets.GenericViewSet):
|
||||
},
|
||||
status=drf_status.HTTP_200_OK,
|
||||
)
|
||||
|
||||
|
||||
class RoomViewSet(
|
||||
mixins.CreateModelMixin,
|
||||
mixins.RetrieveModelMixin,
|
||||
mixins.ListModelMixin,
|
||||
viewsets.GenericViewSet,
|
||||
):
|
||||
"""Application-delegated API for room management.
|
||||
|
||||
Provides JWT-authenticated access to room operations for external applications
|
||||
acting on behalf of users. All operations are scope-based and filtered to the
|
||||
authenticated user's accessible rooms.
|
||||
|
||||
Supported operations:
|
||||
- list: List rooms the user has access to (requires 'rooms:list' scope)
|
||||
- retrieve: Get room details (requires 'rooms:retrieve' scope)
|
||||
- create: Create a new room owned by the user (requires 'rooms:create' scope)
|
||||
"""
|
||||
|
||||
authentication_classes = [authentication.ApplicationJWTAuthentication]
|
||||
permission_classes = [
|
||||
api.permissions.IsAuthenticated & permissions.HasRequiredRoomScope
|
||||
]
|
||||
queryset = models.Room.objects.all()
|
||||
serializer_class = serializers.RoomSerializer
|
||||
|
||||
def list(self, request, *args, **kwargs):
|
||||
"""Limit listed rooms to the ones related to the authenticated user."""
|
||||
|
||||
user = self.request.user
|
||||
|
||||
if user.is_authenticated:
|
||||
queryset = (
|
||||
self.filter_queryset(self.get_queryset()).filter(users=user).distinct()
|
||||
)
|
||||
else:
|
||||
queryset = self.get_queryset().none()
|
||||
|
||||
page = self.paginate_queryset(queryset)
|
||||
if page is not None:
|
||||
serializer = self.get_serializer(page, many=True)
|
||||
return self.get_paginated_response(serializer.data)
|
||||
|
||||
serializer = self.get_serializer(queryset, many=True)
|
||||
return drf_response.Response(serializer.data)
|
||||
|
||||
def perform_create(self, serializer):
|
||||
"""Set the current user as owner of the newly created room."""
|
||||
room = serializer.save()
|
||||
models.ResourceAccess.objects.create(
|
||||
resource=room,
|
||||
user=self.request.user,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
# Log for auditing
|
||||
logger.info(
|
||||
"Room created via application: room_id=%s, user_id=%s, client_id=%s",
|
||||
room.id,
|
||||
self.request.user.id,
|
||||
getattr(self.request.auth, "client_id", "unknown"),
|
||||
)
|
||||
|
||||
334
src/backend/core/tests/test_external_api_rooms.py
Normal file
334
src/backend/core/tests/test_external_api_rooms.py
Normal file
@@ -0,0 +1,334 @@
|
||||
"""
|
||||
Tests for external API /room endpoint
|
||||
"""
|
||||
|
||||
# pylint: disable=W0621
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
import jwt
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core.factories import (
|
||||
RoomFactory,
|
||||
UserFactory,
|
||||
)
|
||||
from core.models import ApplicationScope, RoleChoices, Room
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def generate_test_token(user, scopes):
|
||||
"""Generate a valid JWT token for testing."""
|
||||
now = datetime.now(timezone.utc)
|
||||
scope_string = " ".join(scopes)
|
||||
|
||||
payload = {
|
||||
"iss": settings.APPLICATION_JWT_ISSUER,
|
||||
"aud": settings.APPLICATION_JWT_AUDIENCE,
|
||||
"iat": now,
|
||||
"exp": now + timedelta(seconds=settings.APPLICATION_JWT_EXPIRATION_SECONDS),
|
||||
"client_id": "test-client-id",
|
||||
"scope": scope_string,
|
||||
"user_id": str(user.id),
|
||||
"delegated": True,
|
||||
}
|
||||
|
||||
return jwt.encode(
|
||||
payload,
|
||||
settings.APPLICATION_JWT_SECRET_KEY,
|
||||
algorithm=settings.APPLICATION_JWT_ALG,
|
||||
)
|
||||
|
||||
|
||||
def test_api_rooms_list_requires_authentication():
|
||||
"""Listing rooms without authentication should return 401."""
|
||||
client = APIClient()
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_api_rooms_list_with_valid_token(settings):
|
||||
"""Listing rooms with valid token should succeed."""
|
||||
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
|
||||
user = UserFactory()
|
||||
room = RoomFactory(users=[(user, RoleChoices.OWNER)])
|
||||
|
||||
# Generate valid token
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_LIST])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["count"] == 1
|
||||
assert response.data["results"][0]["id"] == str(room.id)
|
||||
|
||||
|
||||
def test_api_rooms_list_with_expired_token(settings):
|
||||
"""Listing rooms with expired token should return 401."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
settings.APPLICATION_JWT_EXPIRATION_SECONDS = 0
|
||||
|
||||
user = UserFactory()
|
||||
|
||||
# Generate expired token
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_CREATE])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "expired" in str(response.data).lower()
|
||||
|
||||
|
||||
def test_api_rooms_list_with_invalid_token():
|
||||
"""Listing rooms with invalid token should return 401."""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION="Bearer invalid-token-123")
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_api_rooms_list_missing_scope(settings):
|
||||
"""Listing rooms without required scope should return 403."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
|
||||
user = UserFactory()
|
||||
|
||||
# Token without ROOMS_LIST scope
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_CREATE])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "Insufficient permissions. Required scope: rooms:list" in str(response.data)
|
||||
|
||||
|
||||
def test_api_rooms_list_filters_by_user(settings):
|
||||
"""List should only return rooms accessible to the authenticated user."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
|
||||
user1 = UserFactory()
|
||||
user2 = UserFactory()
|
||||
|
||||
room1 = RoomFactory(users=[(user1, RoleChoices.OWNER)])
|
||||
room2 = RoomFactory(users=[(user2, RoleChoices.OWNER)])
|
||||
room3 = RoomFactory(users=[(user1, RoleChoices.MEMBER)])
|
||||
|
||||
token = generate_test_token(user1, [ApplicationScope.ROOMS_LIST])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data["count"] == 2
|
||||
returned_ids = [r["id"] for r in response.data["results"]]
|
||||
assert str(room1.id) in returned_ids
|
||||
assert str(room3.id) in returned_ids
|
||||
assert str(room2.id) not in returned_ids
|
||||
|
||||
|
||||
def test_api_rooms_retrieve_requires_scope(settings):
|
||||
"""Retrieving a room requires ROOMS_RETRIEVE scope."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
|
||||
user = UserFactory()
|
||||
room = RoomFactory(users=[(user, RoleChoices.OWNER)])
|
||||
|
||||
# Token without ROOMS_RETRIEVE scope
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_LIST])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get(f"/external-api/v1.0/rooms/{room.id}/")
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "Insufficient permissions. Required scope: rooms:retrieve" in str(
|
||||
response.data
|
||||
)
|
||||
|
||||
|
||||
def test_api_rooms_retrieve_success(settings):
|
||||
"""Retrieving a room with correct scope should succeed."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
settings.APPLICATION_BASE_URL = "http://your-application.com"
|
||||
settings.ROOM_TELEPHONY_ENABLED = True
|
||||
settings.ROOM_TELEPHONY_PHONE_NUMBER = "+1-555-0100"
|
||||
settings.ROOM_TELEPHONY_DEFAULT_COUNTRY = "US"
|
||||
|
||||
user = UserFactory()
|
||||
room = RoomFactory(users=[(user, RoleChoices.OWNER)])
|
||||
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_RETRIEVE])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get(f"/external-api/v1.0/rooms/{room.id}/")
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
assert response.data == {
|
||||
"id": str(room.id),
|
||||
"name": room.name,
|
||||
"slug": room.slug,
|
||||
"access_level": str(room.access_level),
|
||||
"url": f"http://your-application.com/{room.slug}",
|
||||
"telephony": {
|
||||
"enabled": True,
|
||||
"phone_number": "+1-555-0100",
|
||||
"pin_code": room.pin_code,
|
||||
"default_country": "US",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_api_rooms_create_requires_scope(settings):
|
||||
"""Creating a room requires ROOMS_CREATE scope."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
user = UserFactory()
|
||||
|
||||
# Token without ROOMS_CREATE scope
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_LIST])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.post("/external-api/v1.0/rooms/", {}, format="json")
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "Insufficient permissions. Required scope: rooms:create" in str(
|
||||
response.data
|
||||
)
|
||||
|
||||
|
||||
def test_api_rooms_create_success(settings):
|
||||
"""Creating a room with correct scope should succeed."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
|
||||
user = UserFactory()
|
||||
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_CREATE])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.post("/external-api/v1.0/rooms/", {}, format="json")
|
||||
|
||||
assert response.status_code == 201
|
||||
assert "id" in response.data
|
||||
assert "slug" in response.data
|
||||
|
||||
# Verify room was created with user as owner
|
||||
room = Room.objects.get(id=response.data["id"])
|
||||
assert room.get_role(user) == RoleChoices.OWNER
|
||||
assert room.access_level == "trusted"
|
||||
|
||||
|
||||
def test_api_rooms_response_no_url(settings):
|
||||
"""Response should not include url field when APPLICATION_BASE_URL is None."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
settings.APPLICATION_BASE_URL = None
|
||||
|
||||
user = UserFactory()
|
||||
room = RoomFactory(users=[(user, RoleChoices.OWNER)])
|
||||
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_RETRIEVE])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get(f"/external-api/v1.0/rooms/{room.id}/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "url" not in response.data
|
||||
assert response.data["id"] == str(room.id)
|
||||
|
||||
|
||||
def test_api_rooms_response_no_telephony(settings):
|
||||
"""Response should not include telephony field when ROOM_TELEPHONY_ENABLED is False."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
settings.ROOM_TELEPHONY_ENABLED = False
|
||||
|
||||
user = UserFactory()
|
||||
room = RoomFactory(users=[(user, RoleChoices.OWNER)])
|
||||
|
||||
token = generate_test_token(user, [ApplicationScope.ROOMS_RETRIEVE])
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get(f"/external-api/v1.0/rooms/{room.id}/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "telephony" not in response.data
|
||||
assert response.data["id"] == str(room.id)
|
||||
|
||||
|
||||
def test_api_rooms_token_without_delegated_flag(settings):
|
||||
"""Token without delegated flag should be rejected."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
user = UserFactory()
|
||||
|
||||
# Generate token without delegated flag
|
||||
now = datetime.now(timezone.utc)
|
||||
payload = {
|
||||
"iss": settings.APPLICATION_JWT_ISSUER,
|
||||
"aud": settings.APPLICATION_JWT_AUDIENCE,
|
||||
"iat": now,
|
||||
"exp": now + timedelta(hours=1),
|
||||
"client_id": "test-client",
|
||||
"scope": "rooms:list",
|
||||
"user_id": str(user.id),
|
||||
"delegated": False, # Not delegated
|
||||
}
|
||||
token = jwt.encode(
|
||||
payload,
|
||||
settings.APPLICATION_JWT_SECRET_KEY,
|
||||
algorithm=settings.APPLICATION_JWT_ALG,
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "Invalid token type." in str(response.data)
|
||||
|
||||
|
||||
def test_api_rooms_token_missing_client_id(settings):
|
||||
"""Token without client_id should be rejected."""
|
||||
settings.APPLICATION_JWT_SECRET_KEY = "devKey"
|
||||
user = UserFactory()
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
payload = {
|
||||
"iss": settings.APPLICATION_JWT_ISSUER,
|
||||
"aud": settings.APPLICATION_JWT_AUDIENCE,
|
||||
"iat": now,
|
||||
"exp": now + timedelta(hours=1),
|
||||
"scope": "rooms:list",
|
||||
"user_id": str(user.id),
|
||||
"delegated": True,
|
||||
# Missing client_id
|
||||
}
|
||||
token = jwt.encode(
|
||||
payload,
|
||||
settings.APPLICATION_JWT_SECRET_KEY,
|
||||
algorithm=settings.APPLICATION_JWT_ALG,
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
|
||||
response = client.get("/external-api/v1.0/rooms/")
|
||||
|
||||
assert response.status_code == 401
|
||||
assert "Invalid token claims." in str(response.data)
|
||||
@@ -26,6 +26,12 @@ external_router.register(
|
||||
basename="external_application",
|
||||
)
|
||||
|
||||
external_router.register(
|
||||
"rooms",
|
||||
external_viewsets.RoomViewSet,
|
||||
basename="external_room",
|
||||
)
|
||||
|
||||
urlpatterns = [
|
||||
path(
|
||||
f"api/{settings.API_VERSION}/",
|
||||
|
||||
@@ -8,6 +8,7 @@ Utils functions used in the core app
|
||||
import hashlib
|
||||
import json
|
||||
import random
|
||||
import secrets
|
||||
import string
|
||||
from typing import List, Optional
|
||||
from uuid import uuid4
|
||||
@@ -280,3 +281,14 @@ def generate_client_secret() -> str:
|
||||
Cryptographically secure client secret
|
||||
"""
|
||||
return generate_secure_token(settings.APPLICATION_CLIENT_SECRET_LENGTH)
|
||||
|
||||
|
||||
def generate_room_slug():
|
||||
"""Generate a random room slug in the format 'xxx-xxxx-xxx'."""
|
||||
|
||||
sizes = [3, 4, 3]
|
||||
parts = [
|
||||
"".join(secrets.choice(string.ascii_lowercase) for _ in range(size))
|
||||
for size in sizes
|
||||
]
|
||||
return "-".join(parts)
|
||||
|
||||
@@ -704,6 +704,11 @@ class Base(Configuration):
|
||||
environ_name="APPLICATION_JWT_TOKEN_TYPE",
|
||||
environ_prefix=None,
|
||||
)
|
||||
APPLICATION_BASE_URL = values.Value(
|
||||
None,
|
||||
environ_name="APPLICATION_BASE_URL",
|
||||
environ_prefix=None,
|
||||
)
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
@property
|
||||
|
||||
Reference in New Issue
Block a user