feat: ownership visibility helpers for config entities

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NfPpSpccTWBfZg1FTveyaU
This commit is contained in:
2026-07-09 16:42:00 -04:00
co-authored by Claude Fable 5
parent 960822918a
commit 98b50dbbf4
2 changed files with 119 additions and 0 deletions
+61
View File
@@ -0,0 +1,61 @@
"""Ownership rules for config-file entities (hosts, channels, threshold configs).
Rule: an entry whose config dict has a non-empty ``owner`` is private to that
owner — visible, usable, and editable only by the owner (and admins). An
entry with no ``owner`` is global: usable by everyone, editable by admins only.
The scoped-merge helpers implement the non-admin save path for
``POST /api/0/config``: the caller's visible subset of a section is replaced
by the submitted payload; everything else is preserved untouched.
"""
from typing import Any, Dict
class ScopedMergeError(Exception):
"""A non-admin payload violated an ownership rule; message names the entry."""
def is_global(cfg: Any) -> bool:
"""True when *cfg* has no owner (usable by everyone)."""
return not (isinstance(cfg, dict) and cfg.get("owner"))
def user_can_use(cfg: Any, username: str) -> bool:
"""True when *username* may use/see this entry (global or own)."""
return is_global(cfg) or cfg.get("owner") == username
def _as_list(value: Any) -> list:
if value is None:
return []
if isinstance(value, str):
return [value]
return list(value)
def user_hosts(hosts_cfg: Any, username: str) -> Dict[str, Any]:
"""Subset of *hosts_cfg* where *username* is owner or manager."""
result = {}
for name, cfg in (hosts_cfg or {}).items():
if not isinstance(cfg, dict):
continue
if cfg.get("owner") == username or username in _as_list(cfg.get("managers")):
result[name] = cfg
return result
def user_channels(channels_cfg: Any, username: str) -> Dict[str, Any]:
"""Subset of channels usable by *username*: global + own."""
return {
name: cfg for name, cfg in (channels_cfg or {}).items()
if isinstance(cfg, dict) and user_can_use(cfg, username)
}
def user_threshold_configs(threshold_cfgs: Any, username: str) -> Dict[str, Any]:
"""Subset of threshold configs usable by *username*: global + own."""
return {
name: cfg for name, cfg in (threshold_cfgs or {}).items()
if isinstance(cfg, dict) and user_can_use(cfg, username)
}