andreas/heartbeat
Public Access
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

feat: per-user filtering of settings sections

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NfPpSpccTWBfZg1FTveyaU
This commit is contained in:
andreas
2026-07-09 16:48:58 -04:00
co-authored by Claude Fable 5
parent 07fefab861
commit 39c4d45bd7
2 changed files with 142 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hbd/server/settings.py
+53 -14
View File
@@ -21,6 +21,8 @@ editable bool Reserved for future use — currently always False
sensitive bool True when the raw value must never be shown
"""
from . import config_access
# Credential field names that should always be masked.
_SECRET_KEYS = frozenset({
"password", "token", "user_key", "api_key", "secret",
@@ -140,9 +142,14 @@ def _sanitize_channel(name, cfg):
# Public API
# ---------------------------------------------------------------------------
def get_settings_sections(config: dict, threshold_checker=None) -> list:
def get_settings_sections(config: dict, threshold_checker=None, user=None) -> list:
"""Return ordered list of setting sections for the settings page.
*user* is an object with ``username``/``admin`` attributes, or None for
the unauthenticated (admin-equivalent) view. Non-admins get only the
channels/hosts/thresholds sections, filtered to global items plus what
they own or manage.
Each section:
{
"title": str,
@@ -162,6 +169,9 @@ def get_settings_sections(config: dict, threshold_checker=None) -> list:
"sensitive": bool,
}
"""
is_admin = user is None or getattr(user, "admin", False)
username = getattr(user, "username", None)
def field(key, label, ftype, description="", editable=False, sensitive=False):
raw = config.get(key)
if sensitive:
@@ -200,6 +210,8 @@ def get_settings_sections(config: dict, threshold_checker=None) -> list:
for ch_name, ch_cfg in sorted((config.get("notification_channels") or {}).items()):
if not isinstance(ch_cfg, dict):
continue
if not is_admin and not config_access.user_can_use(ch_cfg, username):
continue
ch_type = ch_cfg.get("type", "")
fields = []
for k, v in ch_cfg.items():
@@ -219,7 +231,7 @@ def get_settings_sections(config: dict, threshold_checker=None) -> list:
"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)),
"editable": is_admin or ch_cfg.get("owner") == username,
"min_level": ch_cfg.get("min_level", "WARNING"),
"fields": fields,
})
@@ -252,9 +264,14 @@ def get_settings_sections(config: dict, threshold_checker=None) -> list:
}
threshold_config_list = []
raw_threshold_cfgs = config.get("threshold_configs") or {}
if threshold_checker is not None:
if threshold_checker.threshold_configs:
for cfg_name, cfg_metrics in sorted(threshold_checker.threshold_configs.items()):
raw_cfg = raw_threshold_cfgs.get(cfg_name)
tc_owner = raw_cfg.get("owner") if isinstance(raw_cfg, dict) else None
if not is_admin and tc_owner and tc_owner != username:
continue
# 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.
@@ -266,25 +283,37 @@ def get_settings_sections(config: dict, threshold_checker=None) -> list:
[_tc_to_row(tc) for tc in display_metrics.values()],
key=lambda m: m["metric"],
)
threshold_config_list.append({"name": cfg_name, "metrics": metrics})
threshold_config_list.append({
"name": cfg_name,
"metrics": metrics,
"owner": tc_owner,
"editable": is_admin or (tc_owner is not None and tc_owner == username),
})
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})
threshold_config_list.append({"name": "default", "metrics": metrics,
"owner": None, "editable": is_admin})
# ---- Hosts summary ----------------------------------------------------
hosts_list = []
for hname, hcfg in sorted((config.get("hosts") or {}).items()):
if not isinstance(hcfg, dict):
continue
managers = hcfg.get("managers", [])
if isinstance(managers, str):
managers = [managers]
if not is_admin and hcfg.get("owner") != username and username not in managers:
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", []),
"is_owner": is_admin or hcfg.get("owner") == username,
"managers": managers,
"monitors": hcfg.get("monitors", []),
"threshold_configs": (
list(v) if isinstance(v := hcfg.get("threshold_config"), list)
@@ -309,7 +338,7 @@ def get_settings_sections(config: dict, threshold_checker=None) -> list:
"logo": pattrs.get("logo", ""),
})
return [
sections = [
{
"id": "network",
"title": "Network",
@@ -483,16 +512,26 @@ def get_settings_sections(config: dict, threshold_checker=None) -> list:
},
]
if not is_admin:
sections = [s for s in sections if s["id"] in ("channels", "hosts", "thresholds")]
for s in sections:
s["fields"] = []
return sections
def get_settings_data(config: dict, threshold_checker=None) -> dict:
def get_settings_data(config: dict, threshold_checker=None, user=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())
sections = get_settings_sections(config, threshold_checker=threshold_checker, user=user)
is_admin = user is None or getattr(user, "admin", False)
username = getattr(user, "username", None)
channels = config.get("notification_channels") or {}
threshold_cfgs = config.get("threshold_configs") or {}
if not is_admin:
channels = config_access.user_channels(channels, username)
threshold_cfgs = config_access.user_threshold_configs(threshold_cfgs, username)
return {
"sections": sections,
"all_channel_names": all_channel_names,
"all_usernames": all_usernames,
"all_threshold_configs": all_threshold_configs,
"all_channel_names": sorted(channels.keys()),
"all_usernames": sorted((config.get("users") or {}).keys()),
"all_threshold_configs": sorted(threshold_cfgs.keys()),
}
tests/test_settings_sections.py
+89 -10
View File
@@ -24,7 +24,7 @@ def test_sections_have_section_mode():
sections = settings_mod.get_settings_sections(CFG)
for s in sections:
assert "section_mode" in s, f"Section {s['id']} missing section_mode"
assert s["section_mode"] in ("form", "yaml", "channels", "hosts")
assert s["section_mode"] in ("form", "yaml", "channels", "hosts", "thresholds")
def test_sections_have_api_section():
@@ -42,15 +42,13 @@ def test_network_section_has_editable_fields():
assert len(editable) >= 2 # hbd_port, ws_port at minimum
def test_yaml_sections_have_correct_mode():
def test_thresholds_and_dns_section_modes():
sections = settings_mod.get_settings_sections(CFG)
yaml_sections = {s["id"]: s for s in sections if s["section_mode"] == "yaml"}
assert "channels" not in yaml_sections # now uses "channels" mode
assert "hosts" not in yaml_sections # now uses "hosts" mode
assert "thresholds" in yaml_sections
assert "dns" in yaml_sections
assert yaml_sections["thresholds"]["api_section"] == "thresholds"
assert yaml_sections["dns"]["api_section"] == "dns"
by_id = {s["id"]: s for s in sections}
assert by_id["thresholds"]["section_mode"] == "thresholds"
assert by_id["thresholds"]["api_section"] == "thresholds"
assert by_id["dns"]["section_mode"] == "form"
assert by_id["dns"]["api_section"] == "dns"
def test_hosts_section_uses_hosts_mode():
@@ -70,7 +68,7 @@ def test_channels_section_uses_channels_mode():
assert ch["name"] == "pushover_ops"
assert ch["type"] == "pushover"
assert "owner" in ch
assert "private" in ch
assert "editable" in ch
def test_channel_type_schemas_exported():
@@ -112,3 +110,84 @@ def test_users_section_has_user_list():
assert users_sec["users"][0]["username"] == "alice"
# Password hash never exposed
assert "password" not in users_sec["users"][0]
# ---------------------------------------------------------------------------
# Per-user filtering (owners/managers on the settings page)
# ---------------------------------------------------------------------------
from types import SimpleNamespace # noqa: E402
MULTI_CFG = {
**CFG,
"users": {
"alice": {"full_name": "Alice", "admin": True, "password": "x"},
"bob": {"full_name": "Bob", "admin": False, "password": "x"},
},
"notification_channels": {
"global_ch": {"type": "pushover", "token": "t", "user": "u"},
"bob_ch": {"type": "pushover", "token": "t", "user": "u", "owner": "bob"},
"carol_ch": {"type": "pushover", "token": "t", "user": "u", "owner": "carol"},
},
"threshold_configs": {
"bob_tc": {"owner": "bob", "thresholds": {}},
"carol_tc": {"owner": "carol", "thresholds": {}},
},
"hosts": {
"bobhost": {"owner": "bob"},
"managedhost": {"owner": "carol", "managers": ["bob"]},
"carolhost": {"owner": "carol"},
},
}
BOB = SimpleNamespace(username="bob", admin=False)
ADMIN = SimpleNamespace(username="alice", admin=True)
def test_nonadmin_sees_only_three_sections():
sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
assert [s["id"] for s in sections] == ["channels", "hosts", "thresholds"]
def test_nonadmin_sections_hide_admin_fields():
sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
for s in sections:
assert s["fields"] == []
def test_nonadmin_hosts_filtered_with_is_owner():
sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
hosts = next(s for s in sections if s["id"] == "hosts")["hosts"]
by_name = {h["name"]: h for h in hosts}
assert set(by_name) == {"bobhost", "managedhost"}
assert by_name["bobhost"]["is_owner"] is True
assert by_name["managedhost"]["is_owner"] is False
def test_nonadmin_channels_filtered_with_editable():
sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
chans = {c["name"]: c for c in next(s for s in sections if s["id"] == "channels")["channels"]}
assert set(chans) == {"global_ch", "bob_ch"}
assert chans["bob_ch"]["editable"] is True
assert chans["global_ch"]["editable"] is False
def test_admin_sees_everything_with_editable():
sections = settings_mod.get_settings_sections(MULTI_CFG, user=ADMIN)
ids = [s["id"] for s in sections]
assert "network" in ids and "users" in ids
chans = {c["name"]: c for c in next(s for s in sections if s["id"] == "channels")["channels"]}
assert set(chans) == {"global_ch", "bob_ch", "carol_ch"}
assert all(c["editable"] for c in chans.values())
def test_settings_data_pickers_filtered_for_nonadmin():
data = settings_mod.get_settings_data(MULTI_CFG, user=BOB)
assert data["all_channel_names"] == ["bob_ch", "global_ch"]
assert data["all_threshold_configs"] == ["bob_tc"]
def test_no_user_means_admin_view():
sections = settings_mod.get_settings_sections(MULTI_CFG) # auth disabled
assert len(sections) > 3