From 450814daca3b4f716ee466b0eeb66a377b44a7cc Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Tue, 12 May 2026 23:56:56 -0400 Subject: [PATCH] chore: remove docs/superpowers from repo Add to .gitignore to keep local copies untracked. Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 1 + .../plans/2026-04-25-plugin-error-checking.md | 602 ----- .../plans/2026-05-08-gitea-oauth2.md | 781 ------ .../plans/2026-05-09-config-editor.md | 2130 ----------------- .../plans/2026-05-09-multi-oauth-providers.md | 1028 -------- .../2026-05-10-host-overview-info-section.md | 539 ----- ...2026-04-25-plugin-error-checking-design.md | 92 - .../specs/2026-05-08-gitea-oauth2-design.md | 184 -- .../specs/2026-05-09-config-editor-design.md | 210 -- ...2026-05-09-multi-oauth-providers-design.md | 149 -- ...05-10-host-overview-info-section-design.md | 135 -- 11 files changed, 1 insertion(+), 5850 deletions(-) delete mode 100644 docs/superpowers/plans/2026-04-25-plugin-error-checking.md delete mode 100644 docs/superpowers/plans/2026-05-08-gitea-oauth2.md delete mode 100644 docs/superpowers/plans/2026-05-09-config-editor.md delete mode 100644 docs/superpowers/plans/2026-05-09-multi-oauth-providers.md delete mode 100644 docs/superpowers/plans/2026-05-10-host-overview-info-section.md delete mode 100644 docs/superpowers/specs/2026-04-25-plugin-error-checking-design.md delete mode 100644 docs/superpowers/specs/2026-05-08-gitea-oauth2-design.md delete mode 100644 docs/superpowers/specs/2026-05-09-config-editor-design.md delete mode 100644 docs/superpowers/specs/2026-05-09-multi-oauth-providers-design.md delete mode 100644 docs/superpowers/specs/2026-05-10-host-overview-info-section-design.md diff --git a/.gitignore b/.gitignore index 24d83a1..be4e25d 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,4 @@ uv.lock .hb.yaml .superpowers/ rndc-key +docs/superpowers/ diff --git a/docs/superpowers/plans/2026-04-25-plugin-error-checking.md b/docs/superpowers/plans/2026-04-25-plugin-error-checking.md deleted file mode 100644 index 300e005..0000000 --- a/docs/superpowers/plans/2026-04-25-plugin-error-checking.md +++ /dev/null @@ -1,602 +0,0 @@ -# Plugin Error Checking 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:** Improve plugin error checking in hbc, especially for nagios_runner, and fix logger messages silently discarded in daemon mode. - -**Architecture:** Three focused changes across three files: (1) `hbd/client/plugin.py` gains a `skip_reason` attribute on Plugin and updated PluginLoader messaging; (2) `hbd/client/plugins/nagios_runner.py` gains async subprocess execution, stderr capture, signal-killed process handling, and init-time command path validation; (3) `hbd/client/main.py` gains proper post-fork logging reconfiguration to syslog. - -**Tech Stack:** Python 3.11+, asyncio, `logging.handlers.SysLogHandler`, pytest - ---- - -## File Map - -| Action | Path | What changes | -|---|---|---| -| Modify | `hbd/client/plugin.py` | `Plugin.__init__` gains `skip_reason`; `PluginLoader` checks it | -| Modify | `hbd/client/plugins/nagios_runner.py` | async subprocess, stderr, signal codes, init validation, `skip_reason` | -| Modify | `hbd/client/main.py` | `_reconfigure_logging_for_daemon()` helper; remove redundant syslog calls | -| Create | `tests/test_plugin.py` | PluginLoader messaging tests | -| Create | `tests/test_nagios_runner.py` | NagiosRunnerPlugin behaviour tests | - -Run tests throughout with: -```bash -python -m pytest tests/test_plugin.py tests/test_nagios_runner.py -v -``` - ---- - -## Task 1: Plugin.skip_reason + PluginLoader messaging - -**Files:** -- Modify: `hbd/client/plugin.py:40-48` (Plugin.__init__) -- Modify: `hbd/client/plugin.py:369-381` (PluginLoader.load_from_directory) -- Create: `tests/test_plugin.py` - -- [ ] **Step 1: Write failing tests** - -Create `tests/test_plugin.py`: - -```python -import asyncio -import logging -import textwrap - -from hbd.client.plugin import Plugin, PluginLoader, PluginRegistry - - -def test_plugin_skip_reason_defaults_none(tmp_path): - plugin_code = textwrap.dedent(""" - from hbd.client.plugin import MonitorPlugin - - class MinimalPlugin(MonitorPlugin): - name = "minimal" - version = "1.0.0" - interval = 60 - - async def initialize(self): - return True - - async def _collect_metrics(self): - return {} - """) - (tmp_path / "minimal.py").write_text(plugin_code) - registry = PluginRegistry() - loader = PluginLoader(registry) - asyncio.run(loader.load_from_directory(tmp_path)) - plugin = registry.get("minimal") - assert plugin is not None - assert plugin.skip_reason is None - - -def test_loader_logs_info_when_skip_reason_set(tmp_path, caplog): - plugin_code = textwrap.dedent(""" - from hbd.client.plugin import MonitorPlugin - - class SkippablePlugin(MonitorPlugin): - name = "skippable" - version = "1.0.0" - interval = 60 - - async def initialize(self): - self.skip_reason = "not configured in yaml" - return False - - async def _collect_metrics(self): - return {} - """) - (tmp_path / "skippable.py").write_text(plugin_code) - registry = PluginRegistry() - loader = PluginLoader(registry) - - with caplog.at_level(logging.INFO, logger="plugin.loader"): - count = asyncio.run(loader.load_from_directory(tmp_path)) - - assert count == 0 - assert any("skipped: not configured in yaml" in r.message for r in caplog.records) - assert not any("failed initialization" in r.message for r in caplog.records) - - -def test_loader_logs_warning_when_no_skip_reason(tmp_path, caplog): - plugin_code = textwrap.dedent(""" - from hbd.client.plugin import MonitorPlugin - - class FailPlugin(MonitorPlugin): - name = "fail" - version = "1.0.0" - interval = 60 - - async def initialize(self): - return False - - async def _collect_metrics(self): - return {} - """) - (tmp_path / "fail_plugin.py").write_text(plugin_code) - registry = PluginRegistry() - loader = PluginLoader(registry) - - with caplog.at_level(logging.WARNING, logger="plugin.loader"): - count = asyncio.run(loader.load_from_directory(tmp_path)) - - assert count == 0 - assert any("failed initialization" in r.message for r in caplog.records) -``` - -- [ ] **Step 2: Run tests to verify they fail** - -```bash -python -m pytest tests/test_plugin.py -v -``` -Expected: `test_plugin_skip_reason_defaults_none` FAILS (attribute missing), others may error. - -- [ ] **Step 3: Add `skip_reason` to `Plugin.__init__`** - -In `hbd/client/plugin.py`, in `Plugin.__init__` (around line 46), add one line: - -```python -def __init__(self, config: Optional[Dict[str, Any]] = None): - self.config = config or {} - self.logger = logging.getLogger(f"plugin.{self.name}") - self._initialized = False - self.skip_reason: Optional[str] = None -``` - -- [ ] **Step 4: Update PluginLoader messaging** - -In `hbd/client/plugin.py`, replace the `if not initialized:` block (around line 372): - -```python - if not initialized: - if plugin.skip_reason: - self.logger.info( - f"Plugin {plugin.name} skipped: {plugin.skip_reason}" - ) - else: - self.logger.warning( - f"Plugin {plugin.name} failed initialization, skipping" - ) - continue -``` - -- [ ] **Step 5: Run tests to verify they pass** - -```bash -python -m pytest tests/test_plugin.py -v -``` -Expected: all 3 tests PASS. - -- [ ] **Step 6: Commit** - -```bash -git add hbd/client/plugin.py tests/test_plugin.py -git commit -m "feat: add skip_reason to Plugin; improve PluginLoader init messaging" -``` - ---- - -## Task 2: NagiosRunnerPlugin — skip_reason when no commands - -**Files:** -- Modify: `hbd/client/plugins/nagios_runner.py:88-105` (initialize) -- Modify: `tests/test_nagios_runner.py` (create) - -- [ ] **Step 1: Write failing test** - -Create `tests/test_nagios_runner.py`: - -```python -import asyncio -import logging -import os -import stat - -import pytest - -from hbd.client.plugins.nagios_runner import ( - NagiosRunnerPlugin, - NAGIOS_OK, - NAGIOS_WARNING, - NAGIOS_CRITICAL, - NAGIOS_UNKNOWN, -) - - -def test_no_commands_sets_skip_reason(): - plugin = NagiosRunnerPlugin(config={"commands": []}) - result = asyncio.run(plugin.initialize()) - assert result is False - assert plugin.skip_reason is not None - assert "nagios_runner.commands" in plugin.skip_reason -``` - -- [ ] **Step 2: Run test to verify it fails** - -```bash -python -m pytest tests/test_nagios_runner.py::test_no_commands_sets_skip_reason -v -``` -Expected: FAIL — `plugin.skip_reason` is `None`. - -- [ ] **Step 3: Set skip_reason in NagiosRunnerPlugin.initialize()** - -In `hbd/client/plugins/nagios_runner.py`, replace the early-return block in `initialize()` (around line 96): - -```python - if not self.commands: - self.skip_reason = "no commands configured (add nagios_runner.commands to config)" - self.logger.info("No Nagios commands configured") - return False -``` - -- [ ] **Step 4: Run test to verify it passes** - -```bash -python -m pytest tests/test_nagios_runner.py::test_no_commands_sets_skip_reason -v -``` -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add hbd/client/plugins/nagios_runner.py tests/test_nagios_runner.py -git commit -m "feat: set skip_reason on nagios_runner when no commands configured" -``` - ---- - -## Task 3: NagiosRunnerPlugin — async subprocess, stderr capture, negative return codes - -**Files:** -- Modify: `hbd/client/plugins/nagios_runner.py` (imports + `_run_nagios_plugin`) -- Modify: `tests/test_nagios_runner.py` - -- [ ] **Step 1: Write failing tests** - -Append to `tests/test_nagios_runner.py`: - -```python -def test_stderr_used_when_stdout_empty(tmp_path): - script = tmp_path / "check_err.sh" - script.write_text("#!/bin/sh\necho 'error from stderr' >&2\nexit 2\n") - script.chmod(script.stat().st_mode | stat.S_IEXEC) - - config = {"commands": [{"name": "t", "command": str(script)}], "timeout": 5} - plugin = NagiosRunnerPlugin(config=config) - asyncio.run(plugin.initialize()) - data = asyncio.run(plugin._collect_metrics()) - - assert "error from stderr" in data["t_output"] - assert data["t_status_code"] == NAGIOS_CRITICAL - - -def test_stderr_appended_when_both_present(tmp_path): - script = tmp_path / "check_both.sh" - script.write_text("#!/bin/sh\necho 'OK - all good'\necho 'extra detail' >&2\nexit 0\n") - script.chmod(script.stat().st_mode | stat.S_IEXEC) - - config = {"commands": [{"name": "t", "command": str(script)}], "timeout": 5} - plugin = NagiosRunnerPlugin(config=config) - asyncio.run(plugin.initialize()) - data = asyncio.run(plugin._collect_metrics()) - - assert "OK - all good" in data["t_output"] - assert "extra detail" in data["t_output"] - assert data["t_status_code"] == NAGIOS_OK - - -def test_negative_returncode_maps_to_unknown(): - # kill -9 $$ kills the shell itself; asyncio sees returncode -9 - config = {"commands": [{"name": "t", "command": "kill -9 $$"}], "timeout": 5} - plugin = NagiosRunnerPlugin(config=config) - asyncio.run(plugin.initialize()) - data = asyncio.run(plugin._collect_metrics()) - - assert data["t_status_code"] == NAGIOS_UNKNOWN - assert "signal" in data["t_output"].lower() -``` - -- [ ] **Step 2: Run tests to verify they fail** - -```bash -python -m pytest tests/test_nagios_runner.py::test_stderr_used_when_stdout_empty \ - tests/test_nagios_runner.py::test_stderr_appended_when_both_present \ - tests/test_nagios_runner.py::test_negative_returncode_maps_to_unknown -v -``` -Expected: all FAIL — current implementation ignores stderr and doesn't handle negative codes. - -- [ ] **Step 3: Update imports in nagios_runner.py** - -Replace the import block at the top of `hbd/client/plugins/nagios_runner.py`: - -```python -import asyncio -import os -import re -from typing import Any, Dict, List, Optional, Tuple - -from hbd.client.plugin import MonitorPlugin -``` - -(Remove `import subprocess`; add `import asyncio` and `import os`.) - -- [ ] **Step 4: Upgrade collection log level from DEBUG to INFO** - -In `hbd/client/plugins/nagios_runner.py`, in `_collect_metrics()`, change the debug log (around line 144) so results are visible at INFO level: - -```python - self.logger.info( - f"Executed {name}: {STATUS_NAMES.get(status_code, 'UNKNOWN')} - {output[:50]}" - ) -``` - -- [ ] **Step 5: Replace `_run_nagios_plugin` with async implementation** - -Replace the entire `_run_nagios_plugin` method in `hbd/client/plugins/nagios_runner.py`: - -```python - async def _run_nagios_plugin( - self, - command: str - ) -> Tuple[int, str, Dict[str, Any]]: - """Execute a Nagios plugin and parse its output.""" - try: - proc = await asyncio.create_subprocess_shell( - command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - try: - stdout_bytes, stderr_bytes = await asyncio.wait_for( - proc.communicate(), timeout=self.timeout - ) - except asyncio.TimeoutError: - proc.kill() - await proc.communicate() - self.logger.error(f"Command timed out: {command}") - return NAGIOS_UNKNOWN, f"Command timed out after {self.timeout}s", {} - - status_code = proc.returncode - - if status_code < 0: - return NAGIOS_UNKNOWN, f"Process killed by signal {-status_code}", {} - - if status_code > 3: - status_code = NAGIOS_UNKNOWN - - stdout = stdout_bytes.decode(errors="replace").strip() - stderr = stderr_bytes.decode(errors="replace").strip() - - # Parse perfdata from stdout before mixing in stderr - perfdata = self._parse_perfdata(stdout) - - # Build status message - status_part = stdout.split('|')[0].strip() if '|' in stdout else stdout - - if not stdout and stderr: - output_msg = stderr - elif stdout and stderr: - output_msg = f"{status_part} [stderr: {stderr}]" - else: - output_msg = status_part - - return status_code, output_msg, perfdata - - except Exception as e: - self.logger.error(f"Error executing command: {e}") - return NAGIOS_UNKNOWN, f"Execution error: {str(e)}", {} -``` - -Also remove the now-unused `self.shell` line from `__init__` (the `shell` config key is no longer used since `create_subprocess_shell` always uses a shell): - -In `NagiosRunnerPlugin.__init__`, remove: -```python - self.shell: bool = config.get("shell", True) if config else True -``` - -- [ ] **Step 6: Run tests to verify they pass** - -```bash -python -m pytest tests/test_nagios_runner.py -v -``` -Expected: all tests PASS including the 3 new ones. - -- [ ] **Step 7: Commit** - -```bash -git add hbd/client/plugins/nagios_runner.py tests/test_nagios_runner.py -git commit -m "feat: async subprocess in nagios_runner with stderr capture and signal handling" -``` - ---- - -## Task 4: NagiosRunnerPlugin — command path validation at init - -**Files:** -- Modify: `hbd/client/plugins/nagios_runner.py` (initialize) -- Modify: `tests/test_nagios_runner.py` - -- [ ] **Step 1: Write failing tests** - -Append to `tests/test_nagios_runner.py`: - -```python -def test_absolute_path_not_found_warns(caplog): - fake_cmd = "/nonexistent_hbc_test_path/check_something" - config = {"commands": [{"name": "t", "command": fake_cmd}]} - plugin = NagiosRunnerPlugin(config=config) - - with caplog.at_level(logging.WARNING, logger="plugin.nagios_runner"): - asyncio.run(plugin.initialize()) - - assert any("not found" in r.message for r in caplog.records) - - -def test_absolute_path_not_executable_warns(caplog, tmp_path): - non_exec = tmp_path / "check_test" - non_exec.write_text("#!/bin/sh\necho OK\n") - non_exec.chmod(0o644) # readable but not executable - - config = {"commands": [{"name": "t", "command": str(non_exec)}]} - plugin = NagiosRunnerPlugin(config=config) - - with caplog.at_level(logging.WARNING, logger="plugin.nagios_runner"): - asyncio.run(plugin.initialize()) - - assert any("not executable" in r.message for r in caplog.records) - - -def test_relative_path_not_checked(caplog): - # Relative paths (resolved via PATH) must not generate warnings - config = {"commands": [{"name": "t", "command": "echo OK"}]} - plugin = NagiosRunnerPlugin(config=config) - - with caplog.at_level(logging.WARNING, logger="plugin.nagios_runner"): - asyncio.run(plugin.initialize()) - - assert not any( - "not found" in r.message or "not executable" in r.message - for r in caplog.records - ) -``` - -- [ ] **Step 2: Run tests to verify they fail** - -```bash -python -m pytest tests/test_nagios_runner.py::test_absolute_path_not_found_warns \ - tests/test_nagios_runner.py::test_absolute_path_not_executable_warns \ - tests/test_nagios_runner.py::test_relative_path_not_checked -v -``` -Expected: `test_absolute_path_not_found_warns` and `test_absolute_path_not_executable_warns` FAIL (no warnings logged); `test_relative_path_not_checked` may pass. - -- [ ] **Step 3: Add command path validation to `initialize()`** - -In `hbd/client/plugins/nagios_runner.py`, extend `initialize()` by adding validation after the existing "log each command" loop (after line 103, before `return True`): - -```python - # Validate absolute command paths early - for cmd_config in self.commands: - name = cmd_config.get("name", "unnamed") - command = cmd_config.get("command", "") - if not command: - continue - exe = command.split()[0] - if os.path.isabs(exe): - if not os.path.isfile(exe): - self.logger.warning( - f"Command '{name}': executable not found: {exe}" - ) - elif not os.access(exe, os.X_OK): - self.logger.warning( - f"Command '{name}': executable not executable: {exe}" - ) -``` - -- [ ] **Step 4: Run full test suite to verify all pass** - -```bash -python -m pytest tests/test_plugin.py tests/test_nagios_runner.py -v -``` -Expected: all tests PASS. - -- [ ] **Step 5: Commit** - -```bash -git add hbd/client/plugins/nagios_runner.py tests/test_nagios_runner.py -git commit -m "feat: validate absolute command paths at nagios_runner init" -``` - ---- - -## Task 5: Daemon mode logging — route to syslog after fork - -**Files:** -- Modify: `hbd/client/main.py` (new helper + updated daemon block) - -No automated test for daemonization itself (fork behaviour is hard to unit-test). Manual verification steps are provided below. - -- [ ] **Step 1: Add `_reconfigure_logging_for_daemon` helper** - -In `hbd/client/main.py`, add this function just before `def build_parser()` (around line 589): - -```python -def _reconfigure_logging_for_daemon(log_level: int) -> None: - """Replace StreamHandlers (now writing to /dev/null) with a SysLogHandler.""" - from logging.handlers import SysLogHandler - - root = logging.getLogger() - for handler in root.handlers[:]: - root.removeHandler(handler) - handler.close() - - try: - syslog_handler = SysLogHandler( - address="/dev/log", - facility=SysLogHandler.LOG_DAEMON, - ) - except OSError: - syslog_handler = SysLogHandler( - address=("localhost", 514), - facility=SysLogHandler.LOG_DAEMON, - ) - # Attach the fallback first so the warning reaches syslog - syslog_handler.setFormatter( - logging.Formatter("hbc[%(process)d]: %(name)s %(levelname)s: %(message)s") - ) - root.addHandler(syslog_handler) - root.setLevel(log_level) - logging.warning("/dev/log not found, using syslog UDP localhost:514") - return - - syslog_handler.setFormatter( - logging.Formatter("hbc[%(process)d]: %(name)s %(levelname)s: %(message)s") - ) - root.addHandler(syslog_handler) - root.setLevel(log_level) -``` - -- [ ] **Step 2: Update the daemon block in `main()`** - -In `hbd/client/main.py`, replace the entire `if args.daemon:` block (lines 664–675): - -```python - if args.daemon: - print("Daemonizing...") - daemonize() - _reconfigure_logging_for_daemon(log_level) - logging.info(f"hbc starting, sending heartbeat to {', '.join(args.hosts)}") -``` - -This removes the `import syslog`, `syslog.openlog()`, and `syslog.syslog()` calls (now handled by the logging system) and removes the no-op second `logging.basicConfig()` call. - -- [ ] **Step 3: Run existing test suite to confirm no regressions** - -```bash -python -m pytest tests/test_plugin.py tests/test_nagios_runner.py -v -``` -Expected: all tests still PASS. - -- [ ] **Step 4: Manual smoke test — verify syslog output in daemon mode** - -```bash -# In one terminal, tail syslog -sudo journalctl -f -t hbc - -# In another terminal, start hbc in daemon mode (replace HOST with a real or dummy host) -python -m hbd.client.main -d -v localhost - -# Expected in journalctl output: -# hbc[]: hbc.main INFO: Starting hbc for -> ['localhost'] -# hbc[]: hbc.main INFO: hbc starting, sending heartbeat to localhost -# hbc[]: plugin.loader INFO: ... - -# Stop the daemon -pkill -f "hbd.client.main" -``` - -- [ ] **Step 5: Commit** - -```bash -git add hbd/client/main.py -git commit -m "fix: reconfigure logging to syslog after daemonize() instead of no-op basicConfig" -``` diff --git a/docs/superpowers/plans/2026-05-08-gitea-oauth2.md b/docs/superpowers/plans/2026-05-08-gitea-oauth2.md deleted file mode 100644 index b68f843..0000000 --- a/docs/superpowers/plans/2026-05-08-gitea-oauth2.md +++ /dev/null @@ -1,781 +0,0 @@ -# Gitea OAuth2 Authentication 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:** Add Gitea as an OAuth2 login provider that coexists with password auth, auto-provisioning new users on first login. - -**Architecture:** A new `oauth.py` module owns all Gitea-specific logic (CSRF state, URL building, token exchange, user-info fetch). `users.py` gains one function to upsert an OAuth-sourced user. `http.py` gets two new route handlers and a small login-page change. No new dependencies — `aiohttp.ClientSession` is already used in the codebase. - -**Tech Stack:** Python 3.12, aiohttp 3.x, pytest, pytest-asyncio - ---- - -## File Map - -| Action | Path | Responsibility | -|--------|------|----------------| -| Modify | `hbd/server/config.py` | Add `"oauth": {}` default | -| Create | `hbd/server/oauth.py` | CSRF state, URL builder, token exchange, user-info fetch | -| Modify | `hbd/server/users.py` | Add `provision_oauth_user()` | -| Modify | `hbd/server/http.py` | Import oauth, two new routes, login page button | -| Create | `tests/test_oauth.py` | All new unit tests | - ---- - -## Task 1: Add config default and `is_enabled()` - -**Files:** -- Modify: `hbd/server/config.py:34` (after the `"users"` line) -- Create: `hbd/server/oauth.py` -- Create: `tests/test_oauth.py` - -- [ ] **Step 1: Write the failing test** - -Create `tests/test_oauth.py`: - -```python -import pytest -from hbd.server import oauth - - -CFG_OFF = {} -CFG_ON = { - "oauth": { - "gitea": { - "url": "https://git.example.com", - "client_id": "cid", - "client_secret": "csec", - } - } -} -CFG_PARTIAL = {"oauth": {"gitea": {"url": "https://git.example.com"}}} - - -def test_is_enabled_when_all_keys_present(): - assert oauth.is_enabled(CFG_ON) is True - - -def test_is_enabled_false_when_no_oauth_key(): - assert oauth.is_enabled(CFG_OFF) is False - - -def test_is_enabled_false_when_partial_config(): - assert oauth.is_enabled(CFG_PARTIAL) is False -``` - -- [ ] **Step 2: Run to confirm failure** - -``` -pytest tests/test_oauth.py -v -``` - -Expected: `ModuleNotFoundError: No module named 'hbd.server.oauth'` - -- [ ] **Step 3: Add config default** - -In `hbd/server/config.py`, add after the `"default_owner"` line (currently line 35): - -```python - # OAuth2 providers - "oauth": {}, # oauth.gitea.{url,client_id,client_secret} -``` - -- [ ] **Step 4: Create `hbd/server/oauth.py` with `is_enabled`** - -```python -"""Gitea OAuth2 support. - -Config shape (in ~/.hb.yaml): - - oauth: - gitea: - url: https://git.example.com - client_id: - client_secret: - -Register a Gitea OAuth2 application at: - Gitea → Settings → Applications → OAuth2 -Set the redirect URI to: - https:///login/oauth/gitea/callback -""" - -import logging -import secrets -import time - -import aiohttp - -logger = logging.getLogger(__name__) - -STATE_TTL = 600 # 10 minutes - -# state_token -> expiry timestamp -_states: dict[str, float] = {} - - -class OAuthError(Exception): - """Raised when the OAuth2 flow fails for any reason.""" - - -def _gitea_cfg(config: dict) -> dict: - """Return the gitea sub-dict or {} if absent/incomplete.""" - return config.get("oauth", {}).get("gitea", {}) - - -def is_enabled(config: dict) -> bool: - """Return True when all three required Gitea OAuth keys are present.""" - g = _gitea_cfg(config) - return bool(g.get("url") and g.get("client_id") and g.get("client_secret")) -``` - -- [ ] **Step 5: Run to confirm tests pass** - -``` -pytest tests/test_oauth.py -v -``` - -Expected: 3 passed - -- [ ] **Step 6: Commit** - -```bash -git add hbd/server/config.py hbd/server/oauth.py tests/test_oauth.py -git commit -m "feat: add oauth module skeleton and is_enabled()" -``` - ---- - -## Task 2: CSRF state management - -**Files:** -- Modify: `hbd/server/oauth.py` (add `make_state`, `validate_state`) -- Modify: `tests/test_oauth.py` (add state tests) - -- [ ] **Step 1: Write the failing tests** - -Append to `tests/test_oauth.py`: - -```python -import time as time_mod - - -def test_make_state_returns_unique_tokens(): - s1 = oauth.make_state() - s2 = oauth.make_state() - assert s1 != s2 - assert len(s1) == 64 # 32 bytes hex - - -def test_validate_state_valid(): - state = oauth.make_state() - assert oauth.validate_state(state) is True - - -def test_validate_state_consumed_on_use(): - state = oauth.make_state() - oauth.validate_state(state) - assert oauth.validate_state(state) is False # replay rejected - - -def test_validate_state_unknown(): - assert oauth.validate_state("notastate") is False - - -def test_validate_state_expired(monkeypatch): - state = oauth.make_state() - # Wind expiry into the past - monkeypatch.setitem(oauth._states, state, time_mod.time() - 1) - assert oauth.validate_state(state) is False -``` - -- [ ] **Step 2: Run to confirm failure** - -``` -pytest tests/test_oauth.py -v -k "state" -``` - -Expected: `AttributeError: module 'hbd.server.oauth' has no attribute 'make_state'` - -- [ ] **Step 3: Implement state functions** - -Add to `hbd/server/oauth.py` after the `_states` dict definition: - -```python -def make_state() -> str: - """Generate a CSRF state token, store it with TTL, and return it.""" - _purge_states() - token = secrets.token_hex(32) - _states[token] = time.time() + STATE_TTL - return token - - -def validate_state(state: str) -> bool: - """Return True if *state* is known and unexpired; always removes it.""" - expiry = _states.pop(state, None) - if expiry is None: - return False - return time.time() < expiry - - -def _purge_states() -> None: - now = time.time() - expired = [k for k, exp in list(_states.items()) if exp < now] - for k in expired: - del _states[k] -``` - -- [ ] **Step 4: Run to confirm tests pass** - -``` -pytest tests/test_oauth.py -v -``` - -Expected: 8 passed - -- [ ] **Step 5: Commit** - -```bash -git add hbd/server/oauth.py tests/test_oauth.py -git commit -m "feat: add OAuth2 CSRF state management" -``` - ---- - -## Task 3: `provision_oauth_user` in users.py - -**Files:** -- Modify: `hbd/server/users.py` (add `provision_oauth_user`) -- Modify: `tests/test_oauth.py` (add provisioning tests) - -- [ ] **Step 1: Write the failing tests** - -Append to `tests/test_oauth.py`: - -```python -from hbd.server import users as users_mod -from hbd.server.users import User - - -def _reset_users(entries=None): - users_mod.users = entries or {} - - -def test_provision_oauth_user_new(): - _reset_users() - user = users_mod.provision_oauth_user("gituser", "Git User", "https://example.com/avatar.png") - assert user.username == "gituser" - assert user.full_name == "Git User" - assert user.avatar == "https://example.com/avatar.png" - assert user.admin is False - assert user.password_hash == "" - assert "gituser" in users_mod.users - - -def test_provision_oauth_user_no_password_login(): - _reset_users() - user = users_mod.provision_oauth_user("gituser", "Git User", "") - assert user.check_password("anything") is False - - -def test_provision_oauth_user_existing_updates_profile(): - existing = User( - username="alice", - full_name="Old Name", - avatar="old.png", - password_hash="pbkdf2:sha256:1:salt:abc", - admin=True, - notification_channels=["chan1"], - ) - _reset_users({"alice": existing}) - user = users_mod.provision_oauth_user("alice", "New Name", "new.png") - assert user.full_name == "New Name" - assert user.avatar == "new.png" - # Preserved - assert user.admin is True - assert user.password_hash == "pbkdf2:sha256:1:salt:abc" - assert user.notification_channels == ["chan1"] - - -def test_provision_oauth_user_does_not_overwrite_with_empty(): - existing = User(username="bob", full_name="Bob", avatar="bob.png") - _reset_users({"bob": existing}) - user = users_mod.provision_oauth_user("bob", "", "") - assert user.full_name == "Bob" - assert user.avatar == "bob.png" -``` - -- [ ] **Step 2: Run to confirm failure** - -``` -pytest tests/test_oauth.py -v -k "provision" -``` - -Expected: `AttributeError: module 'hbd.server.users' has no attribute 'provision_oauth_user'` - -- [ ] **Step 3: Implement `provision_oauth_user`** - -Add to `hbd/server/users.py` after the `authenticate()` function (after line 187): - -```python -def provision_oauth_user(username: str, full_name: str, avatar: str) -> "User": - """Create or update a user sourced from an OAuth2 provider. - - New users are inserted with no password_hash — they can only authenticate - via OAuth. Existing users (e.g. defined in config with a password) have - their display name and avatar refreshed; all other attributes are preserved. - """ - user = users.get(username) - if user is None: - user = User(username=username, full_name=full_name, avatar=avatar) - users[username] = user - logger.info("Provisioned OAuth user %r", username) - else: - if full_name: - user.full_name = full_name - if avatar: - user.avatar = avatar - return user -``` - -- [ ] **Step 4: Run to confirm tests pass** - -``` -pytest tests/test_oauth.py -v -``` - -Expected: 12 passed - -- [ ] **Step 5: Commit** - -```bash -git add hbd/server/users.py tests/test_oauth.py -git commit -m "feat: add provision_oauth_user() to users module" -``` - ---- - -## Task 4: URL builder, token exchange, and user-info fetch - -**Files:** -- Modify: `hbd/server/oauth.py` (add `authorization_url`, `exchange_code`, `fetch_user`) -- Modify: `tests/test_oauth.py` (add async tests with mocked HTTP) - -- [ ] **Step 1: Write the failing tests** - -Append to `tests/test_oauth.py`: - -```python -import pytest -from unittest.mock import AsyncMock, MagicMock, patch -from urllib.parse import urlparse, parse_qs - - -def test_authorization_url_shape(): - state = "teststate" - redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback" - url = oauth.authorization_url(CFG_ON, state, redirect_uri) - parsed = urlparse(url) - qs = parse_qs(parsed.query) - assert parsed.scheme == "https" - assert parsed.netloc == "git.example.com" - assert parsed.path == "/login/oauth/authorize" - assert qs["client_id"] == ["cid"] - assert qs["state"] == ["teststate"] - assert qs["redirect_uri"] == [redirect_uri] - assert qs["scope"] == ["user:email"] - assert qs["response_type"] == ["code"] - - -@pytest.mark.asyncio -async def test_exchange_code_returns_token(): - redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback" - mock_response = AsyncMock() - mock_response.status = 200 - mock_response.json = AsyncMock(return_value={"access_token": "tok123"}) - - mock_session = MagicMock() - mock_session.post = MagicMock(return_value=AsyncMock( - __aenter__=AsyncMock(return_value=mock_response), - __aexit__=AsyncMock(return_value=False), - )) - - with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( - __aenter__=AsyncMock(return_value=mock_session), - __aexit__=AsyncMock(return_value=False), - )): - token = await oauth.exchange_code(CFG_ON, "mycode", redirect_uri) - assert token == "tok123" - - -@pytest.mark.asyncio -async def test_exchange_code_raises_on_error_status(): - redirect_uri = "https://hbd.example.com/login/oauth/gitea/callback" - mock_response = AsyncMock() - mock_response.status = 401 - mock_response.text = AsyncMock(return_value="unauthorized") - - mock_session = MagicMock() - mock_session.post = MagicMock(return_value=AsyncMock( - __aenter__=AsyncMock(return_value=mock_response), - __aexit__=AsyncMock(return_value=False), - )) - - with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( - __aenter__=AsyncMock(return_value=mock_session), - __aexit__=AsyncMock(return_value=False), - )): - with pytest.raises(oauth.OAuthError): - await oauth.exchange_code(CFG_ON, "badcode", redirect_uri) - - -@pytest.mark.asyncio -async def test_fetch_user_returns_profile(): - mock_response = AsyncMock() - mock_response.status = 200 - mock_response.json = AsyncMock(return_value={ - "login": "alice", - "full_name": "Alice Smith", - "avatar_url": "https://git.example.com/avatars/alice.png", - }) - - mock_session = MagicMock() - mock_session.get = MagicMock(return_value=AsyncMock( - __aenter__=AsyncMock(return_value=mock_response), - __aexit__=AsyncMock(return_value=False), - )) - - with patch("hbd.server.oauth.aiohttp.ClientSession", return_value=AsyncMock( - __aenter__=AsyncMock(return_value=mock_session), - __aexit__=AsyncMock(return_value=False), - )): - profile = await oauth.fetch_user(CFG_ON, "tok123") - assert profile == { - "login": "alice", - "full_name": "Alice Smith", - "avatar_url": "https://git.example.com/avatars/alice.png", - } -``` - -- [ ] **Step 2: Run to confirm failure** - -``` -pytest tests/test_oauth.py -v -k "url or exchange or fetch" -``` - -Expected: `AttributeError: module 'hbd.server.oauth' has no attribute 'authorization_url'` - -- [ ] **Step 3: Implement the three functions** - -Add to `hbd/server/oauth.py`: - -```python -import urllib.parse - - -def authorization_url(config: dict, state: str, redirect_uri: str) -> str: - """Return the Gitea OAuth2 authorization URL to redirect the browser to.""" - g = _gitea_cfg(config) - params = urllib.parse.urlencode({ - "client_id": g["client_id"], - "redirect_uri": redirect_uri, - "response_type": "code", - "scope": "user:email", - "state": state, - }) - return f"{g['url'].rstrip('/')}/login/oauth/authorize?{params}" - - -async def exchange_code(config: dict, code: str, redirect_uri: str) -> str: - """Exchange an authorization *code* for a Gitea access token. - - Returns the access token string. Raises OAuthError on any failure. - """ - g = _gitea_cfg(config) - url = f"{g['url'].rstrip('/')}/login/oauth/access_token" - payload = { - "client_id": g["client_id"], - "client_secret": g["client_secret"], - "code": code, - "grant_type": "authorization_code", - "redirect_uri": redirect_uri, - } - timeout = aiohttp.ClientTimeout(total=10) - try: - async with aiohttp.ClientSession(timeout=timeout) as session: - async with session.post(url, json=payload, headers={"Accept": "application/json"}) as resp: - if resp.status != 200: - text = await resp.text() - raise OAuthError(f"Token exchange failed ({resp.status}): {text}") - data = await resp.json() - except aiohttp.ClientError as exc: - raise OAuthError(f"Token exchange network error: {exc}") from exc - token = data.get("access_token") - if not token: - raise OAuthError(f"No access_token in response: {data}") - return token - - -async def fetch_user(config: dict, token: str) -> dict: - """Fetch the authenticated user's profile from Gitea. - - Returns a dict with keys: login, full_name, avatar_url. - Raises OAuthError on any failure. - """ - g = _gitea_cfg(config) - url = f"{g['url'].rstrip('/')}/api/v1/user" - timeout = aiohttp.ClientTimeout(total=10) - try: - async with aiohttp.ClientSession(timeout=timeout) as session: - async with session.get(url, headers={"Authorization": f"token {token}"}) as resp: - if resp.status != 200: - text = await resp.text() - raise OAuthError(f"User fetch failed ({resp.status}): {text}") - data = await resp.json() - except aiohttp.ClientError as exc: - raise OAuthError(f"User fetch network error: {exc}") from exc - return { - "login": data.get("login", ""), - "full_name": data.get("full_name", ""), - "avatar_url": data.get("avatar_url", ""), - } -``` - -Also add `import urllib.parse` at the top of `oauth.py` (alongside the existing imports). - -- [ ] **Step 4: Run to confirm tests pass** - -``` -pytest tests/test_oauth.py -v -``` - -Expected: 17 passed - -- [ ] **Step 5: Commit** - -```bash -git add hbd/server/oauth.py tests/test_oauth.py -git commit -m "feat: add authorization_url, exchange_code, fetch_user to oauth module" -``` - ---- - -## Task 5: HTTP routes — redirect and callback - -**Files:** -- Modify: `hbd/server/http.py` - -`http.py` defines all handlers inside `async def start(...)`. The two new handlers go in the same block, just before the `app = web.Application()` line (~line 900). The import goes at the top of the file. - -- [ ] **Step 1: Add the import** - -In `hbd/server/http.py`, add after the existing local imports (after `from . import users as users_mod`): - -```python -from . import oauth as oauth_mod -``` - -- [ ] **Step 2: Add the two route handlers** - -In `hbd/server/http.py`, add the two handlers immediately before the `app = web.Application()` line: - -```python - async def oauth_gitea_redirect(request): - """GET /login/oauth/gitea — kick off the Gitea OAuth2 flow.""" - if not oauth_mod.is_enabled(config): - return web.Response(status=404, text="OAuth not configured") - state = oauth_mod.make_state() - redirect_uri = f"{request.url.origin()}/login/oauth/gitea/callback" - raise web.HTTPFound(oauth_mod.authorization_url(config, state, redirect_uri)) - - async def oauth_gitea_callback(request): - """GET /login/oauth/gitea/callback — handle Gitea's redirect back.""" - if not oauth_mod.is_enabled(config): - return web.Response(status=404, text="OAuth not configured") - code = request.rel_url.query.get("code", "") - state = request.rel_url.query.get("state", "") - if not code or not state: - return web.Response(status=400, text="Missing code or state") - if not oauth_mod.validate_state(state): - raise web.HTTPFound("/login?error=1") - redirect_uri = f"{request.url.origin()}/login/oauth/gitea/callback" - try: - token = await oauth_mod.exchange_code(config, code, redirect_uri) - profile = await oauth_mod.fetch_user(config, token) - except oauth_mod.OAuthError as exc: - logger.warning("OAuth error: %s", exc) - raise web.HTTPFound("/login?error=1") - user = users_mod.provision_oauth_user( - profile["login"], - profile["full_name"], - profile["avatar_url"], - ) - session_token = users_mod.create_session(user.username) - resp = web.HTTPFound("/") - resp.set_cookie( - SESSION_COOKIE, - session_token, - max_age=users_mod.SESSION_TTL, - httponly=True, - samesite="Lax", - ) - raise resp -``` - -- [ ] **Step 3: Register the routes** - -In `hbd/server/http.py`, add to the route list after the existing auth routes (after `web.post("/api/0/auth/logout", api_logout)`): - -```python - web.get("/login/oauth/gitea", oauth_gitea_redirect), - web.get("/login/oauth/gitea/callback", oauth_gitea_callback), -``` - -- [ ] **Step 4: Manual smoke test** - -Start the server locally with OAuth configured in `~/.hb.yaml`: - -```yaml -oauth: - gitea: - url: https://your-gitea-instance.example.com - client_id: your-client-id - client_secret: your-client-secret -``` - -Visit `http://localhost:50004/login/oauth/gitea` — confirm you are redirected to Gitea's authorization page. - -- [ ] **Step 5: Commit** - -```bash -git add hbd/server/http.py -git commit -m "feat: add Gitea OAuth2 redirect and callback routes" -``` - ---- - -## Task 6: Login page — "Sign in with Gitea" button - -**Files:** -- Modify: `hbd/server/http.py` (update `login_page` handler, ~line 625) - -- [ ] **Step 1: Replace the login page HTML** - -In `hbd/server/http.py`, find the `html = f"""` block inside `login_page` and replace it with: - -```python - gitea_button = "" - if oauth_mod.is_enabled(config): - gitea_url = _gitea_cfg_url(config) - gitea_button = f""" -
or
- - Sign in with Gitea - """ - - html = f""" - - - - Heartbeat — Login - - - -
-

