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>
216 lines
8.3 KiB
Python
216 lines
8.3 KiB
Python
"""fail2ban-webui – view, ban and unban fail2ban IPs on several hosts.
|
||
|
||
The hosts are reached exclusively via SSH with a restricted key; on every host
|
||
the agent (forced command f2b-webctl) decides what is allowed.
|
||
"""
|
||
import collections
|
||
import copy
|
||
import ipaddress
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import secrets
|
||
import threading
|
||
import urllib.parse
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from flask import Flask, abort, g, jsonify, render_template, request, session
|
||
from werkzeug.middleware.proxy_fix import ProxyFix
|
||
|
||
from . import auth, i18n
|
||
from .backend import Agents, Geo, loki_history
|
||
from .config import Settings
|
||
|
||
log = logging.getLogger("webui")
|
||
JAIL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,40}$") # same rule as the agent
|
||
_audit_lock = threading.Lock()
|
||
|
||
|
||
def load_secret_key(s):
|
||
if s.secret_key:
|
||
return s.secret_key
|
||
path = os.path.join(s.data_dir, "secret_key")
|
||
try:
|
||
with open(path, encoding="utf-8") as fh:
|
||
return fh.read().strip()
|
||
except FileNotFoundError:
|
||
key = secrets.token_urlsafe(48)
|
||
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
||
fh.write(key)
|
||
return key
|
||
|
||
|
||
def create_app(s=None):
|
||
s = s or Settings()
|
||
if not s.hosts:
|
||
s.load_hosts()
|
||
s.validate_auth()
|
||
|
||
app = Flask(__name__)
|
||
app.json.ensure_ascii = False
|
||
app.json.sort_keys = False # hosts in the order of hosts.yaml
|
||
app.config.update(
|
||
SETTINGS=s,
|
||
SECRET_KEY=load_secret_key(s),
|
||
SESSION_COOKIE_NAME="f2bwebui_session",
|
||
SESSION_COOKIE_HTTPONLY=True,
|
||
SESSION_COOKIE_SAMESITE="Lax", # Strict would drop the cookie on the OIDC redirect back
|
||
SESSION_COOKIE_SECURE=s.cookie_secure,
|
||
PERMANENT_SESSION_LIFETIME=timedelta(hours=s.session_hours),
|
||
)
|
||
if s.trusted_proxy_hops > 0:
|
||
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=s.trusted_proxy_hops, x_proto=s.trusted_proxy_hops)
|
||
|
||
geo = Geo(s)
|
||
agents = Agents(s, geo)
|
||
app.extensions["webui"] = {"geo": geo, "agents": agents}
|
||
|
||
# Language first – the auth guard already needs g.t for its messages
|
||
@app.before_request
|
||
def select_language():
|
||
g.lang = i18n.pick(request.cookies.get("lang"), s.default_lang)
|
||
g.t = lambda key, **kw: i18n.translate(g.lang, key, **kw)
|
||
|
||
auth.init_app(app, s)
|
||
|
||
@app.context_processor
|
||
def template_globals():
|
||
return {
|
||
"t": g.t, "lang": g.lang, "languages": i18n.languages(), "user": g.get("user"),
|
||
"csrf": auth.csrf_token(), "geo_enabled": geo.available or s.geoip_auto_download,
|
||
"logout_local": s.auth_modes != ["proxy"], "logout_url": s.proxy_logout_url,
|
||
"version": s.version,
|
||
}
|
||
|
||
# -----------------------------------------------------------------
|
||
# Helpers
|
||
# -----------------------------------------------------------------
|
||
def localize(obj):
|
||
"""Turns error_key/error_args from the backend into text in the request language."""
|
||
if isinstance(obj, dict):
|
||
if "error_key" in obj:
|
||
obj = {k: v for k, v in obj.items() if k not in ("error_key", "error_args")} | {
|
||
"error": g.t(obj["error_key"], **obj.get("error_args", {}))}
|
||
return {k: localize(v) for k, v in obj.items()}
|
||
if isinstance(obj, list):
|
||
return [localize(v) for v in obj]
|
||
return obj
|
||
|
||
def valid_ip(value):
|
||
try:
|
||
return str(ipaddress.ip_address(str(value).strip()))
|
||
except ValueError:
|
||
abort(400, description=g.t("err_invalid_ip", value=repr(value)))
|
||
|
||
def audit(action, **fields):
|
||
entry = {
|
||
"ts": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"),
|
||
"action": action, "user": g.user, "client": request.remote_addr, **fields,
|
||
}
|
||
line = json.dumps(entry, ensure_ascii=False)
|
||
log.info("AUDIT %s", line)
|
||
with _audit_lock, open(s.audit_log, "a", encoding="utf-8") as fh:
|
||
fh.write(line + "\n")
|
||
|
||
def check_write_request():
|
||
if not auth.csrf_valid(request.headers.get("X-CSRF-Token")):
|
||
abort(403, description=g.t("err_csrf"))
|
||
origin = request.headers.get("Origin")
|
||
if origin and urllib.parse.urlparse(origin).netloc != request.host:
|
||
abort(403, description=g.t("err_origin"))
|
||
if not request.is_json:
|
||
abort(415, description=g.t("err_json"))
|
||
|
||
# -----------------------------------------------------------------
|
||
# Routes
|
||
# -----------------------------------------------------------------
|
||
@app.after_request
|
||
def security_headers(resp):
|
||
resp.headers["X-Content-Type-Options"] = "nosniff"
|
||
resp.headers["X-Frame-Options"] = "DENY"
|
||
resp.headers["Referrer-Policy"] = "no-referrer"
|
||
resp.headers["Content-Security-Policy"] = (
|
||
"default-src 'self'; img-src 'self' data:; style-src 'self'; script-src 'self'; "
|
||
"connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
|
||
resp.headers["Cache-Control"] = "no-store"
|
||
return resp
|
||
|
||
for code in (400, 401, 403, 404, 415):
|
||
@app.errorhandler(code)
|
||
def json_error(err):
|
||
if request.path.startswith("/api/") or request.is_json:
|
||
return jsonify({"ok": False, "error": err.description}), err.code
|
||
return render_template("error.html", code=err.code, message=err.description), err.code
|
||
|
||
@app.get("/")
|
||
def index():
|
||
return render_template(
|
||
"index.html", hosts=list(s.hosts), manual_jail=s.manual_jail,
|
||
i18n=i18n.catalog(g.lang),
|
||
client_config={
|
||
"grafana": s.grafana_url, "grafanaDashboard": s.grafana_dashboard,
|
||
"grafanaLokiUid": s.grafana_loki_uid, "loki": bool(s.loki_url),
|
||
"lokiSelector": s.loki_selector, "lokiDays": s.loki_days, "lang": g.lang,
|
||
})
|
||
|
||
@app.get("/healthz")
|
||
def healthz():
|
||
return "ok"
|
||
|
||
@app.get("/api/bans")
|
||
def api_bans():
|
||
return jsonify(localize(copy.deepcopy(agents.bans(force=request.args.get("refresh") == "1"))))
|
||
|
||
@app.post("/api/ban")
|
||
def api_ban():
|
||
check_write_request()
|
||
body = request.get_json(silent=True) or {}
|
||
ip = valid_ip(body.get("ip"))
|
||
targets = [h for h in body.get("hosts", []) if h in s.hosts]
|
||
note = str(body.get("note", ""))[:200]
|
||
if not targets:
|
||
abort(400, description=g.t("err_no_host"))
|
||
results = agents.run_many(targets, "ban", ip)
|
||
audit("ban", ip=ip, hosts=targets, note=note,
|
||
ok={h: r.get("ok", False) for h, r in results.items()})
|
||
agents.bans(force=True)
|
||
return jsonify({"ok": all(r.get("ok") for r in results.values()), "results": localize(results)})
|
||
|
||
@app.post("/api/unban")
|
||
def api_unban():
|
||
check_write_request()
|
||
body = request.get_json(silent=True) or {}
|
||
ip = valid_ip(body.get("ip"))
|
||
host = body.get("host")
|
||
jail = str(body.get("jail", ""))
|
||
targets = list(s.hosts) if host == "*" else [host] if host in s.hosts else []
|
||
if not targets:
|
||
abort(400, description=g.t("err_unknown_host"))
|
||
if jail != "*" and not JAIL_RE.fullmatch(jail):
|
||
abort(400, description=g.t("err_invalid_jail"))
|
||
args = ("unban-all", ip) if jail == "*" else ("unban", jail, ip)
|
||
results = agents.run_many(targets, *args)
|
||
audit("unban", ip=ip, hosts=targets, jail=jail,
|
||
ok={h: r.get("ok", False) for h, r in results.items()})
|
||
agents.bans(force=True)
|
||
return jsonify({"ok": all(r.get("ok") for r in results.values()), "results": localize(results)})
|
||
|
||
@app.get("/api/history")
|
||
def api_history():
|
||
ip = valid_ip(request.args.get("ip"))
|
||
result, status = loki_history(s, ip)
|
||
return jsonify(localize({**result, "ip": ip, "geo": geo.lookup(ip)})), status
|
||
|
||
@app.get("/api/audit")
|
||
def api_audit():
|
||
try:
|
||
with open(s.audit_log, encoding="utf-8") as fh:
|
||
entries = [json.loads(line) for line in collections.deque(fh, 50) if line.strip()]
|
||
except FileNotFoundError:
|
||
entries = []
|
||
return jsonify({"ok": True, "entries": list(reversed(entries))})
|
||
|
||
return app
|