From 5c6f46208187203b4b0ce3bd7a2be3df897c17d3 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Thu, 9 Jul 2026 16:43:08 -0400 Subject: [PATCH] feat: scoped host merge for non-admin config saves Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NfPpSpccTWBfZg1FTveyaU --- hbd/server/config_access.py | 80 +++++++++++++++++++++ tests/test_config_access.py | 134 ++++++++++++++++++++++++++++++++++++ 2 files changed, 214 insertions(+) diff --git a/hbd/server/config_access.py b/hbd/server/config_access.py index 936e207..ef547b9 100644 --- a/hbd/server/config_access.py +++ b/hbd/server/config_access.py @@ -59,3 +59,83 @@ def user_threshold_configs(threshold_cfgs: Any, username: str) -> Dict[str, Any] name: cfg for name, cfg in (threshold_cfgs or {}).items() if isinstance(cfg, dict) and user_can_use(cfg, username) } + + +_HOST_SETTING_KEYS = ("watch", "dyndns", "notification_channels", "threshold_config") +_HOST_ACCESS_KEYS = ("owner", "managers", "monitors") +_HOST_ALLOWED_KEYS = frozenset(_HOST_SETTING_KEYS) | frozenset(_HOST_ACCESS_KEYS) + + +def _check_added_assignments(host_name: str, old_cfg: Any, new_cfg: dict, + key: str, usable: set) -> None: + """Every value under *key* not already on the host must be in *usable*.""" + old_vals = set(_as_list((old_cfg or {}).get(key))) + for v in _as_list(new_cfg.get(key)): + if v not in old_vals and v not in usable: + raise ScopedMergeError( + f"host {host_name!r}: {key} entry {v!r} is not available to you") + + +def merge_hosts_scoped(existing: Any, payload: Any, username: str, + channels_cfg: Any, threshold_cfgs: Any) -> Dict[str, Any]: + """Return a new hosts section with *username*'s visible subset replaced by *payload*. + + Hosts the user cannot see are preserved untouched. Visible hosts missing + from the payload are deleted (owners only). Managers may change setting + keys but not access keys and may not delete. New hosts get their owner + forced to *username*. Newly added channel/threshold assignments must be + global or owned by the user. Violations raise ScopedMergeError. + """ + existing = existing or {} + payload = payload or {} + visible = user_hosts(existing, username) + usable_channels = set(user_channels(channels_cfg, username)) + usable_tcs = set(user_threshold_configs(threshold_cfgs, username)) | {"default"} + + result: Dict[str, Any] = {n: c for n, c in existing.items() if n not in visible} + + for name, entry in payload.items(): + if not isinstance(entry, dict): + raise ScopedMergeError(f"host {name!r}: invalid entry") + unknown = set(entry) - _HOST_ALLOWED_KEYS + if unknown: + raise ScopedMergeError( + f"host {name!r}: fields not permitted: {', '.join(sorted(unknown))}") + if name in existing and name not in visible: + raise ScopedMergeError(f"host {name!r}: you are not an owner or manager") + + old = visible.get(name) + if old is None: + new_cfg = dict(entry) + new_cfg["owner"] = username + else: + new_cfg = dict(old) + for key in _HOST_SETTING_KEYS: + if key in entry: + new_cfg[key] = entry[key] + else: + new_cfg.pop(key, None) + if old.get("owner") == username: + for key in _HOST_ACCESS_KEYS: + if key in entry: + new_cfg[key] = entry[key] + else: + new_cfg.pop(key, None) + else: + if "owner" in entry and (entry.get("owner") or None) != (old.get("owner") or None): + raise ScopedMergeError( + f"host {name!r}: only the owner may change ownership") + for key in ("managers", "monitors"): + if key in entry and set(_as_list(entry[key])) != set(_as_list(old.get(key))): + raise ScopedMergeError( + f"host {name!r}: only the owner may change {key}") + + _check_added_assignments(name, old, new_cfg, "notification_channels", usable_channels) + _check_added_assignments(name, old, new_cfg, "threshold_config", usable_tcs) + result[name] = new_cfg + + for name, old in visible.items(): + if name not in payload and old.get("owner") != username: + raise ScopedMergeError(f"host {name!r}: only the owner may delete a host") + + return result diff --git a/tests/test_config_access.py b/tests/test_config_access.py index 7e4c968..33a8f44 100644 --- a/tests/test_config_access.py +++ b/tests/test_config_access.py @@ -56,3 +56,137 @@ TCS = { def test_user_threshold_configs_global_plus_own(): assert set(ca.user_threshold_configs(TCS, "alice")) == {"default", "alice_tc"} assert set(ca.user_threshold_configs(TCS, "bob")) == {"default"} + + +# --------------------------------------------------------------------------- +# merge_hosts_scoped +# --------------------------------------------------------------------------- + +MERGE_HOSTS = { + "mine": {"owner": "alice", "watch": True, "notification_channels": ["global_ch"]}, + "managed": {"owner": "bob", "managers": ["alice"], "watch": True, + "notification_channels": ["bob_ch"]}, + "foreign": {"owner": "bob", "watch": False}, +} +MERGE_CHANNELS = { + "global_ch": {"type": "pushover"}, + "alice_ch": {"type": "email", "owner": "alice"}, + "bob_ch": {"type": "signal", "owner": "bob"}, +} +MERGE_TCS = {"alice_tc": {"owner": "alice"}, "bob_tc": {"owner": "bob"}} + + +def _merge_hosts(payload, existing=None): + return ca.merge_hosts_scoped( + existing if existing is not None else dict(MERGE_HOSTS), + payload, "alice", MERGE_CHANNELS, MERGE_TCS, + ) + + +def _full_payload(**overrides): + """Payload covering alice's full visible subset, with per-host overrides.""" + p = { + "mine": {"owner": "alice", "watch": True, "notification_channels": ["global_ch"]}, + "managed": {"watch": True, "notification_channels": ["bob_ch"]}, + } + p.update(overrides) + return p + + +def test_merge_hosts_preserves_foreign_hosts(): + result = _merge_hosts(_full_payload()) + assert result["foreign"] == {"owner": "bob", "watch": False} + + +def test_merge_hosts_manager_edits_settings(): + result = _merge_hosts(_full_payload(managed={"watch": False, "dyndns": True, + "notification_channels": ["bob_ch"]})) + assert result["managed"]["watch"] is False + assert result["managed"]["dyndns"] is True + # access fields carried over untouched + assert result["managed"]["owner"] == "bob" + assert result["managed"]["managers"] == ["alice"] + + +def test_merge_hosts_manager_cannot_change_owner(): + with pytest.raises(ca.ScopedMergeError, match="managed"): + _merge_hosts(_full_payload(managed={"watch": True, "owner": "alice"})) + + +def test_merge_hosts_manager_cannot_change_managers(): + with pytest.raises(ca.ScopedMergeError, match="managers"): + _merge_hosts(_full_payload(managed={"watch": True, "managers": ["alice", "carol"]})) + + +def test_merge_hosts_manager_same_access_values_ok(): + # Submitting unchanged access fields is not a violation. + result = _merge_hosts(_full_payload(managed={"watch": True, "owner": "bob", + "managers": ["alice"], + "notification_channels": ["bob_ch"]})) + assert result["managed"]["owner"] == "bob" + + +def test_merge_hosts_manager_cannot_delete(): + payload = {"mine": {"owner": "alice", "watch": True}} # 'managed' missing + with pytest.raises(ca.ScopedMergeError, match="managed"): + _merge_hosts(payload) + + +def test_merge_hosts_owner_can_delete(): + payload = _full_payload() + del payload["mine"] + result = _merge_hosts(payload) + assert "mine" not in result + assert "managed" in result and "foreign" in result + + +def test_merge_hosts_owner_can_transfer_ownership(): + result = _merge_hosts(_full_payload(mine={"owner": "bob", "watch": True})) + assert result["mine"]["owner"] == "bob" + + +def test_merge_hosts_new_host_owner_forced(): + result = _merge_hosts(_full_payload(newhost={"watch": True, "owner": "bob"})) + assert result["newhost"]["owner"] == "alice" + + +def test_merge_hosts_cannot_touch_foreign_host(): + with pytest.raises(ca.ScopedMergeError, match="foreign"): + _merge_hosts(_full_payload(foreign={"watch": True})) + + +def test_merge_hosts_cannot_add_foreign_private_channel(): + with pytest.raises(ca.ScopedMergeError, match="bob_ch"): + _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True, + "notification_channels": ["bob_ch"]})) + + +def test_merge_hosts_keeps_preexisting_foreign_assignment(): + # 'managed' already has bob_ch; keeping it is fine. + result = _merge_hosts(_full_payload()) + assert result["managed"]["notification_channels"] == ["bob_ch"] + + +def test_merge_hosts_can_add_global_and_own_channel(): + result = _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True, + "notification_channels": ["global_ch", "alice_ch"]})) + assert result["mine"]["notification_channels"] == ["global_ch", "alice_ch"] + + +def test_merge_hosts_threshold_assignment_validated(): + with pytest.raises(ca.ScopedMergeError, match="bob_tc"): + _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True, + "threshold_config": ["bob_tc"]})) + result = _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True, + "threshold_config": ["default", "alice_tc"]})) + assert result["mine"]["threshold_config"] == ["default", "alice_tc"] + + +def test_merge_hosts_rejects_unknown_fields(): + with pytest.raises(ca.ScopedMergeError, match="sneaky"): + _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True, "sneaky": 1})) + + +def test_merge_hosts_owner_clearing_list_removes_key(): + result = _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True})) + assert "notification_channels" not in result["mine"]