121 lines
5.4 KiB
Python
121 lines
5.4 KiB
Python
import httpx
|
|
|
|
from app.agent_runs import AgentRunRepository
|
|
from app.constants import CHECKMK_DT_ENABLED, CHECKMK_DT_SECRET, CHECKMK_DT_USER, FLAG_ON
|
|
from app.downtime import CheckmkDowntimeClient
|
|
from tests.test_web import (VIEWER, admin, app, client, csrf_from, login_ok, post, # noqa: F401
|
|
users, viewer)
|
|
|
|
DT_SECRET = "downtime-geheimnis-xyz"
|
|
|
|
|
|
def make_run(app, question="Warum <b>CPU</b>?", steps=None, abort=None):
|
|
repo = AgentRunRepository(app.state.db)
|
|
steps = steps if steps is not None else [
|
|
{"number": 1, "tool": "get_service_status", "arguments": {"host": "web01"}, "result": "<script>alert(1)</script>",
|
|
"duration": 0.5, "error": None}]
|
|
return repo.insert(1_700_000_000, "telegram", "111", "ask", question, "Antwort <i>x</i>", abort, steps)
|
|
|
|
|
|
def test_agent_pages_require_admin(client, app, viewer):
|
|
run_id = make_run(app)
|
|
assert client.get("/agent").status_code == 303
|
|
login_ok(client, VIEWER)
|
|
assert client.get("/agent").status_code == 403
|
|
assert client.get(f"/agent/{run_id}").status_code == 403
|
|
assert "/agent" not in client.get("/").text
|
|
|
|
|
|
def test_agent_list_and_detail_for_admin_escaped(client, app, admin):
|
|
run_id = make_run(app, abort="step_limit")
|
|
login_ok(client)
|
|
assert 'href="/agent"' in client.get("/").text
|
|
page = client.get("/agent")
|
|
assert page.status_code == 200 and f"/agent/{run_id}" in page.text and "Schrittlimit" in page.text
|
|
assert "<b>CPU</b>" not in page.text and "<b>CPU</b>" in page.text
|
|
detail = client.get(f"/agent/{run_id}")
|
|
assert detail.status_code == 200 and "get_service_status" in detail.text and "host=web01" in detail.text
|
|
assert "<script>alert(1)</script>" not in detail.text and "<script>" in detail.text
|
|
assert "Antwort <i>x</i>" in detail.text and "<script" not in detail.text.replace("<script src", "")
|
|
|
|
|
|
def test_agent_detail_unknown_run_is_404(client, admin):
|
|
login_ok(client)
|
|
assert client.get("/agent/999").status_code == 404
|
|
assert client.get("/agent/abc").status_code == 422
|
|
|
|
|
|
def test_agent_empty_list(client, admin):
|
|
login_ok(client)
|
|
assert "Noch keine Läufe" in client.get("/agent").text
|
|
|
|
|
|
def dt_form(**overrides):
|
|
data = {"checkmk_dt_enabled": "1", "checkmk_dt_user": "dtuser", "checkmk_dt_secret": DT_SECRET,
|
|
"checkmk_dt_max_hours": "12"}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def test_downtime_section_is_admin_only_and_has_warning(client, viewer, admin):
|
|
login_ok(client, VIEWER)
|
|
assert client.get("/settings/checkmk_dt").status_code == 403
|
|
client.cookies.clear()
|
|
login_ok(client)
|
|
page = client.get("/settings/checkmk_dt")
|
|
assert page.status_code == 200 and "einzige Funktion, die in Checkmk schreibt" in page.text
|
|
assert "checked" not in page.text.split('name="checkmk_dt_enabled"')[1].split(">")[0]
|
|
|
|
|
|
def test_downtime_section_saves_encrypted_secret_and_never_shows_it(client, app, admin):
|
|
login_ok(client)
|
|
assert post(client, "/settings/checkmk_dt", "/settings/checkmk_dt", **dt_form()).status_code == 303
|
|
store = app.state.store
|
|
assert store.get(CHECKMK_DT_ENABLED) == FLAG_ON and store.get(CHECKMK_DT_USER) == "dtuser"
|
|
assert store.get(CHECKMK_DT_SECRET) == DT_SECRET
|
|
raw = app.state.db.query("SELECT value, is_secret FROM settings WHERE key=?", (CHECKMK_DT_SECRET,))[0]
|
|
assert raw["is_secret"] == 1 and DT_SECRET not in raw["value"]
|
|
assert DT_SECRET not in client.get("/settings/checkmk_dt").text
|
|
audit_rows = app.state.db.query("SELECT old_value, new_value FROM audit WHERE target=?", (CHECKMK_DT_SECRET,))
|
|
assert audit_rows and all(DT_SECRET not in str(tuple(r)) for r in audit_rows)
|
|
assert "gesetzt" in client.get("/settings/checkmk_dt").text
|
|
|
|
|
|
def test_downtime_section_validation(client, app, admin):
|
|
login_ok(client)
|
|
for override in ({"checkmk_dt_max_hours": "0"}, {"checkmk_dt_max_hours": "999"}, {"checkmk_dt_max_hours": "x"},
|
|
{"checkmk_dt_user": "", "checkmk_dt_secret": ""}, {"checkmk_dt_user": "a b"}):
|
|
response = post(client, "/settings/checkmk_dt", "/settings/checkmk_dt", **dt_form(**override))
|
|
assert response.status_code == 422, override
|
|
assert app.state.store.get(CHECKMK_DT_ENABLED) is None
|
|
|
|
|
|
def test_downtime_default_is_off(app):
|
|
assert app.state.store.get(CHECKMK_DT_ENABLED) is None
|
|
assert app.state.runtime.chatops.downtime.enabled() is False
|
|
|
|
|
|
def test_downtime_connection_test_uses_downtime_user_and_get_only(client, app, admin, monkeypatch):
|
|
seen = []
|
|
|
|
def handler(request):
|
|
seen.append((request.method, request.url.path, request.headers["authorization"]))
|
|
return httpx.Response(200, json={"versions": {"checkmk": "2.3.0"}})
|
|
|
|
original = CheckmkDowntimeClient.__init__
|
|
|
|
def patched(self, *args, **kwargs):
|
|
kwargs["transport"] = httpx.MockTransport(handler)
|
|
original(self, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(CheckmkDowntimeClient, "__init__", patched)
|
|
store = app.state.store
|
|
store.set("checkmk.url", "https://cmk.example.de", "t")
|
|
store.set("checkmk.site", "prod", "t")
|
|
login_ok(client)
|
|
token = csrf_from(client, "/settings/checkmk_dt")
|
|
response = client.post("/settings/checkmk_dt/test", data={"csrf": token, **dt_form()})
|
|
body = response.json()
|
|
assert body["ok"] is True and "nur lesend" in body["detail"] and DT_SECRET not in body["detail"]
|
|
assert seen == [("GET", "/prod/check_mk/api/1.0/version", f"Bearer dtuser {DT_SECRET}")]
|