Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 20 additions & 15 deletions tests/pytest/restapi/test_debug_websocket_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ def test_a_non_license_command_also_requires_a_live_token(socketio, app, client,


# ---------------------------------------------------------------------------
# 3. The license FCs require the admin role
# 3. The license FCs are open to any authenticated role
# ---------------------------------------------------------------------------


Expand All @@ -198,30 +198,35 @@ def test_an_admin_can_read_the_anchor(socketio, app, admin_token, anchor):


@pytest.mark.parametrize("command_hex", ["48", "49 00 62" + " 00" * 98, "4A"])
def test_a_non_admin_cannot_use_the_license_fcs(
def test_a_non_admin_can_use_the_license_fcs(
socketio, app, client, admin_token, anchor, command_hex
):
"""Role `user` must not read the anchor nor write a license.
"""Role `user` may run every license FC (decision 2026-08-25).

Reading it once is enough to derive that board's licensing identity and its
possession key offline, forever -- the anchor does not rotate, so revoking the
account afterwards takes nothing back.
The purchase is authorized by the Edge account on the /buy page, never by the
runtime role, so admin-gating these only stopped an operator from activating a
licence they had already paid for. Any authenticated role now reads the anchor
(0x48), writes (0x49) and reads back (0x4A) the blob. The old refusal
("Admin privileges required") must never come back.
"""
user_token = _user_token(client, admin_token)
ws = _connect(socketio, app, token=user_token)
assert ws.is_connected(_NAMESPACE) # a `user` may still debug variables
assert ws.is_connected(_NAMESPACE)

refused = _command(ws, command_hex)
assert refused["success"] is False
assert refused["error"] == "Admin privileges required"
assert "data" not in refused
assert _ANCHOR.hex() not in str(refused)
assert _ANCHOR.decode() not in str(refused)
response = _command(ws, command_hex)

assert response.get("error") != "Admin privileges required"
# 0x48 must actually hand the `user` the real anchor, same as the admin path.
if command_hex == "48":
assert response["success"] is True
parts = response["data"].split()
assert parts[:3] == ["48", "7E", "10"]
assert bytes(int(p, 16) for p in parts[3:]) == anchor


def test_a_non_admin_can_still_run_ordinary_debug_commands(socketio, app, client, admin_token):
"""The role gate is scoped to the license FCs -- it is not a general lockout,
which is what would make it get reverted."""
"""A `user` role debugs variables just like an admin -- there is no role
lockout anywhere on this channel."""
user_token = _user_token(client, admin_token)
ws = _connect(socketio, app, token=user_token)

Expand Down
42 changes: 13 additions & 29 deletions webserver/debug_websocket.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
"""

from flask import request
from flask_jwt_extended import current_user, verify_jwt_in_request
from flask_jwt_extended import verify_jwt_in_request
from jwt import ExpiredSignatureError
from flask_socketio import SocketIO, emit

Expand Down Expand Up @@ -58,22 +58,6 @@ def _reverify_session_token() -> bool:
return False


def _current_user_is_admin() -> bool:
"""True when the re-verified token belongs to an admin account.

Uses the role mechanism that already exists in the REST API (the User model's
``is_admin()``, resolved through the JWT user_lookup_loader) -- @jwt_required
alone does not look at the role, so a plain ``user`` account could read the
anchor of any board and overwrite its license. Must be called AFTER
_reverify_session_token(), which is what populates ``current_user``.
"""
user = current_user
if not user:
return False
checker = getattr(user, "is_admin", None)
return bool(checker and checker())


def init_debug_websocket(app, unix_client_instance):
"""
Initialize the WebSocket server for debug communication.
Expand Down Expand Up @@ -212,18 +196,18 @@ def handle_debug_command(data):
)
return

# The license FCs are a trust boundary of their own: 0x48 hands out
# the raw anchor (from which the licensing identity and the
# possession key are derived, offline and forever) and 0x49 writes
# the license blob. Require the admin role for them -- @jwt_required
# alone never looks at the role.
if is_license_command(command_hex) and not _current_user_is_admin():
logger.warning("License FC refused for a non-admin account: %s", command_hex)
emit(
"debug_response",
{"success": False, "error": "Admin privileges required"},
)
return
# The license FCs are open to any AUTHENTICATED role, not just admin
# (decision 2026-08-25). They were admin-gated on the theory that the
# anchor read (0x48) and the blob write (0x49) were a trust boundary,
# but that gate protected the wrong thing: the PURCHASE is authorized
# by the Edge account on the /buy page, never by the runtime role, so
# requiring admin here only stopped an operator from activating a
# licence they had already paid for. What stays open is low-risk: the
# anchor is the board's serial (baremetal exposes it with no auth at
# all), the blob is node-locked and useless on another device, and a
# bad write is recoverable (the entitlement lives in the backend; a
# refresh rewrites the correct blob). JWT re-verification above still
# applies, so "any role" means any logged-in user, never anonymous.

# License function codes (0x48/0x49/0x4A) operate on host files
# (/proc anchor + conf/<plugin>.license) and are resolved here in
Expand Down