258 lines
8.6 KiB
Python
258 lines
8.6 KiB
Python
import re
|
|
import secrets
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import audit, totp
|
|
from app.checkmk import CheckmkClient
|
|
from app.crypto import generate_master_key
|
|
from app.fileconfig import MASTER_KEY_FILE, SESSION_SECRET_FILE, FileConfig
|
|
from app.main import create_app
|
|
from app.users import ROLE_ADMIN, ROLE_VIEWER
|
|
|
|
CSRF_RE = re.compile(r'name="csrf" value="([^"]+)"')
|
|
PASSWORD = "correct horse battery"
|
|
ADMIN = "admin1"
|
|
VIEWER = "viewer1"
|
|
CHECKMK_SECRET_VALUE = "geheimes-automation-secret"
|
|
LOGIN_FAIL_LIMIT = 5
|
|
|
|
|
|
@pytest.fixture
|
|
def app(tmp_path):
|
|
(tmp_path / MASTER_KEY_FILE).write_text(generate_master_key())
|
|
(tmp_path / SESSION_SECRET_FILE).write_text(secrets.token_urlsafe(48))
|
|
config = FileConfig(config_dir=tmp_path, db_path=tmp_path / "t.db", bind_host="127.0.0.1",
|
|
bind_port=8080, secure_cookies=False)
|
|
return create_app(config)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
return TestClient(app, follow_redirects=False)
|
|
|
|
|
|
@pytest.fixture
|
|
def users(app):
|
|
return app.state.users
|
|
|
|
|
|
def csrf_from(client, path):
|
|
response = client.get(path)
|
|
match = CSRF_RE.search(response.text)
|
|
assert match, f"kein CSRF-Token auf {path} (Status {response.status_code})"
|
|
return match.group(1)
|
|
|
|
|
|
def login(client, username, password):
|
|
token = csrf_from(client, "/login")
|
|
return client.post("/login", data={"csrf": token, "username": username, "password": password})
|
|
|
|
|
|
def login_ok(client, username=ADMIN, password=PASSWORD):
|
|
response = login(client, username, password)
|
|
assert response.status_code == 303
|
|
return response
|
|
|
|
|
|
def post(client, path, form_page, **data):
|
|
token = csrf_from(client, form_page)
|
|
return client.post(path, data={"csrf": token, **data})
|
|
|
|
|
|
def checkmk_form(**overrides):
|
|
data = {
|
|
"checkmk_url": "https://checkmk.example.de",
|
|
"checkmk_site": "prod",
|
|
"checkmk_user": "automation",
|
|
"checkmk_secret": CHECKMK_SECRET_VALUE,
|
|
"checkmk_ca_file": "",
|
|
"checkmk_verify_tls": "1",
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
@pytest.fixture
|
|
def admin(users):
|
|
return users.create(ADMIN, PASSWORD, ROLE_ADMIN, actor="test")
|
|
|
|
|
|
@pytest.fixture
|
|
def viewer(users):
|
|
return users.create(VIEWER, PASSWORD, ROLE_VIEWER, actor="test")
|
|
|
|
|
|
def test_unauthenticated_redirects_to_login(client):
|
|
response = client.get("/")
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login"
|
|
|
|
|
|
def test_login_ok(client, admin):
|
|
login_ok(client)
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert "Dashboard" in response.text
|
|
|
|
|
|
def test_wrong_password(client, app, admin):
|
|
response = login(client, ADMIN, "falsches passwort!!")
|
|
assert response.status_code == 401
|
|
assert "login.failed" in [e["action"] for e in audit.recent(app.state.db)]
|
|
|
|
|
|
def test_lockout_after_failures(client, admin):
|
|
for _ in range(LOGIN_FAIL_LIMIT):
|
|
assert login(client, ADMIN, "falsches passwort!!").status_code == 401
|
|
assert login(client, ADMIN, PASSWORD).status_code == 429
|
|
|
|
|
|
def test_post_without_csrf_forbidden(client, admin):
|
|
response = client.post("/login", data={"username": ADMIN, "password": PASSWORD})
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_viewer_forbidden_on_admin_pages(client, viewer):
|
|
login_ok(client, VIEWER)
|
|
for path in ("/users", "/audit", "/settings/checkmk"):
|
|
assert client.get(path).status_code == 403
|
|
assert client.get("/").status_code == 200
|
|
|
|
|
|
def test_admin_saves_settings_secret_encrypted(client, app, admin):
|
|
login_ok(client)
|
|
response = post(client, "/settings/checkmk", "/settings/checkmk", **checkmk_form())
|
|
assert response.status_code == 303
|
|
store = app.state.store
|
|
assert store.get("checkmk.secret") == CHECKMK_SECRET_VALUE
|
|
raw = app.state.db.query("SELECT value, is_secret FROM settings WHERE key='checkmk.secret'")[0]
|
|
assert raw["is_secret"] == 1
|
|
assert CHECKMK_SECRET_VALUE not in raw["value"]
|
|
page = client.get("/settings/checkmk")
|
|
assert CHECKMK_SECRET_VALUE not in page.text
|
|
assert "gesetzt" in page.text
|
|
assert CHECKMK_SECRET_VALUE not in repr(audit.recent(app.state.db))
|
|
|
|
|
|
def test_empty_secret_keeps_stored_value(client, app, admin):
|
|
login_ok(client)
|
|
post(client, "/settings/checkmk", "/settings/checkmk", **checkmk_form())
|
|
post(client, "/settings/checkmk", "/settings/checkmk", **checkmk_form(checkmk_secret="", checkmk_site="test"))
|
|
assert app.state.store.get("checkmk.secret") == CHECKMK_SECRET_VALUE
|
|
assert app.state.store.get("checkmk.site") == "test"
|
|
|
|
|
|
def test_settings_validation_error(client, app, admin):
|
|
login_ok(client)
|
|
response = post(client, "/settings/checkmk", "/settings/checkmk",
|
|
**checkmk_form(checkmk_url="checkmk.example.de"))
|
|
assert response.status_code == 422
|
|
assert "http://" in response.text
|
|
assert app.state.store.get("checkmk.url") is None
|
|
|
|
|
|
def test_settings_unknown_section(client, admin):
|
|
login_ok(client)
|
|
assert client.get("/settings/gibtsnicht").status_code == 404
|
|
|
|
|
|
def test_settings_test_endpoint(client, admin, monkeypatch):
|
|
async def fake_version(self):
|
|
return {"versions": {"checkmk": "2.3.0"}}
|
|
|
|
async def fake_problems(self):
|
|
return []
|
|
|
|
monkeypatch.setattr(CheckmkClient, "version", fake_version)
|
|
monkeypatch.setattr(CheckmkClient, "problem_services", fake_problems)
|
|
login_ok(client)
|
|
token = csrf_from(client, "/settings/checkmk")
|
|
response = client.post("/settings/checkmk/test", data=checkmk_form(), headers={"X-CSRF-Token": token})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["ok"] is True
|
|
assert "2.3.0" in body["detail"]
|
|
|
|
|
|
def test_settings_test_endpoint_reports_validation(client, admin):
|
|
login_ok(client)
|
|
token = csrf_from(client, "/settings/checkmk")
|
|
response = client.post("/settings/checkmk/test", data=checkmk_form(checkmk_url="nope"),
|
|
headers={"X-CSRF-Token": token})
|
|
assert response.json()["ok"] is False
|
|
|
|
|
|
def test_session_invalid_after_password_change(client, users, admin):
|
|
login_ok(client)
|
|
assert client.get("/").status_code == 200
|
|
users.set_password(admin.id, "another long password", actor="test")
|
|
response = client.get("/")
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login"
|
|
|
|
|
|
def test_totp_login_flow(client, users, admin):
|
|
secret = totp.generate_secret()
|
|
users.set_totp(admin.id, secret, actor="test")
|
|
response = login(client, ADMIN, PASSWORD)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login/totp"
|
|
assert client.get("/").status_code == 303
|
|
token = csrf_from(client, "/login/totp")
|
|
bad = client.post("/login/totp", data={"csrf": token, "code": "000000"})
|
|
assert bad.status_code == 401
|
|
response = client.post("/login/totp", data={"csrf": token, "code": totp.current_code(secret)})
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/"
|
|
assert client.get("/").status_code == 200
|
|
|
|
|
|
def test_totp_page_requires_pending_login(client):
|
|
response = client.get("/login/totp")
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/login"
|
|
|
|
|
|
def test_create_change_role_delete_user_via_ui(client, users, admin):
|
|
login_ok(client)
|
|
response = post(client, "/users", "/users", username="newbie", password=PASSWORD, role=ROLE_VIEWER)
|
|
assert response.status_code == 303
|
|
created = users.get_by_name("newbie")
|
|
assert created and created.role == ROLE_VIEWER
|
|
|
|
post(client, f"/users/{created.id}/role", "/users", role=ROLE_ADMIN)
|
|
assert users.get(created.id).role == ROLE_ADMIN
|
|
|
|
post(client, f"/users/{created.id}/delete", "/users")
|
|
assert users.get(created.id) is None
|
|
|
|
|
|
def test_create_user_validation_message(client, admin):
|
|
login_ok(client)
|
|
post(client, "/users", "/users", username="newbie", password="kurz", role=ROLE_VIEWER)
|
|
assert "Passwort: mindestens" in client.get("/users").text
|
|
|
|
|
|
def test_last_admin_guard_shows_flash(client, users, admin):
|
|
login_ok(client)
|
|
post(client, f"/users/{admin.id}/role", "/users", role=ROLE_VIEWER)
|
|
assert "letzte aktive Admin" in client.get("/users").text
|
|
|
|
|
|
def test_cannot_delete_or_disable_self(client, users, admin):
|
|
login_ok(client)
|
|
post(client, f"/users/{admin.id}/delete", "/users")
|
|
assert users.get(admin.id) is not None
|
|
post(client, f"/users/{admin.id}/disable", "/users", disabled="1")
|
|
assert not users.get(admin.id).disabled
|
|
|
|
|
|
def test_health_json(client):
|
|
response = client.get("/health")
|
|
assert response.status_code == 503
|
|
body = response.json()
|
|
assert body["ok"] is False
|
|
assert "checks" in body
|