Compare commits

..

3 Commits

Author SHA1 Message Date
leo 5b92ae8f73 wip 2026-07-29 19:02:35 +02:00
leo ac2b5bd4f3 wip 2026-07-29 18:57:48 +02:00
leo 7c92f6054b wip 2026-07-29 18:57:48 +02:00
36 changed files with 1332 additions and 1069 deletions
+4 -2
View File
@@ -10,11 +10,14 @@ and this project adheres to
### Added
- ✨(backend) push recordings to the owner's Drive (POC)
## Fixed
- ✨(summary) report exception type in failure analytics
- ✨(frontend) add configurable documentation menu item
- ✨(frontend) allow promoting authenticated participants
- ✨(frontend) introduce an "unauthenticated" participant badge
- ✨(backend) add roomkit viewset to start a room without WebRTC join
### Changed
@@ -27,7 +30,6 @@ and this project adheres to
- 📝(legal) update terms of service
- 💄(frontend) render Avatar initials in uppercase
- 💄(frontend) improve participant name rendering in the list
- 🚚(backend) rename TelephonyService to SIPManagement
## Fixed
+6 -6
View File
@@ -84,6 +84,7 @@ bootstrap: \
data/media \
data/static \
create-env-files \
create-docker-network \
build \
migrate \
demo \
@@ -117,11 +118,16 @@ down: ## stop and remove containers, networks, images, and volumes
@$(COMPOSE) down
.PHONY: down
create-docker-network: ## create the shared lasuite-network if it doesn't exist
@docker network create lasuite-network || true
.PHONY: create-docker-network
logs: ## display app-dev logs (follow mode)
@$(COMPOSE) logs -f app-dev
.PHONY: logs
run-backend: ## start only the backend application and all needed services
@$(MAKE) create-docker-network
@$(COMPOSE) up --force-recreate -d celery-dev --remove-orphans
@$(COMPOSE) up --force-recreate -d nginx
@echo "Wait for postgresql to be up..."
@@ -389,12 +395,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
-6
View File
@@ -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')
-182
View File
@@ -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
+25 -8
View File
@@ -14,6 +14,9 @@ services:
image: sj26/mailcatcher:latest
ports:
- "1081:1080"
networks:
- default
- lasuite
minio:
user: ${DOCKER_USER:-1000}
@@ -33,6 +36,13 @@ services:
command: minio server --console-address :9001 /data
volumes:
- ./data/media:/data
networks:
default:
# The backend containers also sit on lasuite-network, where Drive's own
# minio answers to "minio" as well. They address this one by an alias no
# other stack uses, so the two can never race in DNS.
aliases:
- meet-minio
createbuckets:
image: minio/mc
@@ -93,6 +103,7 @@ services:
networks:
- resource-server
- default
- lasuite
celery-dev:
user: ${DOCKER_USER:-1000}
@@ -109,6 +120,9 @@ services:
- /app/.venv
depends_on:
- app-dev
networks:
- default
- lasuite
app:
build:
@@ -196,32 +210,32 @@ services:
- env.d/development/kc_postgresql
keycloak:
image: quay.io/keycloak/keycloak:20.0.1
image: quay.io/keycloak/keycloak:26.3.2
volumes:
- ./docker/auth/realm.json:/opt/keycloak/data/import/realm.json
command:
- start-dev
- --features=preview
- --import-realm
- --proxy=edge
- --hostname-url=http://localhost:8083
- --hostname-admin-url=http://localhost:8083/
- --proxy-headers=xforwarded
- --hostname=http://localhost:8083
- --hostname-strict=false
- --hostname-strict-https=false
environment:
KEYCLOAK_ADMIN: admin
KEYCLOAK_ADMIN_PASSWORD: admin
KC_BOOTSTRAP_ADMIN_USERNAME: admin
KC_BOOTSTRAP_ADMIN_PASSWORD: admin
KC_DB: postgres
KC_DB_URL_HOST: kc_postgresql
KC_DB_URL_DATABASE: keycloak
KC_DB_PASSWORD: pass
KC_DB_USERNAME: meet
KC_DB_SCHEMA: public
PROXY_ADDRESS_FORWARDING: 'true'
ports:
- "8080:8080"
depends_on:
- kc_postgresql
networks:
- default
- lasuite
livekit:
image: livekit/livekit-server
@@ -338,3 +352,6 @@ services:
networks:
default:
resource-server:
lasuite:
name: lasuite-network
external: true
+237 -32
View File
@@ -56,7 +56,9 @@
"value": "meet"
}
],
"realmRoles": ["user"]
"realmRoles": [
"user"
]
},
{
"username": "user-e2e-chromium",
@@ -70,7 +72,9 @@
"value": "password-e2e-chromium"
}
],
"realmRoles": ["user"]
"realmRoles": [
"user"
]
},
{
"username": "user-e2e-webkit",
@@ -84,7 +88,9 @@
"value": "password-e2e-webkit"
}
],
"realmRoles": ["user"]
"realmRoles": [
"user"
]
},
{
"username": "user-e2e-firefox",
@@ -98,7 +104,9 @@
"value": "password-e2e-firefox"
}
],
"realmRoles": ["user"]
"realmRoles": [
"user"
]
}
],
"roles": {
@@ -118,9 +126,15 @@
"description": "${role_default-roles}",
"composite": "true",
"composites": {
"realm": ["offline_access", "uma_authorization"],
"realm": [
"offline_access",
"uma_authorization"
],
"client": {
"account": ["view-profile", "manage-account"]
"account": [
"view-profile",
"manage-account"
]
}
},
"clientRole": "false",
@@ -269,7 +283,9 @@
"composite": "true",
"composites": {
"client": {
"realm-management": ["query-clients"]
"realm-management": [
"query-clients"
]
}
},
"clientRole": "true",
@@ -292,7 +308,10 @@
"composite": "true",
"composites": {
"client": {
"realm-management": ["query-users", "query-groups"]
"realm-management": [
"query-users",
"query-groups"
]
}
},
"clientRole": "true",
@@ -368,7 +387,9 @@
"composite": "true",
"composites": {
"client": {
"account": ["view-consent"]
"account": [
"view-consent"
]
}
},
"clientRole": "true",
@@ -400,7 +421,9 @@
"composite": "true",
"composites": {
"client": {
"account": ["manage-account-links"]
"account": [
"manage-account-links"
]
}
},
"clientRole": "true",
@@ -455,7 +478,9 @@
"clientRole": "false",
"containerId": "ccf4fd40-4286-474d-854a-4714282a8bec"
},
"requiredCredentials": ["password"],
"requiredCredentials": [
"password"
],
"otpPolicyType": "totp",
"otpPolicyAlgorithm": "HmacSHA1",
"otpPolicyInitialCounter": 0,
@@ -463,9 +488,14 @@
"otpPolicyLookAheadWindow": 1,
"otpPolicyPeriod": 30,
"otpPolicyCodeReusable": "false",
"otpSupportedApplications": ["totpAppGoogleName", "totpAppFreeOTPName"],
"otpSupportedApplications": [
"totpAppGoogleName",
"totpAppFreeOTPName"
],
"webAuthnPolicyRpEntityName": "keycloak",
"webAuthnPolicySignatureAlgorithms": ["ES256"],
"webAuthnPolicySignatureAlgorithms": [
"ES256"
],
"webAuthnPolicyRpId": "",
"webAuthnPolicyAttestationConveyancePreference": "not specified",
"webAuthnPolicyAuthenticatorAttachment": "not specified",
@@ -475,7 +505,9 @@
"webAuthnPolicyAvoidSameAuthenticatorRegister": "false",
"webAuthnPolicyAcceptableAaguids": [],
"webAuthnPolicyPasswordlessRpEntityName": "keycloak",
"webAuthnPolicyPasswordlessSignatureAlgorithms": ["ES256"],
"webAuthnPolicyPasswordlessSignatureAlgorithms": [
"ES256"
],
"webAuthnPolicyPasswordlessRpId": "",
"webAuthnPolicyPasswordlessAttestationConveyancePreference": "not specified",
"webAuthnPolicyPasswordlessAuthenticatorAttachment": "not specified",
@@ -487,14 +519,19 @@
"scopeMappings": [
{
"clientScope": "offline_access",
"roles": ["offline_access"]
"roles": [
"offline_access"
]
}
],
"clientScopeMappings": {
"account": [
{
"client": "account-console",
"roles": ["manage-account", "view-groups"]
"roles": [
"manage-account",
"view-groups"
]
}
]
},
@@ -509,7 +546,9 @@
"enabled": "true",
"alwaysDisplayInConsole": "false",
"clientAuthenticatorType": "client-secret",
"redirectUris": ["/realms/meet/account/*"],
"redirectUris": [
"/realms/meet/account/*"
],
"webOrigins": [],
"notBefore": 0,
"bearerOnly": "false",
@@ -551,7 +590,9 @@
"enabled": "true",
"alwaysDisplayInConsole": "false",
"clientAuthenticatorType": "client-secret",
"redirectUris": ["/realms/meet/account/*"],
"redirectUris": [
"/realms/meet/account/*"
],
"webOrigins": [],
"notBefore": 0,
"bearerOnly": "false",
@@ -796,8 +837,12 @@
"enabled": "true",
"alwaysDisplayInConsole": "false",
"clientAuthenticatorType": "client-secret",
"redirectUris": ["/admin/meet/console/*"],
"webOrigins": ["+"],
"redirectUris": [
"/admin/meet/console/*"
],
"webOrigins": [
"+"
],
"notBefore": 0,
"bearerOnly": "false",
"consentRequired": "false",
@@ -845,6 +890,142 @@
"offline_access",
"microprofile-jwt"
]
},
{
"clientId": "drive",
"name": "",
"description": "",
"rootUrl": "",
"adminUrl": "",
"baseUrl": "",
"surrogateAuthRequired": false,
"enabled": true,
"alwaysDisplayInConsole": false,
"clientAuthenticatorType": "client-secret",
"secret": "ThisIsAnExampleKeyForDevPurposeOnly",
"redirectUris": [
"http://localhost:3100/*",
"http://localhost:8171/*",
"http://localhost:8085/*"
],
"webOrigins": [
"http://localhost:3100",
"http://localhost:8171",
"http://localhost:8085"
],
"notBefore": 0,
"bearerOnly": false,
"consentRequired": false,
"standardFlowEnabled": true,
"implicitFlowEnabled": false,
"directAccessGrantsEnabled": false,
"serviceAccountsEnabled": false,
"publicClient": false,
"frontchannelLogout": true,
"protocol": "openid-connect",
"attributes": {
"access.token.lifespan": "-1",
"client.secret.creation.time": "1707820779",
"user.info.response.signature.alg": "RS256",
"post.logout.redirect.uris": "http://localhost:3100/*##http://localhost:8171/*##http://localhost:8085/*",
"oauth2.device.authorization.grant.enabled": "false",
"use.jwks.url": "false",
"backchannel.logout.revoke.offline.tokens": "false",
"use.refresh.tokens": "true",
"tls-client-certificate-bound-access-tokens": "false",
"oidc.ciba.grant.enabled": "false",
"backchannel.logout.session.required": "true",
"client_credentials.use_refresh_token": "false",
"acr.loa.map": "{}",
"require.pushed.authorization.requests": "false",
"display.on.consent.screen": "false",
"client.session.idle.timeout": "-1",
"token.response.type.bearer.lower-case": "false"
},
"authenticationFlowBindingOverrides": {},
"fullScopeAllowed": true,
"nodeReRegistrationTimeout": -1,
"defaultClientScopes": [
"web-origins",
"acr",
"roles",
"profile",
"email"
],
"optionalClientScopes": [
"address",
"phone",
"offline_access",
"microprofile-jwt"
]
},
{
"clientId": "deploycenter",
"name": "",
"description": "",
"rootUrl": "",
"adminUrl": "",
"baseUrl": "",
"surrogateAuthRequired": false,
"enabled": true,
"alwaysDisplayInConsole": false,
"clientAuthenticatorType": "client-secret",
"secret": "ThisIsAnExampleKeyForDevPurposeOnly",
"redirectUris": [
"http://localhost:3100/*",
"http://localhost:8171/*",
"http://localhost:8085/*"
],
"webOrigins": [
"http://localhost:3100",
"http://localhost:8171",
"http://localhost:8085"
],
"notBefore": 0,
"bearerOnly": false,
"consentRequired": false,
"standardFlowEnabled": true,
"implicitFlowEnabled": false,
"directAccessGrantsEnabled": false,
"serviceAccountsEnabled": false,
"publicClient": false,
"frontchannelLogout": true,
"protocol": "openid-connect",
"attributes": {
"access.token.lifespan": "-1",
"client.secret.creation.time": "1707820779",
"user.info.response.signature.alg": "RS256",
"post.logout.redirect.uris": "http://localhost:3100/*##http://localhost:8171/*##http://localhost:8085/*",
"oauth2.device.authorization.grant.enabled": "false",
"use.jwks.url": "false",
"backchannel.logout.revoke.offline.tokens": "false",
"use.refresh.tokens": "true",
"tls-client-certificate-bound-access-tokens": "false",
"oidc.ciba.grant.enabled": "false",
"backchannel.logout.session.required": "true",
"client_credentials.use_refresh_token": "false",
"acr.loa.map": "{}",
"require.pushed.authorization.requests": "false",
"display.on.consent.screen": "false",
"client.session.idle.timeout": "-1",
"token.response.type.bearer.lower-case": "false"
},
"authenticationFlowBindingOverrides": {},
"fullScopeAllowed": true,
"nodeReRegistrationTimeout": -1,
"defaultClientScopes": [
"web-origins",
"acr",
"roles",
"profile",
"email"
],
"optionalClientScopes": [
"address",
"phone",
"offline_access",
"microprofile-jwt"
]
}
],
"clientScopes": [
@@ -1382,7 +1563,9 @@
},
"smtpServer": {},
"eventsEnabled": "false",
"eventsListeners": ["jboss-logging"],
"eventsListeners": [
"jboss-logging"
],
"enabledEventTypes": [],
"adminEventsEnabled": "false",
"adminEventsDetailsEnabled": "false",
@@ -1405,7 +1588,9 @@
"subType": "anonymous",
"subComponents": {},
"config": {
"allow-default-scopes": ["true"]
"allow-default-scopes": [
"true"
]
}
},
{
@@ -1415,7 +1600,9 @@
"subType": "anonymous",
"subComponents": {},
"config": {
"max-clients": ["200"]
"max-clients": [
"200"
]
}
},
{
@@ -1425,7 +1612,9 @@
"subType": "authenticated",
"subComponents": {},
"config": {
"allow-default-scopes": ["true"]
"allow-default-scopes": [
"true"
]
}
},
{
@@ -1481,8 +1670,12 @@
"subType": "anonymous",
"subComponents": {},
"config": {
"host-sending-registration-request-must-match": ["true"],
"client-uris-must-match": ["true"]
"host-sending-registration-request-must-match": [
"true"
],
"client-uris-must-match": [
"true"
]
}
}
],
@@ -1501,7 +1694,9 @@
"providerId": "aes-generated",
"subComponents": {},
"config": {
"priority": ["100"]
"priority": [
"100"
]
}
},
{
@@ -1510,8 +1705,12 @@
"providerId": "hmac-generated",
"subComponents": {},
"config": {
"priority": ["100"],
"algorithm": ["HS256"]
"priority": [
"100"
],
"algorithm": [
"HS256"
]
}
},
{
@@ -1520,8 +1719,12 @@
"providerId": "rsa-enc-generated",
"subComponents": {},
"config": {
"priority": ["100"],
"algorithm": ["RSA-OAEP"]
"priority": [
"100"
],
"algorithm": [
"RSA-OAEP"
]
}
},
{
@@ -1530,7 +1733,9 @@
"providerId": "rsa-generated",
"subComponents": {},
"config": {
"priority": ["100"]
"priority": [
"100"
]
}
}
]
-21
View File
@@ -143,24 +143,3 @@ $ 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).
+84
View File
@@ -126,6 +126,90 @@ RECORDING_STORAGE_EVENT_TOKEN = <token>
> Questions? Open an issue on [GitHub](https://github.com/suitenumerique/meet/issues/new?assignees=&labels=bug&template=Bug_report.md) or join our [Matrix community](https://matrix.to/#/#meet-official:matrix.org).
## Push recordings to Drive
Once a recording is over, it can be pushed to the main workspace of the user who
started it in [Drive](https://github.com/suitenumerique/drive), on top of staying
in the object storage. The file is streamed from the object storage to Drive: it
is never fully held in the worker's memory nor written to its disk.
Drive is called as an OIDC resource server, following its
[resource server documentation](https://github.com/suitenumerique/drive/blob/main/docs/resource_server.md):
```mermaid
sequenceDiagram
participant User
participant Backend as Django Backend
participant Worker as Celery Worker
participant Storage as Object Storage
participant Drive
User->>Backend: POST /api/v1.0/rooms/{id}/start-recording/
Backend->>Backend: Park the user's OIDC access token (encrypted)
Note over Backend: Recording in progress...
Storage->>Backend: Storage event notification
Backend->>Worker: Schedule push_recording
Worker->>Drive: GET /items/ (as the user)
Drive-->>Worker: Main workspace
Worker->>Drive: POST /items/{workspace}/children/
Drive-->>Worker: Item + presigned upload URL
Worker->>Storage: GET recording (streamed)
Worker->>Drive: PUT presigned URL (relayed chunk by chunk)
Worker->>Drive: POST /items/{item}/upload-ended/
Worker->>Backend: Drop the parked access token
```
### Special requirements
- Drive configured as an OIDC resource server, accepting Meet's audience
(`OIDC_RS_ALLOWED_AUDIENCES` must contain Meet's client id), with the `items`
endpoint allowing the `list`, `children` and `upload_ended` actions.
- `OIDC_STORE_ACCESS_TOKEN` enabled on Meet, along with
`OIDC_STORE_REFRESH_TOKEN_KEY`, the Fernet key encrypting the parked token.
> [!CAUTION]
> This is a proof of concept: the access token is captured when the recording
> starts and assumed to still be valid when the recording ends. Long recordings
> may therefore fail to be pushed. Exchanging it for a long-lived, narrowly
> scoped token ([RFC 8693](https://datatracker.ietf.org/doc/html/rfc8693)) is the
> intended follow-up.
### Configuration options
| Option | Type | Default | Description |
| ----------------------------------------------------- | ----------- | ------- | -------------------------------------------------------------------------------------------------------------------------------------------------- |
| **RECORDING_PUSH_TO_DRIVE_ENABLED** | Boolean | `False` | Enable pushing recordings to the owner's Drive. |
| **DRIVE_API_BASE_URL** | String | `None` | Base URL of Drive's external API, e.g. `https://drive.example.com/external_api/v1.0`. |
| **RECORDING_PUSH_TO_DRIVE_SIGNED_URL_EXPIRY_SECONDS** | Integer | `3600` | Lifetime of the signed URL the worker downloads the recording from. |
| **OIDC_STORE_ACCESS_TOKEN** | Boolean | `False` | Keep the user's access token in the session, required to call Drive on their behalf. |
| **OIDC_STORE_REFRESH_TOKEN_KEY** | Secret/File | `None` | Fernet key encrypting OIDC tokens at rest. Generate one with `Fernet.generate_key()`. |
| **DRIVE_UPLOAD_STORAGE_NETLOC** | String | `None` | Development only: `host:port` to reach Drive's object storage at, when the domain Drive signs its upload URLs with only resolves from a browser. |
### Local development
Meet and Drive run as two separate compose projects, joined by the external
`lasuite-network` (`make create-docker-network`). Meet's backend containers reach
Drive's nginx at `drive-nginx:8083` and its object storage at `drive-minio:9000`.
On the Drive side:
```bash
OIDC_RESOURCE_SERVER_ENABLED=True
OIDC_RS_CLIENT_ID=drive
OIDC_RS_CLIENT_SECRET=ThisIsAnExampleKeyForDevPurposeOnly
OIDC_RS_AUDIENCE_CLAIM=client_id
OIDC_RS_ALLOWED_AUDIENCES=meet
```
`DRIVE_UPLOAD_STORAGE_NETLOC` is needed there because Drive signs its upload URLs
with `localhost:9100`, which does not resolve from Meet's containers. The
presigned signature covers the `Host` header, so the backend keeps announcing the
signed host and only swaps the address it connects to.
## LiveKit Egress
La Suite Meet uses LiveKit Egress to record room sessions. For reference, see the [LiveKit Egress repository](https://github.com/livekit/egress) and the [official documentation](https://docs.livekit.io/home/egress/overview/).
-3
View File
@@ -8,6 +8,3 @@ class AnalyticsEvent(StrEnum):
# Rooms
ROOM_CREATED = "room_created"
# Roomkit (meeting-room SIP devices)
ROOMKIT_JOINED = "roomkit_joined"
-1
View File
@@ -16,7 +16,6 @@ class FeatureFlag:
"file_upload": "FILE_UPLOAD_ENABLED",
"addons": "ADDONS_ENABLED",
"application": "APPLICATION_ENABLED",
"roomkit": "ROOMKIT_ENABLED",
}
@classmethod
-12
View File
@@ -73,15 +73,3 @@ class CreationCallbackAnonRateThrottle(MonitoredAnonRateThrottle):
"""Throttle Anonymous user requesting room generation callback"""
scope = "creation_callback"
class RoomKitJoinRateThrottle(MonitoredUserRateThrottle):
"""Throttle the LiveKit SIP module requesting roomkit joins.
The roomkit endpoints are authenticated as a machine user, so all requests
share a single throttle bucket. This is not a security measure against
brute-force attacks but a guard against accidental hammering from a buggy
SIP module.
"""
scope = "roomkit_join"
+28
View File
@@ -364,6 +364,33 @@ class RoomViewSet(
room.id,
)
@staticmethod
def _park_drive_credentials(request, recording):
"""Keep the OIDC access token needed to push the recording to Drive later.
Pushing happens long after this request, when the egress is over and the
user is gone, so the token has to be parked now.
POC limitation: we assume the token is still valid by then. The target
design is a token exchange (RFC 8693) performed here, to get a long-lived
token narrowly scoped to that upload.
"""
if not settings.RECORDING_PUSH_TO_DRIVE_ENABLED:
return
access_token = request.session.get("oidc_access_token")
if not access_token:
logger.warning(
"No OIDC access token in session, recording %s will not be pushed "
"to Drive. Is OIDC_STORE_ACCESS_TOKEN enabled?",
recording.id,
)
return
recording.set_owner_access_token(access_token)
@decorators.action(
detail=True,
methods=["post"],
@@ -399,6 +426,7 @@ class RoomViewSet(
role=models.RoleChoices.OWNER,
recording=recording,
)
self._park_drive_credentials(request, recording)
except (DjangoValidationError, IntegrityError):
# DjangoValidationError covers the Python-level check (full_clean);
@@ -0,0 +1,18 @@
# Generated by Django 5.2.14 on 2026-07-20 00:00
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='recording',
name='status',
field=models.CharField(choices=[('initiated', 'Initiated'), ('active', 'Active'), ('stopped', 'Stopped'), ('saved', 'Saved'), ('aborted', 'Aborted'), ('failed', 'Failed'), ('failed_to_start', 'Failed to Start'), ('failed_to_stop', 'Failed to Stop'), ('notification_succeeded', 'Notification succeeded'), ('external_process_successful', 'External process successful'), ('external_process_failed', 'External process failed')], default='initiated', max_length=50),
),
]
@@ -0,0 +1,24 @@
# Generated by Django 5.2.14 on 2026-07-29 00:00
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("core", "0022_alter_recording_status"),
]
operations = [
migrations.AddField(
model_name="recording",
name="owner_access_token",
field=models.TextField(
blank=True,
editable=False,
help_text="Encrypted OIDC access token of the user who started the recording, used to push the recording to their Drive on their behalf. Dropped as soon as the push has been attempted.",
null=True,
verbose_name="Owner access token",
),
),
]
+45 -8
View File
@@ -429,14 +429,7 @@ class Room(Resource):
def save(self, *args, **kwargs):
"""Generate a unique n-digit pin code for new rooms."""
# Roomkit devices also join by PIN, so a PIN is needed as soon as
# either integration is enabled.
if (
(settings.ROOM_TELEPHONY_ENABLED or settings.ROOMKIT_ENABLED)
and not self.pk
and not self.pin_code
):
if settings.ROOM_TELEPHONY_ENABLED and not self.pk and not self.pin_code:
self.pin_code = self.generate_unique_pin_code(
length=settings.ROOM_TELEPHONY_PIN_LENGTH
)
@@ -620,6 +613,17 @@ class Recording(BaseModel):
verbose_name=_("External Process ID"),
help_text=_("ID of the external process associated with the recording."),
)
owner_access_token = models.TextField(
null=True,
blank=True,
editable=False,
verbose_name=_("Owner access token"),
help_text=_(
"Encrypted OIDC access token of the user who started the recording, "
"used to push the recording to their Drive on their behalf. "
"Dropped as soon as the push has been attempted."
),
)
class Meta:
db_table = "meet_recording"
@@ -722,6 +726,39 @@ class Recording(BaseModel):
return self.expired_at < timezone.now()
def set_owner_access_token(self, access_token: str) -> None:
"""Park the OIDC access token of the user who started the recording.
It is stored encrypted, and only long enough for the worker to push the
recording to that user's Drive once the recording is over.
"""
self.owner_access_token = utils.encrypt_secret(access_token)
self.save(update_fields=["owner_access_token", "updated_at"])
def get_owner_access_token(self) -> Optional[str]:
"""Return the parked OIDC access token, or None if there is none left."""
if not self.owner_access_token:
return None
try:
return utils.decrypt_secret(self.owner_access_token)
except utils.SecretDecryptionError:
logger.exception(
"Could not decrypt the access token of recording %s", self.id
)
return None
def clear_owner_access_token(self) -> None:
"""Drop the parked access token, it is a user credential."""
if self.owner_access_token is None:
return
self.owner_access_token = None
self.save(update_fields=["owner_access_token", "updated_at"])
class RecordingAccess(BaseAccess):
"""Relation model to give access to a recording for a user or a team with a role."""
@@ -19,6 +19,7 @@ from livekit import api as livekit_api
from core import models, utils
from core.analytics import UserFeatureFlag, is_user_feature_flag_enabled
from core.tasks.push_recording import push_recording
from core.utils import generate_download_s3_url
logger = logging.getLogger(__name__)
@@ -45,7 +46,17 @@ class NotificationService:
"""Service for processing recordings and notifying external services."""
def notify_external_services(self, recording):
"""Process a recording based on its mode."""
"""Process a recording, then push it to the owner's Drive."""
try:
return self._notify_by_mode(recording)
finally:
# Independent from the mode: the file itself is pushed to the owner's
# Drive, whether it is a screen recording or a transcript's audio.
self._push_recording_to_drive(recording)
def _notify_by_mode(self, recording):
"""Route a recording to the services its mode calls for."""
if recording.mode == models.RecordingModeChoices.TRANSCRIPT:
return self._notify_summary_service(recording)
@@ -222,6 +233,31 @@ class NotificationService:
f"Unknown summary service version: {settings.SUMMARY_SERVICE_VERSION}"
)
@staticmethod
def _push_recording_to_drive(recording: models.Recording):
"""Hand the recording over to the task pushing it to the owner's Drive.
Best effort: a failure here must not compromise the rest of the
notification flow, the recording itself is safe in object storage.
"""
if not settings.RECORDING_PUSH_TO_DRIVE_ENABLED:
return
if not recording.owner_access_token:
logger.warning(
"No access token parked for recording %s, skipping the Drive push",
recording.id,
)
return
try:
push_recording.delay(str(recording.id))
except Exception: # pylint: disable=broad-except
logger.exception(
"Could not schedule the Drive push of recording %s", recording.id
)
@staticmethod
def _notify_summary_service_v1(recording: models.Recording):
"""Notify summary service about a new recording."""
-1
View File
@@ -1 +0,0 @@
"""Meet core roomkit API endpoints for meeting-room (SIP) device integration."""
@@ -1,65 +0,0 @@
"""Authentication for the roomkit API of the Meet core app."""
import logging
import secrets
from django.conf import settings
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from core.recording.event.authentication import MachineUser
logger = logging.getLogger(__name__)
class ServerToServerAuthentication(BaseAuthentication):
"""Custom authentication class for roomkit server-to-server requests.
Validates the Authorization header against the roomkit server-to-server
token. A valid PIN code is intentionally not enough to authenticate: the
endpoints are restricted to the LiveKit SIP module's credentials.
"""
AUTH_HEADER = "Authorization"
TOKEN_TYPE = "Bearer" # noqa S105
def authenticate(self, request):
"""Validate the Bearer token from the Authorization header.
Returns a (MachineUser, token) pair on success, and raises
AuthenticationFailed if the header is missing, malformed, or contains
an invalid token.
"""
required_token = settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN
if not required_token:
raise AuthenticationFailed("Server-to-server token is not configured.")
auth_header = request.headers.get(self.AUTH_HEADER)
if not auth_header:
logger.warning(
"Roomkit authentication failed: missing Authorization header (ip: %s)",
request.META.get("REMOTE_ADDR"),
)
raise AuthenticationFailed("Authorization header is missing.")
# Validate token format and existence
auth_parts = auth_header.split(" ")
if len(auth_parts) != 2 or auth_parts[0] != self.TOKEN_TYPE:
raise AuthenticationFailed("Invalid authorization header.")
token = auth_parts[1]
# Use constant-time comparison to prevent timing attacks
if not secrets.compare_digest(token.encode(), required_token.encode()):
logger.warning(
"Roomkit authentication failed: invalid token (ip: %s)",
request.META.get("REMOTE_ADDR"),
)
raise AuthenticationFailed("Invalid server-to-server token.")
return MachineUser(username="roomkit"), token
def authenticate_header(self, request):
"""Return the WWW-Authenticate header value."""
return f"{self.TOKEN_TYPE} realm='Roomkit server to server'"
-21
View File
@@ -1,21 +0,0 @@
"""Serializers for the roomkit API of the Meet core app."""
# pylint: disable=abstract-method
from django.conf import settings
from rest_framework import serializers
from core.api.serializers import BaseValidationOnlySerializer
class RoomKitJoinSerializer(BaseValidationOnlySerializer):
"""Validate roomkit join requests from the LiveKit SIP module."""
pin_code = serializers.CharField(required=True)
def validate_pin_code(self, value):
"""Ensure the PIN code matches the configured length."""
if len(value) != settings.ROOM_TELEPHONY_PIN_LENGTH:
raise serializers.ValidationError("PIN code length is invalid.")
return value
-89
View File
@@ -1,89 +0,0 @@
"""Roomkit API endpoints for meeting-room (SIP) device integration."""
from logging import getLogger
from rest_framework import decorators, viewsets
from rest_framework import (
exceptions as drf_exceptions,
)
from rest_framework import (
response as drf_response,
)
from rest_framework import (
status as drf_status,
)
from core import analytics, models
from core.api import permissions, throttling
from core.api.feature_flag import FeatureFlag
from core.services.sip_management import SIPException, SIPManagement
from . import authentication, serializers
logger = getLogger(__name__)
class RoomKitViewSet(viewsets.ViewSet):
"""Server-to-server API endpoints for the roomkit integration.
Groups all interactions between roomkit (SIP) devices and the backend,
brokered by the LiveKit SIP module. All endpoints are authenticated
with the roomkit server-to-server tokens.
"""
authentication_classes = [authentication.ServerToServerAuthentication]
permission_classes = [permissions.IsAuthenticated]
@decorators.action(
detail=False,
methods=["post"],
url_path="join",
throttle_classes=[throttling.RoomKitJoinRateThrottle],
)
@FeatureFlag.require("roomkit")
def join(self, request):
"""Prepare a room for a meeting-room (SIP) device joining by PIN code.
Called by the LiveKit SIP module when a meeting-room device dials in
with a PIN code before any WebRTC participant has joined. Resolves the
room by PIN and creates its SIP dispatch rule, so the device can enter
without waiting for a WebRTC user.
The webhook-based creation path is kept: both converge on the same rule
through the shared SIPManagement.
"""
serializer = serializers.RoomKitJoinSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
room = models.Room.objects.get(
pin_code=serializer.validated_data["pin_code"]
)
except models.Room.DoesNotExist as e:
raise drf_exceptions.NotFound("No room found for this PIN code.") from e
try:
created = SIPManagement().ensure_dispatch_rule(room)
except SIPException as e:
raise drf_exceptions.APIException("Could not create dispatch rule.") from e
analytics.capture(
request.user,
analytics.AnalyticsEvent.ROOMKIT_JOINED,
{
"room_id": str(room.pk),
"dispatch_rule_created": created,
},
)
logger.info(
"Roomkit join requested: room_id=%s, dispatch_rule_created=%s",
room.id,
created,
)
return drf_response.Response(
{"status": "success"},
status=drf_status.HTTP_200_OK,
)
+217
View File
@@ -0,0 +1,217 @@
"""Client for La Suite Drive's external API (OIDC resource server).
Drive exposes `/external_api/v1.0/*` to applications holding an end user's OIDC
access token. Uploading a file is a four step dance, documented in Drive's
`docs/resource_server.md`:
1. `GET /items/` to locate the user's main workspace,
2. `POST /items/{workspace_id}/children/` to create the file item, which returns
a presigned upload URL (the `policy`),
3. `PUT {policy}` to push the bytes to Drive's object storage,
4. `POST /items/{item_id}/upload-ended/` to let Drive know the upload is over.
Drive never fetches a URL on our behalf, so the bytes have to transit through
whoever holds the user's token, i.e. us.
"""
import logging
from urllib.parse import urlparse, urlunparse
from django.conf import settings
import requests
logger = logging.getLogger(__name__)
# (connect, read) timeouts, in seconds. The upload one is generous: it covers a
# whole recording being relayed to Drive's object storage.
API_TIMEOUT = (10, 30)
UPLOAD_TIMEOUT = (10, 300)
# Safety net when walking the paginated item list looking for the main workspace.
MAX_WORKSPACE_PAGES = 10
class DriveError(Exception):
"""Raised when Drive's external API cannot fulfill a request."""
class SizedStream:
"""Read-only byte stream of a known size, suitable as a `requests` body.
`requests` falls back to a chunked transfer encoding when it cannot guess the
body size upfront, which presigned S3 uploads reject. Advertising the size
through `__len__` makes it send a plain `Content-Length` instead, while the
underlying stream is still consumed chunk by chunk.
"""
def __init__(self, stream, length: int):
"""Wrap `stream`, whose full content is `length` bytes long."""
self._stream = stream
self._length = length
def __len__(self) -> int:
"""Return the total size of the stream, in bytes."""
return self._length
def __iter__(self):
"""Iterate over the stream, required for `requests` to stream the body."""
return iter(self._stream)
def read(self, amt=None) -> bytes:
"""Read up to `amt` bytes from the stream."""
return self._stream.read(amt)
class DriveClient:
"""Talk to Drive's external API on behalf of a user.
The client is bound to a single user access token: every call is performed
as that user, and Drive applies its own permissions accordingly.
"""
def __init__(self, access_token: str, *, base_url: str | None = None):
"""Prepare a session authenticated with the user's OIDC access token."""
self._base_url = (base_url or settings.DRIVE_API_BASE_URL or "").rstrip("/")
if not self._base_url:
raise DriveError(
"Drive API is not configured, set DRIVE_API_BASE_URL to enable it."
)
if not access_token:
raise DriveError("An access token is required to call Drive.")
self._session = requests.Session()
self._session.headers.update(
{
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
}
)
def __enter__(self):
"""Allow use as a context manager, closing the session on exit."""
return self
def __exit__(self, *args):
"""Close the underlying HTTP session."""
self.close()
def close(self):
"""Release the underlying HTTP session."""
self._session.close()
def _request(self, method, path, **kwargs):
"""Perform an authenticated call to the external API and return its body."""
url = f"{self._base_url}{path}"
kwargs.setdefault("timeout", API_TIMEOUT)
try:
response = self._session.request(method, url, **kwargs)
response.raise_for_status()
except requests.RequestException as exc:
raise DriveError(f"Drive call failed: {method} {url}") from exc
if not response.content:
return None
try:
return response.json()
except ValueError as exc:
raise DriveError(f"Drive returned a non-JSON body for {url}") from exc
def get_main_workspace(self) -> dict:
"""Return the user's main workspace, the default destination for files."""
path = "/items/"
for _page in range(MAX_WORKSPACE_PAGES):
data = self._request("GET", path) or {}
for item in data.get("results") or []:
if item.get("main_workspace"):
return item
next_url = data.get("next")
if not next_url:
break
# `next` is absolute; keep only what follows the API base URL.
path = next_url[len(self._base_url) :]
raise DriveError("No main workspace found for this user.")
def create_file(self, *, parent_id: str, filename: str) -> dict:
"""Create a file item under `parent_id` and return it.
The returned item carries a `policy`: the presigned URL the content has
to be uploaded to.
"""
item = self._request(
"POST",
f"/items/{parent_id}/children/",
json={"type": "file", "filename": filename},
)
if not item or not item.get("policy"):
raise DriveError(
f"Drive did not return an upload policy for file '{filename}'."
)
return item
@staticmethod
def _resolve_upload_target(policy_url: str) -> tuple[str, str | None]:
"""Return the address to connect to, and the `Host` header to send.
Drive signs its upload URLs with the object storage domain meant for its
*browser* clients, which does not necessarily resolve from here — that is
the case in the split docker compose development setup. The presigned
signature covers the `Host` header, so we may only swap the address we
connect to and must keep announcing the original host.
"""
override = settings.DRIVE_UPLOAD_STORAGE_NETLOC
if not override:
return policy_url, None
parsed = urlparse(policy_url)
return urlunparse(parsed._replace(netloc=override)), parsed.netloc
def upload_content(self, *, policy_url: str, stream, content_length, content_type):
"""Push `stream` to the presigned URL, without buffering it as a whole."""
url, host_header = self._resolve_upload_target(policy_url)
headers = {
"Content-Type": content_type,
"Content-Length": str(content_length),
"x-amz-acl": "private",
}
if host_header:
headers["Host"] = host_header
try:
# A bare `requests.put`, not the authenticated session: the presigned
# URL carries its own credentials and the object storage rejects an
# extra `Authorization` header.
response = requests.put(
url,
data=SizedStream(stream, content_length),
headers=headers,
timeout=UPLOAD_TIMEOUT,
)
response.raise_for_status()
except requests.RequestException as exc:
raise DriveError("Upload to Drive's object storage failed.") from exc
def complete_upload(self, item_id: str) -> None:
"""Tell Drive the upload is over, which makes the file available."""
self._request("POST", f"/items/{item_id}/upload-ended/", json={})
+10 -10
View File
@@ -28,7 +28,7 @@ from .room_management import (
RoomManagementException,
RoomNotFoundException,
)
from .sip_management import SIPException, SIPManagement
from .telephony import TelephonyException, TelephonyService
logger = getLogger(__name__)
@@ -107,7 +107,7 @@ class LiveKitEventsService:
)
self.webhook_receiver = api.WebhookReceiver(token_verifier)
self.lobby_service = LobbyService()
self.sip_management = SIPManagement()
self.telephony_service = TelephonyService()
self.recording_events = RecordingEventsService()
self._filter_regex = None
@@ -245,12 +245,12 @@ class LiveKitEventsService:
except models.Room.DoesNotExist as err:
raise ActionFailedError(f"Room with ID {room_id} does not exist") from err
if settings.ROOM_TELEPHONY_ENABLED or settings.ROOMKIT_ENABLED:
if settings.ROOM_TELEPHONY_ENABLED:
try:
self.sip_management.ensure_dispatch_rule(room)
except SIPException as e:
self.telephony_service.create_dispatch_rule(room)
except TelephonyException as e:
raise ActionFailedError(
f"Failed to create sip dispatch rule for room {room_id}"
f"Failed to create telephony dispatch rule for room {room_id}"
) from e
def _handle_room_finished(self, data):
@@ -265,12 +265,12 @@ class LiveKitEventsService:
)
raise ActionFailedError("Failed to process room finished event") from e
if settings.ROOM_TELEPHONY_ENABLED or settings.ROOMKIT_ENABLED:
if settings.ROOM_TELEPHONY_ENABLED:
try:
self.sip_management.delete_dispatch_rule(room_id)
except SIPException as e:
self.telephony_service.delete_dispatch_rule(room_id)
except TelephonyException as e:
raise ActionFailedError(
f"Failed to delete sip dispatch rule for room {room_id}"
f"Failed to delete telephony dispatch rule for room {room_id}"
) from e
try:
@@ -1,9 +1,9 @@
"""SIP management service for managing SIP dispatch rules for room access."""
"""Telephony service for managing SIP dispatch rules for room access."""
from logging import getLogger
from asgiref.sync import async_to_sync
from livekit.api import TwirpError, TwirpErrorCode
from livekit.api import TwirpError
from livekit.protocol.sip import (
CreateSIPDispatchRuleRequest,
DeleteSIPDispatchRuleRequest,
@@ -17,16 +17,12 @@ from core import utils
logger = getLogger(__name__)
class SIPException(Exception):
"""Exception raised when SIP operations fail."""
class TelephonyException(Exception):
"""Exception raised when telephony operations fail."""
class DispatchRuleConflictError(SIPException):
"""Raised when a dispatch rule already exists for the same routing criteria."""
class SIPManagement:
"""Service for managing SIP access through the telephony or roomkit system (SIP)."""
class TelephonyService:
"""Service for managing participant access through the telephony system (SIP)."""
def _rule_name(self, room_id):
"""Generate the rule name for a room based on its ID."""
@@ -36,7 +32,7 @@ class SIPManagement:
async def create_dispatch_rule(self, room):
"""Create a SIP inbound dispatch rule for direct room routing.
Configures livekit-sip to route incoming SIP calls directly to the specified room
Configures telephony to route incoming SIP calls directly to the specified room
using the room's ID and PIN code for authentication.
"""
@@ -55,12 +51,10 @@ class SIPManagement:
try:
await lkapi.sip.create_sip_dispatch_rule(create=request)
except TwirpError as e:
if e.code == TwirpErrorCode.ALREADY_EXISTS:
raise DispatchRuleConflictError("Dispatch rule already exists") from e
logger.exception(
"Unexpected error creating dispatch rule for room %s", room.id
)
raise SIPException("Could not create dispatch rule") from e
raise TelephonyException("Could not create dispatch rule") from e
finally:
await lkapi.aclose()
@@ -85,7 +79,7 @@ class SIPManagement:
)
except TwirpError as e:
logger.exception("Failed to list dispatch rules for room %s", room_id)
raise SIPException("Could not list dispatch rules") from e
raise TelephonyException("Could not list dispatch rules") from e
finally:
await lkapi.aclose()
@@ -100,28 +94,6 @@ class SIPManagement:
if existing_rule.name == rule_name
]
@async_to_sync
async def has_dispatch_rule(self, room_id):
"""Check whether at least one dispatch rule exists for a specific room."""
return bool(await self._list_dispatch_rules_ids(room_id))
def ensure_dispatch_rule(self, room):
"""Create the SIP dispatch rule for a room if it does not already exist.
Returns:
bool: True if a rule was created, False if it already existed.
"""
if self.has_dispatch_rule(room.pk):
return False
try:
self.create_dispatch_rule(room)
except DispatchRuleConflictError:
return False
return True
@async_to_sync
async def delete_dispatch_rule(self, room_id):
"""Delete all SIP inbound dispatch rules associated with a specific room."""
@@ -146,7 +118,7 @@ class SIPManagement:
except TwirpError as e:
logger.exception("Failed to delete dispatch rules for room %s", room_id)
raise SIPException("Could not delete dispatch rules") from e
raise TelephonyException("Could not delete dispatch rules") from e
finally:
await lkapi.aclose()
+11
View File
@@ -0,0 +1,11 @@
"""Asynchronous tasks of the core application.
Importing the task modules here is what makes Celery's `autodiscover_tasks`
register them: it only imports the `core.tasks` package itself, never its
submodules.
"""
from core.tasks.file import process_file_deletion
from core.tasks.push_recording import push_recording
__all__ = ["process_file_deletion", "push_recording"]
+3
View File
@@ -1,4 +1,7 @@
"""Celery-optional task decorator."""
# ruff: noqa: PLC0415
# pylint: disable=import-outside-toplevel
from django.conf import settings
+112
View File
@@ -0,0 +1,112 @@
"""Task pushing a finished recording to its owner's Drive."""
import logging
from django.conf import settings
import requests
from core import models, utils
from core.services.drive import API_TIMEOUT, DriveClient, DriveError
from core.tasks._task import task
logger = logging.getLogger(__name__)
# (connect, read) timeouts for the download leg, in seconds. The read one has to
# accommodate a whole recording being relayed.
DOWNLOAD_TIMEOUT = (API_TIMEOUT[0], 300)
def _build_filename(recording: models.Recording) -> str:
"""Return a human-readable filename for the Drive item."""
return (
f"{recording.room.slug}-"
f"{recording.created_at:%Y-%m-%d-%H-%M}."
f"{recording.extension}"
)
@task
def push_recording(recording_id: str) -> bool:
"""Push a recording to the main workspace of the user who started it.
The recording is streamed straight from object storage to Drive's presigned
URL: it is never fully downloaded to the worker's disk or memory.
The access token parked when the recording started is consumed here, and
dropped afterwards whatever the outcome — it is a user credential, and a
replay would need a fresh one anyway.
Mostly taken from: https://github.com/suitenumerique/drive/blob/main/docs/resource_server.md
"""
try:
recording = models.Recording.objects.select_related("room").get(pk=recording_id)
except models.Recording.DoesNotExist:
logger.error(
"Recording %s does not exist, cannot push it to Drive", recording_id
)
return False
access_token = recording.get_owner_access_token()
if not access_token:
logger.error(
"No access token stored for recording %s, cannot push it to Drive. "
"Was OIDC_STORE_ACCESS_TOKEN enabled when the recording started?",
recording_id,
)
return False
download_url = utils.generate_download_s3_url(
recording.key,
expires_in=settings.RECORDING_PUSH_TO_DRIVE_SIGNED_URL_EXPIRY_SECONDS,
override_domain=False,
)
filename = _build_filename(recording)
try:
with DriveClient(access_token) as drive:
workspace = drive.get_main_workspace()
item = drive.create_file(parent_id=workspace["id"], filename=filename)
# The bytes are relayed chunk by chunk: the recording is never held
# in memory as a whole.
with requests.get(
download_url, stream=True, timeout=DOWNLOAD_TIMEOUT
) as download:
download.raise_for_status()
content_length = download.headers.get("Content-Length")
if content_length is None:
raise DriveError(
"Object storage did not return the recording size, "
"cannot stream it to Drive."
)
drive.upload_content(
policy_url=item["policy"],
stream=download.raw,
content_length=int(content_length),
content_type=download.headers.get(
"Content-Type", "application/octet-stream"
),
)
drive.complete_upload(item["id"])
except (DriveError, requests.RequestException):
logger.exception("Failed to push recording %s to Drive", recording_id)
return False
finally:
recording.clear_owner_access_token()
logger.info(
"Recording %s pushed to Drive as '%s' (item %s)",
recording_id,
filename,
item["id"],
)
return True
@@ -117,7 +117,7 @@ def test_api_files_create_file_authenticated_success():
policy_parsed = urlparse(policy)
assert policy_parsed.scheme == "http"
assert policy_parsed.netloc in ["minio:9000", "localhost:9000"]
assert policy_parsed.netloc in ["meet-minio:9000", "minio:9000", "localhost:9000"]
assert policy_parsed.path == f"/meet-media-storage/tmp/files/{file.id!s}.png"
query_params = parse_qs(policy_parsed.query)
@@ -0,0 +1,324 @@
"""
Test pushing a recording to the owner's Drive.
"""
# pylint: disable=redefined-outer-name,unused-argument
from unittest import mock
from django.test import override_settings
import pytest
import responses
from core import factories, models
from core.recording.event.notification import NotificationService
from core.tasks.push_recording import push_recording
pytestmark = pytest.mark.django_db
DRIVE_API = "https://drive.test/external_api/v1.0"
DOWNLOAD_URL = "https://storage.test/recordings/recording.mp4"
# Signed by Drive with the object storage domain meant for its browser clients.
UPLOAD_URL = "http://drive-storage.test:9100/drive-media/item?X-Amz-Signature=deadbeef"
INTERNAL_UPLOAD_URL = "http://drive-minio:9000/drive-media/item"
WORKSPACE_ID = "11111111-1111-4111-8111-111111111111"
ITEM_ID = "22222222-2222-4222-8222-222222222222"
RECORDING_CONTENT = b"fake-recording-bytes"
@pytest.fixture
def recording_with_token():
"""Return a recording carrying a parked access token."""
recording = factories.RecordingFactory(
mode=models.RecordingModeChoices.SCREEN_RECORDING
)
recording.set_owner_access_token("user-access-token")
return recording
@pytest.fixture
def mocked_download_url():
"""Avoid signing a real S3 URL, the object storage is not the point here."""
with mock.patch(
"core.utils.generate_download_s3_url", return_value=DOWNLOAD_URL
) as patched:
yield patched
@pytest.fixture
def upload():
"""Capture what gets PUT to the presigned URL.
The upload sends a stream, but "responses" drains file-like bodies before
handing the request over, so what lands here is already the bytes.
"""
captured = {}
def callback(request):
captured["url"] = request.url
captured["headers"] = request.headers
captured["body"] = request.body
return 200, {}, ""
captured["callback"] = callback
return captured
def register_download(with_content_length=True):
"""Register the object storage response holding the recording bytes."""
headers = (
{"Content-Length": str(len(RECORDING_CONTENT))} if with_content_length else None
)
responses.add(
responses.GET,
DOWNLOAD_URL,
body=RECORDING_CONTENT,
status=200,
headers=headers,
content_type="video/mp4",
)
def register_drive(upload=None, workspaces=None):
"""Register the Drive API calls of a successful upload."""
responses.add(
responses.GET,
f"{DRIVE_API}/items/",
json={
"results": workspaces
if workspaces is not None
else [
{"id": "shared-workspace", "main_workspace": False},
{"id": WORKSPACE_ID, "main_workspace": True},
],
"next": None,
},
status=200,
)
responses.add(
responses.POST,
f"{DRIVE_API}/items/{WORKSPACE_ID}/children/",
json={"id": ITEM_ID, "policy": UPLOAD_URL},
status=201,
)
if upload is not None:
responses.add_callback(responses.PUT, UPLOAD_URL, callback=upload["callback"])
responses.add_callback(
responses.PUT, INTERNAL_UPLOAD_URL, callback=upload["callback"]
)
responses.add(
responses.POST,
f"{DRIVE_API}/items/{ITEM_ID}/upload-ended/",
json={},
status=200,
)
@override_settings(DRIVE_API_BASE_URL=DRIVE_API, DRIVE_UPLOAD_STORAGE_NETLOC=None)
@responses.activate
def test_push_recording_uploads_to_the_main_workspace(
recording_with_token, mocked_download_url, upload
):
"""The recording is created in the main workspace, uploaded, then confirmed."""
register_download()
register_drive(upload=upload)
assert push_recording(str(recording_with_token.id)) is True
workspaces_call, create_call, ended_call = (
responses.calls[0].request,
responses.calls[1].request,
responses.calls[4].request,
)
assert workspaces_call.url == f"{DRIVE_API}/items/"
assert workspaces_call.headers["Authorization"] == "Bearer user-access-token"
room = recording_with_token.room
expected_filename = (
f"{room.slug}-{recording_with_token.created_at:%Y-%m-%d-%H-%M}.mp4"
)
assert expected_filename in create_call.body.decode()
assert upload["url"] == UPLOAD_URL
assert upload["body"] == RECORDING_CONTENT
assert upload["headers"]["Content-Length"] == str(len(RECORDING_CONTENT))
assert upload["headers"]["Content-Type"] == "video/mp4"
assert upload["headers"]["x-amz-acl"] == "private"
# The presigned URL carries its own credentials, an extra Authorization
# header would make the object storage reject the upload.
assert "Authorization" not in upload["headers"]
assert ended_call.url == f"{DRIVE_API}/items/{ITEM_ID}/upload-ended/"
@override_settings(DRIVE_API_BASE_URL=DRIVE_API, DRIVE_UPLOAD_STORAGE_NETLOC=None)
@responses.activate
def test_push_recording_drops_the_access_token(
recording_with_token, mocked_download_url, upload
):
"""The parked credential does not outlive the push."""
register_download()
register_drive(upload=upload)
push_recording(str(recording_with_token.id))
recording_with_token.refresh_from_db()
assert recording_with_token.owner_access_token is None
@override_settings(DRIVE_API_BASE_URL=DRIVE_API, DRIVE_UPLOAD_STORAGE_NETLOC=None)
@responses.activate
def test_push_recording_drops_the_access_token_on_failure(
recording_with_token, mocked_download_url, upload
):
"""A failed push does not leave the credential behind either."""
register_download()
register_drive(
upload=upload, workspaces=[{"id": "shared", "main_workspace": False}]
)
assert push_recording(str(recording_with_token.id)) is False
recording_with_token.refresh_from_db()
assert recording_with_token.owner_access_token is None
@override_settings(DRIVE_API_BASE_URL=DRIVE_API)
@responses.activate
def test_push_recording_without_parked_token():
"""Without a token there is nobody to push on behalf of, so nothing happens."""
recording = factories.RecordingFactory()
assert push_recording(str(recording.id)) is False
assert not responses.calls
@override_settings(DRIVE_API_BASE_URL=DRIVE_API)
@responses.activate
def test_push_recording_unknown_recording():
"""An unknown recording is reported, not raised."""
assert push_recording("33333333-3333-4333-8333-333333333333") is False
assert not responses.calls
@override_settings(
DRIVE_API_BASE_URL=DRIVE_API, DRIVE_UPLOAD_STORAGE_NETLOC="drive-minio:9000"
)
@responses.activate
def test_push_recording_rewrites_the_upload_host(
recording_with_token, mocked_download_url, upload
):
"""The upload reaches the internal address while announcing the signed host.
The presigned signature covers the Host header, so it has to stay untouched
even when the address we connect to does not.
"""
register_download()
register_drive(upload=upload)
assert push_recording(str(recording_with_token.id)) is True
assert upload["url"].startswith(INTERNAL_UPLOAD_URL)
assert upload["headers"]["Host"] == "drive-storage.test:9100"
assert upload["body"] == RECORDING_CONTENT
@override_settings(DRIVE_API_BASE_URL=DRIVE_API, DRIVE_UPLOAD_STORAGE_NETLOC=None)
@responses.activate
def test_push_recording_without_content_length(
recording_with_token, mocked_download_url, upload
):
"""A size-less download cannot be relayed, and is reported as a failure."""
register_download(with_content_length=False)
register_drive(upload=upload)
assert push_recording(str(recording_with_token.id)) is False
assert "body" not in upload
@override_settings(DRIVE_API_BASE_URL=None)
def test_push_recording_without_drive_configured(
recording_with_token, mocked_download_url
):
"""An unconfigured Drive is reported, not raised."""
assert push_recording(str(recording_with_token.id)) is False
def test_recording_access_token_roundtrip():
"""The parked token is encrypted at rest and read back as-is."""
recording = factories.RecordingFactory()
recording.set_owner_access_token("user-access-token")
recording.refresh_from_db()
assert recording.owner_access_token != "user-access-token"
assert recording.get_owner_access_token() == "user-access-token"
recording.clear_owner_access_token()
recording.refresh_from_db()
assert recording.get_owner_access_token() is None
def test_recording_access_token_undecryptable():
"""A token encrypted with another key is reported as missing, not raised."""
recording = factories.RecordingFactory(owner_access_token="not-a-fernet-token")
assert recording.get_owner_access_token() is None
@pytest.mark.parametrize("enabled", [True, False])
def test_notify_external_services_schedules_the_push(enabled):
"""The push is scheduled from the notification flow, when enabled."""
recording = factories.RecordingFactory(
mode=models.RecordingModeChoices.SCREEN_RECORDING
)
recording.set_owner_access_token("user-access-token")
with (
override_settings(RECORDING_PUSH_TO_DRIVE_ENABLED=enabled),
mock.patch("core.recording.event.notification.push_recording") as mocked_push,
mock.patch.object(
NotificationService, "_notify_user_by_email", return_value=True
),
):
NotificationService().notify_external_services(recording)
assert mocked_push.delay.called is enabled
def test_notify_external_services_without_parked_token():
"""No token means nothing to push with, so no task is scheduled."""
recording = factories.RecordingFactory(
mode=models.RecordingModeChoices.SCREEN_RECORDING
)
with (
override_settings(RECORDING_PUSH_TO_DRIVE_ENABLED=True),
mock.patch("core.recording.event.notification.push_recording") as mocked_push,
mock.patch.object(
NotificationService, "_notify_user_by_email", return_value=True
),
):
NotificationService().notify_external_services(recording)
assert not mocked_push.delay.called
@@ -1 +0,0 @@
"""Tests for the roomkit API of the Meet core app."""
@@ -1,266 +0,0 @@
"""
Test the roomkit join server-to-server API endpoint.
"""
# pylint: disable=redefined-outer-name,unused-argument
from unittest import mock
import pytest
from ...factories import RoomFactory
from ...services.sip_management import SIPException
pytestmark = pytest.mark.django_db
@pytest.fixture
def mock_sip_management():
"""Mock the SIPManagement used by the roomkit viewset."""
with mock.patch("core.roomkit.viewsets.SIPManagement") as mock_service_class:
yield mock_service_class.return_value
def test_join_anonymous(settings, mock_sip_management, client):
"""Requests without an Authorization header should be rejected."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
response = client.post("/api/v1.0/roomkit/join/", {"pin_code": room.pin_code})
assert response.status_code == 401
assert response.json() == {"detail": "Authorization header is missing."}
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_malformed_authorization_header(settings, mock_sip_management, client):
"""Requests with a malformed Authorization header should be rejected."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="testAuthToken",
)
assert response.status_code == 401
assert response.json() == {"detail": "Invalid authorization header."}
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_wrong_bearer(settings, mock_sip_management, client):
"""Requests with an incorrect bearer token should be rejected."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="Bearer wrongAuthToken",
)
assert response.status_code == 401
assert response.json() == {"detail": "Invalid server-to-server token."}
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_token_not_configured(settings, mock_sip_management, client):
"""Requests should be rejected when no server-to-server token is configured."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = None
room = RoomFactory(pin_code="1234567890")
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 401
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_roomkit_disabled(settings, mock_sip_management, client):
"""The endpoint should not be exposed when the roomkit integration is disabled."""
settings.ROOMKIT_ENABLED = False
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 404
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_missing_pin(settings, mock_sip_management, client):
"""Requests without a PIN code should be rejected."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
response = client.post(
"/api/v1.0/roomkit/join/",
{},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 400
assert response.json() == {"pin_code": ["This field is required."]}
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_blank_pin(settings, mock_sip_management, client):
"""Requests with a blank PIN code should be rejected."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": ""},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 400
assert response.json() == {"pin_code": ["This field may not be blank."]}
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_wrong_pin_length(settings, mock_sip_management, client):
"""Requests with a PIN code of unexpected length should be rejected."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
settings.ROOM_TELEPHONY_PIN_LENGTH = 10
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": "123"},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 400
assert response.json() == {"pin_code": ["PIN code length is invalid."]}
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_unknown_pin(settings, mock_sip_management, client):
"""Requests with a PIN matching no room should return 404 and create no rule."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
RoomFactory(pin_code="1234567890")
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": "0987654321"},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 404
assert response.json() == {"detail": "No room found for this PIN code."}
mock_sip_management.ensure_dispatch_rule.assert_not_called()
def test_join_success(settings, mock_sip_management, client):
"""Requests with a valid PIN should create the dispatch rule."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
mock_sip_management.ensure_dispatch_rule.return_value = True
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 200
assert response.json() == {"status": "success"}
mock_sip_management.ensure_dispatch_rule.assert_called_once_with(room)
def test_join_dispatch_rule_already_exists(settings, mock_sip_management, client):
"""Requests should succeed when the dispatch rule already exists (idempotency)."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
mock_sip_management.ensure_dispatch_rule.return_value = False
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 200
assert response.json() == {"status": "success"}
mock_sip_management.ensure_dispatch_rule.assert_called_once_with(room)
def test_join_tracks_analytics_event(settings, mock_sip_management, client):
"""Successful joins should be tracked with an analytics event."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
mock_sip_management.ensure_dispatch_rule.return_value = True
with mock.patch("core.roomkit.viewsets.analytics.capture") as mock_capture:
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="Bearer testAuthToken",
)
assert response.status_code == 200
mock_capture.assert_called_once()
_user, event, properties = mock_capture.call_args[0]
assert str(event) == "roomkit_joined"
assert properties == {
"room_id": str(room.pk),
"dispatch_rule_created": True,
}
def test_join_sip_failure(settings, mock_sip_management, client):
"""Requests should fail with a server error when the sip management service fails."""
settings.ROOMKIT_ENABLED = True
settings.ROOMKIT_SERVER_TO_SERVER_API_TOKEN = "testAuthToken"
room = RoomFactory(pin_code="1234567890")
mock_sip_management.ensure_dispatch_rule.side_effect = SIPException(
"Could not create dispatch rule"
)
with mock.patch("core.roomkit.viewsets.analytics.capture") as mock_capture:
response = client.post(
"/api/v1.0/roomkit/join/",
{"pin_code": room.pin_code},
HTTP_AUTHORIZATION="Bearer testAuthToken",
raise_request_exception=False,
)
assert response.status_code == 500
mock_sip_management.ensure_dispatch_rule.assert_called_once_with(room)
mock_capture.assert_not_called()
@@ -6,6 +6,8 @@ Test LiveKitEvents service.
import uuid
from unittest import mock
from django.test import override_settings
import pytest
from livekit.api import EgressStatus
@@ -21,10 +23,7 @@ from core.services.livekit_events import (
)
from core.services.lobby import LobbyService
from core.services.room_management import RoomManagementException
from core.services.sip_management import (
SIPException,
SIPManagement,
)
from core.services.telephony import TelephonyException, TelephonyService
from core.utils import NotificationError
pytestmark = pytest.mark.django_db
@@ -62,7 +61,7 @@ def test_initialization(
mock_token_verifier.assert_called_once_with(api_key, api_secret)
mock_webhook_receiver.assert_called_once_with(mock_token_verifier.return_value)
assert isinstance(service.lobby_service, LobbyService)
assert isinstance(service.sip_management, SIPManagement)
assert isinstance(service.telephony_service, TelephonyService)
assert isinstance(service.recording_events, RecordingEventsService)
@@ -75,6 +74,8 @@ def test_initialization(
)
@mock.patch("core.utils.notify_participants")
@mock.patch("core.services.room_management.RoomManagement.update_metadata")
# Without storage events, completion falls back to the egress event itself.
@override_settings(RECORDING_STORAGE_EVENT_ENABLE=False)
def test_handle_egress_ended_success(
mock_update_metadata, mock_notify, mode, notification_type, service
):
@@ -158,6 +159,8 @@ def test_handle_egress_updated_non_handled(
)
@mock.patch("core.utils.notify_participants")
@mock.patch("core.services.room_management.RoomManagement.update_metadata")
# Without storage events, completion falls back to the egress event itself.
@override_settings(RECORDING_STORAGE_EVENT_ENABLE=False)
def test_handle_egress_ended_metadata_update_fails(
mock_update_metadata, mock_notify, mode, notification_type, service
):
@@ -472,11 +475,11 @@ def test_handle_egress_ended_ignores_non_savable_recording(
@mock.patch.object(LobbyService, "clear_room_cache")
@mock.patch.object(SIPManagement, "delete_dispatch_rule")
@mock.patch.object(TelephonyService, "delete_dispatch_rule")
def test_handle_room_finished_clears_cache_and_deletes_dispatch_rule(
mock_delete_dispatch_rule, mock_clear_cache, service, settings
):
"""Should clear lobby cache and delete SIP dispatch rule when room finishes."""
"""Should clear lobby cache and delete telephony dispatch rule when room finishes."""
settings.ROOM_TELEPHONY_ENABLED = True
mock_room_name = uuid.uuid4()
mock_data = mock.MagicMock()
@@ -489,31 +492,12 @@ def test_handle_room_finished_clears_cache_and_deletes_dispatch_rule(
@mock.patch.object(LobbyService, "clear_room_cache")
@mock.patch.object(SIPManagement, "delete_dispatch_rule")
def test_handle_room_finished_deletes_dispatch_rule_when_only_roomkit_enabled(
mock_delete_dispatch_rule, mock_clear_cache, service, settings
):
"""Should delete dispatch rule when only roomkit is enabled when room finishes."""
settings.ROOM_TELEPHONY_ENABLED = False
settings.ROOMKIT_ENABLED = True
mock_room_name = uuid.uuid4()
mock_data = mock.MagicMock()
mock_data.room.name = str(mock_room_name)
service._handle_room_finished(mock_data)
mock_delete_dispatch_rule.assert_called_once_with(mock_room_name)
mock_clear_cache.assert_called_once_with(mock_room_name)
@mock.patch.object(LobbyService, "clear_room_cache")
@mock.patch.object(SIPManagement, "delete_dispatch_rule")
@mock.patch.object(TelephonyService, "delete_dispatch_rule")
def test_handle_room_finished_skips_telephony_when_disabled(
mock_delete_dispatch_rule, mock_clear_cache, service, settings
):
"""Should clear lobby cache but skip dispatch rule deletion when telephony is disabled."""
settings.ROOM_TELEPHONY_ENABLED = False
settings.ROOMKIT_ENABLED = False
mock_room_name = uuid.uuid4()
mock_data = mock.MagicMock()
mock_data.room.name = str(mock_room_name)
@@ -527,7 +511,7 @@ def test_handle_room_finished_skips_telephony_when_disabled(
@mock.patch.object(
LobbyService, "clear_room_cache", side_effect=Exception("Test error")
)
@mock.patch.object(SIPManagement, "delete_dispatch_rule")
@mock.patch.object(TelephonyService, "delete_dispatch_rule")
def test_handle_room_finished_raises_error_when_cache_clearing_fails(
mock_delete_dispatch_rule, mock_clear_cache, service, settings
):
@@ -550,9 +534,9 @@ def test_handle_room_finished_raises_error_when_cache_clearing_fails(
@mock.patch.object(LobbyService, "clear_room_cache")
@mock.patch.object(
SIPManagement,
TelephonyService,
"delete_dispatch_rule",
side_effect=SIPException("Test error"),
side_effect=TelephonyException("Test error"),
)
def test_handle_room_finished_raises_error_when_telephony_deletion_fails(
mock_delete_dispatch_rule, mock_clear_cache, service, settings
@@ -563,7 +547,7 @@ def test_handle_room_finished_raises_error_when_telephony_deletion_fails(
mock_data.room.name = "00000000-0000-0000-0000-000000000000"
expected_error = (
"Failed to delete sip dispatch rule for room "
"Failed to delete telephony dispatch rule for room "
"00000000-0000-0000-0000-000000000000"
)
@@ -584,11 +568,11 @@ def test_handle_room_finished_raises_error_for_invalid_room_name(service):
service._handle_room_finished(mock_data)
@mock.patch.object(SIPManagement, "ensure_dispatch_rule")
@mock.patch.object(TelephonyService, "create_dispatch_rule")
def test_handle_room_started_creates_dispatch_rule_successfully(
mock_ensure_dispatch_rule, service, settings
mock_create_dispatch_rule, service, settings
):
"""Should ensure the SIP dispatch rule exists when room starts successfully."""
"""Should create telephony dispatch rule when room starts successfully."""
settings.ROOM_TELEPHONY_ENABLED = True
room = RoomFactory()
mock_data = mock.MagicMock()
@@ -596,75 +580,22 @@ def test_handle_room_started_creates_dispatch_rule_successfully(
service._handle_room_started(mock_data)
mock_ensure_dispatch_rule.assert_called_once_with(room)
mock_create_dispatch_rule.assert_called_once_with(room)
@mock.patch.object(SIPManagement, "ensure_dispatch_rule")
def test_handle_room_started_creates_dispatch_rule_when_only_roomkit_enabled(
mock_ensure_dispatch_rule, service, settings
):
"""Should ensure the dispatch rule exists when only roomkit is enabled during room start."""
settings.ROOM_TELEPHONY_ENABLED = False
settings.ROOMKIT_ENABLED = True
room = RoomFactory()
mock_data = mock.MagicMock()
mock_data.room.name = str(room.id)
service._handle_room_started(mock_data)
mock_ensure_dispatch_rule.assert_called_once_with(room)
@mock.patch.object(SIPManagement, "ensure_dispatch_rule", return_value=False)
def test_handle_room_started_ignores_existing_dispatch_rule(
mock_ensure_dispatch_rule, service, settings
):
"""Should proceed silently when the dispatch rule already exists when room starts."""
settings.ROOM_TELEPHONY_ENABLED = True
room = RoomFactory()
mock_data = mock.MagicMock()
mock_data.room.name = str(room.id)
# ensure_dispatch_rule reports the rule as pre-existing: nothing to raise
service._handle_room_started(mock_data)
mock_ensure_dispatch_rule.assert_called_once_with(room)
@mock.patch.object(
SIPManagement,
"ensure_dispatch_rule",
side_effect=SIPException("Test error"),
)
def test_handle_room_started_raises_error_when_dispatch_rule_creation_fails(
mock_ensure_dispatch_rule, service, settings
):
"""Should raise ActionFailedError when ensuring the dispatch rule fails when room starts."""
settings.ROOM_TELEPHONY_ENABLED = True
room = RoomFactory()
mock_data = mock.MagicMock()
mock_data.room.name = str(room.id)
expected_error = f"Failed to create sip dispatch rule for room {room.id}"
with pytest.raises(ActionFailedError, match=expected_error):
service._handle_room_started(mock_data)
@mock.patch.object(SIPManagement, "ensure_dispatch_rule")
@mock.patch.object(TelephonyService, "create_dispatch_rule")
def test_handle_room_started_skips_dispatch_rule_when_telephony_disabled(
mock_ensure_dispatch_rule, service, settings
mock_create_dispatch_rule, service, settings
):
"""Should skip ensuring the SIP dispatch rule when telephony is disabled during room start."""
"""Should skip creating telephony dispatch rule when telephony is disabled during room start."""
settings.ROOM_TELEPHONY_ENABLED = False
settings.ROOMKIT_ENABLED = False
room = RoomFactory()
mock_data = mock.MagicMock()
mock_data.room.name = str(room.id)
service._handle_room_started(mock_data)
mock_ensure_dispatch_rule.assert_not_called()
mock_create_dispatch_rule.assert_not_called()
def test_handle_room_started_raises_error_for_invalid_room_name(service):
@@ -1,5 +1,5 @@
"""
Test SIP mamagement service.
Test telephony service.
"""
# pylint: disable=W0212
@@ -20,11 +20,7 @@ from livekit.protocol.sip import (
from core.factories import RoomFactory
from core.models import RoomAccessLevel
from core.services.sip_management import (
DispatchRuleConflictError,
SIPException,
SIPManagement,
)
from core.services.telephony import TelephonyException, TelephonyService
pytestmark = pytest.mark.django_db
@@ -39,9 +35,9 @@ def create_mock_livekit_client():
def test_rule_name():
"""Test rule name generation."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
rule_name = sip_management._rule_name(room.id)
rule_name = telephony_service._rule_name(room.id)
assert rule_name == f"SIP_{str(room.id)}"
@@ -49,14 +45,14 @@ def test_rule_name():
@mock.patch("core.utils.create_livekit_client")
def test_create_dispatch_rule_success(mock_client_factory):
"""Test successful dispatch rule creation."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
mock_api.sip.create_sip_dispatch_rule = mock.AsyncMock()
mock_client_factory.return_value = mock_api
sip_management.create_dispatch_rule(room)
telephony_service.create_dispatch_rule(room)
mock_api.sip.create_sip_dispatch_rule.assert_called_once()
create_request = mock_api.sip.create_sip_dispatch_rule.call_args[1]["create"]
@@ -71,7 +67,7 @@ def test_create_dispatch_rule_success(mock_client_factory):
@mock.patch("core.utils.create_livekit_client")
def test_create_dispatch_rule_api_failure(mock_client_factory):
"""Test dispatch rule creation when API fails."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
@@ -80,8 +76,8 @@ def test_create_dispatch_rule_api_failure(mock_client_factory):
)
mock_client_factory.return_value = mock_api
with pytest.raises(SIPException, match="Could not create dispatch rule"):
sip_management.create_dispatch_rule(room)
with pytest.raises(TelephonyException, match="Could not create dispatch rule"):
telephony_service.create_dispatch_rule(room)
mock_api.sip.create_sip_dispatch_rule.assert_called_once()
mock_api.aclose.assert_called_once()
@@ -90,7 +86,7 @@ def test_create_dispatch_rule_api_failure(mock_client_factory):
@mock.patch("core.utils.create_livekit_client")
def test_list_dispatch_rules_ids_success(mock_client_factory):
"""Test successful listing of dispatch rule IDs."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_rules = [
@@ -115,7 +111,7 @@ def test_list_dispatch_rules_ids_success(mock_client_factory):
)
mock_client_factory.return_value = mock_api
result = async_to_sync(sip_management._list_dispatch_rules_ids)(room.id)
result = async_to_sync(telephony_service._list_dispatch_rules_ids)(room.id)
assert len(result) == 2
assert "rule-1" in result
@@ -131,7 +127,7 @@ def test_list_dispatch_rules_ids_success(mock_client_factory):
@mock.patch("core.utils.create_livekit_client")
def test_list_dispatch_rules_ids_empty_response(mock_client_factory):
"""Test listing dispatch rule IDs when no rules exist."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
@@ -140,7 +136,7 @@ def test_list_dispatch_rules_ids_empty_response(mock_client_factory):
)
mock_client_factory.return_value = mock_api
result = async_to_sync(sip_management._list_dispatch_rules_ids)(room.id)
result = async_to_sync(telephony_service._list_dispatch_rules_ids)(room.id)
assert result == []
mock_api.aclose.assert_called_once()
@@ -149,7 +145,7 @@ def test_list_dispatch_rules_ids_empty_response(mock_client_factory):
@mock.patch("core.utils.create_livekit_client")
def test_list_dispatch_rules_ids_no_matching_rules(mock_client_factory):
"""Test listing dispatch rule IDs when no rules match the room."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_rules = [
@@ -167,7 +163,7 @@ def test_list_dispatch_rules_ids_no_matching_rules(mock_client_factory):
)
mock_client_factory.return_value = mock_api
result = async_to_sync(sip_management._list_dispatch_rules_ids)(room.id)
result = async_to_sync(telephony_service._list_dispatch_rules_ids)(room.id)
assert result == []
mock_api.aclose.assert_called_once()
@@ -176,7 +172,7 @@ def test_list_dispatch_rules_ids_no_matching_rules(mock_client_factory):
@mock.patch("core.utils.create_livekit_client")
def test_list_dispatch_rules_ids_api_failure(mock_client_factory):
"""Test listing dispatch rule IDs when API fails."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
@@ -185,34 +181,34 @@ def test_list_dispatch_rules_ids_api_failure(mock_client_factory):
)
mock_client_factory.return_value = mock_api
with pytest.raises(SIPException, match="Could not list dispatch rules"):
async_to_sync(sip_management._list_dispatch_rules_ids)(room.id)
with pytest.raises(TelephonyException, match="Could not list dispatch rules"):
async_to_sync(telephony_service._list_dispatch_rules_ids)(room.id)
mock_api.sip.list_sip_dispatch_rule.assert_called_once()
mock_api.aclose.assert_called_once()
@mock.patch("core.services.sip_management.SIPManagement._list_dispatch_rules_ids")
@mock.patch("core.services.telephony.TelephonyService._list_dispatch_rules_ids")
@mock.patch("core.utils.create_livekit_client")
def test_delete_dispatch_rule_no_rules(mock_client_factory, mock_list_rules):
"""Test deleting dispatch rules when no rules exist."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_list_rules.return_value = []
result = sip_management.delete_dispatch_rule(room.id)
result = telephony_service.delete_dispatch_rule(room.id)
assert result is False
mock_list_rules.assert_called_once_with(room.id)
mock_client_factory.assert_not_called()
@mock.patch("core.services.sip_management.SIPManagement._list_dispatch_rules_ids")
@mock.patch("core.services.telephony.TelephonyService._list_dispatch_rules_ids")
@mock.patch("core.utils.create_livekit_client")
def test_delete_dispatch_rule_single_rule(mock_client_factory, mock_list_rules):
"""Test deleting a single dispatch rule."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_list_rules.return_value = ["rule-1"]
@@ -220,7 +216,7 @@ def test_delete_dispatch_rule_single_rule(mock_client_factory, mock_list_rules):
mock_api.sip.delete_sip_dispatch_rule = mock.AsyncMock()
mock_client_factory.return_value = mock_api
result = sip_management.delete_dispatch_rule(room.id)
result = telephony_service.delete_dispatch_rule(room.id)
assert result is True
mock_api.sip.delete_sip_dispatch_rule.assert_called_once()
@@ -230,11 +226,11 @@ def test_delete_dispatch_rule_single_rule(mock_client_factory, mock_list_rules):
mock_api.aclose.assert_called_once()
@mock.patch("core.services.sip_management.SIPManagement._list_dispatch_rules_ids")
@mock.patch("core.services.telephony.TelephonyService._list_dispatch_rules_ids")
@mock.patch("core.utils.create_livekit_client")
def test_delete_dispatch_rule_multiple_rules(mock_client_factory, mock_list_rules):
"""Test deleting multiple dispatch rules."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_list_rules.return_value = ["rule-1", "rule-2", "rule-3"]
@@ -242,7 +238,7 @@ def test_delete_dispatch_rule_multiple_rules(mock_client_factory, mock_list_rule
mock_api.sip.delete_sip_dispatch_rule = mock.AsyncMock()
mock_client_factory.return_value = mock_api
result = sip_management.delete_dispatch_rule(room.id)
result = telephony_service.delete_dispatch_rule(room.id)
assert result is True
assert mock_api.sip.delete_sip_dispatch_rule.call_count == 3
@@ -257,11 +253,11 @@ def test_delete_dispatch_rule_multiple_rules(mock_client_factory, mock_list_rule
mock_api.aclose.assert_called_once()
@mock.patch("core.services.sip_management.SIPManagement._list_dispatch_rules_ids")
@mock.patch("core.services.telephony.TelephonyService._list_dispatch_rules_ids")
@mock.patch("core.utils.create_livekit_client")
def test_delete_dispatch_rule_partial_failure(mock_client_factory, mock_list_rules):
"""Test deleting multiple dispatch rules when one deletion fails."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_list_rules.return_value = ["rule-1", "rule-2", "rule-3"]
@@ -281,18 +277,18 @@ def test_delete_dispatch_rule_partial_failure(mock_client_factory, mock_list_rul
)
mock_client_factory.return_value = mock_api
with pytest.raises(SIPException, match="Could not delete dispatch rules"):
sip_management.delete_dispatch_rule(room.id)
with pytest.raises(TelephonyException, match="Could not delete dispatch rules"):
telephony_service.delete_dispatch_rule(room.id)
assert mock_api.sip.delete_sip_dispatch_rule.call_count == 2
mock_api.aclose.assert_called_once()
@mock.patch("core.services.sip_management.SIPManagement._list_dispatch_rules_ids")
@mock.patch("core.services.telephony.TelephonyService._list_dispatch_rules_ids")
@mock.patch("core.utils.create_livekit_client")
def test_delete_dispatch_rule_api_failure(mock_client_factory, mock_list_rules):
"""Test deleting dispatch rules when API fails immediately."""
sip_management = SIPManagement()
telephony_service = TelephonyService()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_list_rules.return_value = ["rule-1"]
@@ -302,131 +298,8 @@ def test_delete_dispatch_rule_api_failure(mock_client_factory, mock_list_rules):
)
mock_client_factory.return_value = mock_api
with pytest.raises(SIPException, match="Could not delete dispatch rules"):
sip_management.delete_dispatch_rule(room.id)
with pytest.raises(TelephonyException, match="Could not delete dispatch rules"):
telephony_service.delete_dispatch_rule(room.id)
mock_api.sip.delete_sip_dispatch_rule.assert_called_once()
mock_api.aclose.assert_called_once()
@mock.patch("core.utils.create_livekit_client")
def test_create_dispatch_rule_conflict_raises_dedicated_error(mock_client_factory):
"""Test that a LiveKit conflict error raises DispatchRuleConflictError."""
sip_management = SIPManagement()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
mock_api.sip.create_sip_dispatch_rule = mock.AsyncMock(
side_effect=TwirpError(
msg=(
"Dispatch rule for the same trunk, inbound number, number, and "
"PIN combination already exists in dispatch rule"
),
code="already_exists",
status=409,
)
)
mock_client_factory.return_value = mock_api
with pytest.raises(DispatchRuleConflictError):
sip_management.create_dispatch_rule(room)
mock_api.aclose.assert_called_once()
@mock.patch("core.utils.create_livekit_client")
def test_ensure_dispatch_rule_creates_when_missing(mock_client_factory):
"""Test that ensure_dispatch_rule creates the rule when none exists."""
sip_management = SIPManagement()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
mock_api.sip.list_sip_dispatch_rule = mock.AsyncMock(
return_value=ListSIPDispatchRuleResponse(items=[])
)
mock_api.sip.create_sip_dispatch_rule = mock.AsyncMock()
mock_client_factory.return_value = mock_api
created = sip_management.ensure_dispatch_rule(room)
assert created is True
mock_api.sip.create_sip_dispatch_rule.assert_called_once()
create_request = mock_api.sip.create_sip_dispatch_rule.call_args[1]["create"]
assert isinstance(create_request, CreateSIPDispatchRuleRequest)
assert create_request.name == f"SIP_{str(room.id)}"
assert create_request.rule.dispatch_rule_direct.room_name == str(room.id)
assert create_request.rule.dispatch_rule_direct.pin == str(room.pin_code)
@mock.patch("core.utils.create_livekit_client")
def test_ensure_dispatch_rule_skips_when_existing(mock_client_factory):
"""Test that ensure_dispatch_rule is idempotent when the rule already exists."""
sip_management = SIPManagement()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
existing_rule = SIPDispatchRuleInfo(
sip_dispatch_rule_id="rule-1", name=f"SIP_{str(room.id)}"
)
mock_api = create_mock_livekit_client()
mock_api.sip.list_sip_dispatch_rule = mock.AsyncMock(
return_value=ListSIPDispatchRuleResponse(items=[existing_rule])
)
mock_api.sip.create_sip_dispatch_rule = mock.AsyncMock()
mock_client_factory.return_value = mock_api
created = sip_management.ensure_dispatch_rule(room)
assert created is False
mock_api.sip.create_sip_dispatch_rule.assert_not_called()
@mock.patch("core.utils.create_livekit_client")
def test_ensure_dispatch_rule_returns_false_on_conflict(mock_client_factory):
"""Test that ensure_dispatch_rule tolerates a concurrent rule creation.
If the rule is created by a concurrent caller (e.g. the LiveKit webhook)
between the existence check and the creation, LiveKit rejects the
duplicate and ensure_dispatch_rule reports the rule as already existing.
"""
sip_management = SIPManagement()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
mock_api.sip.list_sip_dispatch_rule = mock.AsyncMock(
return_value=ListSIPDispatchRuleResponse(items=[])
)
mock_api.sip.create_sip_dispatch_rule = mock.AsyncMock(
side_effect=TwirpError(
msg=(
"Dispatch rule for the same trunk, inbound number, number, and "
"PIN combination already exists in dispatch rule"
),
code="already_exists",
status=409,
)
)
mock_client_factory.return_value = mock_api
created = sip_management.ensure_dispatch_rule(room)
assert created is False
@mock.patch("core.utils.create_livekit_client")
def test_ensure_dispatch_rule_raises_on_other_failures(mock_client_factory):
"""Test that ensure_dispatch_rule propagates unexpected LiveKit failures."""
sip_management = SIPManagement()
room = RoomFactory(access_level=RoomAccessLevel.RESTRICTED, pin_code="1234")
mock_api = create_mock_livekit_client()
mock_api.sip.list_sip_dispatch_rule = mock.AsyncMock(
return_value=ListSIPDispatchRuleResponse(items=[])
)
mock_api.sip.create_sip_dispatch_rule = mock.AsyncMock(
side_effect=TwirpError(msg="Internal server error", code="unknown", status=500)
)
mock_client_factory.return_value = mock_api
with pytest.raises(SIPException, match="Could not create dispatch rule"):
sip_management.ensure_dispatch_rule(room)
+2 -15
View File
@@ -184,13 +184,12 @@ def test_models_rooms_is_public_property():
@mock.patch.object(Room, "generate_unique_pin_code")
def test_telephony_and_roomkit_disabled_skips_pin_generation(
def test_telephony_disabled_skips_pin_generation(
mock_generate_unique_pin_code, settings
):
"""Telephony and roomkit both disabled should not generate pin codes."""
"""Telephony disabled should not generate pin codes."""
settings.ROOM_TELEPHONY_ENABLED = False
settings.ROOMKIT_ENABLED = False
room = RoomFactory()
@@ -198,18 +197,6 @@ def test_telephony_and_roomkit_disabled_skips_pin_generation(
assert room.pin_code is None
def test_roomkit_enabled_generates_pin_code(settings):
"""Roomkit enabled alone should generate pin codes, even without telephony."""
settings.ROOM_TELEPHONY_ENABLED = False
settings.ROOMKIT_ENABLED = True
room = RoomFactory()
assert room.pin_code is not None
assert len(room.pin_code) == settings.ROOM_TELEPHONY_PIN_LENGTH
def test_default_and_custom_pin_length(settings):
"""Pin codes should be created with correct configured length."""
-6
View File
@@ -9,7 +9,6 @@ from rest_framework.routers import DefaultRouter, SimpleRouter
from core.addons import viewsets as addons_viewsets
from core.api import get_frontend_configuration, viewsets
from core.external_api import viewsets as external_viewsets
from core.roomkit import viewsets as roomkit_viewsets
# - Main endpoints
router = DefaultRouter()
@@ -20,11 +19,6 @@ router.register("files", viewsets.FileViewSet, basename="files")
router.register(
"resource-accesses", viewsets.ResourceAccessViewSet, basename="resource_accesses"
)
router.register(
"roomkit",
roomkit_viewsets.RoomKitViewSet,
basename="roomkit",
)
router.register(
"addons/sessions",
addons_viewsets.SessionViewSet,
+39
View File
@@ -17,6 +17,7 @@ from typing import List, Optional
from uuid import uuid4
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.core.files.storage import default_storage
import aiohttp
@@ -25,6 +26,7 @@ import botocore
import magic
import phonenumbers
from asgiref.sync import async_to_sync
from cryptography.fernet import Fernet, InvalidToken
from livekit.api import ( # pylint: disable=E0611
AccessToken,
ListRoomsRequest,
@@ -417,6 +419,43 @@ def generate_upload_policy(file):
return policy
class SecretDecryptionError(Exception):
"""Raised when a stored secret cannot be decrypted."""
@lru_cache(maxsize=1)
def get_cipher_suite():
"""Return the Fernet cipher suite used to encrypt secrets at rest.
Deliberately the same key as django-lasuite's OIDC token storage, so a
deployment only has one key to provision for user credentials.
"""
key = settings.OIDC_STORE_REFRESH_TOKEN_KEY
if not key:
raise ImproperlyConfigured("OIDC_STORE_REFRESH_TOKEN_KEY setting is required.")
return Fernet(key)
def encrypt_secret(value: str) -> str:
"""Encrypt a secret meant to be stored at rest."""
return get_cipher_suite().encrypt(value.encode()).decode()
def decrypt_secret(value: str) -> str:
"""Decrypt a secret stored by `encrypt_secret`."""
try:
return get_cipher_suite().decrypt(value.encode()).decode()
except InvalidToken as exc:
raise SecretDecryptionError(
"The stored secret could not be decrypted, was the key rotated?"
) from exc
def generate_download_s3_url(
key: str, *, expires_in: int, override_domain: bool = True
):
+37 -20
View File
@@ -349,11 +349,6 @@ class Base(Configuration):
environ_name="CREATION_CALLBACK_THROTTLE_RATES",
environ_prefix=None,
),
"roomkit_join": values.Value(
default="300/minute",
environ_name="ROOMKIT_JOIN_THROTTLE_RATES",
environ_prefix=None,
),
},
}
MONITORED_THROTTLE_FAILURE_CALLBACK = (
@@ -573,6 +568,20 @@ class Base(Configuration):
OIDC_STORE_ID_TOKEN = values.BooleanValue(
default=True, environ_name="OIDC_STORE_ID_TOKEN", environ_prefix=None
)
# Required to call other La Suite applications on behalf of the user, e.g.
# to push a recording to their Drive.
OIDC_STORE_ACCESS_TOKEN = values.BooleanValue(
default=False, environ_name="OIDC_STORE_ACCESS_TOKEN", environ_prefix=None
)
OIDC_STORE_REFRESH_TOKEN = values.BooleanValue(
default=False, environ_name="OIDC_STORE_REFRESH_TOKEN", environ_prefix=None
)
# Fernet key used to encrypt OIDC tokens at rest, both the refresh token
# django-lasuite stores in the session and the access token parked on a
# recording. Generate one with `Fernet.generate_key()`.
OIDC_STORE_REFRESH_TOKEN_KEY = SecretFileValue(
None, environ_name="OIDC_STORE_REFRESH_TOKEN_KEY", environ_prefix=None
)
ALLOW_LOGOUT_GET_METHOD = values.BooleanValue(
default=True, environ_name="ALLOW_LOGOUT_GET_METHOD", environ_prefix=None
)
@@ -725,6 +734,29 @@ class Base(Configuration):
None, environ_name="RECORDING_MAX_DURATION", environ_prefix=None
)
# Push recordings to Drive
# Once a recording is over, it is pushed to the main workspace of the user who
# started it, using their OIDC access token. It requires OIDC_STORE_ACCESS_TOKEN,
# and Drive to be configured as an OIDC resource server accepting Meet's audience.
RECORDING_PUSH_TO_DRIVE_ENABLED = values.BooleanValue(
False, environ_name="RECORDING_PUSH_TO_DRIVE_ENABLED", environ_prefix=None
)
# Base URL of Drive's external API, e.g. https://drive.example.com/external_api/v1.0
DRIVE_API_BASE_URL = values.Value(
None, environ_name="DRIVE_API_BASE_URL", environ_prefix=None
)
# Lifetime of the signed URL the worker downloads the recording from.
RECORDING_PUSH_TO_DRIVE_SIGNED_URL_EXPIRY_SECONDS = values.PositiveIntegerValue(
60 * 60,
environ_name="RECORDING_PUSH_TO_DRIVE_SIGNED_URL_EXPIRY_SECONDS",
environ_prefix=None,
)
# Development only: host:port to reach Drive's object storage at, when the
# domain Drive signs its upload URLs with is only resolvable from a browser.
DRIVE_UPLOAD_STORAGE_NETLOC = values.Value(
None, environ_name="DRIVE_UPLOAD_STORAGE_NETLOC", environ_prefix=None
)
# Recording encoding options for LiveKit Egress (video composite egress only).
# These settings affect screen recordings handled by VideoCompositeEgressService;
# they are silently ignored by AudioCompositeEgressService (audio-only transcript
@@ -886,21 +918,6 @@ class Base(Configuration):
environ_prefix=None,
)
# Roomkit (meeting-room SIP devices) integration
ROOMKIT_ENABLED = values.BooleanValue(
False,
environ_name="ROOMKIT_ENABLED",
environ_prefix=None,
)
# Server-to-server API token allowing the LiveKit SIP module to call the
# roomkit endpoints (e.g. join a room on behalf of a meeting-room device
# dialing in before any WebRTC participant).
ROOMKIT_SERVER_TO_SERVER_API_TOKEN = SecretFileValue(
None,
environ_name="ROOMKIT_SERVER_TO_SERVER_API_TOKEN",
environ_prefix=None,
)
# Subtitles settings
ROOM_SUBTITLE_ENABLED = values.BooleanValue(
False, environ_name="ROOM_SUBTITLE_ENABLED", environ_prefix=None