Experimental/mirror
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c5da9461a2a68a6a772021eedd6ae8b98c05b9db
mirror/tests/e2e/test_live_mirror.py
T
ducoterra 7988da2b93 feat: decide who may connect at all, by source address 
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
2026-11-01 17:47:31 -05:00

303 lines
12 KiB
Python
Raw Blame History

"""End-to-end: a real `mirror serve`, real Chromium, real captured frames.
Every assertion about "it works" is made on pixels that reached the viewer's
<video>, not on log lines. If the media path breaks, these tests break.
"""
from __future__ import annotations
import re
import time
import pytest
from playwright.sync_api import expect
from .conftest import frames_matching, start_sharing, stop_sharing
pytestmark = [pytest.mark.e2e, pytest.mark.timeout(180)]
@pytest.fixture
def device(base_url, browser, request):
"""Open a page as its own device (isolated context, fresh peer id)."""
opened = []
def _open(path: str = "/", errors: list | None = None):
context = browser.new_context(viewport={"width": 1100, "height": 720})
opened.append(context)
request.addfinalizer(context.close)
page = context.new_page()
if errors is not None:
page.on("pageerror", lambda error: errors.append(str(error)))
page.goto(f"{base_url}{path}", wait_until="domcontentloaded")
expect(page.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
return page
return _open
def tile_ids(page) -> list[str]:
return page.locator(".tile").evaluate_all("els => els.map(e => e.dataset.stream)")
class TestShareThenWatch:
def test_a_shared_window_reaches_a_viewer_as_moving_pixels(self, device):
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
expect(sharer.locator("#stat-resolution")).to_have_text(
re.compile(r"\d{3,4}\s.\s\d{3,4}"), timeout=10_000
)
expect(sharer.locator("#stat-surface")).to_contain_text("Screen", timeout=5_000)
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
frames = frames_matching(viewer, ".tile video")
assert not frames.get("missing"), "no <video> element in the tile"
assert frames["width"] > 0 and frames["height"] > 0, f"no decoded frame size: {frames}"
assert frames["distinct"] >= 2, f"video is frozen: {frames}"
# The sharer learns that someone is watching.
expect(sharer.locator("#viewer-count")).to_have_text("1", timeout=10_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_clicking_a_tile_takes_it_on_stage(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
viewer.locator(".tile").click()
expect(viewer.locator("body")).to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
expect(viewer.locator("#watch-stage video")).to_have_count(1)
# Still moving while staged — staging must not restart the peer connection.
frames = frames_matching(viewer, "#watch-stage video")
assert frames["distinct"] >= 2, f"staged video froze: {frames}"
# And the mirror survives the stage: the sharer still has one viewer.
expect(sharer.locator("#viewer-count")).to_have_text("1", timeout=10_000)
def test_stopping_the_share_clears_the_wall(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
stop_sharing(sharer)
expect(viewer.locator(".tile")).to_have_count(0, timeout=10_000)
expect(viewer.locator("#empty")).to_be_visible()
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"))
def test_closing_the_sharer_tab_clears_the_wall(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
sharer.close()
expect(viewer.locator(".tile")).to_have_count(0, timeout=10_000)
class TestChromeVisibility:
"""The page must never advertise a control that does not apply to its state."""
def test_the_idle_share_page_offers_only_a_start(self, device):
sharer = device("/mirror")
expect(sharer.locator("#start")).to_be_visible()
expect(sharer.locator("#placeholder")).to_be_visible()
expect(sharer.locator("#stop")).not_to_be_visible()
expect(sharer.locator("#livetag")).not_to_be_visible()
start_sharing(sharer)
expect(sharer.locator("#stop")).to_be_visible()
expect(sharer.locator("#livetag")).to_be_visible()
expect(sharer.locator("#start")).not_to_be_visible()
expect(sharer.locator("#placeholder")).not_to_be_visible()
def test_exit_only_appears_on_the_tile_that_is_on_stage(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile")).to_have_count(1, timeout=15_000)
expect(viewer.locator(".tile .exit")).not_to_be_visible()
viewer.locator(".tile").click()
expect(viewer.locator("#watch-stage .exit")).to_be_visible()
viewer.locator("#watch-stage .exit").click()
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
expect(viewer.locator(".tile .exit")).not_to_be_visible()
# Leaving the stage must not have disturbed the stream.
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=20_000)
class TestManyToMany:
def test_one_mirror_fans_out_to_several_viewers(self, device):
errors: list = []
sharer = device("/mirror", errors)
start_sharing(sharer)
viewers = [device("/watch", errors) for _ in range(3)]
for viewer in viewers:
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
expect(sharer.locator("#viewer-count")).to_have_text("3", timeout=15_000)
expect(sharer.locator("#audience li")).to_have_count(3, timeout=10_000)
for viewer in viewers:
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"a viewer froze: {frames}"
# One viewer leaving must not disturb the others.
viewers[0].close()
expect(sharer.locator("#viewer-count")).to_have_text("2", timeout=15_000)
for viewer in viewers[1:]:
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"collateral damage: {frames}"
assert errors == [], f"uncaught browser errors: {errors}"
def test_several_mirrors_land_on_one_wall(self, device):
errors: list = []
sharers = []
for _ in range(2):
page = device("/mirror", errors)
start_sharing(page)
sharers.append(page)
viewer = device("/watch", errors)
expect(viewer.locator(".tile")).to_have_count(2, timeout=15_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=30_000)
ids = tile_ids(viewer)
assert len(ids) == 2 and len(set(ids)) == 2, "tiles must be distinct mirrors"
# Stage the second mirror, then jump to the first.
viewer.locator(".tile").nth(1).click()
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
staged = viewer.locator("#watch-stage .tile").get_attribute("data-stream")
assert staged == ids[1]
viewer.locator("#watch-stage .exit").click()
expect(viewer.locator("body")).not_to_have_class(re.compile("staged"), timeout=5_000)
viewer.locator(".tile").nth(0).click()
expect(viewer.locator("#watch-stage .tile")).to_have_count(1)
assert viewer.locator("#watch-stage .tile").get_attribute("data-stream") == ids[0]
# Both mirrors are still alive and untouched by the staging gymnastics.
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=20_000)
assert errors == [], f"uncaught browser errors: {errors}"
def test_stopping_one_mirror_leaves_the_other_running(self, device):
first = device("/mirror")
start_sharing(first)
second = device("/mirror")
start_sharing(second)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(2, timeout=30_000)
stop_sharing(first)
expect(viewer.locator(".tile")).to_have_count(1, timeout=10_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=20_000)
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"the surviving mirror froze: {frames}"
class TestTimingAndRecovery:
def test_a_viewer_already_on_the_wall_sees_a_late_mirror(self, device):
viewer = device("/watch")
expect(viewer.locator("#empty")).to_be_visible()
sharer = device("/mirror")
start_sharing(sharer)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"late mirror did not animate: {frames}"
def test_a_reloaded_viewer_recovers_the_live_mirror(self, device):
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
viewer.reload(wait_until="domcontentloaded")
expect(viewer.locator("#net")).to_have_attribute("data-state", "online", timeout=10_000)
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
frames = frames_matching(viewer, ".tile video")
assert frames["distinct"] >= 2, f"reload did not restore motion: {frames}"
expect(sharer.locator("#viewer-count")).to_have_text("1", timeout=15_000)
def test_the_index_page_counts_what_is_live(self, device):
index = device("/")
expect(index.locator("#livebar")).to_contain_text("No mirrors live", timeout=10_000)
sharer = device("/mirror")
start_sharing(sharer)
expect(index.locator("#livebar")).to_contain_text("1 mirror", timeout=10_000)
class TestRosterTruthfulness:
def test_the_api_agrees_with_what_the_browser_shows(self, device, base_url):
import httpx
sharer = device("/mirror")
start_sharing(sharer)
viewer = device("/watch")
expect(viewer.locator(".tile[data-state='live']")).to_have_count(1, timeout=25_000)
# Give the stats ticker a beat, then compare server truth to pixels.
time.sleep(1.0)
payload = httpx.get(f"{base_url}/api/streams", timeout=5).json()
assert payload["stats"]["mirrors"] == 1
assert payload["stats"]["watching"] == 1
mirror = payload["mirrors"][0]
assert mirror["width"] and mirror["height"]
assert mirror["displaySurface"] in ("monitor", "window", "browser")
class TestAddressGate:
"""`--allow-cidr` in a real process, refused by a real browser.
``tests/test_app.py::TestAddressGate`` covers the JSON refusal the pages key
their own notice on, the proxy unwrapping, and the socket closing before it
accepts. This covers what a person actually gets when a server refuses them:
an explanation naming the address it judged, not an empty wall that looks
like a bug in the tool.
"""
@pytest.mark.parametrize("path", ["/mirror", "/watch"])
def test_a_refused_address_sees_why_and_none_of_the_app(self, browser, restricted_url, path):
context = browser.new_context(viewport={"width": 1100, "height": 720})
page = context.new_page()
try:
response = page.goto(f"{restricted_url}{path}")
assert response.status == 403
expect(page.locator("body")).to_contain_text("not open to your address")
expect(page.locator("body")).to_contain_text("203.0.113.0/24")
# Which address it judged — without this, "you are not allowed" in
# front of a proxy is undiagnosable.
expect(page.locator("body")).to_contain_text("127.0.0.1")
# None of the app shell leaked past the gate.
expect(page.locator("#start, #grid, #notice")).to_have_count(0)
finally:
context.close()
def test_the_health_probe_is_not_gated(self, restricted_url):
import httpx
assert httpx.get(f"{restricted_url}/healthz", timeout=5).status_code == 200
Reference in New Issue View Git Blame Copy Permalink