"""Settings descriptor: maps config keys to display metadata.
``get_settings_sections(config)`` returns an ordered list of sections, each
containing a list of field descriptors. The template iterates this structure
generically, so adding editability later is a matter of:
1. Setting ``"editable": True`` on a field.
2. Adding the matching ```` in the template
(guided by ``"type"``).
3. Wiring a POST handler in http.py.
Field descriptor keys
---------------------
key str Config key (for future form POST matching)
label str Human-readable label
description str One-line help text shown below the value
value any Sanitized display value (secrets replaced with "•••")
type str One of: text | number | port | boolean | path | duration |
list | secret | size | select
editable bool Reserved for future use — currently always False
sensitive bool True when the raw value must never be shown
"""
# Credential field names that should always be masked.
_SECRET_KEYS = frozenset({
"password", "token", "user_key", "api_key", "secret",
"smtp_password", "smtp_user", "api_password", "access_token",
})
CHANNEL_TYPE_SCHEMAS = {
"pushover": {
"label": "Pushover",
"fields": [
{"key": "token", "label": "App token", "type": "secret", "required": True},
{"key": "user", "label": "User key", "type": "secret", "required": True},
{"key": "sound", "label": "Sound", "type": "text", "required": False},
],
},
"email": {
"label": "E-mail",
"fields": [
{"key": "recipients", "label": "Recipients (comma-separated)", "type": "list", "required": True},
{"key": "sender", "label": "From address", "type": "text", "required": True},
{"key": "smtp_server", "label": "SMTP server", "type": "text", "required": True},
{"key": "smtp_port", "label": "SMTP port", "type": "port", "required": False},
{"key": "smtp_user", "label": "SMTP username", "type": "text", "required": False},
{"key": "smtp_password", "label": "SMTP password", "type": "secret", "required": False},
],
},
"signal": {
"label": "Signal",
"fields": [
{"key": "user", "label": "Sender number", "type": "text", "required": True},
{"key": "recipient", "label": "Recipient number", "type": "text", "required": True},
{"key": "cli_path", "label": "signal-cli path", "type": "text", "required": False},
],
},
"matrix": {
"label": "Matrix",
"fields": [
{"key": "homeserver", "label": "Homeserver URL", "type": "text", "required": True},
{"key": "access_token", "label": "Access token", "type": "secret", "required": True},
{"key": "room_id", "label": "Room ID", "type": "text", "required": True},
],
},
"sms_voipms": {
"label": "SMS (voip.ms)",
"fields": [
{"key": "api_user", "label": "API username", "type": "text", "required": True},
{"key": "api_password", "label": "API password", "type": "secret", "required": True},
{"key": "did", "label": "DID (from)", "type": "text", "required": True},
{"key": "dst", "label": "Destination", "type": "text", "required": True},
],
},
"mattermost": {
"label": "Mattermost",
"fields": [
{"key": "host", "label": "Host", "type": "text", "required": True},
{"key": "token", "label": "Webhook token", "type": "secret", "required": True},
{"key": "channel", "label": "Channel", "type": "text", "required": True},
{"key": "username", "label": "Bot username", "type": "text", "required": False},
{"key": "icon", "label": "Icon URL", "type": "text", "required": False},
],
},
}
_CHANNEL_TYPE_LABELS = {k: v["label"] for k, v in CHANNEL_TYPE_SCHEMAS.items()}
def _mask(value):
"""Return a masked placeholder for sensitive values."""
if not value:
return ""
return "•••"
def _fmt_size(n):
"""Format a byte count as a human-readable string."""
try:
n = int(n)
except (TypeError, ValueError):
return str(n)
for unit in ("B", "KB", "MB", "GB"):
if n < 1024:
return f"{n} {unit}"
n //= 1024
return f"{n} TB"
def _fmt_duration(seconds):
"""Format seconds into a human-readable duration string."""
try:
s = int(seconds)
except (TypeError, ValueError):
return str(seconds)
if s < 60:
return f"{s}s"
if s < 3600:
m, sec = divmod(s, 60)
return f"{m}m {sec}s" if sec else f"{m}m"
h, rem = divmod(s, 3600)
m = rem // 60
return f"{h}h {m}m" if m else f"{h}h"
def _sanitize_channel(name, cfg):
"""Return a sanitized copy of a notification channel config."""
result = {}
for k, v in cfg.items():
if k in _SECRET_KEYS:
result[k] = _mask(v)
elif isinstance(v, list):
result[k] = v
else:
result[k] = v
return result
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def get_settings_sections(config: dict, threshold_checker=None) -> list:
"""Return ordered list of setting sections for the settings page.
Each section:
{
"title": str,
"description": str,
"fields": [ field_descriptor, ... ]
}
Each field_descriptor:
{
"key": str,
"label": str,
"description": str,
"value": display_value,
"raw": raw_config_value, # None for sensitive
"type": str,
"editable": bool,
"sensitive": bool,
}
"""
def field(key, label, ftype, description="", editable=False, sensitive=False):
raw = config.get(key)
if sensitive:
display = _mask(raw)
raw_out = None
elif ftype == "size":
display = _fmt_size(raw)
raw_out = raw
elif ftype == "duration":
display = _fmt_duration(raw)
raw_out = raw
elif ftype == "boolean":
display = bool(raw)
raw_out = raw
elif ftype == "list":
val = raw or []
display = list(val) if not isinstance(val, list) else val
raw_out = display
else:
display = raw if raw is not None else ""
raw_out = raw
return {
"key": key,
"label": label,
"description": description,
"value": display,
"raw": raw_out,
"type": ftype,
"editable": editable,
"sensitive": sensitive,
}
# ---- Notification channels (complex, built separately) ----------------
_METADATA_KEYS = {"type", "owner", "private", "min_level"}
notif_channels = []
for ch_name, ch_cfg in sorted((config.get("notification_channels") or {}).items()):
if not isinstance(ch_cfg, dict):
continue
ch_type = ch_cfg.get("type", "")
fields = []
for k, v in ch_cfg.items():
if k in _METADATA_KEYS:
continue
sensitive = k in _SECRET_KEYS
fields.append({
"key": k,
"label": k.replace("_", " ").title(),
"value": _mask(v) if sensitive else (
", ".join(v) if isinstance(v, list) else str(v)
),
"sensitive": sensitive,
})
notif_channels.append({
"name": ch_name,
"type": ch_type,
"type_label": _CHANNEL_TYPE_LABELS.get(ch_type, ch_type.title()),
"owner": ch_cfg.get("owner"),
"private": bool(ch_cfg.get("private", False)),
"min_level": ch_cfg.get("min_level", "WARNING"),
"fields": fields,
})
# ---- Users (show metadata only, never password hashes) ----------------
users_list = []
for username, attrs in (config.get("users") or {}).items():
if not isinstance(attrs, dict):
continue
users_list.append({
"username": username,
"full_name": attrs.get("full_name", ""),
"admin": bool(attrs.get("admin", False)),
"avatar": attrs.get("avatar", ""),
"notification_channels": attrs.get("notification_channels", []),
})
# ---- Threshold configurations -----------------------------------------
def _tc_to_row(tc):
return {
"metric": tc.metric_path,
"operator": tc.operator.value,
"warning": tc.warning,
"critical": tc.critical,
"hysteresis": tc.hysteresis,
"count": tc.count,
"enabled": tc.enabled,
"display": tc.display or "",
"grace": tc.grace,
}
threshold_config_list = []
if threshold_checker is not None:
if threshold_checker.threshold_configs:
for cfg_name, cfg_metrics in sorted(threshold_checker.threshold_configs.items()):
# For the default config use the merged effective set;
# for named overrides use only the explicitly defined metrics
# (threshold_raw_configs) so inherited defaults are not repeated.
if cfg_name == "default":
display_metrics = cfg_metrics
else:
display_metrics = threshold_checker.threshold_raw_configs.get(cfg_name, cfg_metrics)
metrics = sorted(
[_tc_to_row(tc) for tc in display_metrics.values()],
key=lambda m: m["metric"],
)
threshold_config_list.append({"name": cfg_name, "metrics": metrics})
elif threshold_checker.thresholds:
metrics = sorted(
[_tc_to_row(tc) for tc in threshold_checker.thresholds.values()],
key=lambda m: m["metric"],
)
threshold_config_list.append({"name": "default", "metrics": metrics})
# ---- Hosts summary ----------------------------------------------------
hosts_list = []
for hname, hcfg in sorted((config.get("hosts") or {}).items()):
if not isinstance(hcfg, dict):
continue
hosts_list.append({
"name": hname,
"watch": bool(hcfg.get("watch", True)),
"dyndns": bool(hcfg.get("dyndns", False)),
"owner": hcfg.get("owner", ""),
"managers": hcfg.get("managers", []),
"monitors": hcfg.get("monitors", []),
"threshold_configs": (
list(v) if isinstance(v := hcfg.get("threshold_config"), list)
else ([v] if v else [])
),
"notification_channels": hcfg.get("notification_channels", []),
})
# ---- OAuth providers -------------------------------------------------------
oauth_providers = []
for pname, pattrs in (config.get("oauth") or {}).items():
if not isinstance(pattrs, dict):
continue
cs = pattrs.get("client_secret", "")
oauth_providers.append({
"name": pname,
"type": pattrs.get("type", "gitea"),
"url": pattrs.get("url", ""),
"client_id": pattrs.get("client_id", ""),
"client_secret": "•••" if cs else "",
"label": pattrs.get("label", ""),
"logo": pattrs.get("logo", ""),
})
return [
{
"id": "network",
"title": "Network",
"description": "Ports and bind addresses for all server sockets.",
"section_mode": "form",
"api_section": "server",
"fields": [
field("hb_port", "Heartbeat UDP port", "port",
"UDP port the server listens on for heartbeat datagrams.", editable=True),
field("hbd_host", "HTTP bind address", "text",
"Interface to bind the HTTP server to. Empty = all interfaces.", editable=True),
field("hbd_port", "HTTP API port", "port",
"TCP port for the HTTP API and web UI.", editable=True),
field("ws_port", "WebSocket port", "port",
"TCP port for the plain WebSocket server.", editable=True),
field("wss_port", "Secure WebSocket port", "port",
"TCP port for WSS (TLS WebSocket). Leave empty to disable.", editable=True),
],
},
{
"id": "tls",
"title": "TLS / WebSocket Security",
"description": "Certificate paths used when wss_port is set.",
"section_mode": "form",
"api_section": None,
"fields": [
field("cert_path", "Certificate directory", "path",
"Directory containing the TLS certificate and key files."),
field("wss_pem", "Certificate file", "text",
"Filename of the TLS certificate chain (PEM format)."),
field("wss_key", "Key file", "text",
"Filename of the TLS private key (PEM format)."),
],
},
{
"id": "monitoring",
"title": "Monitoring",
"description": "Heartbeat timing and alert re-notification behaviour.",
"section_mode": "form",
"api_section": "server",
"fields": [
field("interval", "Heartbeat interval", "duration",
"Expected time between heartbeat messages from each client.", editable=True),
field("grace", "Grace period", "number",
"Extra seconds to wait after a missed heartbeat before sending notifications.", editable=True),
field("threshold_renotify_interval", "Re-notify interval", "duration",
"How often to re-send notifications for ongoing threshold alerts.", editable=True),
field("autosave_interval", "Autosave interval", "duration",
"How often the server saves its state to disk."),
field("base_url", "Base URL", "text",
"Base URL for notification links.", editable=True),
],
},
{
"id": "persistence",
"title": "Persistence & Logging",
"description": "State file and event log settings.",
"section_mode": "form",
"api_section": "server",
"fields": [
field("pickfile", "State file", "path",
"Path to the pickle file used to persist host state across restarts.", editable=True),
field("logfile", "Event log", "path",
"Path to the event log file.", editable=True),
],
},
{
"id": "journal",
"title": "Message Journal",
"description": "All received heartbeat and plugin messages are journalled here.",
"section_mode": "form",
"api_section": "server",
"fields": [
field("journal_enabled", "Enabled", "boolean",
"Turn journalling on or off.", editable=True),
field("journal_dir", "Journal directory","path",
"Directory where journal files are written.", editable=True),
field("journal_file", "Journal filename", "text",
"Base filename for the journal (rotated copies get a numeric suffix)."),
field("journal_max_size", "Max file size", "size",
"Rotate the journal when it exceeds this size.", editable=True),
field("journal_max_backups", "Backup count", "number",
"Number of rotated journal files to keep.", editable=True),
],
},
{
"id": "dns",
"title": "Dynamic DNS",
"description": "nsupdate-based DNS registration via nsupdate(8).",
"section_mode": "form",
"api_section": "dns",
"fields": [
field("nsupdate_bin", "nsupdate binary", "path",
"Path to the nsupdate binary.", editable=True),
field("rndc_key", "RNDC key file", "path",
"Path to the rndc key file used to authenticate DNS updates.", editable=True),
field("dyndomains", "Dynamic domains", "list",
"Domains updated via nsupdate when a host with dyndns: true reports in.",
editable=True),
],
},
{
"id": "users",
"title": "Users",
"description": "Accounts defined in the config file. Password hashes are never shown.",
"section_mode": "form",
"api_section": "users",
"users": users_list,
"fields": [
field("default_owner", "Default owner", "text",
"Username that owns hosts with no explicit owner. "
"Falls back to the first admin user.", editable=True),
],
},
{
"id": "oauth",
"title": "OAuth Providers",
"description": "OAuth2 login providers. Client secrets are masked.",
"section_mode": "form",
"api_section": "oauth",
"providers": oauth_providers,
"fields": [],
},
{
"id": "channels",
"title": "Notification Channels",
"description": "Named notification providers. Credentials are masked.",
"section_mode": "channels",
"api_section": "notification_channels",
"channels": notif_channels,
"fields": [
field("default_notification_channels", "Default channels", "list",
"Channels used when a host does not specify its own."),
],
},
{
"id": "hosts",
"title": "Hosts",
"description": "Host definitions loaded from the config file.",
"section_mode": "hosts",
"api_section": "hosts",
"hosts": hosts_list,
"fields": [],
},
{
"id": "thresholds",
"title": "Threshold Configurations",
"description": "Named alert threshold sets. Each defines warning/critical levels per metric.",
"section_mode": "thresholds",
"api_section": "thresholds",
"threshold_configs": threshold_config_list,
"fields": [
field("default_threshold_config", "Default config", "text",
"Threshold config used for hosts with no explicit mapping.", editable=True),
],
},
{
"id": "runtime",
"title": "Runtime",
"description": "Flags set at startup (require restart to change).",
"section_mode": "form",
"api_section": None,
"fields": [
field("foreground", "Foreground mode", "boolean",
"Run in the foreground instead of daemonising."),
field("verbose", "Verbose logging", "boolean",
"Enable verbose log output."),
field("debug", "Debug level", "number",
"0 = off. Higher values increase log verbosity."),
],
},
]
def get_settings_data(config: dict, threshold_checker=None) -> dict:
"""Return sections list + auxiliary data for the settings template."""
sections = get_settings_sections(config, threshold_checker=threshold_checker)
all_channel_names = sorted((config.get("notification_channels") or {}).keys())
all_usernames = sorted((config.get("users") or {}).keys())
all_threshold_configs = sorted((config.get("threshold_configs") or {}).keys())
return {
"sections": sections,
"all_channel_names": all_channel_names,
"all_usernames": all_usernames,
"all_threshold_configs": all_threshold_configs,
}