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

feat: channel ownership rule — owner-presence means private, admin promote/demote

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:45:21 -04:00
co-authored by Claude Fable 5
parent dd939ab86e
commit a149a1e285
2 changed files with 31 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hbd/server/http.py
+14 -16
View File
@@ -20,6 +20,7 @@ from . import users as users_mod
from . import oauth as oauth_mod
from . import ws as ws_mod
from . import configio as configio_mod
from . import config_access
logger = logging.getLogger(__name__)
@@ -1407,19 +1408,13 @@ async def start(
# -------------------------------------------------------------------------
def _visible_channels_for_user(user):
"""Return {name: cfg} of channels visible to user (public + own private)."""
"""Return {name: cfg} of channels visible to user (global + own)."""
all_channels = config.get("notification_channels") or {}
if user is None:
return {}
if user.admin:
return dict(all_channels)
visible = {}
for name, cfg in all_channels.items():
if not isinstance(cfg, dict):
continue
if not cfg.get("private") or cfg.get("owner") == user.username:
visible[name] = cfg
return visible
return config_access.user_channels(all_channels, user.username)
def _build_channel_response(ch_name, ch_cfg):
"""Serialize a channel config dict for the API response."""
@@ -1443,7 +1438,7 @@ async def start(
"type": ch_type,
"type_label": settings_mod._CHANNEL_TYPE_LABELS.get(ch_type, ch_type.title()),
"owner": ch_cfg.get("owner"),
"private": bool(ch_cfg.get("private", False)),
"private": not config_access.is_global(ch_cfg),
"min_level": ch_cfg.get("min_level", "WARNING"),
"fields": fields,
}
@@ -1508,9 +1503,12 @@ async def start(
if body.get("min_level"):
channel_cfg["min_level"] = body["min_level"]
if user.admin:
owner = (body.get("owner") or "").strip()
if owner:
channel_cfg["owner"] = owner
else:
channel_cfg["owner"] = user.username
if body.get("private"):
channel_cfg["private"] = True
try:
disk_data = configio_mod.read_roundtrip(_config_path)
@@ -1575,12 +1573,12 @@ async def start(
if body.get("min_level"):
channel_cfg["min_level"] = body["min_level"]
if owner is not None:
if user.admin:
new_owner = (body.get("owner") or "").strip() if "owner" in body else (owner or "")
if new_owner:
channel_cfg["owner"] = new_owner
elif owner is not None:
channel_cfg["owner"] = owner
if "private" in body:
channel_cfg["private"] = bool(body["private"])
elif existing_on_disk.get("private"):
channel_cfg["private"] = True
configio_mod.apply_channel(disk_data, ch_name, channel_cfg)
configio_mod.write_config(_config_path, disk_data)
tests/test_notification_channels_api.py
+15 -23
View File
@@ -86,61 +86,53 @@ def test_delete_channel_persisted_after_write(tmp_path):
# ---------------------------------------------------------------------------
# Visibility logic (mirrors http.py _visible_channels_for_user)
# Visibility logic (owner-presence rule, hbd.server.config_access)
# ---------------------------------------------------------------------------
from hbd.server import config_access as ca # noqa: E402
def _visible(config, user):
"""Local copy of the visibility helper for unit testing without the HTTP layer."""
all_channels = config.get("notification_channels") or {}
if user.get("admin"):
return set(all_channels.keys())
username = user["username"]
return {
name for name, cfg in all_channels.items()
if isinstance(cfg, dict) and (not cfg.get("private") or cfg.get("owner") == username)
}
return set(ca.user_channels(all_channels, user["username"]))
CONFIG_VISIBILITY = {
"notification_channels": {
"pub_ch": {"type": "pushover", "token": "t", "user": "u"},
"alice_priv": {"type": "email", "owner": "alice", "private": True,
"alice_priv": {"type": "email", "owner": "alice",
"recipients": ["a@a.com"], "sender": "s@a.com", "smtp_server": "s"},
"bob_priv": {"type": "signal", "owner": "bob", "private": True,
"user": "+1", "recipient": "+2"},
"admin_owned": {"type": "pushover", "token": "t2", "user": "u2", "owner": "adminuser"},
"bob_priv": {"type": "signal", "owner": "bob", "user": "+1", "recipient": "+2"},
"stale_flag": {"type": "pushover", "token": "t2", "user": "u2", "private": True},
}
}
def test_public_channel_visible_to_all():
def test_global_channel_visible_to_all():
for uname in ("alice", "bob", "carol"):
user = {"username": uname, "admin": False}
assert "pub_ch" in _visible(CONFIG_VISIBILITY, user)
assert "pub_ch" in _visible(CONFIG_VISIBILITY, {"username": uname, "admin": False})
def test_private_channel_visible_only_to_owner():
def test_owned_channel_visible_only_to_owner():
alice = {"username": "alice", "admin": False}
bob = {"username": "bob", "admin": False}
carol = {"username": "carol", "admin": False}
assert "alice_priv" in _visible(CONFIG_VISIBILITY, alice)
assert "alice_priv" not in _visible(CONFIG_VISIBILITY, bob)
assert "alice_priv" not in _visible(CONFIG_VISIBILITY, carol)
assert "bob_priv" in _visible(CONFIG_VISIBILITY, bob)
assert "bob_priv" not in _visible(CONFIG_VISIBILITY, alice)
def test_admin_sees_all_channels():
admin = {"username": "adminuser", "admin": True}
visible = _visible(CONFIG_VISIBILITY, admin)
assert visible == {"pub_ch", "alice_priv", "bob_priv", "admin_owned"}
assert _visible(CONFIG_VISIBILITY, admin) == {"pub_ch", "alice_priv", "bob_priv", "stale_flag"}
def test_admin_owned_channel_is_public_by_default():
def test_stale_private_flag_without_owner_is_global():
"""Owner-presence is the single signal; a leftover private flag is ignored."""
alice = {"username": "alice", "admin": False}
assert "admin_owned" in _visible(CONFIG_VISIBILITY, alice)
assert "stale_flag" in _visible(CONFIG_VISIBILITY, alice)
# ---------------------------------------------------------------------------