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 <noreply@anthropic.com>
This commit is contained in:
@@ -1173,6 +1173,62 @@ async def start(
|
|||||||
|
|
||||||
return web.json_response({"ok": True})
|
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 = web.Application()
|
||||||
app.add_routes(
|
app.add_routes(
|
||||||
[
|
[
|
||||||
@@ -1189,6 +1245,7 @@ async def start(
|
|||||||
# Users
|
# Users
|
||||||
web.get("/api/0/users", api_users),
|
web.get("/api/0/users", api_users),
|
||||||
web.get("/api/0/users/me", api_user_self),
|
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),
|
web.get("/api/0/users/{username}/avatar", api_user_avatar),
|
||||||
# Config API (admin)
|
# Config API (admin)
|
||||||
web.get("/api/0/config", api_config_get),
|
web.get("/api/0/config", api_config_get),
|
||||||
|
|||||||
@@ -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"]
|
||||||
Reference in New Issue
Block a user