From d3f389cab104d41d71433b33be16fe083c5f61d7 Mon Sep 17 00:00:00 2001 From: Lotus CI Date: Mon, 19 Oct 2026 23:30:15 -0400 Subject: [PATCH] feat(voice-limit-guard): enforce the limit at connect time too (#25) The limit was only checked at token issue, and lk-jwt tokens last an hour, so a token fetched while there was space could be used to connect once the room was full. The reconcile loop now compares each limited room's distinct Matrix users with max_users and RemoveParticipants the people who arrived since the previous sweep, newest first, until it is back within the limit. People already in the call are never removed (lowering a limit mid-call kicks nobody); extra devices of one person never count twice. Fixes #25 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017PCb59uY3UKqDqt7B6hzd5 --- README.md | 1 + livekit/test_voice_limit_guard.py | 78 +++++++++++++++++++++++++++++++ livekit/voice-limit-guard.py | 77 +++++++++++++++++++++++++++++- 3 files changed, 154 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 66bb3c5..a0bcb55 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,7 @@ Every Matrix client must fetch a LiveKit JWT from lk-jwt-service before it can j - **Participant limit** — it forwards to lk-jwt-service, and if a token is issued decodes the JWT to get the LiveKit alias (`video.room`) + requester (`sub`), then asks LiveKit `ListParticipants` how many **distinct Matrix users** are in the room. requester already present (rejoin) → allow · distinct users ≥ limit → **403** · otherwise → allow. - **Publish permissions (screenshare / camera)** — LiveKit is a pure SFU and **cannot cap a publisher's bitrate/framerate** (no such field exists in the grant/config/API — that stays a Lotus-client-cooperative setting). But the JWT's `video.canPublishSources` **is** SFU-enforced for every client. Since the guard holds the LiveKit signing secret, when a room forbids a source it **decodes the issued token, drops `screen_share`/`screen_share_audio` (and/or `camera`) from `canPublishSources`, and re-signs it** (HS256, same key). Microphone is always kept. The SFU then rejects those tracks for **all** clients — nothing to opt into. - **Live (mid-call) enforcement** — the JWT re-sign covers anyone *joining* after a policy change. For people **already in the call**, a background **reconcile loop** (every `GUARD_RECONCILE_INTERVAL`, default 3 s) calls LiveKit `UpdateParticipant` to narrow their `canPublishSources`, which **unpublishes an in-progress screenshare/camera server-side for all clients** and blocks re-publish (confirmed LiveKit 1.9.11 behavior: reducing `can_publish_sources` removes the offending live track). So flipping a room to audio-only kills existing cameras/screenshares within ~one interval. The loop learns each LiveKit room's Matrix id from tokens it issues, only ever **removes** forbidden sources (never grants), preserves every other permission flag (full-replace safety), and no-ops once compliant. Disable with `GUARD_RECONCILE=0`. +- **Connect-time limit (matrix #25)** — the token check can't stop a token fetched while there was space (lk-jwt tokens last an hour) from being used once the room is full. The same reconcile loop compares each limited room's **distinct Matrix users** with `max_users`, and when people who **arrived since the previous sweep** take it over the limit it calls LiveKit `RemoveParticipant` on them, newest first (all their devices). People already in the call are never removed, so **lowering a limit mid-call kicks nobody**; a person's extra devices never count twice. The first sweep of a room only records who is there. Simultaneous joins are already handled at token issue (per-room lock + 30 s slot reservations). - **Fail-open:** any error (admin API down, bad/absent token, LiveKit unreachable, unparseable room id, unexpected JWT shape) returns the upstream response **unchanged**, so calls keep working even if enforcement is degraded. The limit check and the source-policy re-sign are **independent** (a LiveKit-admin outage during the limit count can't skip the source restriction, and vice-versa). Before re-signing, the guard **verifies its own secret actually signed the token** — on a `LIVEKIT_SECRET` mismatch it skips the restriction and passes the original token through (so a secret drift can never emit a token the SFU rejects). A room with no policy set takes a zero-overhead fast path (token untouched). > `LIVEKIT_KEY`/`LIVEKIT_SECRET` live in `/etc/matrix-deploy.env` (0600, not in git). The same pair must diff --git a/livekit/test_voice_limit_guard.py b/livekit/test_voice_limit_guard.py index f743762..9093650 100644 --- a/livekit/test_voice_limit_guard.py +++ b/livekit/test_voice_limit_guard.py @@ -417,3 +417,81 @@ class SlotReservationTests(unittest.TestCase): r = guard.SlotReservations() self.assertIs(r.room_lock("a"), r.room_lock("a")) self.assertIsNot(r.room_lock("a"), r.room_lock("b")) + + +def _p(identity, joined_ms): + return {"identity": identity, "joined_at_ms": str(joined_ms), + "permission": {"canPublish": True, "canPublishSources": ["MICROPHONE"]}} + + +class TestOverLimitRemovals(unittest.TestCase): + """[matrix #25] connect-time enforcement of the voice limit.""" + + def test_first_sweep_records_and_removes_nobody(self): + parts = [_p("@a:x:DEV1", 1), _p("@b:x:DEV1", 2)] + removals, seen = guard.over_limit_removals(parts, 1, None) + self.assertEqual(removals, []) + self.assertEqual(seen, {"@a:x", "@b:x"}) + + def test_a_newcomer_past_the_limit_is_removed_with_all_devices(self): + parts = [_p("@a:x:DEV1", 1), _p("@b:x:DEV1", 5), _p("@b:x:DEV2", 6)] + removals, seen = guard.over_limit_removals(parts, 1, {"@a:x"}) + self.assertEqual(sorted(removals), ["@b:x:DEV1", "@b:x:DEV2"]) + self.assertEqual(seen, {"@a:x"}) + + def test_newest_newcomers_go_first(self): + parts = [_p("@a:x:D", 1), _p("@b:x:D", 10), _p("@c:x:D", 20)] + removals, _ = guard.over_limit_removals(parts, 2, {"@a:x"}) + self.assertEqual(removals, ["@c:x:D"]) + + def test_lowering_the_limit_mid_call_kicks_nobody(self): + parts = [_p("@a:x:D", 1), _p("@b:x:D", 2), _p("@c:x:D", 3)] + removals, seen = guard.over_limit_removals(parts, 1, {"@a:x", "@b:x", "@c:x"}) + self.assertEqual(removals, []) + self.assertEqual(seen, {"@a:x", "@b:x", "@c:x"}) + + def test_extra_devices_of_a_present_user_are_not_newcomers(self): + parts = [_p("@a:x:DEV1", 1), _p("@a:x:DEV2", 9)] + removals, _ = guard.over_limit_removals(parts, 1, {"@a:x"}) + self.assertEqual(removals, []) + + def test_no_limit_removes_nobody(self): + parts = [_p("@a:x:D", 1), _p("@b:x:D", 2)] + self.assertEqual(guard.over_limit_removals(parts, 0, {"@a:x"})[0], []) + + def test_joined_at_seconds_are_understood(self): + parts = [{"identity": "@a:x:D", "joined_at": "100"}, {"identity": "@b:x:D", "joined_at": "200"}, + {"identity": "@c:x:D", "joinedAt": "150"}] + removals, _ = guard.over_limit_removals(parts, 1, set()) + self.assertEqual(sorted(removals), ["@b:x:D", "@c:x:D"]) + + +class TestReconcileRoomLimit(unittest.TestCase): + def setUp(self): + self._orig = (guard.room_state, guard.livekit_list_participants, + guard.livekit_update_participant, guard.livekit_remove_participant) + self.removed, self.updates = [], [] + guard.livekit_remove_participant = lambda a, i: self.removed.append(i) + guard.livekit_update_participant = lambda a, i, p: self.updates.append(i) + guard._seen_users.clear() + + def tearDown(self): + (guard.room_state, guard.livekit_list_participants, + guard.livekit_update_participant, guard.livekit_remove_participant) = self._orig + guard._seen_users.clear() + + def test_a_late_connect_with_an_early_token_is_removed(self): + guard.room_state = lambda rid, max_age=0: {"max_users": 1, "allow_screenshare": True, "allow_camera": True} + guard.livekit_list_participants = lambda a: [_p("@a:x:D", 1)] + guard.reconcile_room("alias", "!room:x") # first sweep: @a is there + self.assertEqual(self.removed, []) + guard.livekit_list_participants = lambda a: [_p("@a:x:D", 1), _p("@late:x:D", 50)] + guard.reconcile_room("alias", "!room:x") + self.assertEqual(self.removed, ["@late:x:D"]) + self.assertEqual(self.updates, []) # nothing forbidden in this room + + def test_unlimited_unrestricted_room_is_not_listed(self): + guard.room_state = lambda rid, max_age=0: {"max_users": 0, "allow_screenshare": True, "allow_camera": True} + guard.livekit_list_participants = lambda a: (_ for _ in ()).throw(AssertionError("listed")) + guard.reconcile_room("alias", "!room:x") + self.assertEqual(self.removed, []) diff --git a/livekit/voice-limit-guard.py b/livekit/voice-limit-guard.py index 24f12e7..806cb22 100644 --- a/livekit/voice-limit-guard.py +++ b/livekit/voice-limit-guard.py @@ -405,6 +405,7 @@ def remember_room(alias: str, room_id: str) -> None: def forget_room(alias: str) -> None: with _alias_lock: _alias_to_room.pop(alias, None) + _seen_users.pop(alias, None) def forbidden_sources(policy: dict) -> set: @@ -451,6 +452,65 @@ def normalize_permission(perm: dict) -> dict: return out +def _joined_at(participant: dict) -> float: + """When this device joined, in ms (ParticipantInfo joined_at_ms, else s).""" + for key in ("joined_at_ms", "joinedAtMs"): + if participant.get(key) not in (None, "", 0, "0"): + return float(participant[key]) + for key in ("joined_at", "joinedAt"): + if participant.get(key) not in (None, "", 0, "0"): + return float(participant[key]) * 1000.0 + return 0.0 + + +def over_limit_removals(participants: list, limit: int, seen): + """[matrix #25] Pure: which devices to remove so the room is back within + its limit, and the set of users now known to be in the room. + + Only users who ARRIVED since the previous sweep (not in `seen`) can be + removed, newest first: a token fetched while there was space can be used + to connect up to an hour later, past the limit, but lowering the limit + mid-call must not kick people already there. Users are distinct Matrix + users (matrix_user), so a person's extra devices never count twice, and + removing a user removes all their devices. `seen is None` means the room + hasn't been swept yet: record it, remove nobody. + """ + first_join = {} + identities = {} + for p in participants: + identity = p.get("identity") or "" + user = matrix_user(identity) + if not user: + continue + identities.setdefault(user, []).append(identity) + joined = _joined_at(p) + first_join[user] = min(first_join.get(user, joined), joined) + users = set(first_join) + if seen is None or limit <= 0 or len(users) <= limit: + return [], users + newcomers = sorted( + (u for u in users if u not in seen), + key=lambda u: (first_join[u], u), + reverse=True, + ) + removed = newcomers[: len(users) - limit] + return [i for u in removed for i in identities[u]], users - set(removed) + + +def livekit_remove_participant(alias: str, identity: str) -> None: + """Disconnect a participant (livekit RemoveParticipant). Already gone -> + Twirp not_found (HTTP 404), which is benign.""" + try: + _livekit_admin_post(alias, "RemoveParticipant", {"room": alias, "identity": identity}) + except urllib.error.HTTPError as exc: + if exc.code != 404: + log(f"remove_participant {identity} in {alias} -> HTTP {exc.code}") + + +# alias -> set of users present at the previous sweep (see over_limit_removals) +_seen_users = {} + + def reconcile_participant(alias: str, participant: dict, forbidden: set) -> bool: """Enforce the forbidden-source policy on one live participant. Returns True if an UpdateParticipant call was issued.""" @@ -483,8 +543,10 @@ def reconcile_room(alias: str, room_id: str) -> None: # Read the policy fresh-ish so a flip is picked up within ~one interval. policy = room_state(room_id, max_age=RECONCILE_INTERVAL) forbidden = forbidden_sources(policy) - if not forbidden: - return # nothing restricted; leave permissions untouched + limit = int(policy.get("max_users", 0) or 0) + if not forbidden and limit <= 0: + _seen_users.pop(alias, None) + return # nothing restricted; leave the room alone try: participants = livekit_list_participants(alias) except urllib.error.HTTPError as exc: @@ -501,7 +563,18 @@ def reconcile_room(alias: str, room_id: str) -> None: # empty_timeout). Don't forget on an empty read — only a 404 (room gone) # prunes — so a transient empty/race can't drop mid-call enforcement. return + # [matrix #25] Connect-time limit: the token check alone can't stop a token + # fetched earlier (valid for an hour) from being used once the room is full. + removals, now_seen = over_limit_removals(participants, limit, _seen_users.get(alias)) + _seen_users[alias] = now_seen + for identity in removals: + livekit_remove_participant(alias, identity) + log(f"removed {identity} from {alias}: over the voice limit of {limit}") + if not forbidden: + return for p in participants: + if p.get("identity") in removals: + continue try: reconcile_participant(alias, p, forbidden) except Exception as exc: