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

feat: scoped non-admin saves through POST /api/0/config

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:46:47 -04:00
co-authored by Claude Fable 5
parent 6b4524b30d
commit 07fefab861
1 changed files with 35 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hbd/server/http.py
+33 -4
View File
@@ -1288,12 +1288,16 @@ async def start(
return web.json_response({"backups": backups}) return web.json_response({"backups": backups})
async def api_config_post(request): async def api_config_post(request):
"""POST /api/0/config — publish staged changes to .hb.yaml. Admin only.""" """POST /api/0/config — publish staged changes to .hb.yaml.
Admins may write any section. Non-admins may submit only 'hosts' and
'thresholds'; their payload is merged into the config scoped to the
entries they own or manage (see hbd.server.config_access).
"""
user, err = _require_auth(request) user, err = _require_auth(request)
if err: if err:
return err return err
if user and not user.admin: is_admin = user is None or user.admin
return web.json_response({"error": "Forbidden"}, status=403)
if not _config_path: if not _config_path:
return web.json_response({"error": "Config path not available"}, status=503) return web.json_response({"error": "Config path not available"}, status=503)
try: try:
@@ -1304,6 +1308,13 @@ async def start(
if not isinstance(payload, dict): if not isinstance(payload, dict):
return web.json_response({"error": "Invalid JSON"}, status=400) return web.json_response({"error": "Invalid JSON"}, status=400)
if not is_admin:
extra = set(payload) - {"hosts", "thresholds"}
if extra:
return web.json_response(
{"error": f"Not permitted to edit: {', '.join(sorted(extra))}"},
status=403)
try: try:
data = configio_mod.read_roundtrip(_config_path) data = configio_mod.read_roundtrip(_config_path)
@@ -1352,18 +1363,36 @@ async def start(
if "thresholds" in payload: if "thresholds" in payload:
tc = payload["thresholds"] tc = payload["thresholds"]
if isinstance(tc, str): if isinstance(tc, str):
if not is_admin:
return web.json_response({"error": "Forbidden"}, status=403)
configio_mod.apply_yaml_section(data, "thresholds", tc) configio_mod.apply_yaml_section(data, "thresholds", tc)
elif isinstance(tc, dict): elif isinstance(tc, dict):
data["threshold_configs"] = _build_threshold_configs_from_form(tc) built = _build_threshold_configs_from_form(tc)
if is_admin:
data["threshold_configs"] = built
else:
data["threshold_configs"] = config_access.merge_threshold_configs_scoped(
data.get("threshold_configs") or {}, built, user.username)
if "hosts" in payload: if "hosts" in payload:
h = payload["hosts"] h = payload["hosts"]
if isinstance(h, dict): if isinstance(h, dict):
if is_admin:
configio_mod.apply_structured_section(data, "hosts", h) configio_mod.apply_structured_section(data, "hosts", h)
else: else:
merged = config_access.merge_hosts_scoped(
dict(data.get("hosts") or {}), h, user.username,
data.get("notification_channels") or {},
data.get("threshold_configs") or {})
configio_mod.apply_structured_section(data, "hosts", merged)
elif is_admin:
configio_mod.apply_yaml_section(data, "hosts", h) configio_mod.apply_yaml_section(data, "hosts", h)
else:
return web.json_response({"error": "Forbidden"}, status=403)
configio_mod.write_config(_config_path, data) configio_mod.write_config(_config_path, data)
except config_access.ScopedMergeError as exc:
return web.json_response({"error": str(exc)}, status=403)
except Exception as exc: except Exception as exc:
logger.error("Config write failed: %s", exc) logger.error("Config write failed: %s", exc)
return web.json_response({"error": str(exc)}, status=500) return web.json_response({"error": str(exc)}, status=500)