From 39c4d45bd791e400535a1adfb3c75cc1cef5bd70 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Thu, 9 Jul 2026 16:48:58 -0400 Subject: [PATCH] feat: per-user filtering of settings sections Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NfPpSpccTWBfZg1FTveyaU --- hbd/server/settings.py | 67 +++++++++++++++++----- tests/test_settings_sections.py | 99 +++++++++++++++++++++++++++++---- 2 files changed, 142 insertions(+), 24 deletions(-) diff --git a/hbd/server/settings.py b/hbd/server/settings.py index 50ae402..1e2541a 100644 --- a/hbd/server/settings.py +++ b/hbd/server/settings.py @@ -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()), } diff --git a/tests/test_settings_sections.py b/tests/test_settings_sections.py index e0800a0..59b4fea 100644 --- a/tests/test_settings_sections.py +++ b/tests/test_settings_sections.py @@ -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