bvandeusenandClaude Opus 5.5
5492336f1e
- New installs start from widgets.RECOMMENDED, packed row by row for the enabled plugins; the edit drawer can reset a dashboard to it. - Host Agent — Resources folds into Hosts — Overview (which gains load 1m), UniFi — Devices into UniFi — Overview; migration 0032 moves existing dashboards onto the survivors. - The Overview labels what it counts (hosts, monitors up/down/pending, alerts firing) and lists firing alerts and down monitors with since-times. - Uptime polls every 5 minutes (poll_seconds) instead of every cycle. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
92 lines
3.6 KiB
Python
92 lines
3.6 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_views(source: str):
|
|
"""(function node, decorator texts) for each route-decorated function."""
|
|
for node in ast.walk(ast.parse(source)):
|
|
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 for d in decorators for m in _ROUTE_METHODS):
|
|
yield node, decorators
|
|
|
|
|
|
def unmarked_routes(source: str) -> list[str]:
|
|
return [n.name for n, decs in route_views(source)
|
|
if not any(d.startswith(("require_role(", "public_route(")) for d in decs)]
|
|
|
|
|
|
def _sources():
|
|
for root in ("steward", "plugins"):
|
|
for path in (_REPO / root).rglob("*.py"):
|
|
if "migrations" not in path.parts:
|
|
yield path, path.read_text()
|
|
|
|
|
|
def test_the_route_guard_can_fail():
|
|
source = (
|
|
"@bp.get('/open')\nasync def open_view(): ...\n"
|
|
"@bp.post('/closed')\n@require_role(UserRole.viewer)\nasync def closed(): ...\n"
|
|
"@bp.get('/login')\n@public_route('login')\nasync def login(): ...\n"
|
|
"def helper(): ...\n")
|
|
assert unmarked_routes(source) == ["open_view"]
|
|
|
|
|
|
def test_every_route_says_who_may_reach_it():
|
|
unmarked = [f"{p.relative_to(_REPO)} {name}" for p, text in _sources()
|
|
for name in unmarked_routes(text)]
|
|
assert not unmarked, "routes with neither require_role nor public_route: " + ", ".join(unmarked)
|
|
|
|
|
|
def test_the_scan_sees_routes():
|
|
names = {n.name for _, text in _sources() for n, _ in route_views(text)}
|
|
assert {"login", "health", "ingest", "widget_history", "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)
|