800 lines
25 KiB
Python
800 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Awaitable, Callable
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from cloud.internal_api.models import AssignmentModel
|
|
from device.manager import DeviceManager
|
|
from host_agent.client import HostAgentAPIError, HostTaskSubmissionUnknownError
|
|
from host_agent.config import HostAgentConfig
|
|
from host_agent.history import ConsoleHistoryStore
|
|
from host_agent.identity import HostIdentityStore
|
|
from host_agent.local_account import LocalAccountStore
|
|
from host_agent.status import AgentStatusTracker
|
|
from host_agent.web.app import SESSION_COOKIE_NAME, create_console_app
|
|
from host_agent.web.auth import SessionManager
|
|
from storage.device_config import DeviceConfigStore
|
|
from storage.task_metadata import TaskMetadataStore
|
|
|
|
CSRF_PATTERN = re.compile(r'name="csrf_token" value="([^"]+)"')
|
|
|
|
TaskSubmissionCallable = Callable[[str, str | None], Awaitable[str]]
|
|
|
|
|
|
def _build_client(
|
|
tmp_path,
|
|
*,
|
|
create_account: bool = True,
|
|
submit_self_task: TaskSubmissionCallable | None = None,
|
|
include_metadata_store: bool = True,
|
|
) -> tuple[TestClient, dict]:
|
|
config = HostAgentConfig(
|
|
control_plane_url="https://control.example",
|
|
host_id="host-a",
|
|
token="secret",
|
|
enrollment_managed=False,
|
|
console_session_ttl_seconds=3600.0,
|
|
)
|
|
manager = DeviceManager()
|
|
config_store = DeviceConfigStore(tmp_path / "devices.sqlite3")
|
|
local_account_store = LocalAccountStore(tmp_path / "host_local_account.json")
|
|
if create_account:
|
|
local_account_store.create("operator", "correct horse battery staple")
|
|
identity_store = HostIdentityStore(tmp_path / "host_identity.json")
|
|
history_store = ConsoleHistoryStore(tmp_path / "history.sqlite3")
|
|
status_tracker = AgentStatusTracker()
|
|
session_manager = SessionManager(ttl_seconds=3600.0)
|
|
metadata_store: TaskMetadataStore | None = None
|
|
if include_metadata_store:
|
|
metadata_store = TaskMetadataStore(
|
|
db_path=tmp_path / "task_metadata.sqlite3"
|
|
)
|
|
|
|
app = create_console_app(
|
|
config=config,
|
|
manager=manager,
|
|
config_store=config_store,
|
|
local_account_store=local_account_store,
|
|
identity_store=identity_store,
|
|
history_store=history_store,
|
|
status_tracker=status_tracker,
|
|
session_manager=session_manager,
|
|
enrollment_client=None,
|
|
submit_self_task=submit_self_task,
|
|
metadata_store=metadata_store,
|
|
)
|
|
client = TestClient(app)
|
|
context = {
|
|
"manager": manager,
|
|
"config_store": config_store,
|
|
"local_account_store": local_account_store,
|
|
"history_store": history_store,
|
|
"session_manager": session_manager,
|
|
"status_tracker": status_tracker,
|
|
"metadata_store": metadata_store,
|
|
}
|
|
return client, context
|
|
|
|
|
|
def _login(
|
|
client: TestClient,
|
|
*,
|
|
username: str = "operator",
|
|
password: str = "correct horse battery staple",
|
|
) -> str:
|
|
response = client.post("/login", data={"username": username, "password": password})
|
|
assert response.status_code == 200
|
|
match = CSRF_PATTERN.search(response.text)
|
|
assert match is not None
|
|
return match.group(1)
|
|
|
|
|
|
def test_unauthenticated_get_root_redirects_to_login(tmp_path) -> None:
|
|
client, _ = _build_client(tmp_path)
|
|
|
|
response = client.get("/", follow_redirects=False)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login"
|
|
|
|
|
|
def test_login_with_no_account_shows_setup_message_and_rejects_post(tmp_path) -> None:
|
|
client, context = _build_client(tmp_path, create_account=False)
|
|
|
|
get_response = client.get("/login")
|
|
assert "device-host-agent setup" in get_response.text
|
|
|
|
post_response = client.post(
|
|
"/login", data={"username": "operator", "password": "anything"}
|
|
)
|
|
|
|
assert "device-host-agent setup" in post_response.text
|
|
assert SESSION_COOKIE_NAME not in client.cookies
|
|
|
|
|
|
def test_login_with_wrong_password_fails_and_sets_no_cookie(tmp_path) -> None:
|
|
client, _ = _build_client(tmp_path)
|
|
|
|
response = client.post("/login", data={"username": "operator", "password": "wrong"})
|
|
|
|
assert response.status_code == 200
|
|
assert "Invalid username or password" in response.text
|
|
assert SESSION_COOKIE_NAME not in client.cookies
|
|
|
|
|
|
def test_login_with_correct_password_sets_cookie_and_dashboard_succeeds(
|
|
tmp_path,
|
|
) -> None:
|
|
client, _ = _build_client(tmp_path)
|
|
|
|
response = client.post(
|
|
"/login",
|
|
data={"username": "operator", "password": "correct horse battery staple"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert SESSION_COOKIE_NAME in client.cookies
|
|
assert "Status" in response.text
|
|
assert "control.example" in response.text
|
|
|
|
|
|
def test_connected_device_without_running_task_shows_connected_not_busy(
|
|
tmp_path,
|
|
) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
_login(client)
|
|
context["manager"].register_device(
|
|
"device-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
status="busy",
|
|
)
|
|
|
|
dashboard_response = client.get("/")
|
|
api_response = client.get("/api/status")
|
|
|
|
assert "connected" in dashboard_response.text
|
|
assert "<td>busy</td>" not in dashboard_response.text
|
|
assert api_response.json()["devices"][0]["status"] == "connected"
|
|
|
|
|
|
def test_device_running_current_assignment_still_shows_busy(tmp_path) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
_login(client)
|
|
context["manager"].register_device(
|
|
"device-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
status="busy",
|
|
)
|
|
context["status_tracker"].mark_assignment_started(
|
|
AssignmentModel(
|
|
task_id="task-a",
|
|
attempt=1,
|
|
lease_id="lease-a",
|
|
lease_expires_at=datetime(2026, 7, 12, tzinfo=UTC),
|
|
host_id="host-a",
|
|
device_id="device-a",
|
|
goal="open settings",
|
|
)
|
|
)
|
|
|
|
dashboard_response = client.get("/")
|
|
api_response = client.get("/api/status")
|
|
|
|
assert "<td>busy</td>" in dashboard_response.text
|
|
assert api_response.json()["devices"][0]["status"] == "busy"
|
|
|
|
|
|
def test_mutating_post_without_csrf_token_is_rejected_and_makes_no_change(
|
|
tmp_path,
|
|
) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
_login(client)
|
|
|
|
response = client.post(
|
|
"/devices/save",
|
|
data={
|
|
"device_id": "device-a",
|
|
"driver_type": "wda",
|
|
"name": "Lab iPhone",
|
|
"connection_info": "{}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert context["config_store"].get("device-a") is None
|
|
assert context["manager"].list_devices() == []
|
|
|
|
|
|
def test_add_device_appears_in_devices_page_and_manager(tmp_path) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/devices/save",
|
|
data={
|
|
"device_id": "device-a",
|
|
"driver_type": "wda",
|
|
"name": "Lab iPhone",
|
|
"connection_info": "{}",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "device-a" in response.text
|
|
assert context["config_store"].get("device-a") is not None
|
|
assert [device.id for device in context["manager"].list_devices()] == ["device-a"]
|
|
|
|
|
|
def test_remove_device_unregisters_from_manager(tmp_path) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
csrf_token = _login(client)
|
|
client.post(
|
|
"/devices/save",
|
|
data={
|
|
"device_id": "device-a",
|
|
"driver_type": "wda",
|
|
"name": "Lab iPhone",
|
|
"connection_info": "{}",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
assert context["config_store"].get("device-a") is not None
|
|
|
|
response = client.post(
|
|
"/devices/remove",
|
|
data={"device_id": "device-a", "csrf_token": csrf_token},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert context["config_store"].get("device-a") is None
|
|
assert context["manager"].list_devices() == []
|
|
|
|
|
|
def test_change_password_wrong_current_fails_old_password_still_works(tmp_path) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/account",
|
|
data={
|
|
"current_password": "wrong",
|
|
"new_password": "new password",
|
|
"confirm_password": "new password",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "incorrect" in response.text
|
|
account_store: LocalAccountStore = context["local_account_store"]
|
|
account = account_store.load()
|
|
assert account is not None
|
|
assert account_store.verify(account, "correct horse battery staple") is True
|
|
|
|
|
|
def test_change_password_correct_succeeds_old_password_no_longer_works(
|
|
tmp_path,
|
|
) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/account",
|
|
data={
|
|
"current_password": "correct horse battery staple",
|
|
"new_password": "new password",
|
|
"confirm_password": "new password",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "Password updated" in response.text
|
|
account_store: LocalAccountStore = context["local_account_store"]
|
|
account = account_store.load()
|
|
assert account is not None
|
|
assert account_store.verify(account, "new password") is True
|
|
assert account_store.verify(account, "correct horse battery staple") is False
|
|
|
|
|
|
def test_history_reflects_entries_written_beforehand(tmp_path) -> None:
|
|
client, context = _build_client(tmp_path)
|
|
context["history_store"].record_heartbeat(device_count=3)
|
|
_login(client)
|
|
|
|
response = client.get("/history")
|
|
|
|
assert response.status_code == 200
|
|
assert "heartbeat: 3 devices" in response.text
|
|
|
|
|
|
def test_logout_invalidates_session_so_subsequent_request_redirects_to_login(
|
|
tmp_path,
|
|
) -> None:
|
|
client, _ = _build_client(tmp_path)
|
|
csrf_token = _login(client)
|
|
|
|
logout_response = client.post(
|
|
"/logout", data={"csrf_token": csrf_token}, follow_redirects=False
|
|
)
|
|
assert logout_response.status_code == 303
|
|
assert logout_response.headers["location"] == "/login"
|
|
|
|
response = client.get("/", follow_redirects=False)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login"
|
|
|
|
|
|
def _make_submission_recorder(
|
|
*,
|
|
task_id: str = "task-cloud-1",
|
|
raise_api_error: HostAgentAPIError | None = None,
|
|
raise_unknown: bool = False,
|
|
history_store: ConsoleHistoryStore | None = None,
|
|
history_fail: bool = False,
|
|
) -> tuple[TaskSubmissionCallable, dict]:
|
|
captured: dict = {}
|
|
|
|
async def submit(goal: str, device_id: str | None) -> str:
|
|
captured["goal"] = goal
|
|
captured["device_id"] = device_id
|
|
if raise_unknown:
|
|
raise HostTaskSubmissionUnknownError("transport failure")
|
|
if raise_api_error is not None:
|
|
raise raise_api_error
|
|
return task_id
|
|
|
|
async def record(*, task_id: str, device_id: str | None) -> None:
|
|
captured.setdefault("history_calls", []).append(
|
|
{"task_id": task_id, "device_id": device_id}
|
|
)
|
|
if history_fail:
|
|
raise RuntimeError("history store unavailable")
|
|
|
|
if history_store is not None:
|
|
original = history_store.record_task_submission
|
|
|
|
def _proxy(*, task_id: str, device_id: str | None) -> None:
|
|
try:
|
|
asyncio_run(record(task_id=task_id, device_id=device_id))
|
|
except RuntimeError:
|
|
pass
|
|
return original(task_id=task_id, device_id=device_id)
|
|
|
|
history_store.record_task_submission = _proxy # type: ignore[method-assign]
|
|
return submit, captured
|
|
|
|
|
|
def asyncio_run(coro):
|
|
import asyncio
|
|
|
|
return asyncio.get_event_loop().run_until_complete(coro) if asyncio.get_event_loop().is_running() else asyncio.run(coro)
|
|
|
|
|
|
def test_tasks_page_renders_submission_form_with_device_options(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder()
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
_login(client)
|
|
|
|
response = client.get("/tasks")
|
|
|
|
assert response.status_code == 200
|
|
assert "Submit task to current Host" in response.text
|
|
assert 'action="/tasks/submit"' in response.text
|
|
assert "Automatic (let Host choose" in response.text
|
|
assert "Lab iPhone" in response.text
|
|
|
|
|
|
def test_authenticated_automatic_submission_invokes_submit_with_none_device(
|
|
tmp_path,
|
|
) -> None:
|
|
submit, captured = _make_submission_recorder(task_id="task-cloud-auto")
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "__automatic__",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/tasks?submitted=1&task_id=task-cloud-auto"
|
|
assert captured == {"goal": "open settings", "device_id": None}
|
|
|
|
|
|
def test_authenticated_explicit_device_submission_uses_runtime_id(tmp_path) -> None:
|
|
submit, captured = _make_submission_recorder(task_id="task-cloud-explicit")
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-cloud-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Cloud iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open mail",
|
|
"device_id": "device-cloud-a",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert (
|
|
response.headers["location"]
|
|
== "/tasks?submitted=1&task_id=task-cloud-explicit"
|
|
)
|
|
assert captured == {"goal": "open mail", "device_id": "device-cloud-a"}
|
|
|
|
|
|
def test_redirected_tasks_page_renders_task_id_confirmation(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder()
|
|
client, _ = _build_client(tmp_path, submit_self_task=submit)
|
|
_login(client)
|
|
|
|
response = client.get("/tasks?submitted=1&task_id=task-cloud-1")
|
|
|
|
assert response.status_code == 200
|
|
assert "task-cloud-1" in response.text
|
|
assert "Cloud task ID" in response.text
|
|
|
|
|
|
def test_confirmed_submission_records_history_audit(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder(task_id="task-cloud-audit")
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-cloud-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Cloud iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "audit me",
|
|
"device_id": "device-cloud-a",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
entries = context["history_store"].list_recent()
|
|
assert entries[0]["kind"] == "task_submission"
|
|
assert entries[0]["detail"] == {
|
|
"task_id": "task-cloud-audit",
|
|
"device_id": "device-cloud-a",
|
|
}
|
|
|
|
|
|
def test_unauthenticated_submission_redirects_to_login_without_calling_client(
|
|
tmp_path,
|
|
) -> None:
|
|
submit, captured = _make_submission_recorder()
|
|
client, _ = _build_client(tmp_path, submit_self_task=submit)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={"goal": "open settings", "device_id": "__automatic__"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login"
|
|
assert captured == {}
|
|
|
|
|
|
def test_csrf_invalid_submission_is_rejected(tmp_path) -> None:
|
|
submit, captured = _make_submission_recorder()
|
|
client, _ = _build_client(tmp_path, submit_self_task=submit)
|
|
_login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "__automatic__",
|
|
"csrf_token": "wrong-token",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert captured == {}
|
|
|
|
|
|
def test_blank_goal_rejected_before_client_invocation(tmp_path) -> None:
|
|
submit, captured = _make_submission_recorder()
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": " ",
|
|
"device_id": "__automatic__",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "Goal cannot be empty" in response.text
|
|
assert captured == {}
|
|
|
|
|
|
def test_stale_device_selection_rejected_before_client_invocation(tmp_path) -> None:
|
|
submit, captured = _make_submission_recorder()
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value],
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "device-that-was-removed",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "no longer registered" in response.text
|
|
assert captured == {}
|
|
|
|
|
|
def test_cloud_definitive_rejection_renders_safe_error_without_calling_history(
|
|
tmp_path,
|
|
) -> None:
|
|
submit, captured = _make_submission_recorder(
|
|
raise_api_error=HostAgentAPIError(
|
|
403, "Host self-submission is disabled"
|
|
),
|
|
)
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "__automatic__",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 502
|
|
assert "Host self-submission is disabled" in response.text
|
|
entries = context["history_store"].list_recent()
|
|
assert all(entry["kind"] != "task_submission" for entry in entries)
|
|
assert captured == {"goal": "open settings", "device_id": None}
|
|
|
|
|
|
def test_transport_uncertain_response_redirects_to_unknown_outcome(
|
|
tmp_path,
|
|
) -> None:
|
|
submit, captured = _make_submission_recorder(raise_unknown=True)
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "__automatic__",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/tasks?outcome=unknown"
|
|
entries = context["history_store"].list_recent()
|
|
assert all(entry["kind"] != "task_submission" for entry in entries)
|
|
assert captured == {"goal": "open settings", "device_id": None}
|
|
|
|
|
|
def test_unknown_outcome_query_shows_safe_message(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder()
|
|
client, _ = _build_client(tmp_path, submit_self_task=submit)
|
|
_login(client)
|
|
|
|
response = client.get("/tasks?outcome=unknown")
|
|
|
|
assert response.status_code == 200
|
|
assert "Submission outcome is unknown" in response.text
|
|
assert "Check the Cloud console" in response.text
|
|
|
|
|
|
def test_submission_form_absent_when_client_missing(tmp_path) -> None:
|
|
client, _ = _build_client(tmp_path, submit_self_task=None)
|
|
_login(client)
|
|
|
|
response = client.get("/tasks")
|
|
|
|
assert response.status_code == 200
|
|
assert 'action="/tasks/submit"' not in response.text
|
|
assert "submission client is not available yet" in response.text
|
|
|
|
|
|
def test_submit_when_client_missing_returns_503_without_history(tmp_path) -> None:
|
|
client, context = _build_client(tmp_path, submit_self_task=None)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "__automatic__",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
entries = context["history_store"].list_recent()
|
|
assert all(entry["kind"] != "task_submission" for entry in entries)
|
|
|
|
|
|
def test_audit_failure_does_not_break_successful_submission(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder(
|
|
task_id="task-cloud-audit-fail",
|
|
history_store=None,
|
|
)
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
|
|
def boom(*, task_id: str, device_id: str | None) -> None:
|
|
raise RuntimeError("audit DB offline")
|
|
|
|
context["history_store"].record_task_submission = boom # type: ignore[method-assign]
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "__automatic__",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert (
|
|
response.headers["location"]
|
|
== "/tasks?submitted=1&task_id=task-cloud-audit-fail"
|
|
)
|
|
|
|
|
|
def test_failed_submission_never_writes_successful_audit(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder(
|
|
raise_api_error=HostAgentAPIError(422, "ownership mismatch"),
|
|
)
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "open settings",
|
|
"device_id": "device-runtime-a",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
entries = context["history_store"].list_recent()
|
|
assert all(entry["kind"] != "task_submission" for entry in entries)
|
|
|
|
|
|
def test_tasks_page_autoescapes_goal_device_and_error_text(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder(
|
|
raise_api_error=HostAgentAPIError(403, "<img src=x onerror=alert(1)>"),
|
|
)
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"<script>alert('x')</script>",
|
|
lambda: object(), # type: ignore[arg-type,return-value]
|
|
name='"><img src=x onerror=alert(1)>',
|
|
)
|
|
csrf_token = _login(client)
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": "<script>alert('x')</script>",
|
|
"device_id": "<script>alert('x')</script>",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 502
|
|
html = response.text
|
|
assert "<script>alert('x')</script>" in html
|
|
assert "<img src=x onerror=alert(1)>" not in html
|
|
assert "<script>alert('x')</script>" not in html
|
|
assert "<script>alert" in html
|
|
assert "alert(1)" in html # auto-escaped as text, not executable
|
|
|
|
|
|
def test_submitted_redirect_does_not_include_goal_text(tmp_path) -> None:
|
|
submit, _ = _make_submission_recorder(task_id="task-cloud-clean")
|
|
client, context = _build_client(tmp_path, submit_self_task=submit)
|
|
context["manager"].register_device(
|
|
"device-runtime-a",
|
|
lambda: object(), # type: ignore[arg-type,return-value],
|
|
name="Lab iPhone",
|
|
)
|
|
csrf_token = _login(client)
|
|
secret_goal = "rotate bearer-token-deadbeef-very-secret-12345"
|
|
|
|
response = client.post(
|
|
"/tasks/submit",
|
|
data={
|
|
"goal": secret_goal,
|
|
"device_id": "__automatic__",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert secret_goal not in response.headers["location"]
|
|
assert "bearer-token-deadbeef" not in response.headers["location"]
|
|
follow = client.get(response.headers["location"])
|
|
assert secret_goal not in follow.text
|
|
assert "bearer-token-deadbeef" not in follow.text
|