538 lines
17 KiB
Python
538 lines
17 KiB
Python
"""Software-only tests for the RF-001 HTTP control channel."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
import tools.rf001_http_control as control_module
|
|
from tools.rf001_http_control import (
|
|
AuthenticationError,
|
|
ControlHTTPServer,
|
|
ControlState,
|
|
PROTOCOL_VERSION,
|
|
ProtocolError,
|
|
RequestAuthenticator,
|
|
canonical_json,
|
|
request_json,
|
|
signed_headers,
|
|
)
|
|
|
|
|
|
TOKEN = b"software-only-test-token-0123456789abcdef0123456789abcdef"
|
|
|
|
|
|
def test_bootstrap_registration_uses_distinct_fixed_identifier(
|
|
tmp_path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
releases = tmp_path / "releases"
|
|
releases.mkdir()
|
|
package = releases / "rf001-bootstrap-2026.09.29.1.zip"
|
|
package.write_bytes(b"fixed bootstrap package")
|
|
monkeypatch.setattr(control_module, "DEFAULT_RELEASES_ROOT", releases)
|
|
state = ControlState(tmp_path / "state.json")
|
|
registered = state.register_bootstrap_release(
|
|
str(package),
|
|
release_id="rf001-bootstrap-2026.09.29.1",
|
|
agent_version="2026.09.29.1",
|
|
content_manifest_sha256="a" * 64,
|
|
token=TOKEN,
|
|
)
|
|
assert registered["package_kind"] == "BOOTSTRAP"
|
|
assert registered["rf_tx"] == 0
|
|
replacement = releases / "replacement.zip"
|
|
replacement.write_bytes(b"different bootstrap package")
|
|
with pytest.raises(ProtocolError, match="already registered differently"):
|
|
state.register_bootstrap_release(
|
|
str(replacement),
|
|
release_id="rf001-bootstrap-2026.09.29.1",
|
|
agent_version="2026.09.29.1",
|
|
content_manifest_sha256="b" * 64,
|
|
token=TOKEN,
|
|
)
|
|
with pytest.raises(ProtocolError, match="release_id"):
|
|
state.register_bootstrap_release(
|
|
str(package),
|
|
release_id="rf001-agent-2026.09.29.1",
|
|
agent_version="2026.09.29.1",
|
|
content_manifest_sha256="a" * 64,
|
|
token=TOKEN,
|
|
)
|
|
|
|
|
|
def test_request_authentication_rejects_replay_and_tampering() -> None:
|
|
authenticator = RequestAuthenticator(TOKEN, window_seconds=60)
|
|
body = canonical_json({"agent_id": "rx-user-pc", "rf_tx": 0})
|
|
target = "/api/v1/agent/check-in"
|
|
headers = signed_headers(
|
|
TOKEN,
|
|
"POST",
|
|
target,
|
|
body,
|
|
timestamp=1_000,
|
|
nonce="a" * 32,
|
|
)
|
|
|
|
authenticator.verify("POST", target, body, headers, now=1_000)
|
|
with pytest.raises(AuthenticationError, match="already used"):
|
|
authenticator.verify("POST", target, body, headers, now=1_000)
|
|
|
|
changed_headers = signed_headers(
|
|
TOKEN,
|
|
"POST",
|
|
target,
|
|
body,
|
|
timestamp=1_000,
|
|
nonce="b" * 32,
|
|
)
|
|
with pytest.raises(AuthenticationError, match="mismatch"):
|
|
authenticator.verify(
|
|
"POST",
|
|
target,
|
|
body + b" ",
|
|
changed_headers,
|
|
now=1_000,
|
|
)
|
|
|
|
|
|
def test_state_leases_inventory_and_records_idempotent_result(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
task = state.enqueue_inventory("rx-user-pc", include_hackrf_info=True)
|
|
response = state.check_in(
|
|
{
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"agent_id": "rx-user-pc",
|
|
"inventory": {"hostname": "User-PC", "rf_tx": 0},
|
|
"rf_tx": 0,
|
|
}
|
|
)
|
|
assert response["task"] == task
|
|
assert response["rf_tx"] == 0
|
|
|
|
result = {
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"task_id": task["task_id"],
|
|
"agent_id": "rx-user-pc",
|
|
"status": "COMPLETE",
|
|
"started_at": "2026-09-23T00:00:00+00:00",
|
|
"finished_at": "2026-09-23T00:00:01+00:00",
|
|
"result": {"hostname": "User-PC", "rf_tx": 0},
|
|
"errors": [],
|
|
"rf_tx": 0,
|
|
}
|
|
accepted = state.record_result(result)
|
|
assert accepted == {"accepted": True, "duplicate": False, "rf_tx": 0}
|
|
duplicate = state.record_result(result)
|
|
assert duplicate == {"accepted": True, "duplicate": True, "rf_tx": 0}
|
|
|
|
conflicting = dict(result)
|
|
conflicting["status"] = "FAILED"
|
|
with pytest.raises(ProtocolError, match="conflicting"):
|
|
state.record_result(conflicting)
|
|
|
|
|
|
def test_state_leases_fixed_offline_verify_task(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
expected_head = "a" * 40
|
|
task = state.enqueue_offline_verify(
|
|
"rx-user-pc",
|
|
expected_head=expected_head,
|
|
)
|
|
response = state.check_in(
|
|
{
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"agent_id": "rx-user-pc",
|
|
"inventory": {"hostname": "User-PC", "rf_tx": 0},
|
|
"rf_tx": 0,
|
|
}
|
|
)
|
|
assert response["task"] == task
|
|
assert task["kind"] == "OFFLINE_VERIFY"
|
|
assert task["expected_head"] == expected_head
|
|
assert task["rf_tx"] == 0
|
|
|
|
|
|
def test_offline_verify_rejects_invalid_expected_head(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
with pytest.raises(ProtocolError, match="expected_head"):
|
|
state.enqueue_offline_verify("rx-user-pc", expected_head="main")
|
|
|
|
|
|
def test_rx_capture_requires_exact_local_authorization(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
with pytest.raises(ProtocolError, match="exact RX authorization"):
|
|
state.enqueue_rx_capture(
|
|
"rx-user-pc",
|
|
expected_head="a" * 40,
|
|
authorization="yes",
|
|
)
|
|
|
|
task = state.enqueue_rx_capture(
|
|
"rx-user-pc",
|
|
expected_head="a" * 40,
|
|
authorization="AUTHORIZE RF RX",
|
|
)
|
|
assert task["kind"] == "RX_CAPTURE_START"
|
|
assert task["rx_serial"] == "000000000000000097c063dc2e5327d3"
|
|
assert task["rx_authorized"] is True
|
|
assert task["automatic_retry"] is False
|
|
assert task["rf_tx"] == 0
|
|
|
|
|
|
def test_expired_rx_capture_lease_is_not_retried(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
task = state.enqueue_rx_capture(
|
|
"rx-user-pc",
|
|
expected_head="a" * 40,
|
|
authorization="AUTHORIZE RF RX",
|
|
)
|
|
check_in = {
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"agent_id": "rx-user-pc",
|
|
"inventory": {"hostname": "User-PC", "rf_tx": 0},
|
|
"rf_tx": 0,
|
|
}
|
|
|
|
first = state.check_in(check_in)
|
|
assert first["task"] == task
|
|
|
|
state._data["tasks"][0]["leased_at_epoch"] = int(time.time()) - 31
|
|
second = state.check_in(check_in)
|
|
|
|
assert second["task"] is None
|
|
stored = state.snapshot()["tasks"][0]
|
|
assert stored["state"] == "LEASE_EXPIRED"
|
|
assert "lease_expired_at" in stored
|
|
assert stored["task_id"] == task["task_id"]
|
|
|
|
|
|
def test_status_and_decode_tasks_use_fixed_session(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
session_id = "RF001A_20260923T150000Z_1234ABCD"
|
|
status = state.enqueue_rx_status("rx-user-pc", session_id=session_id)
|
|
decode = state.enqueue_capture_decode(
|
|
"rx-user-pc",
|
|
session_id=session_id,
|
|
tx_started_at="2026-09-23T15:00:05+00:00",
|
|
tx_finished_at="2026-09-23T15:00:07+00:00",
|
|
)
|
|
diagnostics = state.enqueue_capture_diagnostics(
|
|
"rx-user-pc",
|
|
session_id=session_id,
|
|
tx_started_at="2026-09-23T15:00:05+00:00",
|
|
tx_finished_at="2026-09-23T15:00:07+00:00",
|
|
)
|
|
assert status["kind"] == "RX_CAPTURE_STATUS"
|
|
assert decode["kind"] == "CAPTURE_DECODE"
|
|
assert diagnostics["kind"] == "CAPTURE_DIAGNOSTICS"
|
|
assert status["session_id"] == decode["session_id"] == session_id
|
|
assert diagnostics["session_id"] == session_id
|
|
assert status["rf_tx"] == decode["rf_tx"] == diagnostics["rf_tx"] == 0
|
|
|
|
with pytest.raises(ProtocolError, match="invalid session_id"):
|
|
state.enqueue_rx_status("rx-user-pc", session_id="..\\outside")
|
|
|
|
|
|
def test_staircase_task_has_only_fixed_hashes_and_no_retry(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
session_id = "RF001A_20260925T125829Z_227EADB4"
|
|
task = state.enqueue_staircase_search(
|
|
"rx-user-pc",
|
|
session_id=session_id,
|
|
expected_rx_sha256="2" * 64,
|
|
expected_reference_sha256="f" * 64,
|
|
)
|
|
assert task["kind"] == "STAIRCASE_SEARCH"
|
|
assert task["session_id"] == session_id
|
|
assert task["expected_rx_sha256"] == "2" * 64
|
|
assert task["expected_reference_sha256"] == "f" * 64
|
|
assert task["hardware_accessed"] is False
|
|
assert task["automatic_retry"] is False
|
|
assert task["rf_tx"] == 0
|
|
assert "capture_path" not in task
|
|
assert "reference_path" not in task
|
|
assert "command" not in task
|
|
|
|
with pytest.raises(ProtocolError, match="invalid session_id"):
|
|
state.enqueue_staircase_search(
|
|
"rx-user-pc",
|
|
session_id="..\\outside",
|
|
expected_rx_sha256="2" * 64,
|
|
expected_reference_sha256="f" * 64,
|
|
)
|
|
with pytest.raises(ProtocolError, match="expected_rx_sha256"):
|
|
state.enqueue_staircase_search(
|
|
"rx-user-pc",
|
|
session_id=session_id,
|
|
expected_rx_sha256="bad",
|
|
expected_reference_sha256="f" * 64,
|
|
)
|
|
with pytest.raises(ProtocolError, match="expected_reference_sha256"):
|
|
state.enqueue_staircase_search(
|
|
"rx-user-pc",
|
|
session_id=session_id,
|
|
expected_rx_sha256="2" * 64,
|
|
expected_reference_sha256="bad",
|
|
)
|
|
|
|
|
|
def test_long_task_heartbeat_renews_expired_lease_without_reissue(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
task = state.enqueue_staircase_search(
|
|
"rx-user-pc",
|
|
session_id="RF001A_20260925T125829Z_227EADB4",
|
|
expected_rx_sha256="2" * 64,
|
|
expected_reference_sha256="f" * 64,
|
|
)
|
|
check_in = {
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"agent_id": "rx-user-pc",
|
|
"inventory": {"agent_version": "test", "rf_tx": 0},
|
|
"rf_tx": 0,
|
|
}
|
|
assert state.check_in(check_in)["task"] == task
|
|
state._data["tasks"][0]["leased_at_epoch"] = int(time.time()) - 300
|
|
active = {
|
|
**check_in,
|
|
"active_task": {
|
|
"task_id": task["task_id"],
|
|
"kind": "STAIRCASE_SEARCH",
|
|
"state": "RUNNING",
|
|
"phase": "SEARCHING",
|
|
"processed_samples": 400,
|
|
"total_samples": 1000,
|
|
"hardware_accessed": False,
|
|
"rf_tx": 0,
|
|
},
|
|
}
|
|
response = state.check_in(active)
|
|
assert response["task"] is None
|
|
stored = state.snapshot()["tasks"][0]
|
|
assert stored["state"] == "LEASED"
|
|
assert stored["progress"]["processed_samples"] == 400
|
|
|
|
|
|
def test_update_approval_is_bound_to_registered_release_and_not_repeated(
|
|
tmp_path,
|
|
) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
release_id = "rf001-agent-2026.09.29.2"
|
|
state._data["releases"][release_id] = {
|
|
"manifest": {
|
|
"schema_version": 1,
|
|
"release_id": release_id,
|
|
"agent_version": "2026.09.29.2",
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"package_kind": "AGENT_RELEASE",
|
|
"package_sha256": "a" * 64,
|
|
"package_bytes": 1234,
|
|
"content_manifest_sha256": "b" * 64,
|
|
},
|
|
"manifest_signature": "c" * 64,
|
|
"package_path": str(tmp_path / "release.zip"),
|
|
}
|
|
task = state.enqueue_agent_update(
|
|
"rx-user-pc",
|
|
release_id=release_id,
|
|
validity_seconds=3600,
|
|
)
|
|
assert task["kind"] == "UPDATE_AGENT"
|
|
assert task["release_id"] == release_id
|
|
assert task["package_sha256"] == "a" * 64
|
|
assert task["automatic_retry"] is False
|
|
assert task["hardware_accessed"] is False
|
|
assert task["rf_tx"] == 0
|
|
with pytest.raises(ProtocolError, match="already has an update task"):
|
|
state.enqueue_agent_update(
|
|
"rx-user-pc",
|
|
release_id=release_id,
|
|
validity_seconds=3600,
|
|
)
|
|
|
|
|
|
def test_queued_update_waits_for_active_software_task(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
running = state.enqueue_staircase_search(
|
|
"rx-user-pc",
|
|
session_id="RF001A_20260925T125829Z_227EADB4",
|
|
expected_rx_sha256="2" * 64,
|
|
expected_reference_sha256="f" * 64,
|
|
)
|
|
check_in = {
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"agent_id": "rx-user-pc",
|
|
"inventory": {"agent_version": "test", "rf_tx": 0},
|
|
"rf_tx": 0,
|
|
}
|
|
assert state.check_in(check_in)["task"] == running
|
|
release_id = "rf001-agent-2026.09.29.2"
|
|
state._data["releases"][release_id] = {
|
|
"manifest": {
|
|
"schema_version": 1,
|
|
"release_id": release_id,
|
|
"agent_version": "2026.09.29.2",
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"package_kind": "AGENT_RELEASE",
|
|
"package_sha256": "a" * 64,
|
|
"package_bytes": 1234,
|
|
"content_manifest_sha256": "b" * 64,
|
|
},
|
|
"manifest_signature": "c" * 64,
|
|
"package_path": str(tmp_path / "release.zip"),
|
|
}
|
|
update = state.enqueue_agent_update(
|
|
"rx-user-pc", release_id=release_id, validity_seconds=3600
|
|
)
|
|
response = state.check_in(
|
|
{
|
|
**check_in,
|
|
"active_task": {
|
|
"task_id": running["task_id"],
|
|
"kind": "STAIRCASE_SEARCH",
|
|
"state": "RUNNING",
|
|
"phase": "SEARCHING",
|
|
"processed_samples": 1,
|
|
"total_samples": 10,
|
|
"hardware_accessed": False,
|
|
"rf_tx": 0,
|
|
},
|
|
}
|
|
)
|
|
assert response["task"] is None
|
|
stored = next(
|
|
task
|
|
for task in state.snapshot()["tasks"]
|
|
if task["task_id"] == update["task_id"]
|
|
)
|
|
assert stored["state"] == "QUEUED"
|
|
assert stored["update_status"] == "WAITING_FOR_IDLE"
|
|
assert stored["blocked_by_task_id"] == running["task_id"]
|
|
|
|
|
|
def test_http_server_end_to_end_inventory_without_rf(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
server = ControlHTTPServer(
|
|
("127.0.0.1", 0),
|
|
state=state,
|
|
token=TOKEN,
|
|
allowed_agent_ip="127.0.0.1",
|
|
)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
base_url = f"http://127.0.0.1:{server.server_address[1]}"
|
|
try:
|
|
task = request_json(
|
|
base_url,
|
|
TOKEN,
|
|
"POST",
|
|
"/api/v1/local/tasks/inventory",
|
|
{"agent_id": "rx-user-pc", "include_hackrf_info": True},
|
|
)
|
|
assert task["kind"] == "INVENTORY"
|
|
assert task["rf_tx"] == 0
|
|
|
|
check_in = request_json(
|
|
base_url,
|
|
TOKEN,
|
|
"POST",
|
|
"/api/v1/agent/check-in",
|
|
{
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"agent_id": "rx-user-pc",
|
|
"inventory": {"hostname": "User-PC", "rf_tx": 0},
|
|
"rf_tx": 0,
|
|
},
|
|
)
|
|
assert check_in["task"] == task
|
|
assert check_in["rf_tx"] == 0
|
|
|
|
result = {
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"task_id": task["task_id"],
|
|
"agent_id": "rx-user-pc",
|
|
"status": "COMPLETE",
|
|
"started_at": "2026-09-23T00:00:00+00:00",
|
|
"finished_at": "2026-09-23T00:00:01+00:00",
|
|
"result": {"hostname": "User-PC", "rf_tx": 0},
|
|
"errors": [],
|
|
"rf_tx": 0,
|
|
}
|
|
acknowledgement = request_json(
|
|
base_url,
|
|
TOKEN,
|
|
"POST",
|
|
"/api/v1/agent/result",
|
|
result,
|
|
)
|
|
assert acknowledgement["accepted"] is True
|
|
assert acknowledgement["rf_tx"] == 0
|
|
|
|
status = request_json(
|
|
base_url,
|
|
TOKEN,
|
|
"GET",
|
|
"/api/v1/local/status",
|
|
)
|
|
assert status["results"][task["task_id"]] == result
|
|
assert status["tasks"][0]["state"] == "COMPLETE"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def test_result_cannot_claim_rf_tx(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
task = state.enqueue_inventory("rx-user-pc", include_hackrf_info=False)
|
|
result = {
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"task_id": task["task_id"],
|
|
"agent_id": "rx-user-pc",
|
|
"status": "COMPLETE",
|
|
"result": {},
|
|
"errors": [],
|
|
"rf_tx": 1,
|
|
}
|
|
with pytest.raises(ProtocolError, match="rf_tx = 0"):
|
|
state.record_result(result)
|
|
|
|
|
|
def test_cancelled_task_is_not_released_or_completed(tmp_path) -> None:
|
|
state = ControlState(tmp_path / "state.json")
|
|
task = state.enqueue_rx_capture(
|
|
"rx-user-pc",
|
|
expected_head="a" * 40,
|
|
authorization="AUTHORIZE RF RX",
|
|
)
|
|
cancellation = state.cancel_task(
|
|
task["task_id"],
|
|
reason="software packaging failure before hardware access",
|
|
)
|
|
assert cancellation["state"] == "CANCELLED"
|
|
assert cancellation["rf_tx"] == 0
|
|
|
|
response = state.check_in(
|
|
{
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"agent_id": "rx-user-pc",
|
|
"inventory": {"hostname": "User-PC", "rf_tx": 0},
|
|
"rf_tx": 0,
|
|
}
|
|
)
|
|
assert response["task"] is None
|
|
|
|
result = {
|
|
"protocol_version": PROTOCOL_VERSION,
|
|
"task_id": task["task_id"],
|
|
"agent_id": "rx-user-pc",
|
|
"status": "FAILED",
|
|
"result": {"hardware_accessed": False, "rf_tx": 0},
|
|
"errors": ["packaging failure"],
|
|
"rf_tx": 0,
|
|
}
|
|
with pytest.raises(ProtocolError, match="cancelled"):
|
|
state.record_result(result)
|