q792602257/agentic-mobile-control
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(host-agent): migrate local console to Jinja2 templates with autoescape
Tests / Test failed: 4, passed: 744
Details

Browse Source 
Replace hand-written f-string + html.escape() rendering in the Host Agent
local console with a module-level Jinja2 Environment configured with
select_autoescape(["html","xml"]). XSS safety now holds by mechanism
rather than per-call discipline — every operator-controlled field
(device name, connection_info, task summary, etc.) is escaped by the
engine uniformly.

Eight templates under host_agent/web/templates/ replace the former
_chrome(), _CSS, escape(), and per-page _xxx_body() helpers: base.html
(header/nav/CSS + {% block body %}), login, dashboard (with the polling
<script> preserved byte-identically inside {% raw %}), devices, account,
history, tasks_list, and task_detail. The task-list and task-detail
templates — added by the just-landed task-execution-progress-visibility
change — were also migrated here rather than left in f-string form,
since this change removes the shared helpers they depended on.

URLs, auth/session/CSRF semantics, redirects, and /api/status JSON are
unchanged. 15 new template tests cover render-smoke, XSS probing, script
byte-identity, and no-autoescape-bypass guards. Tasks 8.1-8.6 (manual
browser verification) remain.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
q792602257
2026-07-14 13:05:19 +08:00
co-authored by Claude Opus 4.6
parent ec261d57c2
commit 8381b3068a
20 changed files with 1286 additions and 394 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/device-host-agent/tests/host_agent/web/__baseline__/dashboard_script.txt
+43
View File
@@ -0,0 +1,43 @@
<script>
(function () {
function render(data) {
var hb = data.status.last_heartbeat;
document.getElementById("last-heartbeat").textContent = hb
? (hb.ok ? "ok" : "failed") + " at " + hb.at + " (" + hb.device_count + " devices)"
: "never";
var current = data.status.current_assignment;
document.getElementById("current-assignment").textContent = current
? current.task_id + " on " + current.device_id + " (started " + current.started_at + ")"
: "none";
var progress = data.status.progress;
document.getElementById("current-progress").textContent = progress
? "step " + progress.step_index + " \u2014 " + progress.step_status + ": " + progress.summary
: "";
var policy = data.status.host_policy;
document.getElementById("host-policy").textContent = policy
? "revision " + policy.revision + "; self-submission "
+ (policy.self_submission_enabled ? "enabled" : "disabled")
+ "; max active tasks " + (policy.max_active_tasks || "unlimited")
+ "; daily token budget " + (policy.daily_token_budget || "unmetered")
: "no Cloud policy cached";
var body = document.getElementById("device-status-body");
body.innerHTML = "";
data.devices.forEach(function (device) {
var row = document.createElement("tr");
["id", "name", "driver_type", "status"].forEach(function (key) {
var cell = document.createElement("td");
cell.textContent = device[key] || "";
row.appendChild(cell);
});
body.appendChild(row);
});
}
function poll() {
fetch("/api/status", { credentials: "same-origin" })
.then(function (response) { return response.ok ? response.json() : null; })
.then(function (data) { if (data) render(data); })
.catch(function () {});
}
setInterval(poll, 5000);
})();
</script>
apps/device-host-agent/tests/host_agent/web/__init__.py
View File
apps/device-host-agent/tests/host_agent/web/conftest.py
+230
View File
@@ -0,0 +1,230 @@
"""Shared fixtures and context factories for Jinja2 template tests."""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
import pytest
from host_agent.config import HostAgentConfig
from host_agent.web.app import _ENV
from host_agent.web.auth import SessionState
XSS_PROBE = "<script>alert(1)</script>"
_ESCAPED_PROBE = "&lt;script&gt;alert(1)&lt;/script&gt;"
@pytest.fixture
def env() -> Any:
"""The module-level Jinja2 Environment from host_agent.web.app."""
return _ENV
@pytest.fixture
def sample_session() -> SessionState:
"""A representative logged-in session."""
return SessionState(
username="operator",
csrf_token="test-csrf-token",
expires_at=datetime(2030, 1, 1, tzinfo=UTC),
)
@pytest.fixture
def xss_probe() -> str:
return XSS_PROBE
# ---------------------------------------------------------------------------
# Context factories
# ---------------------------------------------------------------------------
def make_login_context(
*, account: Any = None, error: str | None = None
) -> dict[str, Any]:
return {
"title": "Login",
"session": None,
"account": account,
"error": error,
}
def make_dashboard_context(
session: SessionState,
*,
devices: list[dict[str, Any]] | None = None,
identity: Any = None,
heartbeat_text: str = "never",
assignment_text: str = "none",
progress_text: str = "",
policy_text: str = "no Cloud policy cached",
config: HostAgentConfig | None = None,
) -> dict[str, Any]:
if devices is None:
devices = [
{
"id": "dev-1",
"name": "Pixel 8",
"driver_type": "wda",
"display_status": "connected",
},
{
"id": "dev-2",
"name": "iPhone 15",
"driver_type": "wda",
"display_status": "busy",
},
]
if config is None:
config = HostAgentConfig(control_plane_url="http://localhost:8080")
return {
"title": "Status",
"session": session,
"identity": identity,
"devices": devices,
"config": config,
"heartbeat_text": heartbeat_text,
"assignment_text": assignment_text,
"progress_text": progress_text,
"policy_text": policy_text,
}
def make_devices_context(
session: SessionState,
*,
devices: list[dict[str, Any]] | None = None,
edit_record: dict[str, Any] | None = None,
connection_info_json: str = "{}",
error: str | None = None,
) -> dict[str, Any]:
if devices is None:
devices = [
{
"device_id": "dev-1",
"name": "Pixel 8",
"driver_type": "wda",
"cloud_device_id": "cloud-1",
"connection_info": {"port": 8100},
},
]
if edit_record is None:
edit_record = {
"device_id": "dev-1",
"name": "Pixel 8",
"driver_type": "wda",
"connection_info": {"port": 8100},
}
connection_info_json = '{"port": 8100}'
return {
"title": "Devices",
"session": session,
"devices": devices,
"csrf_token": session.csrf_token,
"edit_record": edit_record,
"connection_info_json": connection_info_json,
"error": error,
}
def make_account_context(
session: SessionState,
*,
message: str | None = None,
error: str | None = None,
) -> dict[str, Any]:
return {
"title": "Account",
"session": session,
"csrf_token": session.csrf_token,
"message": message,
"error": error,
}
def make_history_context(
session: SessionState,
*,
entries: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
if entries is None:
entries = [
{
"occurred_at": "2026-01-01T00:00:00Z",
"kind": "assignment",
"summary": "Task abc-123 started on dev-1",
},
]
return {
"title": "History",
"session": session,
"entries": entries,
}
def make_tasks_list_context(
session: SessionState,
*,
tasks: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
if tasks is None:
tasks = [
{
"id": "task-001",
"status": "completed",
"device_id": "dev-1",
"created_at": "2026-01-01T00:00:00Z",
"updated_at": "2026-01-01T00:05:00Z",
},
]
return {
"title": "Tasks",
"session": session,
"tasks": tasks,
}
def make_task_detail_context(
session: SessionState,
*,
task: dict[str, Any] | None = None,
task_rows: list[tuple[str, Any]] | None = None,
timeline_steps: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
if task is None:
task = {
"id": "task-001",
"goal": "Open settings",
"device_id": "dev-1",
"status": "completed",
"created_at": "2026-01-01T00:00:00Z",
"updated_at": "2026-01-01T00:05:00Z",
}
if task_rows is None:
task_rows = [
("id", task["id"]),
("goal", task["goal"]),
("device_id", task["device_id"]),
("status", task["status"]),
]
if timeline_steps is None:
timeline_steps = [
{
"index": 0,
"timestamp": "2026-01-01T00:01:00Z",
"prompt": "Tap the Settings icon",
"tool_call_text": '{"action": "tap", "x": 100, "y": 200}',
"result_text": '{"ok": true}',
"screenshot_src": None,
},
]
return {
"title": "Task task-001",
"session": session,
"task": task,
"task_rows": task_rows,
"timeline_steps": timeline_steps,
}
apps/device-host-agent/tests/host_agent/web/test_templates.py
+256
View File
@@ -0,0 +1,256 @@
"""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_text": XSS_PROBE,
"result_text": XSS_PROBE,
"screenshot_src": None,
}
],
)
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 "&lt;script&gt;alert(1)&lt;/script&gt;" in result
assert "<script>alert(1)</script>" not in result