bvandeusenandClaude Opus 5.5
6a1bc437c9
#5033 "Share links open only their own dashboard; every route without a role is on an explicit allow-list". Any live share token in ?s= made the caller a viewer on every viewer route — host pages, container logs, any dashboard — and the host-agent widget routes had no role check at all. - steward/dashboard/share.py: a share token opens, read-only, exactly the widget fragment URLs its own dashboard loads — same path, same configured parameters; only the time window (hours/range) may change. - require_role marks its view; public_route marks the deliberately open ones (login, setup, health, the share page, agent ingest, install.sh, agent.py) with the reason. A before_request refuses any view with neither, so a route that forgets its decorator is closed, not open. - /plugins/host_agent/widget and /widget/history now require viewer. - The history widget's range buttons keep the rest of the query (and so a share link's token) instead of rebuilding it; the uptime widget's "Full SLA" link shows to logged-in users, not to share viewers. - tests/test_route_access.py: a source scan that fails on any unmarked route in core or bundled plugins, the default-deny refusal, and the share matcher's cases. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
"""Who can reach a route: every view carries require_role or public_route,
|
|
an unmarked one is refused, and a share link opens only its own widgets."""
|
|
import ast
|
|
import pathlib
|
|
|
|
import pytest
|
|
from werkzeug.datastructures import MultiDict
|
|
|
|
import steward
|
|
from steward.dashboard.share import share_allows
|
|
|
|
_REPO = pathlib.Path(steward.__file__).parent.parent
|
|
_ROUTE_METHODS = ("get", "post", "put", "patch", "delete", "route", "websocket")
|
|
|
|
|
|
def _route_decorated_views():
|
|
files = [p for root in ("steward", "plugins") for p in (_REPO / root).rglob("*.py")
|
|
if "migrations" not in p.parts]
|
|
for path in files:
|
|
tree = ast.parse(path.read_text())
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
continue
|
|
decorators = [ast.unparse(d) for d in node.decorator_list]
|
|
if any(f".{m}('/" in d or f'.{m}("/' in d for d in decorators for m in _ROUTE_METHODS):
|
|
yield path, node, decorators
|
|
|
|
|
|
def test_every_route_says_who_may_reach_it():
|
|
unmarked = [f"{p.relative_to(_REPO)}:{n.lineno} {n.name}"
|
|
for p, n, decs in _route_decorated_views()
|
|
if not any(d.startswith(("require_role(", "public_route(")) for d in decs)]
|
|
assert not unmarked, "routes with neither require_role nor public_route: " + ", ".join(unmarked)
|
|
|
|
|
|
def test_the_scan_sees_routes():
|
|
names = {n.name for _, n, _ in _route_decorated_views()}
|
|
assert {"login", "health", "ingest", "widget_table", "public_share"} <= names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unmarked_view_is_refused(app, client):
|
|
async def unmarked():
|
|
return "open"
|
|
app.add_url_rule("/__unmarked", "unmarked", unmarked)
|
|
assert (await client.get("/__unmarked")).status_code == 403
|
|
assert (await client.get("/health")).status_code == 200
|
|
|
|
|
|
WIDGETS = ["/plugins/host_agent/widget/history?host_id=h1&hours=6&wid=7",
|
|
"/monitors/widget?wid=8"]
|
|
|
|
|
|
@pytest.mark.parametrize("path,args,allowed", [
|
|
("/plugins/host_agent/widget/history", {"host_id": "h1", "hours": "6", "wid": "7", "s": "t"}, True),
|
|
("/plugins/host_agent/widget/history", {"host_id": "h1", "hours": "24", "wid": "7", "s": "t"}, True),
|
|
("/monitors/widget", {"wid": "8", "s": "t"}, True),
|
|
# the same widget aimed at another host
|
|
("/plugins/host_agent/widget/history", {"host_id": "h2", "hours": "6", "wid": "7", "s": "t"}, False),
|
|
# a parameter the widget was not configured with
|
|
("/monitors/widget", {"wid": "8", "host_id": "h1", "s": "t"}, False),
|
|
# a configured parameter left off
|
|
("/plugins/host_agent/widget/history", {"hours": "6", "wid": "7", "s": "t"}, False),
|
|
# any other page
|
|
("/hosts/", {"s": "t"}, False),
|
|
("/plugins/docker/logs", {"wid": "8", "s": "t"}, False),
|
|
])
|
|
def test_share_opens_only_its_widgets(path, args, allowed):
|
|
assert share_allows(WIDGETS, path, MultiDict(args)) is allowed
|
|
|
|
|
|
def test_share_refuses_a_repeated_parameter():
|
|
args = MultiDict([("wid", "8"), ("wid", "9"), ("s", "t")])
|
|
assert not share_allows(WIDGETS, "/monitors/widget", args)
|