# Config Editor Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Allow admins to edit the full `.hb.yaml` config through the Settings page UI, and allow regular users to manage their own notification channels and profile fields through the Profile page. The YAML file remains the single authoritative source; comments are preserved on every write. **Architecture:** A new `configio.py` module handles all YAML I/O using `ruamel.yaml` for comment-preserving round-trip reads and atomic writes with backup rotation. New HTTP endpoints (`GET/POST /api/0/config`, `PUT /api/0/users/me`) serve the browser. The Settings page accumulates staged edits in JS state (one dict per logical section) and writes them all in a single `POST /api/0/config`; the Profile page writes immediately via `PUT /api/0/users/me`. **Tech Stack:** `ruamel.yaml>=0.18` (new), `aiohttp` + `jinja2` + `pytest` (existing), vanilla JS in templates. **Spec:** `docs/superpowers/specs/2026-05-09-config-editor-design.md` --- ## File Map | Action | File | Responsibility | |--------|------|----------------| | Create | `hbd/server/configio.py` | All `.hb.yaml` I/O: read_roundtrip, write_config, list_backups, apply_structured_section, apply_yaml_section | | Create | `tests/test_configio.py` | Unit tests for configio | | Modify | `pyproject.toml` | Add `ruamel.yaml>=0.18` to server deps | | Modify | `hbd/server/http.py` | Add config API handlers, PUT /api/0/users/me, update profile_page, update route table | | Modify | `hbd/server/settings.py` | Add `section_mode`, `api_section`, `editable` flags; add oauth section; pass `all_channel_names` | | Modify | `hbd/server/templates/settings.html` | Form inputs, YAML editors, Stage/Publish/Discard JS, pending banner, rollback modal | | Modify | `hbd/server/templates/profile.html` | Identity form, change-password form, notification channels checkboxes | --- ## Task 1: `configio.py` — YAML I/O module + ruamel.yaml dependency **Files:** - Create: `hbd/server/configio.py` - Create: `tests/test_configio.py` - Modify: `pyproject.toml` - [ ] **Step 1: Add ruamel.yaml to pyproject.toml** In `pyproject.toml`, find the `[project.optional-dependencies]` `server` list and add the new dep: ```toml server = [ "websockets>=13.2", "mattermostdriver>=7.3.0", "aiohttp>=3.11", "Jinja2>=3.1.6", "matrix-nio>=0.24", "ruamel.yaml>=0.18", ] ``` - [ ] **Step 2: Write the failing tests** Create `tests/test_configio.py`: ```python import glob import os import pytest from hbd.server import configio SAMPLE_YAML = """\ # Server configuration hbd_port: 50004 # HTTP API port interval: 20 users: alice: full_name: Alice Smith admin: true notification_channels: pushover_ops: type: pushover token: abc123 """ def test_read_roundtrip_loads_values(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) assert data["hbd_port"] == 50004 assert data["interval"] == 20 assert data["users"]["alice"]["full_name"] == "Alice Smith" def test_write_config_creates_backup(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) data["interval"] = 30 configio.write_config(str(f), data) backups = configio.list_backups(str(f)) assert len(backups) == 1 assert ".bak." in backups[0] def test_write_config_preserves_comments(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) data["interval"] = 30 configio.write_config(str(f), data) content = f.read_text() assert "# Server configuration" in content assert "# HTTP API port" in content def test_write_config_atomically_replaces_file(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) data["interval"] = 99 configio.write_config(str(f), data) assert not (tmp_path / ".hb.yaml.tmp").exists() data2 = configio.read_roundtrip(str(f)) assert data2["interval"] == 99 def test_write_config_backup_rotation(tmp_path): cfg = tmp_path / ".hb.yaml" cfg.write_text(SAMPLE_YAML) # Pre-create 10 existing backups with old timestamps for i in range(10): (tmp_path / f".hb.yaml.bak.20260101-{i:06d}").write_text("old") data = configio.read_roundtrip(str(cfg)) configio.write_config(str(cfg), data) backups = configio.list_backups(str(cfg)) assert len(backups) == 10 assert not (tmp_path / ".hb.yaml.bak.20260101-000000").exists() def test_list_backups_newest_first(tmp_path): cfg = tmp_path / ".hb.yaml" cfg.write_text(SAMPLE_YAML) for i in range(3): (tmp_path / f".hb.yaml.bak.20260101-{i:02d}0000").write_text("b") backups = configio.list_backups(str(cfg)) assert len(backups) == 3 assert backups == sorted(backups, reverse=True) def test_apply_structured_section_server_updates_keys(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) configio.apply_structured_section(data, "server", {"interval": 60, "hbd_port": 8080}) assert data["interval"] == 60 assert data["hbd_port"] == 8080 def test_apply_structured_section_server_ignores_unknown_keys(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) configio.apply_structured_section(data, "server", {"interval": 60, "not_a_key": "x"}) assert "not_a_key" not in data def test_apply_structured_section_users_replaces_dict(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) new_users = {"bob": {"full_name": "Bob Jones", "admin": False}} configio.apply_structured_section(data, "users", new_users) assert "alice" not in data["users"] assert data["users"]["bob"]["full_name"] == "Bob Jones" def test_apply_yaml_section_notification_channels(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) new_yaml = "email_ops:\n type: email\n recipients: [ops@example.com]\n" configio.apply_yaml_section(data, "notification_channels", new_yaml) assert "email_ops" in data["notification_channels"] assert "pushover_ops" not in data["notification_channels"] def test_apply_yaml_section_thresholds_maps_to_threshold_configs(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) configio.apply_yaml_section(data, "thresholds", "default:\n cpu: 80\n") assert "threshold_configs" in data assert data["threshold_configs"]["default"]["cpu"] == 80 def test_apply_yaml_section_dns_replaces_each_key(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) configio.apply_yaml_section( data, "dns", "nsupdate_bin: /usr/bin/nsupdate\ndyndomains: [dyn.example.com]\n" ) assert data["nsupdate_bin"] == "/usr/bin/nsupdate" assert data["dyndomains"] == ["dyn.example.com"] def test_apply_yaml_section_unknown_raises(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) with pytest.raises(ValueError, match="Unknown YAML section"): configio.apply_yaml_section(data, "nope", "x: 1\n") def test_apply_structured_section_unknown_raises(tmp_path): f = tmp_path / ".hb.yaml" f.write_text(SAMPLE_YAML) data = configio.read_roundtrip(str(f)) with pytest.raises(ValueError, match="Unknown structured section"): configio.apply_structured_section(data, "nope", {"x": 1}) ``` - [ ] **Step 3: Run tests to confirm they fail** ``` pytest tests/test_configio.py -v ``` Expected: ImportError or ModuleNotFoundError — `configio` does not exist yet. - [ ] **Step 4: Install ruamel.yaml** ```bash pip install "ruamel.yaml>=0.18" ``` - [ ] **Step 5: Implement `hbd/server/configio.py`** ```python """YAML round-trip read/write for .hb.yaml, with backup and atomic writes.""" import glob import os import threading from datetime import datetime from ruamel.yaml import YAML _write_lock = threading.Lock() _yaml = YAML() _yaml.preserve_quotes = True # Top-level keys managed by the 'server' logical section _SERVER_KEYS = [ "hbd_port", "hbd_host", "ws_port", "wss_port", "hb_port", "interval", "grace", "base_url", "threshold_renotify_interval", "logfile", "pidfile", "pickfile", "journal_enabled", "journal_dir", "journal_max_size", "journal_max_backups", "default_owner", ] # Top-level keys managed by the 'dns' logical section _DNS_KEYS = ["nsupdate_bin", "dyndomains", "dyndnshosts", "drophosts"] def read_roundtrip(path: str): """Load .hb.yaml with ruamel.yaml, preserving comments and ordering.""" with open(path, "r", encoding="utf-8") as f: return _yaml.load(f) def write_config(path: str, data) -> None: """Backup current file then atomically write data. Backup naming: {path}.bak.YYYYMMDD-HHMMSS Rotation: keep the 10 most recent backups, delete older ones. Atomic write: write to {path}.tmp then os.replace({path}.tmp, path). Acquires _write_lock for the full backup+write sequence. """ with _write_lock: ts = datetime.now().strftime("%Y%m%d-%H%M%S") backup_path = f"{path}.bak.{ts}" if os.path.exists(path): with open(path, "rb") as src, open(backup_path, "wb") as dst: dst.write(src.read()) backups = sorted(glob.glob(f"{path}.bak.*"), reverse=True) for old in backups[10:]: os.unlink(old) tmp = f"{path}.tmp" with open(tmp, "w", encoding="utf-8") as f: _yaml.dump(data, f) os.replace(tmp, path) def list_backups(path: str) -> list: """Return backup paths sorted newest-first.""" return sorted(glob.glob(f"{path}.bak.*"), reverse=True) def apply_structured_section(data, section: str, values: dict) -> None: """Merge a dict of scalar/list values into data for the named logical section. For 'server': updates each known key individually, preserving comments on unchanged keys. For 'users': replaces the entire users dict. """ if section == "server": for key in _SERVER_KEYS: if key in values: data[key] = values[key] elif section == "users": data["users"] = values else: raise ValueError(f"Unknown structured section: {section!r}") def apply_yaml_section(data, section: str, yaml_text: str) -> None: """Replace the named logical section by parsing yaml_text.""" parsed = _yaml.load(yaml_text) if section == "notification_channels": data["notification_channels"] = parsed elif section == "thresholds": data["threshold_configs"] = parsed elif section == "hosts": data["hosts"] = parsed elif section == "dns": if parsed: for key in _DNS_KEYS: if key in parsed: data[key] = parsed[key] else: raise ValueError(f"Unknown YAML section: {section!r}") ``` - [ ] **Step 6: Run tests to confirm they pass** ``` pytest tests/test_configio.py -v ``` Expected: all 14 tests PASS. - [ ] **Step 7: Commit** ```bash git add hbd/server/configio.py tests/test_configio.py pyproject.toml git commit -m "feat: add configio module for comment-preserving YAML round-trip writes" ``` --- ## Task 2: Config read API (GET /api/0/config, section/{name}, /backups) **Files:** - Modify: `hbd/server/http.py` (add import, `_mask_config_for_api`, three handlers, routes) - [ ] **Step 1: Write failing tests** Create `tests/test_http_config_api.py`: ```python """Tests for the config read/write API helpers in http.py.""" import pytest from hbd.server import http def test_mask_config_for_api_masks_user_passwords(): config = { "hbd_port": 50004, "interval": 20, "users": { "alice": {"full_name": "Alice", "admin": True, "password": "pbkdf2:sha256:abc"}, }, "oauth": {}, } result = http._mask_config_for_api(config) assert result["users"]["alice"]["password"] == "•••" assert result["users"]["alice"]["full_name"] == "Alice" def test_mask_config_for_api_masks_oauth_client_secret(): config = { "hbd_port": 50004, "interval": 20, "users": {}, "oauth": { "gitea": {"type": "gitea", "url": "https://git.example.com", "client_id": "cid", "client_secret": "verysecret"}, }, } result = http._mask_config_for_api(config) assert result["oauth"]["gitea"]["client_secret"] == "•••" assert result["oauth"]["gitea"]["client_id"] == "cid" def test_mask_config_for_api_includes_server_keys(): config = {"hbd_port": 50004, "interval": 20, "users": {}, "oauth": {}} result = http._mask_config_for_api(config) assert result["server"]["hbd_port"] == 50004 assert result["server"]["interval"] == 20 def test_mask_config_for_api_no_password_in_users_leaves_no_key(): config = { "hbd_port": 50004, "users": {"bob": {"full_name": "Bob", "admin": False}}, "oauth": {}, } result = http._mask_config_for_api(config) assert "password" not in result["users"]["bob"] ``` - [ ] **Step 2: Run tests to confirm they fail** ``` pytest tests/test_http_config_api.py -v ``` Expected: ImportError — `http._mask_config_for_api` does not exist yet. - [ ] **Step 3: Add import and helper to `hbd/server/http.py`** After the existing imports block (after `from . import ws as ws_mod`), add: ```python from . import configio as configio_mod ``` After the `_can_own_host` function and before the `start()` function, add: ```python def _mask_config_for_api(config) -> dict: """Return a JSON-serializable config dict with secrets masked.""" _SERVER_KEYS = [ "hbd_port", "hbd_host", "ws_port", "wss_port", "hb_port", "interval", "grace", "base_url", "threshold_renotify_interval", "logfile", "pidfile", "pickfile", "journal_enabled", "journal_dir", "journal_max_size", "journal_max_backups", "default_owner", ] result = {} result["server"] = {k: config.get(k) for k in _SERVER_KEYS} users = {} for username, attrs in (config.get("users") or {}).items(): u = dict(attrs) if "password" in u: u["password"] = "•••" users[username] = u result["users"] = users oauth = {} for name, attrs in (config.get("oauth") or {}).items(): o = dict(attrs) if "client_secret" in o: o["client_secret"] = "•••" oauth[name] = o result["oauth"] = oauth return result ``` - [ ] **Step 4: Run tests to confirm helper tests pass** ``` pytest tests/test_http_config_api.py -v ``` Expected: all 4 tests PASS. - [ ] **Step 5: Add the three GET handlers inside `start()` in `http.py`** Add these three handler functions inside `start()`, just before the `app = web.Application()` line (alongside the other handler closures): ```python # ------------------------------------------------------------------------- # Config API (admin only) # ------------------------------------------------------------------------- _config_path = getattr(config, "_config_path", "") or "" async def api_config_get(request): """GET /api/0/config — full config as JSON, secrets masked. Admin only.""" user, err = _require_auth(request) if err: return err if user and not user.admin: return web.json_response({"error": "Admin only"}, status=403) return web.json_response(_mask_config_for_api(config)) async def api_config_section_get(request): """GET /api/0/config/section/{name} — raw YAML text for a YAML-editor section.""" user, err = _require_auth(request) if err: return err if user and not user.admin: return web.json_response({"error": "Admin only"}, status=403) if not _config_path: return web.json_response({"error": "Config path not available"}, status=503) name = request.match_info["name"] _DNS_KEYS = ["nsupdate_bin", "dyndomains", "dyndnshosts", "drophosts"] _YAML_EXTRACTORS = { "notification_channels": lambda d: d.get("notification_channels") or {}, "thresholds": lambda d: d.get("threshold_configs") or {}, "hosts": lambda d: d.get("hosts") or {}, "dns": lambda d: {k: d[k] for k in _DNS_KEYS if k in d}, } if name not in _YAML_EXTRACTORS: return web.json_response({"error": "Unknown section"}, status=404) import io as _io from ruamel.yaml import YAML as _YAML data = configio_mod.read_roundtrip(_config_path) section_data = _YAML_EXTRACTORS[name](data) _sy = _YAML() _sy.preserve_quotes = True buf = _io.StringIO() _sy.dump(section_data, buf) return web.json_response({"yaml": buf.getvalue()}) async def api_config_backups_get(request): """GET /api/0/config/backups — list of backup paths, newest first.""" user, err = _require_auth(request) if err: return err if user and not user.admin: return web.json_response({"error": "Admin only"}, status=403) if not _config_path: return web.json_response({"backups": []}) backups = configio_mod.list_backups(_config_path) return web.json_response({"backups": backups}) ``` - [ ] **Step 6: Register the new routes in `app.add_routes()`** In the `app.add_routes([...])` call, add after the existing Users routes: ```python # Config API (admin) web.get("/api/0/config", api_config_get), web.get("/api/0/config/section/{name}", api_config_section_get), web.get("/api/0/config/backups", api_config_backups_get), ``` - [ ] **Step 7: Run the full test suite to confirm no regressions** ``` pytest tests/ -v ``` Expected: all existing tests PASS, 4 new tests in test_http_config_api.py PASS. - [ ] **Step 8: Commit** ```bash git add hbd/server/http.py tests/test_http_config_api.py git commit -m "feat: add config read API (GET /api/0/config, /section/{name}, /backups)" ``` --- ## Task 3: Config write API (POST /api/0/config + POST /api/0/config/rollback) **Files:** - Modify: `hbd/server/http.py` (two new handlers + routes) - Modify: `tests/test_http_config_api.py` (add write tests) - [ ] **Step 1: Write failing tests** Append to `tests/test_http_config_api.py`: ```python # ---- configio integration for write path ---- def test_write_path_applies_server_section(tmp_path): cfg = tmp_path / ".hb.yaml" cfg.write_text("hbd_port: 50004\ninterval: 20\nusers: {}\n") from hbd.server import configio data = configio.read_roundtrip(str(cfg)) configio.apply_structured_section(data, "server", {"interval": 60}) configio.write_config(str(cfg), data) data2 = configio.read_roundtrip(str(cfg)) assert data2["interval"] == 60 assert data2["hbd_port"] == 50004 # unchanged def test_write_path_applies_yaml_section(tmp_path): cfg = tmp_path / ".hb.yaml" cfg.write_text( "hbd_port: 50004\nnotification_channels:\n old_ch:\n type: email\n" ) from hbd.server import configio data = configio.read_roundtrip(str(cfg)) configio.apply_yaml_section(data, "notification_channels", "new_ch:\n type: pushover\n") configio.write_config(str(cfg), data) data2 = configio.read_roundtrip(str(cfg)) assert "new_ch" in data2["notification_channels"] assert "old_ch" not in data2["notification_channels"] def test_write_path_hashes_plaintext_password(tmp_path): cfg = tmp_path / ".hb.yaml" cfg.write_text("hbd_port: 50004\nusers:\n alice:\n full_name: Alice\n admin: true\n password: pbkdf2:sha256:old\n") from hbd.server import configio from hbd.server import users as users_mod data = configio.read_roundtrip(str(cfg)) # Simulate what the POST handler does: hash plaintext password new_users = {"alice": {"full_name": "Alice", "admin": True, "password": "newplaintext"}} for username, attrs in new_users.items(): pw = attrs.get("password", "") if pw and not pw.startswith("pbkdf2:"): attrs["password"] = users_mod.hash_password(pw) configio.apply_structured_section(data, "users", new_users) configio.write_config(str(cfg), data) data2 = configio.read_roundtrip(str(cfg)) assert data2["users"]["alice"]["password"].startswith("pbkdf2:") assert data2["users"]["alice"]["password"] != "newplaintext" def test_rollback_restores_backup(tmp_path): cfg = tmp_path / ".hb.yaml" cfg.write_text("hbd_port: 50004\ninterval: 20\n") from hbd.server import configio # Make a change to create a backup data = configio.read_roundtrip(str(cfg)) data["interval"] = 99 configio.write_config(str(cfg), data) backups = configio.list_backups(str(cfg)) assert len(backups) == 1 # Read the backup and write it back (simulating rollback) backup_data = configio.read_roundtrip(backups[0]) configio.write_config(str(cfg), backup_data) restored = configio.read_roundtrip(str(cfg)) assert restored["interval"] == 20 ``` - [ ] **Step 2: Run tests to confirm they pass (they test configio, not http)** ``` pytest tests/test_http_config_api.py -v ``` Expected: all tests PASS (these test configio directly, not the HTTP handler closures). - [ ] **Step 3: Add the two POST handlers inside `start()` in `http.py`** Add these after `api_config_backups_get` and before `app = web.Application()`: ```python async def api_config_post(request): """POST /api/0/config — publish staged changes to .hb.yaml. Admin only.""" user, err = _require_auth(request) if err: return err if user and not user.admin: return web.json_response({"error": "Admin only"}, status=403) if not _config_path: return web.json_response({"error": "Config path not available"}, status=503) try: payload = await request.json() except Exception: return web.json_response({"error": "Invalid JSON"}, status=400) try: data = configio_mod.read_roundtrip(_config_path) if "server" in payload: configio_mod.apply_structured_section(data, "server", payload["server"]) if "users" in payload: # Hash any plaintext passwords; preserve existing hashes when omitted existing_users = data.get("users") or {} users_payload = payload["users"] for username, attrs in users_payload.items(): pw = attrs.get("password", "") if pw and not pw.startswith("pbkdf2:"): attrs["password"] = users_mod.hash_password(pw) elif not pw: existing_hash = (existing_users.get(username) or {}).get("password", "") if existing_hash: attrs["password"] = existing_hash else: attrs.pop("password", None) configio_mod.apply_structured_section(data, "users", users_payload) if "oauth" in payload: data["oauth"] = payload["oauth"] for section in ("notification_channels", "thresholds", "hosts", "dns"): if section in payload: configio_mod.apply_yaml_section(data, section, payload[section]) configio_mod.write_config(_config_path, data) except Exception as exc: logger.error("Config write failed: %s", exc) return web.json_response({"error": str(exc)}, status=500) if hasattr(config, "reload"): await config.reload() users_mod.load_users(config) return web.json_response({"ok": True}) async def api_config_rollback(request): """POST /api/0/config/rollback — restore a backup. Admin only.""" user, err = _require_auth(request) if err: return err if user and not user.admin: return web.json_response({"error": "Admin only"}, status=403) if not _config_path: return web.json_response({"error": "Config path not available"}, status=503) try: body = await request.json() except Exception: return web.json_response({"error": "Invalid JSON"}, status=400) backup = body.get("backup", "") expected_prefix = _config_path + ".bak." if not backup or not backup.startswith(expected_prefix) or not os.path.exists(backup): return web.json_response({"error": "Invalid or missing backup"}, status=400) try: backup_data = configio_mod.read_roundtrip(backup) configio_mod.write_config(_config_path, backup_data) except Exception as exc: logger.error("Rollback failed: %s", exc) return web.json_response({"error": str(exc)}, status=500) if hasattr(config, "reload"): await config.reload() users_mod.load_users(config) return web.json_response({"ok": True}) ``` - [ ] **Step 4: Register the new routes** In `app.add_routes([...])`, add after the GET config routes: ```python web.post("/api/0/config", api_config_post), web.post("/api/0/config/rollback", api_config_rollback), ``` - [ ] **Step 5: Run the full test suite** ``` pytest tests/ -v ``` Expected: all tests PASS. - [ ] **Step 6: Commit** ```bash git add hbd/server/http.py tests/test_http_config_api.py git commit -m "feat: add config write API (POST /api/0/config, POST /api/0/config/rollback)" ``` --- ## Task 4: User self-service API (PUT /api/0/users/me) **Files:** - Modify: `hbd/server/http.py` (one new handler + route) - Create: `tests/test_http_users_me.py` - [ ] **Step 1: Write failing tests** Create `tests/test_http_users_me.py`: ```python """Tests for PUT /api/0/users/me logic.""" import pytest from hbd.server import users as users_mod def test_hash_password_roundtrip(): h = users_mod.hash_password("mysecret") assert h.startswith("pbkdf2:sha256:") assert users_mod.authenticate.__doc__ is not None # module loaded def test_password_change_requires_correct_current(tmp_path): cfg = tmp_path / ".hb.yaml" initial_hash = users_mod.hash_password("oldpass") cfg.write_text( f"hbd_port: 50004\nusers:\n alice:\n full_name: Alice\n admin: true\n password: {initial_hash}\n" ) users_mod.load_users({"users": {"alice": {"full_name": "Alice", "admin": True, "password": initial_hash}}}) # Correct current password authenticates assert users_mod.authenticate("alice", "oldpass") is not None # Wrong current password does not authenticate assert users_mod.authenticate("alice", "wrongpass") is None def test_put_users_me_writes_new_fields(tmp_path): """Simulate the write path: read config, update user, write back.""" initial_hash = users_mod.hash_password("secret") yaml_content = ( "hbd_port: 50004\n" f"users:\n alice:\n full_name: Old Name\n admin: true\n password: {initial_hash}\n" ) cfg = tmp_path / ".hb.yaml" cfg.write_text(yaml_content) from hbd.server import configio data = configio.read_roundtrip(str(cfg)) # Simulate handler updating full_name and avatar user_entry = dict(data["users"]["alice"]) user_entry["full_name"] = "New Name" user_entry["avatar"] = "/img/alice.png" data["users"]["alice"] = user_entry configio.write_config(str(cfg), data) result = configio.read_roundtrip(str(cfg)) assert result["users"]["alice"]["full_name"] == "New Name" assert result["users"]["alice"]["avatar"] == "/img/alice.png" assert result["users"]["alice"]["password"] == initial_hash # unchanged def test_put_users_me_changes_password(tmp_path): initial_hash = users_mod.hash_password("oldpass") cfg = tmp_path / ".hb.yaml" cfg.write_text( f"hbd_port: 50004\nusers:\n alice:\n full_name: Alice\n password: {initial_hash}\n" ) from hbd.server import configio data = configio.read_roundtrip(str(cfg)) new_hash = users_mod.hash_password("newpass") data["users"]["alice"]["password"] = new_hash configio.write_config(str(cfg), data) result = configio.read_roundtrip(str(cfg)) # Load users from new config and authenticate with new password new_config = {"users": dict(result["users"])} users_mod.load_users(new_config) assert users_mod.authenticate("alice", "newpass") is not None assert users_mod.authenticate("alice", "oldpass") is None def test_put_users_me_notification_channels(tmp_path): cfg = tmp_path / ".hb.yaml" cfg.write_text( "hbd_port: 50004\n" "notification_channels:\n pushover_ops:\n type: pushover\n" "users:\n alice:\n full_name: Alice\n notification_channels: []\n" ) from hbd.server import configio data = configio.read_roundtrip(str(cfg)) data["users"]["alice"]["notification_channels"] = ["pushover_ops"] configio.write_config(str(cfg), data) result = configio.read_roundtrip(str(cfg)) assert result["users"]["alice"]["notification_channels"] == ["pushover_ops"] ``` - [ ] **Step 2: Run tests to confirm they pass (they test configio + users_mod)** ``` pytest tests/test_http_users_me.py -v ``` Expected: all 5 tests PASS. - [ ] **Step 3: Add the PUT /api/0/users/me handler inside `start()` in `http.py`** Add after `api_config_rollback` and before `app = web.Application()`: ```python async def api_user_self_put(request): """PUT /api/0/users/me — update own full_name, avatar, notification_channels, password.""" user, err = _require_auth(request) if err: return err if user is None: return web.json_response({"error": "Authentication required"}, status=401) if not _config_path: return web.json_response({"error": "Config path not available"}, status=503) try: body = await request.json() except Exception: return web.json_response({"error": "Invalid JSON"}, status=400) username = user.username password_change = body.get("password") if password_change: current_pw = password_change.get("current", "") new_pw = password_change.get("new", "") if not new_pw: return web.json_response({"error": "New password cannot be empty"}, status=400) if not users_mod.authenticate(username, current_pw): return web.json_response({"error": "Current password incorrect"}, status=403) try: data = configio_mod.read_roundtrip(_config_path) if "users" not in data or data["users"] is None: data["users"] = {} user_entry = dict(data["users"].get(username) or {}) if "full_name" in body: user_entry["full_name"] = str(body["full_name"]) if "avatar" in body: user_entry["avatar"] = str(body["avatar"]) if "notification_channels" in body: user_entry["notification_channels"] = list(body["notification_channels"]) if password_change: user_entry["password"] = users_mod.hash_password(password_change["new"]) data["users"][username] = user_entry configio_mod.write_config(_config_path, data) except Exception as exc: logger.error("User self-update failed: %s", exc) return web.json_response({"error": str(exc)}, status=500) if hasattr(config, "reload"): await config.reload() users_mod.load_users(config) return web.json_response({"ok": True}) ``` - [ ] **Step 4: Register the PUT route in `app.add_routes()`** Change the existing `web.get("/api/0/users/me", api_user_self)` line to add the PUT route immediately after it: ```python web.get("/api/0/users/me", api_user_self), web.put("/api/0/users/me", api_user_self_put), ``` - [ ] **Step 5: Run full test suite** ``` pytest tests/ -v ``` Expected: all tests PASS. - [ ] **Step 6: Commit** ```bash git add hbd/server/http.py tests/test_http_users_me.py git commit -m "feat: add PUT /api/0/users/me for user self-service profile updates" ``` --- ## Task 5: Settings page backend (settings.py + http.py profile_page) **Files:** - Modify: `hbd/server/settings.py` (add `section_mode`, `api_section`, `editable` flags; add oauth section; add `all_channel_names` output) - Modify: `hbd/server/http.py` (pass `all_channel_names` to settings_page and profile_page) - [ ] **Step 1: Write failing tests** Create `tests/test_settings_sections.py`: ```python import pytest from hbd.server import settings as settings_mod CFG = { "hbd_port": 50004, "interval": 20, "grace": 2, "users": { "alice": {"full_name": "Alice Smith", "admin": True, "password": "pbkdf2:sha256:abc", "notification_channels": ["pushover_ops"]}, }, "oauth": { "gitea": {"type": "gitea", "url": "https://git.example.com", "client_id": "cid", "client_secret": "csec", "label": "Sign in with Gitea"}, }, "notification_channels": { "pushover_ops": {"type": "pushover", "token": "tok", "user": "usr"}, }, "hosts": {}, } def test_sections_have_section_mode(): sections = settings_mod.get_settings_sections(CFG) for s in sections: assert "section_mode" in s, f"Section {s['id']} missing section_mode" assert s["section_mode"] in ("form", "yaml") def test_sections_have_api_section(): sections = settings_mod.get_settings_sections(CFG) for s in sections: assert "api_section" in s, f"Section {s['id']} missing api_section" def test_network_section_has_editable_fields(): sections = settings_mod.get_settings_sections(CFG) network = next(s for s in sections if s["id"] == "network") assert network["section_mode"] == "form" assert network["api_section"] == "server" editable = [f for f in network["fields"] if f["editable"]] assert len(editable) >= 2 # hbd_port, ws_port at minimum def test_yaml_sections_have_correct_mode(): sections = settings_mod.get_settings_sections(CFG) yaml_sections = {s["id"]: s for s in sections if s["section_mode"] == "yaml"} assert "channels" in yaml_sections assert "hosts" in yaml_sections assert "thresholds" in yaml_sections assert "dns" in yaml_sections assert yaml_sections["channels"]["api_section"] == "notification_channels" assert yaml_sections["hosts"]["api_section"] == "hosts" assert yaml_sections["thresholds"]["api_section"] == "thresholds" assert yaml_sections["dns"]["api_section"] == "dns" def test_oauth_section_exists(): sections = settings_mod.get_settings_sections(CFG) oauth = next((s for s in sections if s["id"] == "oauth"), None) assert oauth is not None assert oauth["section_mode"] == "form" assert oauth["api_section"] == "oauth" assert len(oauth["providers"]) == 1 assert oauth["providers"][0]["name"] == "gitea" assert oauth["providers"][0]["client_secret"] == "•••" def test_all_channel_names_returned(): sections = settings_mod.get_settings_sections(CFG) # all_channel_names is a top-level key in the return value # We return it alongside sections in a dict now result = settings_mod.get_settings_data(CFG) assert "all_channel_names" in result assert "pushover_ops" in result["all_channel_names"] def test_users_section_has_user_list(): sections = settings_mod.get_settings_sections(CFG) users_sec = next(s for s in sections if s["id"] == "users") assert users_sec["section_mode"] == "form" assert users_sec["api_section"] == "users" assert len(users_sec["users"]) == 1 assert users_sec["users"][0]["username"] == "alice" # Password hash never exposed assert "password" not in users_sec["users"][0] ``` - [ ] **Step 2: Run tests to confirm they fail** ``` pytest tests/test_settings_sections.py -v ``` Expected: FAIL — `section_mode`, `api_section` keys missing; `oauth` section absent; `get_settings_data` not defined. - [ ] **Step 3: Update `hbd/server/settings.py`** Replace the module with the updated version. Key changes from the current file: **a) Add `get_settings_data()` wrapper function** (at the end of the file, replacing nothing — add alongside `get_settings_sections`): ```python def get_settings_data(config: dict, threshold_checker=None) -> dict: """Return sections list + auxiliary data for the settings template.""" sections = get_settings_sections(config, threshold_checker=threshold_checker) all_channel_names = sorted((config.get("notification_channels") or {}).keys()) return {"sections": sections, "all_channel_names": all_channel_names} ``` **b) Add `section_mode` and `api_section` to every section dict** in `get_settings_sections`. The complete updated `return [...]` block is: ```python # ---- OAuth providers (built separately) -------------------------------- oauth_providers = [] for pname, pattrs in (config.get("oauth") or {}).items(): if not isinstance(pattrs, dict): continue oauth_providers.append({ "name": pname, "type": pattrs.get("type", "gitea"), "url": pattrs.get("url", ""), "client_id": pattrs.get("client_id", ""), "client_secret": _mask(pattrs.get("client_secret", "")) or "•••" if pattrs.get("client_secret") else "", "label": pattrs.get("label", ""), "logo": pattrs.get("logo", ""), }) return [ { "id": "network", "title": "Network", "description": "Ports and bind addresses for all server sockets.", "section_mode": "form", "api_section": "server", "fields": [ field("hb_port", "Heartbeat UDP port", "port", "UDP port the server listens on for heartbeat datagrams.", editable=True), field("hbd_host", "HTTP bind address", "text", "Interface to bind the HTTP server to. Empty = all interfaces.", editable=True), field("hbd_port", "HTTP API port", "port", "TCP port for the HTTP API and web UI.", editable=True), field("ws_port", "WebSocket port", "port", "TCP port for the plain WebSocket server.", editable=True), field("wss_port", "Secure WebSocket port", "port", "TCP port for WSS (TLS WebSocket). Leave empty to disable.", editable=True), ], }, { "id": "tls", "title": "TLS / WebSocket Security", "description": "Certificate paths used when wss_port is set. Restart required after changes.", "section_mode": "form", "api_section": None, "fields": [ field("cert_path", "Certificate directory", "path", "Directory containing the TLS certificate and key files."), field("wss_pem", "Certificate file", "text", "Filename of the TLS certificate chain (PEM format)."), field("wss_key", "Key file", "text", "Filename of the TLS private key (PEM format)."), ], }, { "id": "monitoring", "title": "Monitoring", "description": "Heartbeat timing and alert re-notification behaviour.", "section_mode": "form", "api_section": "server", "fields": [ field("interval", "Heartbeat interval", "number", "Expected seconds between heartbeat messages from each client.", editable=True), field("grace", "Grace multiplier", "number", "A host is marked overdue after interval × grace seconds of silence.", editable=True), field("threshold_renotify_interval", "Re-notify interval", "number", "Seconds between threshold re-notifications for ongoing alerts.", editable=True), field("base_url", "Base URL", "text", "Base URL for notification links (e.g. https://hbd.example.com).", editable=True), ], }, { "id": "persistence", "title": "Persistence & Logging", "description": "State file and event log settings.", "section_mode": "form", "api_section": "server", "fields": [ field("pickfile", "State file", "path", "Path to the pickle file used to persist host state across restarts.", editable=True), field("logfile", "Event log", "path", "Path to the event log file.", editable=True), ], }, { "id": "journal", "title": "Message Journal", "description": "All received heartbeat and plugin messages are journalled here.", "section_mode": "form", "api_section": "server", "fields": [ field("journal_enabled", "Enabled", "boolean", "Turn journalling on or off.", editable=True), field("journal_dir", "Journal directory", "path", "Directory where journal files are written.", editable=True), field("journal_file", "Journal filename", "text", "Base filename for the journal.", editable=True), field("journal_max_size", "Max file size", "size", "Rotate the journal when it exceeds this size.", editable=True), field("journal_max_backups", "Backup count", "number", "Number of rotated journal files to keep.", editable=True), ], }, { "id": "users", "title": "Users", "description": "Accounts defined in the config file. Password hashes are never shown.", "section_mode": "form", "api_section": "users", "users": users_list, "fields": [ field("default_owner", "Default owner", "text", "Username that owns hosts with no explicit owner. " "Falls back to the first admin user.", editable=True), ], }, { "id": "oauth", "title": "OAuth Providers", "description": "OAuth2 login providers. Client secrets are masked.", "section_mode": "form", "api_section": "oauth", "providers": oauth_providers, "fields": [], }, { "id": "channels", "title": "Notification Channels", "description": "Named notification providers — edit raw YAML to add or change channels.", "section_mode": "yaml", "api_section": "notification_channels", "channels": notif_channels, "fields": [], }, { "id": "hosts", "title": "Hosts", "description": "Host definitions — edit raw YAML to add or change hosts.", "section_mode": "yaml", "api_section": "hosts", "hosts": hosts_list, "fields": [], }, { "id": "thresholds", "title": "Threshold Configurations", "description": "Named alert threshold sets — edit raw YAML to modify.", "section_mode": "yaml", "api_section": "thresholds", "threshold_configs": threshold_config_list, "fields": [], }, { "id": "dns", "title": "Dynamic DNS", "description": "nsupdate-based DNS registration — edit raw YAML.", "section_mode": "yaml", "api_section": "dns", "fields": [], }, { "id": "runtime", "title": "Runtime", "description": "Flags set at startup (require restart to change).", "section_mode": "form", "api_section": None, "fields": [ field("foreground", "Foreground mode", "boolean", "Run in the foreground instead of daemonising."), field("verbose", "Verbose logging", "boolean", "Enable verbose log output."), field("debug", "Debug level", "number", "0 = off. Higher values increase log verbosity."), ], }, ] ``` **c) Update `settings_page` handler in `http.py`** to use `get_settings_data` and pass `all_channel_names`: ```python async def settings_page(request): """GET /settings — editable config view.""" current_user, _ = _require_auth_redirect(request) if current_user and not current_user.admin: raise web.HTTPForbidden(reason="Admin access required") pkg_dir = os.path.dirname(__file__) templates_dir = config.get("templates_dir", os.path.join(pkg_dir, "templates")) env = jinja2.Environment(loader=jinja2.FileSystemLoader(templates_dir)) tmpl = env.get_template("settings.html") settings_data = settings_mod.get_settings_data(config, threshold_checker=threshold_checker) body = tmpl.render( title="Settings - Heartbeat", sections=settings_data["sections"], all_channel_names=settings_data["all_channel_names"], current_user=current_user.to_dict() if current_user else None, active_page="settings", ) return web.Response(text=body, content_type="text/html") ``` **d) Update `profile_page` handler in `http.py`** to pass `all_channel_names`: After the existing `notif_channels` list is built (around line 855), add: ```python all_channel_names = sorted((config.get("notification_channels") or {}).keys()) ``` And add `all_channel_names=all_channel_names` to the `tmpl.render(...)` call. - [ ] **Step 4: Run tests to confirm they pass** ``` pytest tests/test_settings_sections.py -v ``` Expected: all 7 tests PASS. - [ ] **Step 5: Run full test suite** ``` pytest tests/ -v ``` Expected: all tests PASS. - [ ] **Step 6: Commit** ```bash git add hbd/server/settings.py hbd/server/http.py tests/test_settings_sections.py git commit -m "feat: add section_mode, api_section, editable flags and oauth section to settings" ``` --- ## Task 6: Settings page frontend (settings.html) **Files:** - Modify: `hbd/server/templates/settings.html` This task modifies the template to add: form inputs for editable fields, YAML textareas for yaml sections, the users CRUD table, the OAuth CRUD table, the pending-changes banner, Stage/Publish/Discard JS, and the rollback modal. - [ ] **Step 1: Add CSS for new elements** In the ``): ```css /* ---- Editable inputs ---- */ .field-input { width: 100%; max-width: 360px; border: 1px solid #ccc; border-radius: 4px; padding: 4px 8px; font-size: 0.88em; box-sizing: border-box; font-family: inherit; } .field-input:focus { border-color: #0066cc; outline: none; box-shadow: 0 0 0 2px rgba(0,102,204,.15); } /* ---- Section footer (Stage Changes button) ---- */ .section-footer { padding: 10px 20px; border-top: 1px solid #f0f0f0; display: flex; justify-content: flex-end; } /* ---- Pending changes banner ---- */ .pending-banner { position: sticky; top: 8px; z-index: 100; background: #fffbe6; border: 1px solid #e8c840; border-radius: 6px; padding: 10px 16px; display: flex; align-items: center; justify-content: space-between; font-size: 0.87em; margin-bottom: 16px; box-shadow: 0 2px 8px rgba(0,0,0,.08); } .pending-banner .pending-msg { color: #7a6000; } .pending-banner .pending-actions { display: flex; gap: 8px; } /* ---- YAML editor ---- */ .yaml-editor { width: 100%; font-family: monospace; font-size: 0.83em; border: 1px solid #ccc; border-radius: 4px; padding: 8px; box-sizing: border-box; background: #fafafa; resize: vertical; min-height: 140px; } .yaml-editor:focus { border-color: #0066cc; outline: none; } /* ---- Button styles ---- */ .btn { border: none; border-radius: 4px; padding: 5px 12px; font-size: 0.85em; cursor: pointer; } .btn-primary { background: #0066cc; color: #fff; } .btn-primary:hover { background: #0055aa; } .btn-success { background: #2a7a2a; color: #fff; } .btn-success:hover { background: #226622; } .btn-secondary { background: #888; color: #fff; } .btn-secondary:hover { background: #666; } .btn-danger { background: transparent; color: #c62828; border: 1px solid #e0e0e0; border-radius: 4px; padding: 2px 7px; font-size: 0.82em; cursor: pointer; } .btn-danger:hover { background: #fce4ec; } /* ---- CRUD table for users / oauth ---- */ .crud-table { width: 100%; border-collapse: collapse; font-size: 0.83em; } .crud-table th { background: #f5f5f5; padding: 6px 10px; text-align: left; font-weight: 600; color: #555; font-size: .78em; text-transform: uppercase; letter-spacing: .03em; border-bottom: 1px solid #e0e0e0; } .crud-table td { padding: 6px 10px; border-bottom: 1px solid #f0f0f0; vertical-align: top; } .crud-table tbody tr:last-child td { border-bottom: none; } .crud-table .field-input { max-width: none; } /* ---- Rollback modal ---- */ .modal-overlay { position: fixed; inset: 0; background: rgba(0,0,0,.4); display: flex; align-items: center; justify-content: center; z-index: 1000; } .modal-box { background: #fff; border-radius: 8px; padding: 24px; min-width: 340px; max-width: 520px; width: 90%; box-shadow: 0 8px 32px rgba(0,0,0,.18); } .modal-box h3 { margin: 0 0 12px; font-size: 1em; } .backup-row { display: flex; align-items: center; justify-content: space-between; padding: 6px 0; border-bottom: 1px solid #f0f0f0; font-size: .87em; } .backup-row:last-child { border-bottom: none; } ``` - [ ] **Step 2: Replace the subtitle and add the pending banner + rollback link + modal HTML** Replace the existing `

` line and the `

` opening: ```html

Edit server configuration — changes are staged until you publish them to .hb.yaml.

``` - [ ] **Step 3: Replace the sidebar nav to add rollback link** Replace the existing `