443 lines
16 KiB
Python
443 lines
16 KiB
Python
"""Live view: WebRTC signalling relayed between a viewer's socket and a device's.
|
|
|
|
The server routes by session, holds a device to two
|
|
viewers, ends a session when either socket goes or no answer comes, and hands
|
|
out TURN credentials in coturn's REST scheme.
|
|
"""
|
|
|
|
import asyncio
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fluksio.api.routes import oauth
|
|
from fluksio.cloud.connector import CloudConnector
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow import live as live_view
|
|
from tests.api.routes.test_cameras import REPORT, wait_for
|
|
from tests.api.routes.test_devices import _pair, _panels
|
|
|
|
PREFIX = f"{settings.API_V1_STR}/devices"
|
|
SOCKET = f"{settings.API_V1_STR}/flows/ws"
|
|
LIVE_REPORT = {
|
|
**REPORT,
|
|
"ports": [
|
|
*REPORT["ports"],
|
|
{"port": "live", "direction": "in", "dtype": "video", "on_demand": True},
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def cams(client: TestClient, superuser_token_headers: dict[str, str]):
|
|
"""``cam1`` reports a live port, ``cam2`` does not. Yields their headers."""
|
|
for bucket in [b for b in oauth._hits if b.startswith("pair:")]:
|
|
del oauth._hits[bucket]
|
|
_panels(
|
|
client, superuser_token_headers, {"panels": [{"id": "cam1"}, {"id": "cam2"}]}
|
|
)
|
|
cam1 = _pair(client, superuser_token_headers, "cam1")
|
|
cam2 = _pair(client, superuser_token_headers, "cam2")
|
|
for headers, report in ((cam1, LIVE_REPORT), (cam2, REPORT)):
|
|
device = "cam1" if headers is cam1 else "cam2"
|
|
put = client.put(f"{PREFIX}/{device}/report", headers=headers, json=report)
|
|
assert put.status_code == 200, put.text
|
|
yield cam1, cam2
|
|
for device in ("cam1", "cam2"):
|
|
client.delete(f"{PREFIX}/{device}", headers=superuser_token_headers)
|
|
|
|
|
|
def _viewer(client: TestClient, headers: dict[str, str]) -> Any:
|
|
ticket = client.post(f"{settings.API_V1_STR}/flows/ws-ticket", headers=headers)
|
|
return client.websocket_connect(f"{SOCKET}?ticket={ticket.json()['ticket']}")
|
|
|
|
|
|
def _next(ws: Any, kind: str) -> dict[str, Any]:
|
|
"""The next message of this type on a socket, skipping the rest."""
|
|
while True:
|
|
message = ws.receive_json()
|
|
if message.get("type") == kind:
|
|
return message
|
|
|
|
|
|
def _start(ws: Any, session: str, device: str = "cam1") -> None:
|
|
ws.send_json(
|
|
{"type": "live", "device": device, "session": session, "action": "start"}
|
|
)
|
|
|
|
|
|
def test_signalling_is_relayed_both_ways(
|
|
client: TestClient, superuser_token_headers: dict[str, str], cams
|
|
) -> None:
|
|
"""start, the device's offer, the answer, candidates both ways, bye."""
|
|
cam1, _ = cams
|
|
with (
|
|
client.websocket_connect(SOCKET, headers=cam1) as device,
|
|
_viewer(client, superuser_token_headers) as viewer,
|
|
):
|
|
device.receive_json()
|
|
_start(viewer, "session-one")
|
|
start = _next(device, "signal")
|
|
assert start["action"] == "start"
|
|
assert start["session"] == "session-one"
|
|
assert start["ice_servers"] == []
|
|
|
|
device.send_json(
|
|
{
|
|
"type": "signal",
|
|
"session": "session-one",
|
|
"action": "sdp",
|
|
"sdp": "v=0 offer",
|
|
}
|
|
)
|
|
assert _next(viewer, "live") == {
|
|
"type": "live",
|
|
"device": "cam1",
|
|
"session": "session-one",
|
|
"action": "sdp",
|
|
"sdp": "v=0 offer",
|
|
"ice_servers": [],
|
|
}
|
|
viewer.send_json(
|
|
{
|
|
"type": "live",
|
|
"device": "cam1",
|
|
"session": "session-one",
|
|
"action": "sdp",
|
|
"sdp": "v=0 answer",
|
|
}
|
|
)
|
|
assert _next(device, "signal")["sdp"] == "v=0 answer"
|
|
viewer.send_json(
|
|
{
|
|
"type": "live",
|
|
"device": "cam1",
|
|
"session": "session-one",
|
|
"action": "candidate",
|
|
"candidate": "candidate:1 1 udp 1 10.0.0.2 5000 typ host",
|
|
}
|
|
)
|
|
assert _next(device, "signal")["candidate"].startswith("candidate:1 ")
|
|
device.send_json(
|
|
{
|
|
"type": "signal",
|
|
"session": "session-one",
|
|
"action": "candidate",
|
|
"candidate": "candidate:2 1 udp 1 10.0.0.3 5000 typ host",
|
|
}
|
|
)
|
|
assert _next(viewer, "live")["candidate"].startswith("candidate:2 ")
|
|
|
|
viewer.send_json(
|
|
{
|
|
"type": "live",
|
|
"device": "cam1",
|
|
"session": "session-one",
|
|
"action": "bye",
|
|
}
|
|
)
|
|
bye = _next(device, "signal")
|
|
assert (bye["action"], bye["session"]) == ("bye", "session-one")
|
|
assert live_view.sessions.count("cam1") == 0
|
|
|
|
|
|
def test_a_third_viewer_is_busy_and_closing_ends_the_sessions(
|
|
client: TestClient, superuser_token_headers: dict[str, str], cams
|
|
) -> None:
|
|
cam1, _ = cams
|
|
with client.websocket_connect(SOCKET, headers=cam1) as device:
|
|
device.receive_json()
|
|
with _viewer(client, superuser_token_headers) as viewer:
|
|
_start(viewer, "session-a")
|
|
_start(viewer, "session-b")
|
|
_start(viewer, "session-c")
|
|
busy = _next(viewer, "live")
|
|
assert (busy["session"], busy["action"]) == ("session-c", "busy")
|
|
asked = {_next(device, "signal")["session"] for _ in range(2)}
|
|
assert asked == {"session-a", "session-b"}
|
|
# The viewer's socket closed: the device hears bye for both.
|
|
ended = {_next(device, "signal")["session"] for _ in range(2)}
|
|
assert ended == {"session-a", "session-b"}
|
|
wait_for(lambda: live_view.sessions.count("cam1") == 0)
|
|
|
|
with _viewer(client, superuser_token_headers) as viewer:
|
|
with client.websocket_connect(SOCKET, headers=cam1) as device:
|
|
device.receive_json()
|
|
_start(viewer, "session-d")
|
|
_next(device, "signal")
|
|
# The device answers busy itself, past its own cap.
|
|
device.send_json(
|
|
{"type": "signal", "session": "session-d", "action": "busy"}
|
|
)
|
|
assert _next(viewer, "live")["action"] == "busy"
|
|
|
|
_start(viewer, "session-e")
|
|
_next(device, "signal")
|
|
# The device's socket closed.
|
|
bye = _next(viewer, "live")
|
|
assert (bye["session"], bye["action"], bye["reason"]) == (
|
|
"session-e",
|
|
"bye",
|
|
"offline",
|
|
)
|
|
|
|
|
|
def test_a_session_without_an_answer_times_out(
|
|
client: TestClient,
|
|
superuser_token_headers: dict[str, str],
|
|
cams,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(live_view, "ANSWER_TIMEOUT_S", 0.2)
|
|
cam1, _ = cams
|
|
with (
|
|
client.websocket_connect(SOCKET, headers=cam1) as device,
|
|
_viewer(client, superuser_token_headers) as viewer,
|
|
):
|
|
device.receive_json()
|
|
_start(viewer, "session-slow")
|
|
_next(device, "signal")
|
|
bye = _next(viewer, "live")
|
|
assert (bye["action"], bye["reason"]) == ("bye", "timeout")
|
|
told = _next(device, "signal")
|
|
assert (told["action"], told["reason"]) == ("bye", "timeout")
|
|
|
|
|
|
def test_only_a_person_starts_and_only_the_device_speaks_for_its_session(
|
|
client: TestClient, superuser_token_headers: dict[str, str], cams
|
|
) -> None:
|
|
cam1, cam2 = cams
|
|
with _viewer(client, superuser_token_headers) as viewer:
|
|
# Not connected, then connected without a live port.
|
|
_start(viewer, "session-off")
|
|
assert _next(viewer, "live")["reason"] == "offline"
|
|
with client.websocket_connect(SOCKET, headers=cam2) as other:
|
|
other.receive_json()
|
|
_start(viewer, "session-old", device="cam2")
|
|
assert _next(viewer, "live")["reason"] == "unsupported"
|
|
|
|
with client.websocket_connect(SOCKET, headers=cam1) as device:
|
|
device.receive_json()
|
|
# A device's credential cannot start one.
|
|
_start(other, "session-dev")
|
|
_start(viewer, "session-ok")
|
|
assert _next(device, "signal")["session"] == "session-ok"
|
|
assert live_view.sessions.count("cam1") == 1
|
|
|
|
# Another device cannot speak for it.
|
|
other.send_json(
|
|
{
|
|
"type": "signal",
|
|
"session": "session-ok",
|
|
"action": "sdp",
|
|
"sdp": "x",
|
|
}
|
|
)
|
|
time.sleep(0.1)
|
|
device.send_json(
|
|
{
|
|
"type": "signal",
|
|
"session": "session-ok",
|
|
"action": "sdp",
|
|
"sdp": "y",
|
|
}
|
|
)
|
|
assert _next(viewer, "live")["sdp"] == "y"
|
|
|
|
# Nor can another viewer, for a session it does not hold.
|
|
with _viewer(client, superuser_token_headers) as stranger:
|
|
stranger.send_json(
|
|
{
|
|
"type": "live",
|
|
"device": "cam1",
|
|
"session": "session-ok",
|
|
"action": "bye",
|
|
}
|
|
)
|
|
time.sleep(0.1)
|
|
assert live_view.sessions.count("cam1") == 1
|
|
|
|
|
|
def test_ice_servers_carry_coturn_credentials(
|
|
client: TestClient,
|
|
superuser_token_headers: dict[str, str],
|
|
cams,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""``<expiry>:<who>``, and the base64 HMAC-SHA1 of it under the secret."""
|
|
monkeypatch.setattr(settings, "STUN_URLS", ["stun:turn.example.com:3478"])
|
|
monkeypatch.setattr(
|
|
settings,
|
|
"TURN_URLS",
|
|
["turn:turn.example.com:3478", "turns:turn.example.com:5349?transport=tcp"],
|
|
)
|
|
monkeypatch.setattr(settings, "TURN_SECRET", "s3cret")
|
|
cam1, _ = cams
|
|
|
|
def check(server: dict[str, Any], who: str) -> None:
|
|
expiry, _, name = server["username"].partition(":")
|
|
assert name == who
|
|
assert 0 < int(expiry) - time.time() <= live_view.ICE_TTL_S
|
|
mac = hmac.new(b"s3cret", server["username"].encode(), hashlib.sha1)
|
|
assert server["credential"] == base64.b64encode(mac.digest()).decode()
|
|
|
|
answer = client.get(f"{PREFIX}/cam1/ice", headers=superuser_token_headers)
|
|
assert answer.status_code == 200, answer.text
|
|
stun, turn = answer.json()["ice_servers"]
|
|
assert stun == {"urls": ["stun:turn.example.com:3478"]}
|
|
me = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers)
|
|
check(turn, me.json()["id"])
|
|
|
|
assert (
|
|
client.get(f"{PREFIX}/nope/ice", headers=superuser_token_headers).status_code
|
|
== 404
|
|
)
|
|
# A device gets its servers in `start`, not here.
|
|
assert client.get(f"{PREFIX}/cam1/ice", headers=cam1).status_code == 403
|
|
|
|
with (
|
|
client.websocket_connect(SOCKET, headers=cam1) as device,
|
|
_viewer(client, superuser_token_headers) as viewer,
|
|
):
|
|
device.receive_json()
|
|
_start(viewer, "session-ice")
|
|
check(_next(device, "signal")["ice_servers"][1], "device-cam1")
|
|
# The viewer's come with the offer, minted for this session's user.
|
|
device.send_json(
|
|
{"type": "signal", "session": "session-ice", "action": "sdp", "sdp": "v=0"}
|
|
)
|
|
check(_next(viewer, "live")["ice_servers"][1], me.json()["id"])
|
|
|
|
|
|
def test_a_full_outbox_ends_the_session_instead_of_dropping() -> None:
|
|
"""A socket that stops reading loses its session, never a message."""
|
|
|
|
class Devices:
|
|
def reported_ports(self, _device: str) -> dict:
|
|
return {"live": {}}
|
|
|
|
async def run() -> None:
|
|
registry = live_view.LiveSessions()
|
|
camera, viewer = live_view.Endpoint(), live_view.Endpoint()
|
|
registry.attach("cam9", camera)
|
|
registry.heard(
|
|
{
|
|
"type": "live",
|
|
"device": "cam9",
|
|
"session": "s-overflow",
|
|
"action": "start",
|
|
},
|
|
viewer,
|
|
person="someone",
|
|
device="",
|
|
devices=Devices(),
|
|
superuser=True,
|
|
)
|
|
assert registry.count("cam9") == 1
|
|
# The viewer never reads, and the device trickles on.
|
|
for index in range(live_view.OUTBOX + 1):
|
|
registry.heard(
|
|
{
|
|
"type": "signal",
|
|
"session": "s-overflow",
|
|
"action": "candidate",
|
|
"candidate": f"candidate:{index} 1 udp 1 10.0.0.9 5000 typ host",
|
|
},
|
|
camera,
|
|
person="",
|
|
device="cam9",
|
|
devices=None,
|
|
)
|
|
assert registry.count("cam9") == 0
|
|
start, bye = [json.loads(await camera.next()) for _ in range(2)]
|
|
assert start["action"] == "start"
|
|
assert (bye["action"], bye["reason"]) == ("bye", "overflow")
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_the_tunnel_carries_signalling(
|
|
client: TestClient, superuser_token_headers: dict[str, str], cams
|
|
) -> None:
|
|
"""A viewer through the portal, a device on this network, and back."""
|
|
cam1, _ = cams
|
|
ticket = client.post(
|
|
f"{settings.API_V1_STR}/flows/ws-ticket", headers=superuser_token_headers
|
|
).json()["ticket"]
|
|
connector = CloudConnector(client.app)
|
|
sent: list[dict[str, Any]] = []
|
|
|
|
class Socket:
|
|
async def send(self, text: str) -> None:
|
|
sent.append(json.loads(text))
|
|
|
|
def said() -> list[dict[str, Any]]:
|
|
return [
|
|
json.loads(frame["text"])
|
|
for frame in sent
|
|
if frame["op"] == "ws_msg" and '"live"' in frame["text"]
|
|
]
|
|
|
|
frame = {"id": "t1", "path": "/api/v1/flows/ws", "query": f"ticket={ticket}"}
|
|
assert client.portal is not None
|
|
with client.websocket_connect(SOCKET, headers=cam1) as device:
|
|
device.receive_json()
|
|
stream = client.portal.start_task_soon(connector._stream, Socket(), frame)
|
|
try:
|
|
wait_for(lambda: "t1" in connector._listeners)
|
|
hear = connector._listeners["t1"]
|
|
# The stream subscribes to the bus once it has said its snapshot.
|
|
wait_for(lambda: any(f["op"] == "ws_msg" for f in sent))
|
|
time.sleep(0.1)
|
|
client.portal.call(
|
|
hear,
|
|
json.dumps(
|
|
{
|
|
"type": "live",
|
|
"device": "cam1",
|
|
"session": "tunnel-1",
|
|
"action": "start",
|
|
}
|
|
),
|
|
)
|
|
assert _next(device, "signal")["session"] == "tunnel-1"
|
|
device.send_json(
|
|
{"type": "signal", "session": "tunnel-1", "action": "sdp", "sdp": "v=0"}
|
|
)
|
|
offer = wait_for(lambda: said())[0]
|
|
assert (offer["session"], offer["sdp"]) == ("tunnel-1", "v=0")
|
|
finally:
|
|
stream.cancel()
|
|
wait_for(stream.done)
|
|
# The stream went: the device hears bye.
|
|
assert _next(device, "signal")["action"] == "bye"
|
|
|
|
|
|
def test_frames_held_for_a_stream_go_with_it(client: TestClient) -> None:
|
|
"""A stream refused before it listened leaves nothing behind."""
|
|
connector = CloudConnector(client.app)
|
|
sent: list[str] = []
|
|
|
|
class Socket:
|
|
async def send(self, text: str) -> None:
|
|
sent.append(text)
|
|
|
|
frame = {"id": "t2", "path": "/api/v1/flows/ws", "query": "ticket=spent"}
|
|
assert client.portal is not None
|
|
|
|
def open_and_speak() -> None:
|
|
connector._start_stream(Socket(), frame)
|
|
# What the link does with a frame for a stream not listening yet,
|
|
# here before the stream has even started to run.
|
|
connector._early["t2"] = ['{"type": "media", "names": []}']
|
|
|
|
client.portal.call(open_and_speak)
|
|
wait_for(lambda: "t2" not in connector._streams)
|
|
assert "t2" not in connector._early
|