ducoterra
7988da2b93
mirror has no accounts, so on a real network the address a connection arrives
from is not a nicety — it is the entire access control. Before this change the
answer was "everyone who can open a TCP connection", which is fine for a laptop
demo and alarming for a box someone leaves running.
`mirror serve --allow-cidr 10.2.0.0/22` (or MIRROR_ALLOW_CIDRS) now restricts
pages, the JSON API and the signaling socket to the networks named, and it is
strict where it needs to be: a malformed CIDR exits 2 with one line instead of
starting a server whose list silently failed to parse, because the two
outcomes of that typo are "nobody can connect" and "everybody can connect".
Empty remains the only permissive state, on purpose.
The rules that matter live in src/mirror/access.py and are the ones that get
either outcome:
* X-Forwarded-For is believed only from MIRROR_TRUSTED_PROXIES (loopback by
default), and only its left-most value. An untrusted peer that sends the
header gets judged on its own address — there is a test for exactly that,
because it is the shape of a bypass.
* The socket is closed before it accepts, so a refused address never creates
a roster entry to clean up behind it.
* An address that cannot be parsed fails closed, and a v6 address is never
"inside" a v4 network (comparing the two raises TypeError, which would be
a 500 on a dual-stack host).
* /healthz is exempt: the container health probe comes from loopback, which
a production list otherwise excludes.
The refused client is told why, and which address it was judged as — "403" in
front of a proxy is undiagnosable otherwise. Browsers get a page, fetches get
JSON, and the page is self-contained because a refused address cannot load
/static/app.css any more than it can load a page. net.js keys its own notice on
the /api/config 403 and stops: without that the pill reads "offline,
reconnecting in 32s" forever for a decision that will never change.
The static files drop their only inline script and style attribute (the landing
page's counter is /static/home.js, its fine print is .fineprint) so the
deployment can send `script-src 'self'; style-src 'self'` with no
'unsafe-inline'. That policy was verified through a real nginx edge carrying the
exact header set — capture through SRTP to decoded, painted frames at
1280x720, eight distinct, zero securitypolicyviolation events — which also
caught the edge sending `Cache-Control: no-store, no-store` and needed
proxy_hide_header, since nginx appends rather than replaces.
24 tests added; 176 pass, 2 skipped, pyright and ruff clean.
LLM-Generated: true
493 lines
23 KiB
Python
493 lines
23 KiB
Python
"""HTTP surface + signaling socket tests. No browser required."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
from mirror.access import parse_cidrs
|
|
from mirror.app import create_app
|
|
from mirror.config import Settings
|
|
|
|
STATIC = Path(__file__).resolve().parent.parent / "src/mirror/static"
|
|
|
|
|
|
def gated(*allow, client_address="203.0.113.9", **kwargs):
|
|
"""A client speaking as ``client_address`` to a server restricted to ``allow``."""
|
|
settings = Settings(allow_cidrs=parse_cidrs(",".join(allow)), **kwargs)
|
|
return TestClient(create_app(settings), client=(client_address, 44444))
|
|
|
|
|
|
EXTERNAL_REF = re.compile(r"""(?:src|href)\s*=\s*["']https?://""", re.IGNORECASE)
|
|
|
|
|
|
def handshake(socket, role, name=""):
|
|
socket.send_json({"type": "hello", "role": role, "name": name})
|
|
message = socket.receive_json()
|
|
assert message["type"] == "welcome", message
|
|
return message["peerId"]
|
|
|
|
|
|
def wait_for(socket, kind):
|
|
"""Read frames until one has ``type == kind``.
|
|
|
|
The roster legitimately pushes ``stream_updated`` frames whenever the
|
|
audience changes, so tests must not assume a one-message-per-action cadence.
|
|
Any ``error`` frame seen on the way is a failure worth surfacing.
|
|
"""
|
|
seen = []
|
|
while True:
|
|
message = socket.receive_json()
|
|
if message.get("type") == kind:
|
|
return message
|
|
assert message.get("type") != "error", (
|
|
f"unexpected error while waiting for {kind}: {message}"
|
|
)
|
|
seen.append(message)
|
|
assert len(seen) < 25, f"never saw {kind!r}; got {seen}"
|
|
|
|
|
|
class TestPages:
|
|
@pytest.mark.parametrize(
|
|
("path", "marker"),
|
|
[
|
|
("/", "Watch everything"),
|
|
("/mirror", "Share my screen"),
|
|
("/share", "Share my screen"),
|
|
("/watch", "tile-template"),
|
|
],
|
|
)
|
|
def test_pages_render(self, client, path, marker):
|
|
response = client.get(path)
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("text/html")
|
|
assert marker in response.text
|
|
|
|
def test_pages_are_not_cacheable(self, client):
|
|
assert client.get("/watch").headers["cache-control"] == "no-store"
|
|
|
|
def test_the_two_pages_use_the_media_apis_they_claim(self):
|
|
share = (STATIC / "share.js").read_text()
|
|
watch = (STATIC / "watch.js").read_text()
|
|
assert "getDisplayMedia" in share
|
|
assert "RTCPeerConnection" in share and "RTCPeerConnection" in watch
|
|
assert "new WebSocket" in (STATIC / "net.js").read_text()
|
|
|
|
def test_nothing_is_loaded_from_a_cdn(self):
|
|
"""The no-CDN rule is a hard product requirement, so assert it."""
|
|
offenders = []
|
|
for file in sorted(STATIC.iterdir()):
|
|
text = file.read_text()
|
|
if EXTERNAL_REF.search(text):
|
|
offenders.append(file.name)
|
|
if "@import" in text or "fonts.googleapis" in text:
|
|
offenders.append(f"{file.name} (@import)")
|
|
assert offenders == []
|
|
|
|
@pytest.mark.parametrize("asset", ["app.css", "net.js", "share.js", "watch.js"])
|
|
def test_static_assets_are_served(self, client, asset):
|
|
assert client.get(f"/static/{asset}").status_code == 200
|
|
|
|
def test_unknown_page_is_a_404(self, client):
|
|
assert client.get("/nope").status_code == 404
|
|
|
|
|
|
class TestClientScripts:
|
|
"""Guards for the hand-written ES modules: they must parse, and the ids they
|
|
reach for must exist. Both of these have already bitten once."""
|
|
|
|
NODE = shutil.which("node")
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is not installed")
|
|
@pytest.mark.parametrize("script", ["net.js", "share.js", "watch.js"])
|
|
def test_scripts_parse_as_es_modules(self, script, tmp_path):
|
|
copy = tmp_path / "subject.mjs"
|
|
copy.write_text((STATIC / script).read_text())
|
|
result = subprocess.run([self.NODE, "--check", str(copy)], capture_output=True, text=True)
|
|
assert result.returncode == 0, f"{script}: {result.stderr}"
|
|
|
|
@pytest.mark.parametrize(
|
|
("script", "page"),
|
|
[("share.js", "mirror.html"), ("watch.js", "watch.html")],
|
|
)
|
|
def test_every_element_the_script_reaches_for_exists(self, script, page):
|
|
html = (STATIC / page).read_text()
|
|
wanted = set(re.findall(r"""\$\(\"#([A-Za-z0-9_-]+)\"\)""", (STATIC / script).read_text()))
|
|
missing = {name for name in wanted if f'id="{name}"' not in html}
|
|
assert not missing, f"{script} looks up {sorted(missing)} which {page} does not render"
|
|
|
|
def test_both_pages_load_the_stylesheet_and_their_module(self):
|
|
for page, script in (("mirror.html", "share.js"), ("watch.html", "watch.js")):
|
|
html = (STATIC / page).read_text()
|
|
assert 'href="/static/app.css"' in html
|
|
assert f'src="/static/{script}"' in html
|
|
assert 'type="module"' in html
|
|
|
|
def test_the_two_pages_share_no_element_ids_but_the_shared_chrome(self):
|
|
"""Both pages consume one stylesheet, so a duplicated id means a colliding rule.
|
|
|
|
This is not hypothetical: a `#stage` on both pages once made the share
|
|
page's preview inherit `display: none` from the watch page's overlay.
|
|
"""
|
|
shared = {"net", "notice", "toast"}
|
|
ids = {
|
|
page: set(re.findall(r'id="([A-Za-z0-9_-]+)"', (STATIC / page).read_text()))
|
|
for page in ("mirror.html", "watch.html", "index.html")
|
|
}
|
|
assert ids["mirror.html"] & ids["watch.html"] == shared
|
|
assert ids["index.html"] & ids["mirror.html"] <= shared | {"livebar"}
|
|
for page, found in ids.items():
|
|
assert len(found) == len(set(found)), f"{page} declares an id twice"
|
|
|
|
def test_videos_are_muted_and_inline_so_autoplay_is_not_blocked(self):
|
|
for page in ("mirror.html", "watch.html"):
|
|
for tag in re.findall(r"<video[^>]*>", (STATIC / page).read_text()):
|
|
assert "muted" in tag, f"{page}: {tag} would be blocked by autoplay policy"
|
|
assert "playsinline" in tag, f"{page}: {tag} would hijack iOS into fullscreen"
|
|
|
|
|
|
class TestStylesheet:
|
|
def test_hidden_always_means_gone(self):
|
|
"""Every component that sets its own `display` outranks the UA's `[hidden]`.
|
|
|
|
Without this rule the share page shows the LIVE badge and the Stop button
|
|
while nothing is being shared, and every wall tile offers an Exit button
|
|
it is not on stage for.
|
|
"""
|
|
css = (STATIC / "app.css").read_text()
|
|
assert re.search(r"\[hidden\]\s*\{[^}]*display:\s*none\s*!important", css), (
|
|
"app.css must declare a global `[hidden] { display: none !important }`"
|
|
)
|
|
# Components that set `display` and are toggled with `hidden` in JS.
|
|
for selector in (".btn", ".livetag", ".icon-btn", ".toast", ".net"):
|
|
assert selector in css, f"{selector} is toggled by `hidden`; it lives in app.css"
|
|
|
|
def test_no_rule_reintroduces_a_conflicting_display_for_hidden(self):
|
|
"""Every rule whose selector mentions `[hidden]` must actually hide."""
|
|
css = re.sub(r"/\*.*?\*/", "", (STATIC / "app.css").read_text(), flags=re.S)
|
|
offenders = [
|
|
selector.strip()
|
|
for selector, body in re.findall(r"([^{}]+)\{([^{}]*)\}", css)
|
|
if "[hidden]" in selector and not re.search(r"display\s*:\s*none", body)
|
|
]
|
|
assert offenders == []
|
|
|
|
|
|
class TestAddressGate:
|
|
"""mirror has no accounts; the address is the access control."""
|
|
|
|
ALLOW = "10.2.0.0/22"
|
|
|
|
def test_unset_allows_everyone_which_is_the_development_default(self):
|
|
with gated(client_address="203.0.113.9") as client:
|
|
assert client.get("/watch").status_code == 200
|
|
|
|
@pytest.mark.parametrize("path", ["/", "/mirror", "/watch", "/api/streams", "/static/app.css"])
|
|
def test_a_refused_address_gets_no_page_no_roster_no_assets(self, path):
|
|
with gated(self.ALLOW) as client:
|
|
assert client.get(path).status_code == 403
|
|
|
|
def test_the_probe_stays_answerable_from_anywhere(self):
|
|
"""The container health check comes from loopback, which the list excludes."""
|
|
with gated(self.ALLOW) as client:
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
def test_a_browser_gets_an_explanation_rather_than_json(self):
|
|
with gated(self.ALLOW) as client:
|
|
response = client.get("/mirror", headers={"accept": "text/html"})
|
|
assert response.status_code == 403
|
|
assert "text/html" in response.headers["content-type"]
|
|
body = response.text
|
|
assert self.ALLOW in body
|
|
assert "203.0.113.9" in body # which address was judged, or this is undiagnosable
|
|
|
|
def test_the_refusal_page_cannot_depend_on_assets_the_client_cannot_fetch(self):
|
|
with gated(self.ALLOW) as client:
|
|
body = client.get("/", headers={"accept": "text/html"}).text
|
|
assert not EXTERNAL_REF.search(body)
|
|
assert "/static/" not in body
|
|
|
|
def test_a_fetch_gets_the_machine_readable_refusal(self):
|
|
with gated(self.ALLOW) as client:
|
|
response = client.get("/api/config")
|
|
assert response.status_code == 403
|
|
assert response.json() == {
|
|
"error": "forbidden",
|
|
"message": "This mirror only accepts connections from 10.2.0.0/22.",
|
|
"you": "203.0.113.9",
|
|
}
|
|
|
|
def test_a_refusal_is_never_cacheable(self):
|
|
with gated(self.ALLOW) as client:
|
|
for headers in ({"accept": "text/html"}, {}):
|
|
response = client.get("/api/config", headers=headers)
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
def test_the_socket_closes_before_it_accepts(self):
|
|
with (
|
|
gated(self.ALLOW) as client,
|
|
pytest.raises(WebSocketDisconnect),
|
|
client.websocket_connect("/ws") as socket,
|
|
):
|
|
socket.send_json({"type": "hello", "role": "viewer"})
|
|
|
|
def test_the_real_address_is_in_and_connects(self):
|
|
with gated(self.ALLOW, client_address="10.2.1.10") as client:
|
|
assert client.get("/watch").status_code == 200
|
|
with client.websocket_connect("/ws") as socket:
|
|
assert handshake(socket, "viewer")
|
|
|
|
def test_a_trusted_local_proxy_carries_the_client_through(self):
|
|
"""The fleet shape: nginx on the same host, X-Forwarded-For set by nginx."""
|
|
with gated(self.ALLOW, client_address="127.0.0.1") as client:
|
|
assert client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"}).status_code == 200
|
|
with client.websocket_connect("/ws", headers={"x-forwarded-for": "10.2.1.5"}) as socket:
|
|
assert handshake(socket, "viewer")
|
|
|
|
def test_a_trusted_proxy_carrying_a_foreign_client_is_still_refused(self):
|
|
with gated(self.ALLOW, client_address="127.0.0.1") as client:
|
|
assert client.get("/watch", headers={"x-forwarded-for": "8.8.8.8"}).status_code == 403
|
|
with (
|
|
pytest.raises(WebSocketDisconnect),
|
|
client.websocket_connect("/ws", headers={"x-forwarded-for": "8.8.8.8"}),
|
|
):
|
|
pass
|
|
|
|
def test_an_untrusted_peer_cannot_write_its_own_pass(self):
|
|
with gated(self.ALLOW, client_address="203.0.113.9") as client:
|
|
response = client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"})
|
|
assert response.status_code == 403
|
|
assert response.json()["you"] == "203.0.113.9"
|
|
|
|
def test_both_address_families_can_be_admitted_side_by_side(self):
|
|
with gated(f"{self.ALLOW}, fd00::/64", client_address="fd00::1234") as client:
|
|
assert client.get("/watch").status_code == 200
|
|
|
|
|
|
class TestJsonApi:
|
|
def test_config_advertises_the_media_defaults(self, client):
|
|
body = client.get("/api/config").json()
|
|
assert body["iceServers"] == []
|
|
assert body["maxBitrateBps"] == 2_000_000
|
|
assert "version" in body
|
|
|
|
def test_config_is_a_whitelist(self, client):
|
|
body = client.get("/api/config").json()
|
|
assert set(body) == {"version", "iceServers", "maxBitrateBps", "pingIntervalMs"}
|
|
|
|
def test_streams_starts_empty(self, client):
|
|
body = client.get("/api/streams").json()
|
|
assert body["mirrors"] == []
|
|
assert body["stats"]["mirrors"] == 0
|
|
|
|
def test_healthz(self, client):
|
|
assert client.get("/healthz").json()["status"] == "ok"
|
|
|
|
|
|
class TestSignalingSocket:
|
|
def test_hello_must_come_first(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_json({"type": "publish", "label": "sneaky"})
|
|
message = socket.receive_json()
|
|
assert message["type"] == "error"
|
|
assert message["code"] == "hello_required"
|
|
assert message["fatal"] is True
|
|
|
|
def test_malformed_json_is_rejected(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_text("{not json")
|
|
assert socket.receive_json()["code"] == "bad_json"
|
|
|
|
def test_a_non_object_frame_is_rejected(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_text("[1, 2, 3]")
|
|
assert socket.receive_json()["code"] == "bad_json"
|
|
|
|
def test_an_unknown_role_is_refused(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
socket.send_json({"type": "hello", "role": "spectator"})
|
|
assert socket.receive_json()["code"] == "bad_role"
|
|
|
|
def test_unknown_message_type_is_reported_but_survivable(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
handshake(socket, "viewer", "V")
|
|
socket.send_json({"type": "telepathy"})
|
|
assert socket.receive_json()["code"] == "bad_message"
|
|
socket.send_json({"type": "ping"})
|
|
assert socket.receive_json()["type"] == "pong"
|
|
|
|
def test_ping_roundtrips(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
handshake(socket, "viewer")
|
|
socket.send_json({"type": "ping"})
|
|
assert socket.receive_json() == {"type": "pong"}
|
|
|
|
def test_an_oversized_frame_is_refused(self, client):
|
|
with client.websocket_connect("/ws") as socket:
|
|
handshake(socket, "viewer")
|
|
socket.send_text(json.dumps({"type": "ping", "padding": "x" * 70_000}))
|
|
assert socket.receive_json()["code"] == "too_large"
|
|
|
|
def test_roster_flows_from_sharer_to_viewer(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "Linux Box")
|
|
viewer_id = handshake(viewer, "viewer", "Phone")
|
|
|
|
sharer.send_json({"type": "publish", "label": "Terminal", "width": 1280, "height": 720})
|
|
added = wait_for(viewer, "stream_added")["stream"]
|
|
assert added["id"] == sharer_id
|
|
assert added["label"] == "Terminal"
|
|
assert added["name"] == "Linux Box"
|
|
assert added["width"] == 1280
|
|
|
|
viewer.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
joined = wait_for(sharer, "peer_joined")
|
|
assert joined["viewerId"] == viewer_id
|
|
assert joined["name"] == "Phone"
|
|
|
|
sharer.send_json({"type": "signal", "to": viewer_id, "payload": {"sdp": {"s": "o"}}})
|
|
relayed = wait_for(viewer, "signal")
|
|
assert relayed["from"] == sharer_id
|
|
assert relayed["payload"] == {"sdp": {"s": "o"}}
|
|
|
|
viewer.send_json({"type": "signal", "to": sharer_id, "payload": {"c": "ice"}})
|
|
assert wait_for(sharer, "signal")["payload"] == {"c": "ice"}
|
|
|
|
def test_capture_metadata_survives_the_wire(self, client):
|
|
"""Regression: `displaySurface` once arrived as "" because the model expected
|
|
snake_case and unknown keys were being ignored."""
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json(
|
|
{
|
|
"type": "publish",
|
|
"label": "proj.gl — zsh",
|
|
"displaySurface": "window",
|
|
"width": 1920,
|
|
"height": 1080,
|
|
"audio": True,
|
|
}
|
|
)
|
|
stream = wait_for(viewer, "stream_added")["stream"]
|
|
assert stream["displaySurface"] == "window"
|
|
assert stream["audio"] is True
|
|
assert (stream["width"], stream["height"]) == (1920, 1080)
|
|
assert client.get("/api/streams").json()["mirrors"][0]["displaySurface"] == "window"
|
|
|
|
def test_both_viewers_of_one_mirror_see_their_own_leg(self, client):
|
|
with (
|
|
client.websocket_connect("/ws") as sharer,
|
|
client.websocket_connect("/ws") as one,
|
|
client.websocket_connect("/ws") as two,
|
|
):
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
first = handshake(one, "viewer", "one")
|
|
second = handshake(two, "viewer", "two")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(one, "stream_added")
|
|
wait_for(two, "stream_added")
|
|
|
|
one.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
two.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
assert wait_for(sharer, "peer_joined")["viewerId"] == first
|
|
assert wait_for(sharer, "peer_joined")["viewerId"] == second
|
|
|
|
sharer.send_json({"type": "signal", "to": first, "payload": {"n": 1}})
|
|
sharer.send_json({"type": "signal", "to": second, "payload": {"n": 2}})
|
|
assert wait_for(one, "signal")["payload"] == {"n": 1}
|
|
assert wait_for(two, "signal")["payload"] == {"n": 2}
|
|
|
|
def test_a_mirror_disappears_for_viewers_when_the_sharer_leaves(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer", "V")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
wait_for(sharer, "peer_joined")
|
|
|
|
sharer.close()
|
|
assert wait_for(viewer, "stream_removed")["streamId"] == sharer_id
|
|
|
|
def test_unpublish_removes_the_mirror(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
sharer.send_json({"type": "unpublish"})
|
|
assert wait_for(viewer, "stream_removed")["streamId"] == sharer_id
|
|
assert client.get("/api/streams").json()["mirrors"] == []
|
|
|
|
def test_signalling_a_peer_that_never_subscribed_is_refused(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer")
|
|
viewer_id = handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
assert wait_for(viewer, "stream_added")["stream"]["sharerId"] == sharer_id
|
|
# The viewer knows the mirror exists but has not subscribed to it.
|
|
sharer.send_json({"type": "signal", "to": viewer_id, "payload": {}})
|
|
assert wait_for(sharer, "error")["code"] == "not_subscribed"
|
|
|
|
def test_a_viewer_cannot_reach_a_sharer_it_never_subscribed_to(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "signal", "to": sharer_id, "payload": {}})
|
|
assert wait_for(viewer, "error")["code"] == "not_subscribed"
|
|
|
|
def test_subscribing_to_a_mirror_that_is_not_there_reports_cleanly(self, client):
|
|
with client.websocket_connect("/ws") as viewer:
|
|
handshake(viewer, "viewer")
|
|
viewer.send_json({"type": "subscribe", "streamId": "p404"})
|
|
message = wait_for(viewer, "error")
|
|
assert message["code"] == "unknown_stream"
|
|
assert message["fatal"] is False
|
|
|
|
def test_editing_the_display_name_reaches_viewers(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "Old name")
|
|
handshake(viewer, "viewer")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
sharer.send_json({"type": "hello", "role": "sharer", "name": "Design Laptop"})
|
|
assert wait_for(viewer, "stream_updated")["stream"]["name"] == "Design Laptop"
|
|
assert client.get("/api/streams").json()["mirrors"][0]["sharerId"] == sharer_id
|
|
|
|
def test_stats_track_the_fan_out(self, client):
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer", "V")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "subscribe", "streamId": sharer_id})
|
|
wait_for(sharer, "peer_joined")
|
|
stats = client.get("/api/streams").json()["stats"]
|
|
assert stats["mirrors"] == 1
|
|
assert stats["watching"] == 1
|
|
assert stats["connections"] == 2
|
|
|
|
def test_one_peer_cannot_impersonate_another(self, client):
|
|
"""`to` is addressed by peer id; a viewer must not be able to spoof a sharer."""
|
|
with client.websocket_connect("/ws") as sharer, client.websocket_connect("/ws") as viewer:
|
|
sharer_id = handshake(sharer, "sharer", "A")
|
|
handshake(viewer, "viewer", "V")
|
|
sharer.send_json({"type": "publish", "label": "Screen"})
|
|
wait_for(viewer, "stream_added")
|
|
viewer.send_json({"type": "publish", "label": "forged"})
|
|
assert wait_for(viewer, "error")["code"] == "not_sharer"
|
|
assert client.get("/api/streams").json()["stats"]["mirrors"] == 1
|
|
assert sharer_id and viewer
|