260 lines
8.5 KiB
Python
260 lines
8.5 KiB
Python
"""Render-smoke, XSS-probe, byte-identity, and autoescape tests for the
|
|
Host Agent console Jinja2 templates (Tasks 6.3-6.7)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from host_agent.config import HostAgentConfig
|
|
from host_agent.web.auth import SessionState
|
|
|
|
from .conftest import (
|
|
XSS_PROBE,
|
|
_ESCAPED_PROBE,
|
|
make_account_context,
|
|
make_dashboard_context,
|
|
make_devices_context,
|
|
make_history_context,
|
|
make_login_context,
|
|
make_task_detail_context,
|
|
make_tasks_list_context,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6.3 Render-smoke tests (one per template)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_login_renders(env) -> None:
|
|
html = env.get_template("login.html").render(
|
|
**make_login_context(account={"username": "operator"})
|
|
)
|
|
assert '<form method="post" action="/login">' in html
|
|
|
|
|
|
def test_dashboard_renders(env, sample_session) -> None:
|
|
html = env.get_template("dashboard.html").render(
|
|
**make_dashboard_context(sample_session)
|
|
)
|
|
assert 'id="last-heartbeat"' in html
|
|
|
|
|
|
def test_devices_renders(env, sample_session) -> None:
|
|
html = env.get_template("devices.html").render(
|
|
**make_devices_context(sample_session)
|
|
)
|
|
assert '<form method="post" action="/devices/save">' in html
|
|
|
|
|
|
def test_account_renders(env, sample_session) -> None:
|
|
html = env.get_template("account.html").render(
|
|
**make_account_context(sample_session)
|
|
)
|
|
assert '<form method="post" action="/account">' in html
|
|
|
|
|
|
def test_history_renders(env, sample_session) -> None:
|
|
html = env.get_template("history.html").render(
|
|
**make_history_context(sample_session)
|
|
)
|
|
assert "<table>" in html
|
|
|
|
|
|
def test_tasks_list_renders(env, sample_session) -> None:
|
|
html = env.get_template("tasks_list.html").render(
|
|
**make_tasks_list_context(sample_session)
|
|
)
|
|
assert "<h1>Tasks</h1>" in html
|
|
|
|
|
|
def test_task_detail_renders(env, sample_session) -> None:
|
|
html = env.get_template("task_detail.html").render(
|
|
**make_task_detail_context(sample_session)
|
|
)
|
|
assert "<h2>Timeline</h2>" in html
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6.4 XSS-probe tests (parametrised over templates with operator-influenced
|
|
# string fields set to <script>alert(1)</script>)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_TEMPLATES_WITH_XSS = ["dashboard", "devices", "history", "tasks_list", "task_detail"]
|
|
|
|
|
|
def _xss_context(template: str, session: SessionState) -> dict[str, Any]:
|
|
"""Build a context where every operator-influenced string field is the
|
|
XSS probe string."""
|
|
if template == "dashboard":
|
|
return make_dashboard_context(
|
|
session,
|
|
devices=[
|
|
{
|
|
"id": XSS_PROBE,
|
|
"name": XSS_PROBE,
|
|
"driver_type": XSS_PROBE,
|
|
"display_status": XSS_PROBE,
|
|
}
|
|
],
|
|
heartbeat_text=XSS_PROBE,
|
|
assignment_text=XSS_PROBE,
|
|
progress_text=XSS_PROBE,
|
|
policy_text=XSS_PROBE,
|
|
)
|
|
if template == "devices":
|
|
return make_devices_context(
|
|
session,
|
|
devices=[
|
|
{
|
|
"device_id": XSS_PROBE,
|
|
"name": XSS_PROBE,
|
|
"driver_type": XSS_PROBE,
|
|
"cloud_device_id": XSS_PROBE,
|
|
"connection_info": {},
|
|
}
|
|
],
|
|
edit_record={
|
|
"device_id": XSS_PROBE,
|
|
"name": XSS_PROBE,
|
|
"driver_type": XSS_PROBE,
|
|
"connection_info": {},
|
|
},
|
|
connection_info_json=XSS_PROBE,
|
|
error=XSS_PROBE,
|
|
)
|
|
if template == "history":
|
|
return make_history_context(
|
|
session,
|
|
entries=[
|
|
{
|
|
"occurred_at": XSS_PROBE,
|
|
"kind": XSS_PROBE,
|
|
"summary": XSS_PROBE,
|
|
}
|
|
],
|
|
)
|
|
if template == "tasks_list":
|
|
return make_tasks_list_context(
|
|
session,
|
|
tasks=[
|
|
{
|
|
"id": XSS_PROBE,
|
|
"status": XSS_PROBE,
|
|
"device_id": XSS_PROBE,
|
|
"created_at": XSS_PROBE,
|
|
"updated_at": XSS_PROBE,
|
|
}
|
|
],
|
|
)
|
|
if template == "task_detail":
|
|
return make_task_detail_context(
|
|
session,
|
|
task={
|
|
"id": XSS_PROBE,
|
|
"goal": XSS_PROBE,
|
|
"device_id": XSS_PROBE,
|
|
"status": XSS_PROBE,
|
|
"created_at": XSS_PROBE,
|
|
"updated_at": XSS_PROBE,
|
|
},
|
|
task_rows=[
|
|
("id", XSS_PROBE),
|
|
("goal", XSS_PROBE),
|
|
("status", XSS_PROBE),
|
|
],
|
|
timeline_steps=[
|
|
{
|
|
"index": 0,
|
|
"timestamp": XSS_PROBE,
|
|
"prompt": XSS_PROBE,
|
|
"tool_call": {"action": XSS_PROBE, "description": XSS_PROBE},
|
|
"result": {"ok": XSS_PROBE},
|
|
"before_screenshot_src": None,
|
|
"after_screenshot_src": None,
|
|
"ocr_results": [],
|
|
"ui_tree_nodes": [],
|
|
}
|
|
],
|
|
)
|
|
raise ValueError(f"unknown template {template}")
|
|
|
|
|
|
@pytest.mark.parametrize("template", _TEMPLATES_WITH_XSS)
|
|
def test_xss_probe_is_escaped(env, sample_session, template: str) -> None:
|
|
html = env.get_template(f"{template}.html").render(
|
|
**_xss_context(template, sample_session)
|
|
)
|
|
assert _ESCAPED_PROBE in html, f"escaped probe missing from {template}"
|
|
assert XSS_PROBE not in html, f"raw <script> leaked in {template}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6.5 Byte-identity test for dashboard inline <script>
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_dashboard_script_byte_identity(env, sample_session) -> None:
|
|
config = HostAgentConfig(control_plane_url="http://localhost:8080")
|
|
html = env.get_template("dashboard.html").render(
|
|
title="Status",
|
|
session=sample_session,
|
|
identity=None,
|
|
devices=[],
|
|
config=config,
|
|
heartbeat_text="never",
|
|
assignment_text="none",
|
|
progress_text="",
|
|
policy_text="no Cloud policy cached",
|
|
)
|
|
start = html.index("<script>")
|
|
end = html.index("</script>", start) + len("</script>")
|
|
script_block = html[start:end]
|
|
baseline = (
|
|
(Path(__file__).parent / "__baseline__" / "dashboard_script.txt")
|
|
.read_text(encoding="utf-8")
|
|
.rstrip()
|
|
)
|
|
assert script_block.rstrip() == baseline
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6.6 No-autoescape-bypass test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_no_autoescape_bypass_in_templates() -> None:
|
|
templates_dir = (
|
|
Path(__file__).resolve().parents[3] / "host_agent" / "web" / "templates"
|
|
)
|
|
forbidden = ["| safe", "{% autoescape false %}", "{% endautoescape %}"]
|
|
for tpl_path in templates_dir.glob("*.html"):
|
|
content = tpl_path.read_text(encoding="utf-8")
|
|
for pattern in forbidden:
|
|
assert pattern not in content, f"{pattern} found in {tpl_path.name}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6.7 _ENV autoescape configuration test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_env_autoescape_enabled_for_html(env) -> None:
|
|
"""The Environment's autoescape policy must escape content rendered
|
|
through .html templates."""
|
|
# Verify the autoescape policy is active (truthy or callable).
|
|
assert env.autoescape is True or callable(env.autoescape)
|
|
|
|
# Render the XSS probe through a .html template that interpolates it
|
|
# (login.html interpolates {{ error }}) and confirm it is escaped.
|
|
result = env.get_template("login.html").render(
|
|
title="Login",
|
|
session=None,
|
|
account={"username": "operator"},
|
|
error="<script>alert(1)</script>",
|
|
)
|
|
assert "<script>alert(1)</script>" in result
|
|
assert "<script>alert(1)</script>" not in result
|