schiwagoaandClaude Opus 5.5
2e3f79236c
Web interface to view, ban and unban fail2ban IPs on several hosts via SSH, generalised from the private f2b-web tool. - Web UI (Flask/Waitress): hosts.yaml, local/OIDC/proxy authentication, English/German UI, origin map, optional Loki/Grafana/GeoIP, CLI helpers - Agent f2b-webctl with install.sh/uninstall.sh (dedicated user + sudo, from= restriction, protected networks, manual jail) - Hardened Docker image, examples, documentation, smoke test, testbed Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
155 lines
7.6 KiB
Python
155 lines
7.6 KiB
Python
"""Smoke test for the web UI without Docker or SSH – the agents are simulated.
|
||
|
||
python -m venv .venv && .venv/bin/pip install -r requirements.txt
|
||
.venv/bin/python tests/smoke_test.py
|
||
"""
|
||
import os
|
||
import re
|
||
import sys
|
||
import tempfile
|
||
|
||
ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "app")
|
||
sys.path.insert(0, ROOT)
|
||
tmp = tempfile.mkdtemp()
|
||
with open(os.path.join(tmp, "hosts.yaml"), "w", encoding="utf-8") as fh:
|
||
fh.write("hosts:\n - name: a1\n address: a1.example\n - name: a2\n address: 192.0.2.5\n port: 2222\n user: root\n")
|
||
|
||
from werkzeug.security import generate_password_hash # noqa: E402
|
||
|
||
os.environ.update(DATA_DIR=tmp, CONFIG_FILE=os.path.join(tmp, "hosts.yaml"), AUTH_MODE="local",
|
||
AUTH_USER="admin", AUTH_PASSWORD_HASH=generate_password_hash("correct-horse-battery"),
|
||
LOGIN_MAX_ATTEMPTS="3")
|
||
|
||
from webui import backend, config # noqa: E402
|
||
from webui.app import create_app # noqa: E402
|
||
|
||
calls = []
|
||
|
||
|
||
def fake_ssh(self, name, *args, timeout=None):
|
||
calls.append((name, args))
|
||
if name == "a2" and args == ("list",):
|
||
return backend.fail("err_timeout", host=name)
|
||
if args == ("list",):
|
||
return {"ok": True, "jails": {"sshd": {"bans": [
|
||
{"ip": "203.0.113.7", "start": "2026-10-25 10:00:00", "bantime": 600, "end": "2026-10-25 10:10:00"}],
|
||
"total": 3}}}
|
||
return {"ok": True, "result": "1"}
|
||
|
||
|
||
REAL_SSH = backend.Agents.ssh
|
||
backend.Agents.ssh = fake_ssh
|
||
ok = 0
|
||
|
||
|
||
def check(cond, label):
|
||
global ok
|
||
print(("PASS " if cond else "FAIL ") + label)
|
||
if not cond:
|
||
sys.exit(1)
|
||
ok += 1
|
||
|
||
|
||
app = create_app()
|
||
c = app.test_client()
|
||
|
||
r = c.get("/healthz"); check(r.status_code == 200, "healthz public")
|
||
r = c.get("/"); check(r.status_code == 302 and "/login" in r.location, "index redirects to login")
|
||
r = c.get("/api/bans"); check(r.status_code == 401 and r.json["ok"] is False, "api 401 without login")
|
||
r = c.get("/login"); check(b"Sign in" in r.data and b'<html lang="en">' in r.data, "login page english default")
|
||
csrf = re.search(rb'name="csrf" value="([^"]+)"', r.data).group(1).decode()
|
||
|
||
r = c.post("/login", data={"username": "admin", "password": "x", "csrf": "bad"}, follow_redirects=True)
|
||
check(b"Session expired" in r.data, "login csrf rejected")
|
||
for _ in range(3):
|
||
r = c.post("/login", data={"username": "admin", "password": "wrong", "csrf": csrf}, follow_redirects=True)
|
||
check(b"Invalid username or password" in r.data, "wrong password")
|
||
r = c.post("/login", data={"username": "admin", "password": "correct-horse-battery", "csrf": csrf}, follow_redirects=True)
|
||
check(b"Too many failed attempts" in r.data, "lockout after 3 failures")
|
||
|
||
import logging # noqa: E402
|
||
records = []
|
||
handler = logging.Handler()
|
||
handler.emit = lambda rec: records.append(rec.getMessage())
|
||
logging.getLogger("webui").addHandler(handler)
|
||
sys.modules["webui.auth"]._failures.clear()
|
||
c.post("/login", data={"username": "x\nLOGIN_FAILED user=y ip=8.8.8.8", "password": "w", "csrf": csrf})
|
||
logging.getLogger("webui").removeHandler(handler)
|
||
check(len(records) == 1 and chr(10) not in records[0] and records[0].endswith("ip=127.0.0.1"), "no log injection via user name")
|
||
|
||
from webui import auth # noqa: E402
|
||
auth._failures.clear()
|
||
r = c.post("/login", data={"username": "admin", "password": "correct-horse-battery", "csrf": csrf})
|
||
check(r.status_code == 302 and r.location.endswith("/"), "login ok")
|
||
|
||
r = c.get("/"); check(r.status_code == 200 and b"Current bans" in r.data, "index english")
|
||
page_csrf = re.search(rb'name="csrf-token" content="([^"]+)"', r.data).group(1).decode()
|
||
check(page_csrf != csrf, "session/csrf rotated on login")
|
||
check(b"Content-Security-Policy" not in r.data and "form-action 'self'" in r.headers["Content-Security-Policy"], "csp")
|
||
|
||
r = c.get("/api/bans"); d = r.json
|
||
check(d["hosts"]["a1"]["ok"] and d["hosts"]["a2"]["error"] == "Timeout talking to a2", "bans + localized error en")
|
||
check(len(d["bans"]) == 1 and d["bans"][0]["host"] == "a1", "ban rows")
|
||
|
||
c.set_cookie("lang", "de")
|
||
r = c.get("/"); check(b"Aktuelle Sperren" in r.data and b'<html lang="de">' in r.data and "Schließen".encode() in r.data, "german via cookie")
|
||
r = c.get("/api/bans"); check(r.json["hosts"]["a2"]["error"] == "Zeitüberschreitung bei a2", "localized error de (cached data)")
|
||
|
||
r = c.post("/api/ban", json={"ip": "203.0.113.9", "hosts": ["a1"]}); check(r.status_code == 403, "ban without csrf -> 403")
|
||
r = c.post("/api/ban", json={"ip": "203.0.113.9", "hosts": ["a1"]}, headers={"X-CSRF-Token": page_csrf, "Origin": "https://evil.example"})
|
||
check(r.status_code == 403, "foreign origin -> 403")
|
||
r = c.post("/api/ban", json={"ip": "nope", "hosts": ["a1"]}, headers={"X-CSRF-Token": page_csrf})
|
||
check(r.status_code == 400 and "Ungültige IP" in r.json["error"], "invalid ip de")
|
||
calls.clear()
|
||
r = c.post("/api/ban", json={"ip": "203.0.113.9", "hosts": ["a1", "a2", "evil"], "note": "test"}, headers={"X-CSRF-Token": page_csrf})
|
||
check(r.json["ok"] and sorted(n for n, a in calls if a[0] == "ban") == ["a1", "a2"], "ban on valid hosts only")
|
||
r = c.post("/api/unban", json={"ip": "203.0.113.9", "host": "*", "jail": "*"}, headers={"X-CSRF-Token": page_csrf})
|
||
check(r.json["ok"], "unban everywhere")
|
||
r = c.post("/api/unban", json={"ip": "203.0.113.9", "host": "a1", "jail": "x;rm"}, headers={"X-CSRF-Token": page_csrf})
|
||
check(r.status_code == 400, "bad jail rejected")
|
||
r = c.get("/api/audit"); e = r.json["entries"]
|
||
check(len(e) == 2 and e[0]["action"] == "unban" and e[1]["user"] == "admin", "audit log with user")
|
||
r = c.get("/api/history?ip=203.0.113.9"); check(r.status_code == 404 and "Loki" in r.json["error"], "history without loki")
|
||
|
||
r = c.post("/logout", data={"csrf": page_csrf}); check(r.status_code == 302, "logout")
|
||
r = c.get("/api/bans"); check(r.status_code == 401, "after logout 401")
|
||
|
||
# ssh command line built correctly
|
||
import subprocess # noqa: E402
|
||
captured = {}
|
||
def fake_run(cmd, **kw):
|
||
captured["cmd"] = cmd
|
||
return subprocess.CompletedProcess(cmd, 0, '{"ok": true, "x": 1}' + chr(10), "")
|
||
backend.subprocess.run = fake_run
|
||
res = REAL_SSH(app.extensions["webui"]["agents"], "a2", "ban", "203.0.113.9")
|
||
cmd = captured["cmd"]
|
||
check(res == {"ok": True, "x": 1} and cmd[-4:] == ["root", "--", "192.0.2.5", "ban 203.0.113.9"] and "2222" in cmd, "ssh argv")
|
||
|
||
# config validation
|
||
os.environ["AUTH_MODE"] = "proxy,local"
|
||
try:
|
||
s = config.Settings(); s.validate_auth(); check(False, "proxy,local rejected")
|
||
except config.ConfigError:
|
||
check(True, "proxy,local rejected")
|
||
os.environ["AUTH_MODE"] = "oidc"
|
||
try:
|
||
s = config.Settings(); s.validate_auth(); check(False, "oidc incomplete rejected")
|
||
except config.ConfigError as exc:
|
||
check("OIDC_CLIENT_ID" in str(exc), "oidc incomplete rejected")
|
||
with open(os.path.join(tmp, "bad.yaml"), "w") as fh:
|
||
fh.write("hosts:\n - name: x\n address: -oProxyCommand=evil\n")
|
||
try:
|
||
config.load_hosts(os.path.join(tmp, "bad.yaml"), "f2bweb"); check(False, "option injection in address rejected")
|
||
except config.ConfigError:
|
||
check(True, "option injection in address rejected")
|
||
|
||
# proxy mode
|
||
os.environ.update(AUTH_MODE="proxy", AUTH_PROXY_HEADER="X-Remote-User", AUTH_PROXY_ALLOWED_IPS="10.9.9.9/32")
|
||
p = create_app().test_client()
|
||
r = p.get("/", environ_base={"REMOTE_ADDR": "10.1.1.1"}); check(r.status_code == 403, "proxy: foreign source ip 403")
|
||
r = p.get("/", environ_base={"REMOTE_ADDR": "10.9.9.9"}); check(r.status_code == 401, "proxy: missing user header 401")
|
||
r = p.get("/", environ_base={"REMOTE_ADDR": "10.9.9.9"}, headers={"X-Remote-User": "thomas"})
|
||
check(r.status_code == 200 and b"thomas" in r.data and b"action=\"/logout\"" not in r.data, "proxy: user header accepted, no logout form")
|
||
|
||
print(f"\n{ok} checks passed")
|