stroblmeandClaude Opus 5
4c8339e643
A screen somewhere this installation is not reachable from asks the portal for a code instead, and the portal mints its credential — because a token signed here is one such a device could never present. Where it was minted changes nothing about what it may do. The panel gate moved off the branch that decodes a local panel token and onto whatever claims name a panel, so the portal's and this installation's are bounded by the same check against the same panel's dashboards. A token of that scope naming no panel is refused rather than left holding the account it borrows. The connector marks what arrives on its socket, since that is the only thing that makes it true, and the approval screen now names what is holding a code — approving adopts whatever answers, so it is worth a look first. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017F9RnYCJgASuBTcAjxmnsp
309 lines
10 KiB
Python
309 lines
10 KiB
Python
"""Panels: assignment, pairing, and what a paired credential may reach."""
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import settings
|
|
|
|
PREFIX = f"{settings.API_V1_STR}/panels"
|
|
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
|
|
|
|
|
|
def _panels(client: TestClient, headers: dict[str, str], config: dict) -> None:
|
|
response = client.put(f"{PREFIX}/", headers=headers, json=config)
|
|
assert response.status_code == 200, response.text
|
|
|
|
|
|
def _pair(client: TestClient, headers: dict[str, str], panel: str) -> dict[str, str]:
|
|
"""Walk a device through pairing and return the header it ends up with."""
|
|
started = client.post(f"{PREFIX}/pair").json()
|
|
waiting = client.get(
|
|
f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
|
|
)
|
|
assert waiting.json()["access_token"] is None
|
|
|
|
approved = client.post(
|
|
f"{PREFIX}/{panel}/pair", headers=headers, json={"code": started["code"]}
|
|
)
|
|
assert approved.status_code == 200, approved.text
|
|
|
|
collected = client.get(
|
|
f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
|
|
).json()
|
|
assert collected["panel"] == panel
|
|
return {"Authorization": f"Bearer {collected['access_token']}"}
|
|
|
|
|
|
def test_panels_require_authentication(client: TestClient) -> None:
|
|
assert client.get(f"{PREFIX}/").status_code == 401
|
|
assert client.put(f"{PREFIX}/", json={"panels": []}).status_code == 401
|
|
|
|
|
|
def test_assign_and_read_back(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
for name in ("hall_a", "hall_b"):
|
|
client.post(f"{DASHBOARDS}/{name}", headers=superuser_token_headers)
|
|
|
|
_panels(
|
|
client,
|
|
superuser_token_headers,
|
|
{
|
|
"panels": [
|
|
{"id": "hall", "title": "Hall", "dashboards": ["hall_a", "hall_b"]}
|
|
]
|
|
},
|
|
)
|
|
|
|
stored = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()
|
|
assert stored["panels"][0]["dashboards"] == ["hall_a", "hall_b"]
|
|
# The address a device is pointed at comes from the server, because the
|
|
# browser's own origin is the portal's when someone administers remotely.
|
|
assert stored["frontend_host"] == settings.FRONTEND_HOST.rstrip("/")
|
|
assert (
|
|
client.get(f"{PREFIX}/hall", headers=superuser_token_headers).json()["title"]
|
|
== "Hall"
|
|
)
|
|
assert (
|
|
client.get(f"{PREFIX}/nowhere", headers=superuser_token_headers).status_code
|
|
== 404
|
|
)
|
|
|
|
|
|
def test_duplicate_panel_is_refused(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
response = client.put(
|
|
f"{PREFIX}/",
|
|
headers=superuser_token_headers,
|
|
json={"panels": [{"id": "twice"}, {"id": "twice"}]},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_polling_needs_the_secret(client: TestClient) -> None:
|
|
started = client.post(f"{PREFIX}/pair").json()
|
|
assert len(started["code"]) == 6
|
|
assert (
|
|
client.get(
|
|
f"{PREFIX}/pair/{started['code']}", params={"secret": "wrong"}
|
|
).status_code
|
|
== 404
|
|
)
|
|
assert (
|
|
client.get(f"{PREFIX}/pair/ZZZZZZ", params={"secret": "x"}).status_code == 404
|
|
)
|
|
|
|
|
|
def test_approving_an_unknown_code_or_panel_is_refused(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
_panels(client, superuser_token_headers, {"panels": [{"id": "hall"}]})
|
|
assert (
|
|
client.post(
|
|
f"{PREFIX}/hall/pair",
|
|
headers=superuser_token_headers,
|
|
json={"code": "ZZZZZZ"},
|
|
).status_code
|
|
== 404
|
|
)
|
|
started = client.post(f"{PREFIX}/pair").json()
|
|
assert (
|
|
client.post(
|
|
f"{PREFIX}/nowhere/pair",
|
|
headers=superuser_token_headers,
|
|
json={"code": started["code"]},
|
|
).status_code
|
|
== 404
|
|
)
|
|
|
|
|
|
def test_paired_panel_reaches_only_what_it_shows(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
for name in ("panel_shown", "panel_hidden"):
|
|
client.post(f"{DASHBOARDS}/{name}", headers=superuser_token_headers)
|
|
_panels(
|
|
client,
|
|
superuser_token_headers,
|
|
{"panels": [{"id": "hall", "dashboards": ["panel_shown"]}]},
|
|
)
|
|
|
|
panel_headers = _pair(client, superuser_token_headers, "hall")
|
|
|
|
# What it was assigned, published, plus its own definition and the messages
|
|
# its widgets speak.
|
|
assert (
|
|
client.get(f"{DASHBOARDS}/panel_shown", headers=panel_headers).status_code
|
|
== 200
|
|
)
|
|
assert client.get(f"{PREFIX}/hall", headers=panel_headers).status_code == 200
|
|
assert (
|
|
client.get(
|
|
f"{settings.API_V1_STR}/messages/", headers=panel_headers
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
# The generated client spells the default out, so `?draft=false` is what a
|
|
# browser actually asks with for "the published one".
|
|
for spelling in ("false", "0", "off"):
|
|
assert (
|
|
client.get(
|
|
f"{DASHBOARDS}/panel_shown",
|
|
headers=panel_headers,
|
|
params={"draft": spelling},
|
|
).status_code
|
|
== 200
|
|
), spelling
|
|
|
|
# And nothing else.
|
|
assert (
|
|
client.get(f"{DASHBOARDS}/panel_hidden", headers=panel_headers).status_code
|
|
== 403
|
|
)
|
|
# Fail-closed: a spelling neither side recognises counts as a draft.
|
|
for spelling in ("true", "1", "yes", "maybe"):
|
|
assert (
|
|
client.get(
|
|
f"{DASHBOARDS}/panel_shown",
|
|
headers=panel_headers,
|
|
params={"draft": spelling},
|
|
).status_code
|
|
== 403
|
|
), spelling
|
|
assert client.get(f"{DASHBOARDS}/", headers=panel_headers).status_code == 403
|
|
assert (
|
|
client.get(f"{settings.API_V1_STR}/flows/", headers=panel_headers).status_code
|
|
== 403
|
|
)
|
|
assert (
|
|
client.delete(f"{DASHBOARDS}/panel_shown", headers=panel_headers).status_code
|
|
== 403
|
|
)
|
|
assert client.get(f"{PREFIX}/", headers=panel_headers).status_code == 403
|
|
|
|
|
|
def test_removing_the_panel_revokes_its_credential(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
client.post(f"{DASHBOARDS}/panel_gone", headers=superuser_token_headers)
|
|
_panels(
|
|
client,
|
|
superuser_token_headers,
|
|
{"panels": [{"id": "workshop", "dashboards": ["panel_gone"]}]},
|
|
)
|
|
panel_headers = _pair(client, superuser_token_headers, "workshop")
|
|
assert (
|
|
client.get(f"{DASHBOARDS}/panel_gone", headers=panel_headers).status_code == 200
|
|
)
|
|
|
|
_panels(client, superuser_token_headers, {"panels": []})
|
|
# 401, not 403: there is nothing left to be forbidden from, and the device
|
|
# should go back to the pairing screen rather than retry.
|
|
assert (
|
|
client.get(f"{DASHBOARDS}/panel_gone", headers=panel_headers).status_code == 401
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# A screen that reached the portal but not this installation
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def test_the_device_asking_is_named_before_anyone_approves(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
"""Approving a code adopts whatever holds it, so it is worth a look."""
|
|
started = client.post(
|
|
f"{PREFIX}/pair",
|
|
headers={
|
|
"user-agent": "Mozilla/5.0 (X11; CrOS aarch64)",
|
|
"x-forwarded-for": "203.0.113.7, 10.0.0.1",
|
|
},
|
|
).json()
|
|
|
|
looked = client.get(
|
|
f"{PREFIX}/pair/{started['code']}/device", headers=superuser_token_headers
|
|
)
|
|
assert looked.status_code == 200
|
|
assert "CrOS" in looked.json()["device"]
|
|
# The first hop, not the proxy that relayed it.
|
|
assert "203.0.113.7" in looked.json()["device"]
|
|
assert looked.json()["remote"] is False
|
|
|
|
# Nobody without an account gets to enumerate what is waiting.
|
|
assert client.get(f"{PREFIX}/pair/{started['code']}/device").status_code == 401
|
|
assert (
|
|
client.get(
|
|
f"{PREFIX}/pair/ZZZZZZ/device", headers=superuser_token_headers
|
|
).status_code
|
|
== 404
|
|
)
|
|
|
|
|
|
def test_a_remote_device_is_paired_at_the_portal(
|
|
client: TestClient,
|
|
superuser_token_headers: dict[str, str],
|
|
enrolled: object, # noqa: ARG001 (fixture installs the enrolment)
|
|
monkeypatch, # type: ignore[no-untyped-def]
|
|
) -> None:
|
|
"""A device that arrived through the tunnel gets the portal's credential.
|
|
|
|
It could never present one this installation signed: the portal verifies
|
|
what crosses it, and it verifies against its own key.
|
|
"""
|
|
import httpx
|
|
|
|
from app.api.routes import panels as panels_route
|
|
|
|
calls: list[dict[str, object]] = []
|
|
|
|
def fake_post(url: str, **kwargs: object) -> httpx.Response:
|
|
calls.append({"url": url, **kwargs})
|
|
return httpx.Response(
|
|
200,
|
|
json={"access_token": "minted-by-the-portal", "expires_in": 31536000},
|
|
request=httpx.Request("POST", url),
|
|
)
|
|
|
|
monkeypatch.setattr(panels_route.httpx, "post", fake_post)
|
|
|
|
_panels(client, superuser_token_headers, {"panels": [{"id": "hallway"}]})
|
|
started = client.post(f"{PREFIX}/pair", headers={"x-fluksio-via": "portal"}).json()
|
|
|
|
looked = client.get(
|
|
f"{PREFIX}/pair/{started['code']}/device", headers=superuser_token_headers
|
|
).json()
|
|
assert looked["remote"] is True
|
|
|
|
approved = client.post(
|
|
f"{PREFIX}/hallway/pair",
|
|
headers=superuser_token_headers,
|
|
json={"code": started["code"]},
|
|
)
|
|
assert approved.status_code == 200, approved.text
|
|
|
|
assert calls[0]["url"].endswith("/api/v1/panel-tokens/") # type: ignore[union-attr]
|
|
assert calls[0]["headers"]["Authorization"] == "Bearer installation-token" # type: ignore[index]
|
|
assert calls[0]["json"] == {"panel": "hallway"} # type: ignore[index]
|
|
|
|
collected = client.get(
|
|
f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
|
|
).json()
|
|
assert collected["access_token"] == "minted-by-the-portal"
|
|
|
|
|
|
def test_a_remote_device_needs_an_enrolment(
|
|
client: TestClient, superuser_token_headers: dict[str, str]
|
|
) -> None:
|
|
"""Unenrolled, there is nowhere to ask — and no token to invent locally."""
|
|
_panels(client, superuser_token_headers, {"panels": [{"id": "shed"}]})
|
|
started = client.post(f"{PREFIX}/pair", headers={"x-fluksio-via": "portal"}).json()
|
|
|
|
approved = client.post(
|
|
f"{PREFIX}/shed/pair",
|
|
headers=superuser_token_headers,
|
|
json={"code": started["code"]},
|
|
)
|
|
assert approved.status_code == 409
|