heartbeat/hbd/server/config_access.py

"""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)
    }


_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


def merge_threshold_configs_scoped(existing: Any, payload: Any,
                                   username: str) -> Dict[str, Any]:
    """Return a new threshold_configs section with the user's own configs
    replaced by *payload*.

    Global and foreign-owned configs are preserved and may not appear in the
    payload ('default' included).  Own configs missing from the payload are
    deleted.  Every payload entry gets its owner forced to *username*.
    """
    existing = existing or {}
    payload = payload or {}
    result: Dict[str, Any] = {
        n: c for n, c in existing.items()
        if not (isinstance(c, dict) and c.get("owner") == username)
    }
    for name, entry in payload.items():
        if name == "default":
            raise ScopedMergeError("threshold config 'default' is global and admin-managed")
        old = existing.get(name)
        if old is not None and (not isinstance(old, dict) or old.get("owner") != username):
            raise ScopedMergeError(f"threshold config {name!r}: not owned by you")
        new_cfg = dict(entry) if isinstance(entry, dict) else {}
        new_cfg["owner"] = username
        result[name] = new_cfg
    return result