Heartbeat

- {'

Invalid username, password, or OAuth error.

' if error else ''} -
-
-
- -
{gitea_button} -
- -""" -``` - -- [ ] **Step 2: Add the `_gitea_cfg_url` helper** - -Add this small helper in `hbd/server/http.py` just before the `login_page` handler (around line 600) so the template can read the Gitea display URL without importing internal oauth details: - -```python -def _gitea_cfg_url(config: dict) -> str: - return config.get("oauth", {}).get("gitea", {}).get("url", "") -``` - -Also update the `login_page` handler's `error` logic to show the error when the `?error=1` query param is present (set by the callback on OAuth failure): - -```python - async def login_page(request): - """GET /login — show login form; POST /login — process and redirect.""" - if not users_mod.users_enabled(): - raise web.HTTPFound("/") - - error = "" - if request.method == "POST": - form = await request.post() - username = form.get("username", "") - password = form.get("password", "") - user = users_mod.authenticate(username, password) - if user: - token = users_mod.create_session(username) - redirect_to = request.rel_url.query.get("next", "/") - resp = web.HTTPFound(redirect_to) - resp.set_cookie( - SESSION_COOKIE, - token, - max_age=users_mod.SESSION_TTL, - httponly=True, - samesite="Lax", - ) - raise resp - error = "Invalid username or password." - elif request.rel_url.query.get("error"): - error = "Sign-in failed. Please try again." -``` - -- [ ] **Step 3: Manual verification** - -Start the server with OAuth configured. Visit `/login`. Confirm: -- The "Sign in with Gitea" button appears (green, below a divider) -- Clicking it redirects to Gitea -- After authorising on Gitea, you are redirected back and land on `/` with a valid session cookie - -Without OAuth configured, confirm the button does not appear. - -- [ ] **Step 4: Commit** - -```bash -git add hbd/server/http.py -git commit -m "feat: add Sign in with Gitea button to login page" -``` - ---- - -## Self-Review Notes - -- All 5 spec requirements covered: coexist ✓, auto-provision ✓, regular user ✓, any Gitea user ✓, config-driven ✓ -- `exchange_code` signature in Task 4 matches usage in Task 5 (`config, code, redirect_uri`) ✓ -- `fetch_user` returns `{login, full_name, avatar_url}` — matched in callback handler ✓ -- `validate_state` removes state on use (replay protection) ✓ -- `provision_oauth_user` skips empty strings so existing avatar/name aren't erased ✓ -- `_gitea_cfg_url` is a plain `def`, not `async` — safe to call in template prep ✓ diff --git a/docs/superpowers/plans/2026-05-09-config-editor.md b/docs/superpowers/plans/2026-05-09-config-editor.md deleted file mode 100644 index ed5e0a0..0000000 --- a/docs/superpowers/plans/2026-05-09-config-editor.md +++ /dev/null @@ -1,2130 +0,0 @@ -# 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 `