"""Configuration from environment variables and the hosts file (config/hosts.yaml).""" import ipaddress import os import re from dataclasses import dataclass import yaml NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") USER_RE = re.compile(r"^[a-z_][a-z0-9_-]{0,31}$") AUTH_MODES = {"local", "oidc", "proxy"} class ConfigError(Exception): pass def env(name, default=""): return os.environ.get(name, default).strip() def env_bool(name, default=False): value = env(name).lower() return default if value == "" else value in ("1", "true", "yes", "on") def env_int(name, default): value = env(name) try: return int(value) if value else default except ValueError: raise ConfigError(f"{name} must be an integer, got {value!r}") from None def env_list(name, default=""): return [x.strip() for x in env(name, default).split(",") if x.strip()] @dataclass(frozen=True) class Host: name: str address: str port: int = 22 user: str = "f2bweb" def load_hosts(path, default_user): hosts = [] if os.path.exists(path): with open(path, encoding="utf-8") as fh: data = yaml.safe_load(fh) or {} for item in data.get("hosts") or []: if not isinstance(item, dict): raise ConfigError(f"{path}: every entry below 'hosts' must be a mapping") try: port = int(item.get("port", 22)) except (TypeError, ValueError): raise ConfigError(f"{path}: invalid port {item.get('port')!r}") from None hosts.append(Host( name=str(item.get("name", "")).strip(), address=str(item.get("address", "")).strip(), port=port, user=str(item.get("user") or default_user).strip(), )) # Older single-line form: F2B_HOSTS=name=address,name2=address2 for item in env("F2B_HOSTS").split(","): if "=" in item: name, address = item.split("=", 1) hosts.append(Host(name.strip(), address.strip(), 22, default_user)) seen = set() for h in hosts: if not NAME_RE.fullmatch(h.name): raise ConfigError(f"Invalid host name {h.name!r} (allowed: letters, digits, . _ -)") # No leading dash or whitespace: the address ends up on the ssh command line if not h.address or h.address.startswith("-") or any(c.isspace() for c in h.address): raise ConfigError(f"Invalid address for host {h.name!r}: {h.address!r}") if not 1 <= h.port <= 65535: raise ConfigError(f"Invalid port for host {h.name!r}: {h.port}") if not USER_RE.fullmatch(h.user): raise ConfigError(f"Invalid SSH user for host {h.name!r}: {h.user!r}") if h.name in seen: raise ConfigError(f"Host name {h.name!r} is used twice") seen.add(h.name) if not hosts: raise ConfigError(f"No hosts configured – create {path} (see hosts.yaml.example)") return {h.name: h for h in hosts} class Settings: def __init__(self): data = env("DATA_DIR", "/data") self.data_dir = data self.config_file = env("CONFIG_FILE", "/config/hosts.yaml") self.version = env("APP_VERSION", "dev") # SSH / agents self.ssh_user = env("SSH_USER", "f2bweb") self.ssh_key = env("SSH_KEY", f"{data}/ssh/id_ed25519") self.known_hosts = env("KNOWN_HOSTS", f"{data}/ssh/known_hosts") self.ssh_timeout = env_int("SSH_TIMEOUT", 40) self.cache_seconds = env_int("CACHE_SECONDS", 15) self.manual_jail = env("MANUAL_JAIL", "manual-block") self.audit_log = env("AUDIT_LOG", f"{data}/audit.log") # Web / sessions self.public_url = env("PUBLIC_URL").rstrip("/") self.trusted_proxy_hops = env_int("TRUSTED_PROXY_HOPS", 0) self.default_lang = env("DEFAULT_LANG", "en") self.secret_key = env("SECRET_KEY") self.session_hours = env_int("SESSION_HOURS", 12) self.cookie_secure = env_bool("SESSION_COOKIE_SECURE", self.public_url.startswith("https://")) # Authentication self.auth_modes = [m.lower() for m in env_list("AUTH_MODE", "local")] self.auth_user = env("AUTH_USER", "admin") self.auth_password_hash = env("AUTH_PASSWORD_HASH") hash_file = env("AUTH_PASSWORD_HASH_FILE") if not self.auth_password_hash and hash_file and os.path.exists(hash_file): with open(hash_file, encoding="utf-8") as fh: self.auth_password_hash = fh.read().strip() self.login_max_attempts = env_int("LOGIN_MAX_ATTEMPTS", 5) self.login_lock_minutes = env_int("LOGIN_LOCK_MINUTES", 15) self.proxy_user_header = env("AUTH_PROXY_HEADER") self.proxy_logout_url = env("AUTH_PROXY_LOGOUT_URL") try: self.proxy_allowed_ips = [ipaddress.ip_network(n, strict=False) for n in env_list("AUTH_PROXY_ALLOWED_IPS")] except ValueError as exc: raise ConfigError(f"AUTH_PROXY_ALLOWED_IPS: {exc}") from None issuer = env("OIDC_ISSUER").rstrip("/") self.oidc_discovery_url = env("OIDC_DISCOVERY_URL") or ( f"{issuer}/.well-known/openid-configuration" if issuer else "") self.oidc_client_id = env("OIDC_CLIENT_ID") self.oidc_client_secret = env("OIDC_CLIENT_SECRET") self.oidc_scopes = env("OIDC_SCOPES", "openid email profile") self.oidc_username_claim = env("OIDC_USERNAME_CLAIM", "preferred_username") self.oidc_groups_claim = env("OIDC_GROUPS_CLAIM", "groups") self.oidc_allowed_groups = env_list("OIDC_ALLOWED_GROUPS") self.oidc_allowed_users = env_list("OIDC_ALLOWED_USERS") self.oidc_button_label = env("OIDC_BUTTON_LABEL", "SSO") # Optional integrations self.loki_url = env("LOKI_URL").rstrip("/") self.loki_selector = env("LOKI_SELECTOR", '{job=~".+"}') self.loki_days = env_int("LOKI_DAYS", 7) self.loki_limit = env_int("LOKI_LIMIT", 50) self.loki_user = env("LOKI_USER") self.loki_password = env("LOKI_PASSWORD") self.loki_tenant = env("LOKI_TENANT") self.grafana_url = env("GRAFANA_URL").rstrip("/") self.grafana_dashboard = env("GRAFANA_DASHBOARD") self.grafana_loki_uid = env("GRAFANA_LOKI_UID", "loki") self.geoip_db = env("GEOIP_DB", f"{data}/geoip/dbip-city-lite.mmdb") self.geoip_auto_download = env_bool("GEOIP_AUTO_DOWNLOAD", False) self.hosts = {} def load_hosts(self): self.hosts = load_hosts(self.config_file, self.ssh_user) return self.hosts def validate_auth(self): modes = set(self.auth_modes) if not modes or modes - AUTH_MODES: raise ConfigError(f"AUTH_MODE must be local, oidc, local,oidc or proxy – got {','.join(self.auth_modes)!r}") if "proxy" in modes and len(modes) > 1: raise ConfigError("AUTH_MODE=proxy cannot be combined with other modes") if "local" in modes and not self.auth_password_hash: raise ConfigError("AUTH_MODE=local needs AUTH_PASSWORD_HASH – create one with: " "docker compose run --rm -it fail2ban-webui hash-password") if "oidc" in modes: missing = [n for n, v in (("OIDC_ISSUER or OIDC_DISCOVERY_URL", self.oidc_discovery_url), ("OIDC_CLIENT_ID", self.oidc_client_id), ("OIDC_CLIENT_SECRET", self.oidc_client_secret), ("PUBLIC_URL", self.public_url)) if not v] if missing: raise ConfigError("AUTH_MODE=oidc needs " + ", ".join(missing))