tegwick
9cbdc11f5e
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
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": "<script>x;"})
|
|
assert st == 400 and "<script>x" not in body
|
|
|
|
|
|
def test_secret_in_note_refused(srv):
|
|
t = srv.app.token
|
|
_, loc, _ = req(srv, "POST", "/run/new/0", {"_t": t, "operator": "bernd"})
|
|
req(srv, "POST", loc + "/ping/start", {"_t": t})
|
|
st, _, body = req(srv, "POST", loc + "/ping/fail", {"_t": t, "note": "hvs.ABCDEFGHIJKLMN"})
|
|
assert st == 400 and "secret-looking" in body
|