174 lines
7.2 KiB
Python
174 lines
7.2 KiB
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"]
|
|
|
|
|
|
# ---- 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
|
|
|
|
|
|
def test_write_path_preserves_masked_password(tmp_path):
|
|
"""The "•••" sentinel must preserve the existing hash, not write "•••" to disk."""
|
|
cfg = tmp_path / ".hb.yaml"
|
|
original_hash = "pbkdf2:sha256:original_hash"
|
|
cfg.write_text(
|
|
f"hbd_port: 50004\nusers:\n alice:\n full_name: Alice\n admin: true\n password: {original_hash}\n"
|
|
)
|
|
from hbd.server import configio
|
|
from hbd.server import users as users_mod
|
|
data = configio.read_roundtrip(str(cfg))
|
|
# Simulate what api_config_post does when client sends "•••" back
|
|
existing_users = data.get("users") or {}
|
|
users_payload = {"alice": {"full_name": "Alice", "admin": True, "password": "•••"}}
|
|
for username, attrs in users_payload.items():
|
|
pw = attrs.get("password", "")
|
|
if pw and pw != "•••" and not pw.startswith("pbkdf2:"):
|
|
attrs["password"] = users_mod.hash_password(pw)
|
|
elif not pw or pw == "•••":
|
|
existing_hash = (existing_users.get(username) or {}).get("password", "")
|
|
if existing_hash:
|
|
attrs["password"] = existing_hash
|
|
else:
|
|
attrs.pop("password", None)
|
|
configio.apply_structured_section(data, "users", users_payload)
|
|
configio.write_config(str(cfg), data)
|
|
data2 = configio.read_roundtrip(str(cfg))
|
|
assert data2["users"]["alice"]["password"] == original_hash, (
|
|
f"Expected original hash preserved, got: {data2['users']['alice']['password']!r}"
|
|
)
|
|
|
|
|
|
def test_write_path_preserves_oauth_client_secret(tmp_path):
|
|
"""The "•••" sentinel for oauth client_secret must preserve the existing secret."""
|
|
cfg = tmp_path / ".hb.yaml"
|
|
original_secret = "real_client_secret_value"
|
|
cfg.write_text(
|
|
f"hbd_port: 50004\noauth:\n gitea:\n type: gitea\n url: https://git.example.com\n"
|
|
f" client_id: cid123\n client_secret: {original_secret}\n"
|
|
)
|
|
from hbd.server import configio
|
|
data = configio.read_roundtrip(str(cfg))
|
|
# Simulate what api_config_post does when client sends "•••" back for client_secret
|
|
existing_oauth = data.get("oauth") or {}
|
|
new_oauth = {"gitea": {"type": "gitea", "url": "https://git.example.com", "client_id": "cid123", "client_secret": "•••"}}
|
|
for name, attrs in new_oauth.items():
|
|
cs = attrs.get("client_secret", "")
|
|
if not cs or cs == "•••":
|
|
existing_cs = (existing_oauth.get(name) or {}).get("client_secret", "")
|
|
if existing_cs:
|
|
attrs["client_secret"] = existing_cs
|
|
else:
|
|
attrs.pop("client_secret", None)
|
|
data["oauth"] = new_oauth
|
|
configio.write_config(str(cfg), data)
|
|
data2 = configio.read_roundtrip(str(cfg))
|
|
assert data2["oauth"]["gitea"]["client_secret"] == original_secret, (
|
|
f"Expected original secret preserved, got: {data2['oauth']['gitea']['client_secret']!r}"
|
|
)
|