feat: Phase 2 UI base (login, roles, settings, audit, CLI)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
patrickandClaude Sonnet 5.5 committed
1 parent
6e244234eb
commit
93deb60c7d
30 files changed
+2128
-11
No files matched your search
@@ -0,0 +1,63 @@
|
||||
from app.ratelimit import MAX_FAILURES, WINDOW_SECONDS, FailureLimiter
|
||||
|
||||
|
||||
class Clock:
|
||||
def __init__(self):
|
||||
self.now = 1000.0
|
||||
|
||||
def __call__(self):
|
||||
return self.now
|
||||
|
||||
|
||||
def make():
|
||||
clock = Clock()
|
||||
return FailureLimiter(clock=clock), clock
|
||||
|
||||
|
||||
def test_blocks_after_max_failures():
|
||||
limiter, _ = make()
|
||||
for _ in range(MAX_FAILURES - 1):
|
||||
limiter.record("alice")
|
||||
assert not limiter.blocked("alice")
|
||||
limiter.record("alice")
|
||||
assert limiter.blocked("alice")
|
||||
|
||||
|
||||
def test_keys_are_independent():
|
||||
limiter, _ = make()
|
||||
for _ in range(MAX_FAILURES):
|
||||
limiter.record("alice")
|
||||
assert not limiter.blocked("bob")
|
||||
|
||||
|
||||
def test_block_expires_with_clock():
|
||||
limiter, clock = make()
|
||||
for _ in range(MAX_FAILURES):
|
||||
limiter.record("alice")
|
||||
clock.now += WINDOW_SECONDS - 1
|
||||
assert limiter.blocked("alice")
|
||||
clock.now += 2
|
||||
assert not limiter.blocked("alice")
|
||||
|
||||
|
||||
def test_old_failures_do_not_accumulate():
|
||||
limiter, clock = make()
|
||||
for _ in range(MAX_FAILURES - 1):
|
||||
limiter.record("alice")
|
||||
clock.now += WINDOW_SECONDS + 1
|
||||
limiter.record("alice")
|
||||
assert not limiter.blocked("alice")
|
||||
|
||||
|
||||
def test_reset_clears_block():
|
||||
limiter, _ = make()
|
||||
for _ in range(MAX_FAILURES):
|
||||
limiter.record("alice")
|
||||
limiter.reset("alice")
|
||||
assert not limiter.blocked("alice")
|
||||
|
||||
|
||||
def test_reset_unknown_key_is_noop():
|
||||
limiter, _ = make()
|
||||
limiter.reset("nobody")
|
||||
assert not limiter.blocked("nobody")
|
||||
@@ -0,0 +1,54 @@
|
||||
from app import totp
|
||||
|
||||
RFC_SECRET = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"
|
||||
RFC_TIME = 59
|
||||
RFC_CODE = "287082"
|
||||
|
||||
|
||||
def test_rfc6238_vector():
|
||||
assert totp.current_code(RFC_SECRET, now=RFC_TIME) == RFC_CODE
|
||||
|
||||
|
||||
def test_verify_accepts_current_code():
|
||||
assert totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME)
|
||||
|
||||
|
||||
def test_verify_accepts_spaces():
|
||||
assert totp.verify(RFC_SECRET, "287 082", now=RFC_TIME)
|
||||
|
||||
|
||||
def test_window_previous_and_next_step():
|
||||
assert totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME + totp.STEP_SECONDS)
|
||||
assert totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME - totp.STEP_SECONDS)
|
||||
|
||||
|
||||
def test_outside_window_rejected():
|
||||
assert not totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME + 2 * totp.STEP_SECONDS)
|
||||
assert not totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME - 2 * totp.STEP_SECONDS)
|
||||
|
||||
|
||||
def test_wrong_code_rejected():
|
||||
assert not totp.verify(RFC_SECRET, "000000", now=RFC_TIME)
|
||||
|
||||
|
||||
def test_short_code_rejected():
|
||||
assert not totp.verify(RFC_SECRET, RFC_CODE[:5], now=RFC_TIME)
|
||||
|
||||
|
||||
def test_non_numeric_code_rejected():
|
||||
assert not totp.verify(RFC_SECRET, "28708a", now=RFC_TIME)
|
||||
assert not totp.verify(RFC_SECRET, "", now=RFC_TIME)
|
||||
|
||||
|
||||
def test_generated_secret_roundtrip():
|
||||
secret = totp.generate_secret()
|
||||
assert totp.verify(secret, totp.current_code(secret))
|
||||
|
||||
|
||||
def test_provisioning_uri():
|
||||
uri = totp.provisioning_uri(RFC_SECRET, "max mustermann")
|
||||
assert uri.startswith("otpauth://totp/checkbot:max%20mustermann?")
|
||||
assert f"secret={RFC_SECRET}" in uri
|
||||
assert "issuer=checkbot" in uri
|
||||
assert f"digits={totp.DIGITS}" in uri
|
||||
assert f"period={totp.STEP_SECONDS}" in uri
|
||||
@@ -0,0 +1,155 @@
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from argon2 import PasswordHasher
|
||||
|
||||
from app import audit
|
||||
from app.crypto import SecretBox
|
||||
from app.db import Database
|
||||
from app.users import MIN_PASSWORD_LEN, ROLE_ADMIN, ROLE_VIEWER, UserError, UserStore
|
||||
|
||||
PASSWORD = "correct horse battery"
|
||||
ACTOR = "tester"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db():
|
||||
return Database(":memory:")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store(db):
|
||||
hasher = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1)
|
||||
return UserStore(db, SecretBox(os.urandom(32)), hasher)
|
||||
|
||||
|
||||
def test_create_and_authenticate(store):
|
||||
created = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
assert created.username == "alice"
|
||||
assert created.is_admin
|
||||
assert not created.has_totp
|
||||
assert store.authenticate("alice", PASSWORD).id == created.id
|
||||
assert store.count() == 1
|
||||
|
||||
|
||||
def test_wrong_password(store):
|
||||
store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
assert store.authenticate("alice", "wrong password!!") is None
|
||||
|
||||
|
||||
def test_unknown_user(store):
|
||||
assert store.authenticate("ghost", PASSWORD) is None
|
||||
|
||||
|
||||
def test_password_too_short(store):
|
||||
with pytest.raises(UserError):
|
||||
store.create("alice", "x" * (MIN_PASSWORD_LEN - 1), ROLE_ADMIN, ACTOR)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["ab", "a b c", "ümlaut", "x" * 65, "", "a/b"])
|
||||
def test_invalid_username(store, name):
|
||||
with pytest.raises(UserError):
|
||||
store.create(name, PASSWORD, ROLE_VIEWER, ACTOR)
|
||||
|
||||
|
||||
def test_invalid_role(store):
|
||||
with pytest.raises(UserError):
|
||||
store.create("alice", PASSWORD, "root", ACTOR)
|
||||
|
||||
|
||||
def test_duplicate_case_insensitive(store):
|
||||
store.create("Alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
with pytest.raises(UserError):
|
||||
store.create("alice", PASSWORD, ROLE_VIEWER, ACTOR)
|
||||
|
||||
|
||||
def test_last_admin_guard_set_role(store):
|
||||
admin = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
with pytest.raises(UserError):
|
||||
store.set_role(admin.id, ROLE_VIEWER, ACTOR)
|
||||
|
||||
|
||||
def test_last_admin_guard_set_disabled(store):
|
||||
admin = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
with pytest.raises(UserError):
|
||||
store.set_disabled(admin.id, True, ACTOR)
|
||||
|
||||
|
||||
def test_last_admin_guard_delete(store):
|
||||
admin = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
with pytest.raises(UserError):
|
||||
store.delete(admin.id, ACTOR)
|
||||
|
||||
|
||||
def test_second_admin_allows_changes(store):
|
||||
first = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
store.create("bob", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
store.set_role(first.id, ROLE_VIEWER, ACTOR)
|
||||
assert store.get(first.id).role == ROLE_VIEWER
|
||||
store.set_disabled(first.id, True, ACTOR)
|
||||
assert store.get(first.id).disabled
|
||||
store.delete(first.id, ACTOR)
|
||||
assert store.get(first.id) is None
|
||||
|
||||
|
||||
def test_viewer_can_be_removed_freely(store):
|
||||
store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
viewer = store.create("bob", PASSWORD, ROLE_VIEWER, ACTOR)
|
||||
store.set_disabled(viewer.id, True, ACTOR)
|
||||
store.delete(viewer.id, ACTOR)
|
||||
assert store.count() == 1
|
||||
|
||||
|
||||
def test_set_password_bumps_session_version(store):
|
||||
user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
store.set_password(user.id, "another long password", ACTOR)
|
||||
updated = store.get(user.id)
|
||||
assert updated.session_version == user.session_version + 1
|
||||
assert store.authenticate("alice", PASSWORD) is None
|
||||
assert store.authenticate("alice", "another long password") is not None
|
||||
|
||||
|
||||
def test_set_password_validates_length(store):
|
||||
user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
with pytest.raises(UserError):
|
||||
store.set_password(user.id, "short", ACTOR)
|
||||
|
||||
|
||||
def test_disabled_user_cannot_login(store):
|
||||
store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
bob = store.create("bob", PASSWORD, ROLE_VIEWER, ACTOR)
|
||||
store.set_disabled(bob.id, True, ACTOR)
|
||||
assert store.authenticate("bob", PASSWORD) is None
|
||||
store.set_disabled(bob.id, False, ACTOR)
|
||||
assert store.authenticate("bob", PASSWORD) is not None
|
||||
|
||||
|
||||
def test_unknown_user_id(store):
|
||||
with pytest.raises(UserError):
|
||||
store.set_role(999, ROLE_VIEWER, ACTOR)
|
||||
|
||||
|
||||
def test_totp_secret_encrypted_and_roundtrip(store, db):
|
||||
user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
secret = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"
|
||||
store.set_totp(user.id, secret, ACTOR)
|
||||
raw = db.query("SELECT totp_secret FROM users WHERE id=?", (user.id,))[0]["totp_secret"]
|
||||
assert raw and secret not in raw
|
||||
assert store.totp_secret(user.id) == secret
|
||||
assert store.get(user.id).has_totp
|
||||
store.set_totp(user.id, None, ACTOR)
|
||||
assert store.totp_secret(user.id) is None
|
||||
assert not store.get(user.id).has_totp
|
||||
|
||||
|
||||
def test_audit_entries_without_password(store, db):
|
||||
user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR)
|
||||
store.set_password(user.id, "another long password", ACTOR)
|
||||
store.set_totp(user.id, "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ", ACTOR)
|
||||
entries = audit.recent(db)
|
||||
actions = {e["action"] for e in entries}
|
||||
assert {"user.create", "user.password", "user.totp.enable"} <= actions
|
||||
dump = repr(entries)
|
||||
assert PASSWORD not in dump
|
||||
assert "another long password" not in dump
|
||||
assert "GEZDGNBVGY3TQOJQ" not in dump
|
||||
@@ -0,0 +1,257 @@
|
||||
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
|
||||
Reference in new issue
Block a user