Camera permission and review fixes (openpanel C3, C4)
C4: a device's `viewers` (superusers by default, or members) covers the image and video messages its ports publish. Anyone else gets none of their values on the socket, in the catalogue or a flow's state, 403 for their history, for the frames they hold (the last 8 included) and for TURN credentials, no bytes when asking for them, and `bye: forbidden` for a live view. A device credential stays bounded by its dashboards. The device page sets it in its Camera section, and a tile the viewer may not see says "No access to this camera" and asks the camera for nothing. C3: - A volatile upload is held to MAX_FRAME_BYTES (4 MiB, 413 above it, read without extra copies), and so is a frame on a device's socket. - A frame the engine holds in memory is written to the ring directory before a local worker is handed it, so load_artifact finds it. - A socket the bus dropped events for catches up with a snapshot, demand included, and a device's credential is checked again per frame, cached by the flow store revision and its nonce, so a lost event never keeps an unpaired device's frames coming. - A report is at most 64 KB once its named config_schemas are put in place. - A device away for 60 s has its true bool messages set to false. - Refused frames are logged by fixed keys, a frame is named after its message, `fluksio serve` waits 5 s for open connections, a tile with a stream URL creates no demand, the 8-name cap keeps the newest names. - Every container's log rotates (json-file, 3 x 10 MB).
This commit is contained in:
Melvin Strobl
@@ -0,0 +1,228 @@
|
||||
"""Who may see a device's camera: superusers only, or every member too.
|
||||
|
||||
openpanel ROADMAP C4. ``viewers`` on a device covers its ``image`` and
|
||||
``video`` messages, their bytes, frames on the socket and its live view. A
|
||||
device credential is bounded by its dashboards instead.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from fluksio.core.config import settings
|
||||
from fluksio.flow.demand import demand
|
||||
from tests.api.routes.test_cameras import (
|
||||
CAMERA_NODE,
|
||||
REPORT,
|
||||
_ticket,
|
||||
frame,
|
||||
wait_for,
|
||||
)
|
||||
from tests.api.routes.test_devices import (
|
||||
FLOWS,
|
||||
_dashboard_with,
|
||||
_pair,
|
||||
_panels,
|
||||
_publish_flow,
|
||||
)
|
||||
|
||||
API = settings.API_V1_STR
|
||||
PREFIX = f"{API}/devices"
|
||||
SOCKET = f"{API}/flows/ws"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def camera(client: TestClient, superuser_token_headers: dict[str, str]):
|
||||
"""``cam1`` with a camera wired in ``cams``, and a frame on it."""
|
||||
_panels(client, superuser_token_headers, {"panels": [{"id": "cam1"}]})
|
||||
device = _pair(client, superuser_token_headers, "cam1")
|
||||
reported = client.put(f"{PREFIX}/cam1/report", headers=device, json=REPORT)
|
||||
assert reported.status_code == 200, reported.text
|
||||
_publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE])
|
||||
controller = client.app.state.flow_controller
|
||||
with client.websocket_connect(SOCKET, headers=device) as ws:
|
||||
ws.receive_json()
|
||||
for index in range(3):
|
||||
ws.send_bytes(frame("cams.camera", b"\xff\xd8 frame %d" % index))
|
||||
wait_for(
|
||||
lambda: (controller.values().get("cams.camera") or {})
|
||||
.get("value", {})
|
||||
.get("size")
|
||||
== len(b"\xff\xd8 frame 2")
|
||||
)
|
||||
controller.publish_message("cams.person", True)
|
||||
yield device
|
||||
client.delete(f"{FLOWS}/cams", headers=superuser_token_headers)
|
||||
client.delete(f"{PREFIX}/cam1", headers=superuser_token_headers)
|
||||
|
||||
|
||||
def _viewers(client: TestClient, headers: dict[str, str], viewers: str) -> None:
|
||||
current = next(
|
||||
one
|
||||
for one in client.get(f"{PREFIX}/", headers=headers).json()["data"]
|
||||
if one["id"] == "cam1"
|
||||
)
|
||||
body = {
|
||||
key: current[key] for key in ("id", "title", "dashboards", "touch", "version")
|
||||
}
|
||||
saved = client.put(
|
||||
f"{PREFIX}/cam1", headers=headers, json={**body, "viewers": viewers}
|
||||
)
|
||||
assert saved.status_code == 200, saved.text
|
||||
|
||||
|
||||
def _surfaces(client: TestClient, headers: dict[str, str]) -> dict[str, Any]:
|
||||
"""What one account gets of the camera on every surface but the socket."""
|
||||
controller = client.app.state.flow_controller
|
||||
digest = controller.values()["cams.camera"]["value"]["digest"]
|
||||
catalog = {
|
||||
info["name"]: info
|
||||
for info in client.get(f"{API}/messages/", headers=headers).json()["data"]
|
||||
}
|
||||
return {
|
||||
"catalog": catalog["cams.camera"]["value"] is not None,
|
||||
"person": catalog["cams.person"]["value"],
|
||||
"history": client.get(
|
||||
f"{API}/messages/cams.camera/history", headers=headers
|
||||
).status_code,
|
||||
"flow_history": client.get(
|
||||
f"{FLOWS}/cams/history/camera", headers=headers
|
||||
).status_code,
|
||||
"state": "cams.camera"
|
||||
in client.get(f"{FLOWS}/cams/state", headers=headers).json()["values"],
|
||||
"artifact": client.get(
|
||||
f"{API}/artifacts/{digest}", headers=headers
|
||||
).status_code,
|
||||
"ice": client.get(f"{PREFIX}/cam1/ice", headers=headers).status_code,
|
||||
}
|
||||
|
||||
|
||||
def _live_start(client: TestClient, headers: dict[str, str]) -> str:
|
||||
"""The reason a viewer's live view of cam1 ends with right away."""
|
||||
with client.websocket_connect(_ticket(client, headers)) as ws:
|
||||
ws.receive_json()
|
||||
ws.send_json(
|
||||
{"type": "live", "device": "cam1", "session": "s-access", "action": "start"}
|
||||
)
|
||||
while True:
|
||||
message = ws.receive_json()
|
||||
if message.get("type") == "live":
|
||||
return str(message.get("reason"))
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("camera")
|
||||
def test_a_camera_is_for_superusers_until_opened_to_members(
|
||||
client: TestClient,
|
||||
superuser_token_headers: dict[str, str],
|
||||
normal_user_token_headers: dict[str, str],
|
||||
) -> None:
|
||||
member, admin = normal_user_token_headers, superuser_token_headers
|
||||
open_to_all = {
|
||||
"catalog": True,
|
||||
"person": True,
|
||||
"history": 200,
|
||||
"flow_history": 200,
|
||||
"state": True,
|
||||
"artifact": 200,
|
||||
"ice": 200,
|
||||
}
|
||||
assert _surfaces(client, admin) == open_to_all
|
||||
assert _surfaces(client, member) == {
|
||||
"catalog": False,
|
||||
# Not a camera: its bool is anyone's.
|
||||
"person": True,
|
||||
"history": 403,
|
||||
"flow_history": 403,
|
||||
"state": False,
|
||||
"artifact": 403,
|
||||
"ice": 403,
|
||||
}
|
||||
# The frames before the current one too.
|
||||
store = client.app.state.artifact_store
|
||||
older = store.put([b"\xff\xd8 older"], media_type="image/jpeg", volatile=True)
|
||||
client.app.state.flow_controller.publish_message("cams.camera", older)
|
||||
newer = store.put([b"\xff\xd8 newer"], media_type="image/jpeg", volatile=True)
|
||||
client.app.state.flow_controller.publish_message("cams.camera", newer)
|
||||
fetched = client.get(f"{API}/artifacts/{older['digest']}", headers=member)
|
||||
assert fetched.status_code == 403
|
||||
assert _live_start(client, member) == "forbidden"
|
||||
assert _live_start(client, admin) == "offline"
|
||||
|
||||
_viewers(client, admin, "members")
|
||||
assert _surfaces(client, member) == open_to_all
|
||||
assert _live_start(client, member) == "offline"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("camera")
|
||||
def test_a_members_socket_carries_no_frames_of_a_closed_camera(
|
||||
client: TestClient,
|
||||
superuser_token_headers: dict[str, str],
|
||||
normal_user_token_headers: dict[str, str],
|
||||
camera: dict[str, str],
|
||||
) -> None:
|
||||
"""Neither its values, nor the bytes, nor demand on the device; and the
|
||||
socket is given them once the camera is opened."""
|
||||
with client.websocket_connect(_ticket(client, normal_user_token_headers)) as ws:
|
||||
snapshot = ws.receive_json()
|
||||
assert "cams.camera" not in snapshot["values"]
|
||||
assert "cams.person" in snapshot["values"]
|
||||
ws.send_json({"type": "media", "names": ["cams.camera", "cams.person"]})
|
||||
wait_for(lambda: demand.viewed() == {"cams.person"})
|
||||
assert demand.ports("cam1") == []
|
||||
|
||||
with client.websocket_connect(SOCKET, headers=camera) as device:
|
||||
device.receive_json()
|
||||
device.send_bytes(frame("cams.camera", b"\xff\xd8 not for members"))
|
||||
client.app.state.flow_controller.publish_message("cams.person", False)
|
||||
# The person's flag arrives, the frame before it did not.
|
||||
while True:
|
||||
message = ws.receive_json()
|
||||
names = [
|
||||
event.get("name")
|
||||
for event in message.get("events", [message])
|
||||
if event.get("type") == "message_value"
|
||||
]
|
||||
assert "cams.camera" not in names
|
||||
if "cams.person" in names:
|
||||
break
|
||||
|
||||
_viewers(client, superuser_token_headers, "members")
|
||||
while True:
|
||||
message = ws.receive_json()
|
||||
if message.get("type") == "snapshot":
|
||||
break
|
||||
assert "cams.camera" in message["values"]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("camera")
|
||||
def test_a_wall_panel_shows_the_camera_its_dashboards_carry(
|
||||
client: TestClient, superuser_token_headers: dict[str, str]
|
||||
) -> None:
|
||||
"""Assigned by a superuser, so the panel's own scope is what bounds it."""
|
||||
_dashboard_with(
|
||||
client,
|
||||
superuser_token_headers,
|
||||
"panel_cams",
|
||||
[
|
||||
{
|
||||
"id": "w_cam",
|
||||
"type": "media",
|
||||
"config": {"message": "cams.camera", "dtype": "image"},
|
||||
}
|
||||
],
|
||||
)
|
||||
_panels(
|
||||
client,
|
||||
superuser_token_headers,
|
||||
{"panels": [{"id": "cam1"}, {"id": "hall", "dashboards": ["panel_cams"]}]},
|
||||
)
|
||||
panel = _pair(client, superuser_token_headers, "hall")
|
||||
try:
|
||||
value = client.app.state.flow_controller.values()["cams.camera"]["value"]
|
||||
fetched = client.get(f"{API}/artifacts/{value['digest']}", headers=panel)
|
||||
assert fetched.status_code == 200
|
||||
with client.websocket_connect(SOCKET, headers=panel) as ws:
|
||||
assert "cams.camera" in ws.receive_json()["values"]
|
||||
finally:
|
||||
client.delete(f"{PREFIX}/hall", headers=superuser_token_headers)
|
||||
@@ -244,10 +244,14 @@ def test_a_frame_the_device_may_not_send_is_dropped(
|
||||
current = controller.values()["cams.camera"]["value"]
|
||||
data = client.app.state.artifact_store.volatile.get(current["digest"])
|
||||
assert data == b"\xff\xd8 after"
|
||||
from fluksio.api.routes.flows import FRAME_REFUSALS
|
||||
|
||||
# By fixed keys, never in the words the device chose.
|
||||
said = [r.getMessage() for r in caplog.records if "Dropped a frame" in r.message]
|
||||
assert any("'person' is wired as bool" in line for line in said)
|
||||
assert any("not a message one of its ports publishes" in line for line in said)
|
||||
assert any("only a device sends frames" in line for line in said)
|
||||
for key in ("not_image", "not_wired", "media_type", "not_a_frame"):
|
||||
assert f"Dropped a frame from cam1: {FRAME_REFUSALS[key]}" in said
|
||||
assert any(FRAME_REFUSALS["not_a_device"] in line for line in said)
|
||||
assert not any("elsewhere" in line or "text/html" in line for line in said)
|
||||
|
||||
|
||||
def test_an_unpaired_device_is_closed_and_its_frames_stop(
|
||||
@@ -264,6 +268,106 @@ def test_an_unpaired_device_is_closed_and_its_frames_stop(
|
||||
assert closed.value.code == 1008
|
||||
|
||||
|
||||
def test_a_frame_past_the_cap_is_dropped(
|
||||
client: TestClient, camera, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
controller = client.app.state.flow_controller
|
||||
monkeypatch.setattr(settings, "MAX_FRAME_BYTES", 1000)
|
||||
with client.websocket_connect(SOCKET, headers=camera) as device:
|
||||
device.receive_json()
|
||||
device.send_bytes(frame("cams.camera", b"\xff" * 2000))
|
||||
device.send_bytes(frame("cams.camera", b"\xff\xd8 small"))
|
||||
wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value"))
|
||||
digest = controller.values()["cams.camera"]["value"]["digest"]
|
||||
store = client.app.state.artifact_store
|
||||
assert store.volatile.get(digest) == b"\xff\xd8 small"
|
||||
|
||||
|
||||
def test_a_frame_is_named_after_its_message(client: TestClient, camera) -> None:
|
||||
controller = client.app.state.flow_controller
|
||||
with client.websocket_connect(SOCKET, headers=camera) as device:
|
||||
device.receive_json()
|
||||
device.send_bytes(frame("cams.camera", b"\xff\xd8 named"))
|
||||
value = wait_for(
|
||||
lambda: (controller.values().get("cams.camera") or {}).get("value")
|
||||
)
|
||||
assert value["name"] == "cams.camera"
|
||||
|
||||
|
||||
def test_an_unpaired_device_is_closed_at_its_next_frame(
|
||||
client: TestClient, camera
|
||||
) -> None:
|
||||
"""Without the event an unpair sends, which a full queue may have lost."""
|
||||
from starlette.websockets import WebSocketDisconnect
|
||||
|
||||
with client.websocket_connect(SOCKET, headers=camera) as device:
|
||||
device.receive_json()
|
||||
device.send_bytes(frame("cams.camera", b"\xff\xd8 one"))
|
||||
client.app.state.device_store.bump("cam1")
|
||||
device.send_bytes(frame("cams.camera", b"\xff\xd8 two"))
|
||||
with pytest.raises(WebSocketDisconnect) as closed:
|
||||
while True:
|
||||
device.receive_json()
|
||||
assert closed.value.code == 1008
|
||||
store = client.app.state.artifact_store
|
||||
current = client.app.state.flow_controller.values()["cams.camera"]["value"]
|
||||
assert store.volatile.get(current["digest"]) == b"\xff\xd8 one"
|
||||
|
||||
|
||||
def test_a_device_that_lost_events_catches_up(
|
||||
client: TestClient, camera, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Its demand and its unpairing come again with a snapshot, after the
|
||||
events it did get."""
|
||||
from starlette.websockets import WebSocketDisconnect
|
||||
|
||||
controller = client.app.state.flow_controller
|
||||
with client.websocket_connect(SOCKET, headers=camera) as device:
|
||||
assert device.receive_json()["demand"] == []
|
||||
# Demand changes while the bus drops what the socket has queued.
|
||||
monkeypatch.setattr(event_bus, "dropped", lambda queue: True)
|
||||
demand.want("someone", {"cams.camera"})
|
||||
try:
|
||||
seen = []
|
||||
while not seen or seen[-1].get("type") != "snapshot":
|
||||
seen.append(device.receive_json())
|
||||
assert seen[-1]["demand"] == ["camera"]
|
||||
finally:
|
||||
demand.drop("someone")
|
||||
|
||||
# And an unpair it never heard of closes it at the next batch.
|
||||
client.app.state.device_store.bump("cam1")
|
||||
controller.publish_message("cams.person", True)
|
||||
with pytest.raises(WebSocketDisconnect) as closed:
|
||||
while True:
|
||||
device.receive_json()
|
||||
assert closed.value.code == 1008
|
||||
|
||||
|
||||
def test_a_report_is_bounded_once_its_schemas_are_in_place(
|
||||
client: TestClient, camera
|
||||
) -> None:
|
||||
"""Thirty pins naming one big schema are thirty copies of it, kept."""
|
||||
from fluksio.api.routes.devices import REPORT_MAX, REPORT_STORED_MAX
|
||||
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {f"p{i}": {"type": "integer"} for i in range(250)},
|
||||
}
|
||||
pins = [
|
||||
{"port": f"gpio.{i}", "direction": "in", "dtype": "bool", "config": "pin"}
|
||||
for i in range(60)
|
||||
]
|
||||
report = {**REPORT, "config_schemas": {"pin": schema}, "ports": pins}
|
||||
raw = orjson.dumps(report)
|
||||
assert len(raw) < REPORT_MAX
|
||||
answer = client.put(f"{PREFIX}/cam1/report", headers=camera, content=raw)
|
||||
assert answer.status_code == 413, answer.text
|
||||
kept = client.app.state.device_store.state("cam1").report
|
||||
assert len(orjson.dumps(kept)) < REPORT_STORED_MAX
|
||||
assert kept["board"] == "nano"
|
||||
|
||||
|
||||
def test_a_screen_may_fetch_the_last_few_frames(client: TestClient, camera) -> None:
|
||||
"""A remote screen fetching frame by frame is often one behind."""
|
||||
controller = client.app.state.flow_controller
|
||||
@@ -290,7 +394,8 @@ def test_a_viewer_asks_for_a_few_cameras_at_most() -> None:
|
||||
|
||||
names = [f"cams.c{i}" for i in range(MAX_MEDIA_NAMES + 3)]
|
||||
asked = wanted_names(json.dumps({"type": "media", "names": names}), None)
|
||||
assert asked == set(names[:MAX_MEDIA_NAMES])
|
||||
# The newest, which a page lists last: a tile scrolled back to keeps working.
|
||||
assert asked == set(names[-MAX_MEDIA_NAMES:])
|
||||
|
||||
|
||||
def test_viewers_through_the_portal_count_too(
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import asyncio
|
||||
from datetime import UTC, datetime
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
@@ -519,7 +520,9 @@ async def test_the_live_socket_ends_when_its_client_goes(
|
||||
because only SIGKILL ended the process. This returning is what ends that.
|
||||
"""
|
||||
monkeypatch.setattr(
|
||||
flows_route, "user_from_token", lambda session, token, **_: object()
|
||||
flows_route,
|
||||
"user_from_token",
|
||||
lambda session, token, **_: SimpleNamespace(id="someone", is_superuser=True),
|
||||
)
|
||||
|
||||
class Peer:
|
||||
|
||||
@@ -337,6 +337,7 @@ def test_a_full_outbox_ends_the_session_instead_of_dropping() -> None:
|
||||
person="someone",
|
||||
device="",
|
||||
devices=Devices(),
|
||||
superuser=True,
|
||||
)
|
||||
assert registry.count("cam9") == 1
|
||||
# The viewer never reads, and the device trickles on.
|
||||
|
||||
Reference in New Issue
Block a user