import http.client import re import shutil import threading from pathlib import Path from urllib.parse import urlencode import pytest from rtut import ui ROOT = Path(__file__).resolve().parent.parent @pytest.fixture def srv(tmp_path): d = tmp_path / "runbooks" / "hello"; d.mkdir(parents=True) shutil.copy(ROOT / "runbooks/example-hello/pack.yaml", d / "pack.yaml") s = ui.serve([tmp_path], 0) threading.Thread(target=s.serve_forever, daemon=True).start() yield s s.shutdown() def req(s, method, path, form=None, host="127.0.0.1"): c = http.client.HTTPConnection("127.0.0.1", s.server_address[1]) body = urlencode(form) if form is not None else None c.request(method, path, body, {"Host": host, "Content-Type": "application/x-www-form-urlencoded"}) r = c.getresponse() return r.status, r.getheader("Location"), r.read().decode() def test_index_lists_pack(srv): st, _, body = req(srv, "GET", "/") assert st == 200 and "Example - check a host answers" in body def test_foreign_host_and_missing_token_refused(srv): assert req(srv, "GET", "/", host="evil.example")[0] == 403 assert req(srv, "POST", "/run/new/0", {"operator": "bernd"})[0] == 403 def test_walk_a_run(srv): t = srv.app.token st, loc, _ = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd", "p_host": "localhost", "p_count": "1"}) assert st == 303 and loc.startswith("/run/0/") assert req(srv, "POST", loc + "/ping/start", {"_t": t})[0] == 303 st, _, body = req(srv, "GET", loc) assert "ping -c 1 localhost" in body and "Run check" in body req(srv, "POST", loc + "/ping/verify", {"_t": t, "yes": "1"}) st, _, body = req(srv, "GET", loc) if "passed" in body and "failed" not in body.split("Threat")[0].replace("Mark failed", ""): assert req(srv, "POST", loc + "/ping/done", {"_t": t})[0] == 303 assert "All steps done" in req(srv, "GET", loc)[2] def test_unsafe_parameter_rejected_and_escaped(srv): t = srv.app.token st, _, body = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd", "p_host": "