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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017PCb59uY3UKqDqt7B6hzd5

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, [])

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: