fail2ban-webui/app/webui/cli.py

"""Command line: docker compose run --rm -it fail2ban-webui <command>

  serve          start the web server (default)
  hash-password  create AUTH_PASSWORD_HASH for the .env file
  keygen         create the SSH key in /data/ssh and show the public key
  keyscan        fetch the host keys of all hosts into known_hosts (shows fingerprints)
  check          test the connection to every agent
"""
import argparse
import getpass
import logging
import os
import subprocess
import sys
import threading

from .config import ConfigError, Settings

log = logging.getLogger("webui")


def die(msg, code=1):
    print(f"ERROR: {msg}", file=sys.stderr)
    sys.exit(code)


def ensure_writable(path):
    os.makedirs(path, exist_ok=True)
    probe = os.path.join(path, ".write-test")
    try:
        with open(probe, "w", encoding="utf-8") as fh:
            fh.write("ok")
        os.remove(probe)
    except OSError as exc:
        die(f"{path} is not writable for UID {os.getuid()} ({exc}).\n"
            f"       Fix on the Docker host: chown -R {os.getuid()}:{os.getgid()} ./data")


def cmd_serve(s):
    from waitress import serve
    from .app import create_app

    ensure_writable(s.data_dir)
    try:
        app = create_app(s)
    except ConfigError as exc:
        die(str(exc))
    geo = app.extensions["webui"]["geo"]
    if s.geoip_auto_download:
        threading.Thread(target=geo.download_loop, daemon=True, name="geoip").start()
    log.info("fail2ban-webui %s – hosts: %s – auth: %s – loki: %s – geoip: %s",
             s.version, ", ".join(s.hosts), ",".join(s.auth_modes),
             "on" if s.loki_url else "off", "on" if geo.available or s.geoip_auto_download else "off")
    serve(app, host="0.0.0.0", port=8080, threads=8, ident=None)


def cmd_hash_password(_s):
    from werkzeug.security import generate_password_hash

    if sys.stdin.isatty():
        pw = getpass.getpass("New password: ")
        if pw != getpass.getpass("Repeat password: "):
            die("Passwords do not match.")
    else:
        pw = sys.stdin.readline().rstrip("\r\n")   # \r: input piped from Windows files
    if len(pw) < 12:
        die("Please use at least 12 characters.")
    # Single quotes: docker compose then leaves the $ signs in the hash alone
    print(f"AUTH_PASSWORD_HASH='{generate_password_hash(pw)}'")


def cmd_keygen(s, force=False):
    key = s.ssh_key
    ensure_writable(os.path.dirname(key))
    if os.path.exists(key) and not force:
        print(f"{key} already exists (use --force to replace it).", file=sys.stderr)
    else:
        if force:
            for p in (key, key + ".pub"):
                if os.path.exists(p):
                    os.remove(p)
        subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "fail2ban-webui", "-f", key],
                       check=True)
    with open(key + ".pub", encoding="utf-8") as fh:
        print(fh.read().strip())
    print("\nInstall this public key on every fail2ban host with the agent installer.", file=sys.stderr)


def cmd_keyscan(s, yes=False):
    try:
        s.load_hosts()
    except ConfigError as exc:
        die(str(exc))
    lines, failed = [], []
    for h in s.hosts.values():
        r = subprocess.run(["ssh-keyscan", "-T", "10", "-p", str(h.port), "-t", "ed25519,ecdsa,rsa", h.address],
                           capture_output=True, text=True)
        found = [x for x in r.stdout.splitlines() if x and not x.startswith("#")]
        if not found:
            failed.append(h.name)
            print(f"{h.name:20} {h.address}:{h.port}  NO HOST KEY ({r.stderr.strip()[:120]})")
            continue
        fp = subprocess.run(["ssh-keygen", "-l", "-f", "-"], input="\n".join(found) + "\n",
                            capture_output=True, text=True).stdout.strip()
        print(f"{h.name:20} {h.address}:{h.port}")
        for f in fp.splitlines():
            print(f"    {f}")
        lines.extend(found)
    if not lines:
        die("No host keys received.")
    print("\nCompare these fingerprints with the hosts, e.g.: ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub")
    if not yes:
        if not sys.stdin.isatty():
            die("Run with -it to confirm, or add --yes.")
        if input(f"Write {len(lines)} keys to {s.known_hosts}? [y/N] ").strip().lower() not in ("y", "yes", "j", "ja"):
            die("Aborted, nothing written.", 0)
    ensure_writable(os.path.dirname(s.known_hosts))
    with open(s.known_hosts, "w", encoding="utf-8") as fh:
        fh.write("\n".join(lines) + "\n")
    print(f"Written: {s.known_hosts}")
    if failed:
        die("Missing: " + ", ".join(failed))


def cmd_check(s):
    from .backend import Agents, Geo
    from . import i18n

    try:
        s.load_hosts()
    except ConfigError as exc:
        die(str(exc))
    results = Agents(s, Geo(s)).run_many(list(s.hosts), "list")
    bad = 0
    for name, res in results.items():
        if res.get("ok"):
            jails = res.get("jails", {})
            bans = sum(len(j["bans"]) for j in jails.values())
            print(f"OK    {name:20} agent {res.get('agent_version', '<1.0')}, {len(jails)} jails, {bans} bans")
        else:
            bad += 1
            err = res.get("error") or i18n.translate("en", res.get("error_key", ""), **res.get("error_args", {}))
            print(f"FAIL  {name:20} {err}")
    sys.exit(1 if bad else 0)


def main(argv=None):
    logging.basicConfig(level=logging.INFO, stream=sys.stdout,
                        format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    parser = argparse.ArgumentParser(prog="fail2ban-webui", description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = parser.add_subparsers(dest="cmd")
    sub.add_parser("serve")
    sub.add_parser("hash-password")
    p = sub.add_parser("keygen")
    p.add_argument("--force", action="store_true", help="replace an existing key")
    p = sub.add_parser("keyscan")
    p.add_argument("--yes", action="store_true", help="write without asking")
    sub.add_parser("check")
    args = parser.parse_args(argv)

    try:
        s = Settings()
    except ConfigError as exc:
        die(str(exc))
    cmd = args.cmd or "serve"
    if cmd == "serve":
        try:
            s.load_hosts()
        except ConfigError as exc:
            die(str(exc))
        cmd_serve(s)
    elif cmd == "hash-password":
        cmd_hash_password(s)
    elif cmd == "keygen":
        cmd_keygen(s, args.force)
    elif cmd == "keyscan":
        cmd_keyscan(s, args.yes)
    elif cmd == "check":
        cmd_check(s)