# services/jitsi/launcher-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: jitsi-launcher
namespace: jitsi
data:
app.py: |
import base64
import hashlib
import hmac
import json
import os
import time
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
ISSUER = os.getenv("JWT_ISSUER", "https://sso.bstein.dev/realms/atlas")
AUDIENCE = os.getenv("JWT_AUDIENCE", "jitsi")
APP_ID = os.getenv("JWT_APP_ID", "jitsi")
PUBLIC_URL = os.getenv("PUBLIC_URL", "https://meet.bstein.dev")
SECRET_FILE = os.getenv("JWT_SECRET_FILE", "/var/lib/jitsi-jwt/jwt")
ALLOWED_GROUPS = {g for g in os.getenv("ALLOWED_GROUPS", "").split(",") if g}
TOKEN_TTL = int(os.getenv("JWT_TTL_SECONDS", "600"))
app = FastAPI()
def _b64url(data: bytes) -> bytes:
return base64.urlsafe_b64encode(data).rstrip(b"=")
def _read_secret() -> bytes:
raw = open(SECRET_FILE, "rb").read().strip()
try:
return bytes.fromhex(raw.decode())
except ValueError:
return raw
def _sign(room: str, user: str, groups: list[str]) -> str:
now = int(time.time())
header = {"alg": "HS256", "typ": "JWT"}
payload = {
"iss": ISSUER,
"aud": AUDIENCE,
"sub": "meet.jitsi",
"room": room,
"exp": now + TOKEN_TTL,
"nbf": now - 10,
"context": {
"user": {
"name": user,
"email": user,
"affiliation": "owner",
"groups": groups,
}
},
"app_id": APP_ID,
}
secret = _read_secret()
signing_input = b".".join(
[
_b64url(json.dumps(header, separators=(",", ":")).encode()),
_b64url(json.dumps(payload, separators=(",", ":")).encode()),
]
)
sig = _b64url(hmac.new(secret, signing_input, hashlib.sha256).digest())
return b".".join([signing_input, sig]).decode()
def _render_form(message: str = "") -> HTMLResponse:
body = f"""
Start a Jitsi room
{''+message+'
' if message else ''}
"""
return HTMLResponse(body)
def _extract_groups(request: Request) -> set[str]:
raw = request.headers.get("x-auth-request-groups", "")
# Traefik forwardAuth returns comma-separated groups
return {g.strip() for g in raw.split(",") if g.strip()}
@app.get("/launch")
async def launch(request: Request, room: str | None = None):
user = request.headers.get("x-auth-request-email") or request.headers.get(
"x-auth-request-user", ""
)
groups = _extract_groups(request)
if ALLOWED_GROUPS and not (groups & ALLOWED_GROUPS):
raise HTTPException(status_code=403, detail="forbidden")
if not room:
return _render_form()
room = room.strip()
if not room or "/" in room or ".." in room:
raise HTTPException(status_code=400, detail="invalid room")
token = _sign(room, user or "moderator", sorted(groups))
join_url = f"{PUBLIC_URL}/{room}#config.jwt={token}"
accept = request.headers.get("accept", "")
if "text/html" in accept:
return RedirectResponse(join_url, status_code=302)
return JSONResponse({"room": room, "join_url": join_url, "token": token})
@app.get("/")
async def root():
return RedirectResponse("/launch")
@app.get("/health")
async def health():
return {"status": "ok"}