From 60c692cefc89d66ef7890db32d845a5880aa42b5 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Sat, 9 May 2026 11:45:09 -0400 Subject: [PATCH] feat: add PUT /api/0/users/me for user self-service profile updates Allows any authenticated user to update their own full_name, avatar, notification_channels, and password via the config YAML write path. Co-Authored-By: Claude Sonnet 4.6 --- hbd/server/http.py | 57 +++++++++++++++++++++++++ tests/test_http_users_me.py | 85 +++++++++++++++++++++++++++++++++++++ 2 files changed, 142 insertions(+) create mode 100644 tests/test_http_users_me.py diff --git a/hbd/server/http.py b/hbd/server/http.py index 542bfda..59bd916 100644 --- a/hbd/server/http.py +++ b/hbd/server/http.py @@ -1173,6 +1173,62 @@ async def start( return web.json_response({"ok": True}) + 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) + + if not isinstance(body, dict): + 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}) + app = web.Application() app.add_routes( [ @@ -1189,6 +1245,7 @@ async def start( # Users web.get("/api/0/users", api_users), web.get("/api/0/users/me", api_user_self), + web.put("/api/0/users/me", api_user_self_put), web.get("/api/0/users/{username}/avatar", api_user_avatar), # Config API (admin) web.get("/api/0/config", api_config_get), diff --git a/tests/test_http_users_me.py b/tests/test_http_users_me.py new file mode 100644 index 0000000..30d3db4 --- /dev/null +++ b/tests/test_http_users_me.py @@ -0,0 +1,85 @@ +"""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"]