"""Phase 1 contract-alignment tests for the WolfPanel Agent API client. Verifies the frozen v1 contract on the agent side: * paths under /api/v1/agent/* * X-Agent-Token header on agent-authenticated calls (never Bearer) * register response handling = {agent_token, server_id, status} These tests capture the outgoing urllib Request without doing any network I/O. """ from __future__ import annotations import io import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) class _FakeResponse(io.BytesIO): def __enter__(self): return self def __exit__(self, *exc): return False def _capture_requests(monkeypatch, body: dict | None = None): """Patch urllib so RealApiClient calls are captured, not sent.""" import api_client captured: list = [] def fake_urlopen(request, timeout=None): captured.append(request) return _FakeResponse(json.dumps(body or {}).encode("utf-8")) monkeypatch.setattr(api_client.urllib.request, "urlopen", fake_urlopen) return captured def test_register_targets_v1_path_and_no_auth_header(monkeypatch): import api_client captured = _capture_requests( monkeypatch, {"agent_token": "t", "server_id": 1, "status": "active"} ) client = api_client.RealApiClient("https://api.example.com") resp = client.register({"install_token": "x", "agent_version": "1"}) req = captured[0] assert req.full_url == "https://api.example.com/api/v1/agent/register" # Register is unauthenticated: no agent token header on the wire. assert req.get_header("X-agent-token") is None assert req.get_header("Authorization") is None # Response is surfaced verbatim for the caller to interpret. assert resp == {"agent_token": "t", "server_id": 1, "status": "active"} def test_heartbeat_uses_x_agent_token_header(monkeypatch): import api_client captured = _capture_requests(monkeypatch) client = api_client.RealApiClient("https://api.example.com") client.heartbeat({"agent_version": "1", "status": "online"}, token="secret-agent-token") req = captured[0] assert req.full_url == "https://api.example.com/api/v1/agent/heartbeat" # Frozen agent auth scheme: X-Agent-Token, never Authorization: Bearer. assert req.get_header("X-agent-token") == "secret-agent-token" assert req.get_header("Authorization") is None def test_mock_register_response_shape(): from api_client import MockApiClient resp = MockApiClient().register({"hostname": "h"}) assert set(resp.keys()) == {"agent_token", "server_id", "status"} assert "agent_id" not in resp assert "refresh_token" not in resp def test_store_credentials_persists_server_id_and_token(tmp_path, monkeypatch): monkeypatch.setenv("WOLFPANEL_DEV_HOME", str(tmp_path)) monkeypatch.setenv("WOLFPANEL_API_MOCK", "1") import importlib import config as config_module importlib.reload(config_module) cfg = config_module.load_config() from identity import ensure_dirs, load_agent_token, store_credentials ensure_dirs(cfg) store_credentials(cfg, "42", "wp_agent_token_value") assert load_agent_token(cfg) == "wp_agent_token_value" conf_text = cfg.conf_file.read_text() assert "WOLFPANEL_SERVER_ID=42" in conf_text # The agent token must never be written into the (non-secret) conf file. assert "wp_agent_token_value" not in conf_text # --- Phase 2: discovery session flow --------------------------------------- def test_discovery_endpoints_use_v1_paths_and_token(monkeypatch): import api_client captured = _capture_requests(monkeypatch, {"session_id": 1, "status": "running"}) client = api_client.RealApiClient("https://api.example.com") token = "secret-agent-token" client.open_discovery_session({"scan_type": "full", "expected_categories": []}, token) client.upload_discovery_items({"session_id": 1, "category": "x", "items": []}, token) client.complete_discovery({"session_id": 1, "succeeded_categories": []}, token) client.fail_discovery({"session_id": 1, "error": "x"}, token) urls = [r.full_url for r in captured] assert urls == [ "https://api.example.com/api/v1/agent/discovery/sessions", "https://api.example.com/api/v1/agent/discovery/items", "https://api.example.com/api/v1/agent/discovery/complete", "https://api.example.com/api/v1/agent/discovery/failed", ] for req in captured: assert req.get_header("X-agent-token") == token assert req.get_header("Authorization") is None class _RecordingClient: """Fake ApiClient that records calls; categories in `fail_on` raise ApiError.""" def __init__(self, fail_on=()): from api_client import ApiError self._ApiError = ApiError self.fail_on = set(fail_on) self.opened = None self.items = [] self.completed = None self.failed = None def open_discovery_session(self, payload, token): self.opened = payload return {"session_id": 7, "status": "running"} def upload_discovery_items(self, payload, token): if payload["category"] in self.fail_on: raise self._ApiError("boom") self.items.append(payload) return {"session_id": 7, "accepted": len(payload["items"]), "item_count": 0} def complete_discovery(self, payload, token): self.completed = payload return {"session_id": 7, "status": "completed", "item_count": 0} def fail_discovery(self, payload, token): self.failed = payload return {"session_id": 7, "status": "failed"} def test_upload_inventory_drives_full_session_flow(): from discovery.upload import categories_from_inventory, upload_inventory inventory = { "server_facts": {"hostname": "h"}, # dict -> single item "databases": [{"name": "a"}, {"name": "b"}], # list -> two items "open_ports": [22, 80], # scalars -> wrapped } client = _RecordingClient() result = upload_inventory( None, client, "full", categories_from_inventory(inventory), token="t" ) assert client.opened["scan_type"] == "full" assert client.opened["expected_categories"] == ["databases", "open_ports", "server_facts"] # Scalars are wrapped into dicts for raw storage. ports = next(p for p in client.items if p["category"] == "open_ports") assert ports["items"] == [{"value": 22}, {"value": 80}] assert result["status"] == "completed" assert sorted(client.completed["succeeded_categories"]) == [ "databases", "open_ports", "server_facts", ] assert client.completed["failed_categories"] == [] def test_upload_inventory_reports_failed_category(): from discovery.upload import upload_inventory client = _RecordingClient(fail_on={"databases"}) categories = {"websites": [{"docroot": "/a"}], "databases": [{"name": "x"}]} upload_inventory(None, client, "full", categories, token="t") assert client.completed is not None assert client.completed["succeeded_categories"] == ["websites"] assert client.completed["failed_categories"] == ["databases"] def test_upload_inventory_fails_session_when_nothing_succeeds(): from discovery.upload import upload_inventory client = _RecordingClient(fail_on={"websites"}) upload_inventory(None, client, "full", {"websites": [{"docroot": "/a"}]}, token="t") # No category landed -> session is failed, not completed (no reconcile). assert client.completed is None assert client.failed is not None assert client.failed["session_id"] == 7 class _MiniMonkeypatch: """Minimal monkeypatch for running this file directly (no pytest).""" def __init__(self): self._undo = [] def setattr(self, obj, name, value): self._undo.append((obj, name, getattr(obj, name))) setattr(obj, name, value) def setenv(self, name, value): import os self._undo.append(("env", name, os.environ.get(name))) os.environ[name] = value def undo(self): import os for obj, name, old in reversed(self._undo): if obj == "env": if old is None: os.environ.pop(name, None) else: os.environ[name] = old else: setattr(obj, name, old) self._undo.clear() if __name__ == "__main__": import tempfile import traceback failures = 0 for fname, func in sorted( (k, v) for k, v in dict(globals()).items() if k.startswith("test_") ): mp = _MiniMonkeypatch() params = func.__code__.co_varnames[: func.__code__.co_argcount] kwargs = {} if "monkeypatch" in params: kwargs["monkeypatch"] = mp if "tmp_path" in params: kwargs["tmp_path"] = Path(tempfile.mkdtemp()) try: func(**kwargs) print(f"PASS {fname}") except Exception: # noqa: BLE001 failures += 1 print(f"FAIL {fname}") traceback.print_exc() finally: mp.undo() sys.exit(1 if failures else 0)