Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fa0351084b |
@@ -389,3 +389,31 @@ class NormalizePermissionTest(unittest.TestCase):
|
||||
guard.livekit_update_participant = lambda *a: None
|
||||
participant = {"identity": "@a:x:DEV", "permission": {"canPublish": True, "canPublishSources": ["CAMERA"]}}
|
||||
self.assertFalse(guard.reconcile_participant("!r:x", participant, {"SCREEN_SHARE"}))
|
||||
|
||||
|
||||
class SlotReservationTests(unittest.TestCase):
|
||||
"""[matrix #25] Simultaneous joins can't both pass a full room."""
|
||||
|
||||
def test_second_simultaneous_join_is_blocked(self):
|
||||
r = guard.SlotReservations(ttl=30)
|
||||
present = set()
|
||||
counted = present | r.reserved("room", present, now=0)
|
||||
self.assertFalse(guard.should_block(1, counted, "@a:x"))
|
||||
r.reserve("room", "@a:x", now=0)
|
||||
# @b asks a moment later, before @a has connected to LiveKit.
|
||||
counted = present | r.reserved("room", present, now=1)
|
||||
self.assertTrue(guard.should_block(1, counted, "@b:x"))
|
||||
# @a asking again (second device) is still allowed.
|
||||
self.assertFalse(guard.should_block(1, counted, "@a:x"))
|
||||
|
||||
def test_reservation_expires_and_clears_once_present(self):
|
||||
r = guard.SlotReservations(ttl=30)
|
||||
r.reserve("room", "@a:x", now=0)
|
||||
self.assertEqual(r.reserved("room", set(), now=31), set())
|
||||
r.reserve("room", "@a:x", now=40)
|
||||
self.assertEqual(r.reserved("room", {"@a:x"}, now=41), set())
|
||||
|
||||
def test_room_lock_is_per_room(self):
|
||||
r = guard.SlotReservations()
|
||||
self.assertIs(r.room_lock("a"), r.room_lock("a"))
|
||||
self.assertIsNot(r.room_lock("a"), r.room_lock("b"))
|
||||
|
||||
@@ -145,6 +145,41 @@ def should_block(limit: int, present_users: set, requester: str) -> bool:
|
||||
return len(present_users) >= limit
|
||||
|
||||
|
||||
# [matrix #25] Tokens issued but not yet connected weren't counted, so two
|
||||
# users pressing Join together could both pass a 1-person limit. Each admitted
|
||||
# user holds a short reservation (counted like a present user) until LiveKit
|
||||
# lists them or it expires; checks for a room are serialised with a lock.
|
||||
RESERVATION_TTL = 30.0
|
||||
|
||||
|
||||
class SlotReservations:
|
||||
"""Per-room reservations for admitted-but-not-yet-connected users."""
|
||||
|
||||
def __init__(self, ttl: float = RESERVATION_TTL):
|
||||
self.ttl = ttl
|
||||
self._by_room = {} # alias -> {user: expiry}
|
||||
self._guard = threading.Lock()
|
||||
self._room_locks = {}
|
||||
|
||||
def room_lock(self, alias: str) -> threading.Lock:
|
||||
with self._guard:
|
||||
return self._room_locks.setdefault(alias, threading.Lock())
|
||||
|
||||
def reserved(self, alias: str, present: set, now: float) -> set:
|
||||
"""Unexpired reservations; drops expired ones and users now present."""
|
||||
room = self._by_room.get(alias, {})
|
||||
for user in [u for u, exp in room.items() if exp <= now or u in present]:
|
||||
del room[user]
|
||||
return set(room)
|
||||
|
||||
def reserve(self, alias: str, user: str, now: float) -> None:
|
||||
if user:
|
||||
self._by_room.setdefault(alias, {})[user] = now + self.ttl
|
||||
|
||||
|
||||
RESERVATIONS = SlotReservations()
|
||||
|
||||
|
||||
def allowed_sources(policy: dict):
|
||||
"""Pure decision: the narrowed `canPublishSources` list, or None.
|
||||
|
||||
@@ -618,11 +653,16 @@ class Handler(BaseHTTPRequestHandler):
|
||||
# LiveKit-admin outage here cannot skip the publish-source policy below.
|
||||
if limit > 0 and alias:
|
||||
try:
|
||||
present = livekit_present_users(alias)
|
||||
if should_block(limit, present, requester):
|
||||
log(f"blocked {requester or '?'} from {room_id}: {len(present)}/{limit}")
|
||||
self._send_blocked()
|
||||
return
|
||||
with RESERVATIONS.room_lock(alias):
|
||||
present = livekit_present_users(alias)
|
||||
now = time.monotonic()
|
||||
counted = present | RESERVATIONS.reserved(alias, present, now)
|
||||
if should_block(limit, counted, requester):
|
||||
log(f"blocked {requester or '?'} from {room_id}: {len(counted)}/{limit}")
|
||||
self._send_blocked()
|
||||
return
|
||||
if requester not in present:
|
||||
RESERVATIONS.reserve(alias, requester, now)
|
||||
except Exception as exc:
|
||||
# Fail open on the limit only; still apply the policy below.
|
||||
log(f"limit check error for {room_id}: {exc}")
|
||||
|
||||
Reference in New Issue
Block a user