280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""API tests: authentication, RBAC, client + ACL flows (spec section 64)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from tests.conftest import login
|
|
|
|
|
|
def test_health_live(client):
|
|
response = client.get("/health/live")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_login_sets_cookie_and_session(client):
|
|
response = login(client, "admin", "admin-password-123")
|
|
assert response.json()["role"] == "super_admin"
|
|
|
|
session = client.get("/api/v1/auth/session")
|
|
assert session.status_code == 200
|
|
assert session.json()["username"] == "admin"
|
|
|
|
|
|
def test_login_wrong_password_audited(client):
|
|
response = client.post(
|
|
"/api/v1/auth/login", json={"username": "admin", "password": "wrong"}
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_unauthenticated_api_rejected(client):
|
|
response = client.get("/api/v1/mqtt/clients")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_auditor_cannot_create_clients(auditor_client):
|
|
response = auditor_client.post(
|
|
"/api/v1/mqtt/clients", json={"username": "ps042"}
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_create_client_returns_password_once(admin_client):
|
|
response = admin_client.post("/api/v1/mqtt/clients", json={"username": "ps042"})
|
|
assert response.status_code == 201
|
|
body = response.json()
|
|
assert body["username"] == "ps042"
|
|
assert len(body["password"]) >= 20 # random password generated
|
|
|
|
listing = admin_client.get("/api/v1/mqtt/clients")
|
|
assert any(c["username"] == "ps042" for c in listing.json())
|
|
|
|
|
|
def test_disable_enable_delete_client(admin_client):
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "sensor-1"})
|
|
|
|
response = admin_client.post("/api/v1/mqtt/clients/sensor-1/disable")
|
|
assert response.status_code == 204
|
|
clients = {c["username"]: c for c in admin_client.get("/api/v1/mqtt/clients").json()}
|
|
assert clients["sensor-1"]["enabled"] is False
|
|
|
|
assert admin_client.post("/api/v1/mqtt/clients/sensor-1/enable").status_code == 204
|
|
|
|
response = admin_client.delete("/api/v1/mqtt/clients/sensor-1")
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_management_identity_is_protected(admin_client):
|
|
"""The dynsec admin (MOSQUITTO_DYNSEC_USERNAME) is the backend's own
|
|
service account: hidden from the listing and immune to password reset,
|
|
disable and delete. A reset previously severed the management channel
|
|
(Mosquitto kicks all connections for the username) — hence the guard.
|
|
"""
|
|
# Create "admin" directly so the filter/guard has something to catch.
|
|
# (The FakeDynsec fixture is empty at test start.)
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "admin"})
|
|
|
|
# Hidden from the listing even though dynsec knows the client.
|
|
listing = admin_client.get("/api/v1/mqtt/clients").json()
|
|
assert all(c["username"] != "admin" for c in listing)
|
|
assert admin_client.get("/api/v1/mqtt/clients/admin").status_code == 404
|
|
|
|
# Every mutation targeting it returns 400, not 2xx.
|
|
assert (
|
|
admin_client.post("/api/v1/mqtt/clients/admin/password:reset").status_code
|
|
== 400
|
|
)
|
|
assert admin_client.post("/api/v1/mqtt/clients/admin/disable").status_code == 400
|
|
assert admin_client.delete("/api/v1/mqtt/clients/admin").status_code == 400
|
|
# Creating a client that shadows the identity is refused as well.
|
|
assert (
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "admin"}).status_code
|
|
== 400
|
|
)
|
|
|
|
|
|
def test_password_reset_still_works_for_normal_clients(admin_client):
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "ps042"})
|
|
|
|
reset = admin_client.post("/api/v1/mqtt/clients/ps042/password:reset")
|
|
assert reset.status_code == 200
|
|
body = reset.json()
|
|
assert body["username"] == "ps042"
|
|
assert len(body["password"]) >= 20 # random password, shown once
|
|
|
|
|
|
def test_role_acl_lifecycle(admin_client):
|
|
admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})
|
|
|
|
acls = [
|
|
{
|
|
"acltype": "publishClientSend",
|
|
"topic": "site/%u/telemetry/#",
|
|
"allow": True,
|
|
"priority": 0,
|
|
},
|
|
{
|
|
"acltype": "subscribePattern",
|
|
"topic": "site/%u/config/#",
|
|
"allow": True,
|
|
"priority": 1,
|
|
},
|
|
]
|
|
response = admin_client.put("/api/v1/mqtt/roles/telemetry", json={"acls": acls})
|
|
assert response.status_code == 200
|
|
|
|
roles = {r["rolename"]: r for r in admin_client.get("/api/v1/mqtt/roles").json()}
|
|
assert roles["telemetry"]["acls"] == acls
|
|
|
|
|
|
def test_group_lifecycle(admin_client):
|
|
response = admin_client.post(
|
|
"/api/v1/mqtt/groups", json={"groupname": "sensors"}
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
groups = {g["groupname"]: g for g in admin_client.get("/api/v1/mqtt/groups").json()}
|
|
assert "sensors" in groups
|
|
|
|
assert admin_client.delete("/api/v1/mqtt/groups/sensors").status_code == 204
|
|
|
|
|
|
def test_client_role_assignment(admin_client):
|
|
admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "assign-test"})
|
|
|
|
response = admin_client.post(
|
|
"/api/v1/mqtt/clients/assign-test/roles",
|
|
json={"rolename": "telemetry", "priority": 5},
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
client = admin_client.get("/api/v1/mqtt/clients/assign-test").json()
|
|
roles = {r["rolename"] for r in client["roles"]}
|
|
assert "telemetry" in roles
|
|
|
|
assert (
|
|
admin_client.delete("/api/v1/mqtt/clients/assign-test/roles/telemetry").status_code
|
|
== 204
|
|
)
|
|
client = admin_client.get("/api/v1/mqtt/clients/assign-test").json()
|
|
assert all(r["rolename"] != "telemetry" for r in client["roles"])
|
|
|
|
|
|
def test_group_member_assignment(admin_client):
|
|
admin_client.post("/api/v1/mqtt/groups", json={"groupname": "sensors"})
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "member-test"})
|
|
|
|
response = admin_client.post(
|
|
"/api/v1/mqtt/groups/sensors/clients",
|
|
json={"username": "member-test", "priority": 1},
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
client = admin_client.get("/api/v1/mqtt/clients/member-test").json()
|
|
groups = {g["groupname"] for g in client["groups"]}
|
|
assert "sensors" in groups
|
|
|
|
assert (
|
|
admin_client.delete("/api/v1/mqtt/groups/sensors/clients/member-test").status_code
|
|
== 204
|
|
)
|
|
client = admin_client.get("/api/v1/mqtt/clients/member-test").json()
|
|
assert all(g["groupname"] != "sensors" for g in client["groups"])
|
|
|
|
|
|
def test_group_role_assignment(admin_client):
|
|
admin_client.post("/api/v1/mqtt/groups", json={"groupname": "sensors"})
|
|
admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"})
|
|
|
|
response = admin_client.post(
|
|
"/api/v1/mqtt/groups/sensors/roles",
|
|
json={"rolename": "telemetry", "priority": 2},
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
assert (
|
|
admin_client.delete("/api/v1/mqtt/groups/sensors/roles/telemetry").status_code
|
|
== 204
|
|
)
|
|
|
|
|
|
def test_role_deletion(admin_client):
|
|
"""deleteRole removes the role everywhere (spec section 13): the broker's
|
|
dynsec deleteRole strips it from every client and group that holds it."""
|
|
admin_client.post("/api/v1/mqtt/roles", json={"rolename": "obsolete"})
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "ps042"})
|
|
admin_client.post("/api/v1/mqtt/clients/ps042/roles", json={"rolename": "obsolete"})
|
|
admin_client.post("/api/v1/mqtt/groups", json={"groupname": "legacy"})
|
|
admin_client.post("/api/v1/mqtt/groups/legacy/roles", json={"rolename": "obsolete"})
|
|
|
|
deleted = admin_client.delete("/api/v1/mqtt/roles/obsolete")
|
|
assert deleted.status_code == 204
|
|
|
|
# Role gone from the listing...
|
|
roles = {r["rolename"] for r in admin_client.get("/api/v1/mqtt/roles").json()}
|
|
assert "obsolete" not in roles
|
|
# ...and the broker's cleanup means clients/groups no longer hold it.
|
|
client = admin_client.get("/api/v1/mqtt/clients/ps042").json()
|
|
assert all(r["rolename"] != "obsolete" for r in client["roles"])
|
|
groups = admin_client.get("/api/v1/mqtt/groups").json()
|
|
legacy = next(g for g in groups if g["groupname"] == "legacy")
|
|
assert all(r["rolename"] != "obsolete" for r in legacy["roles"])
|
|
response = admin_client.post(
|
|
"/api/v1/broker/listeners",
|
|
json={"name": "Production MQTT", "port": 1883, "protocol": "mqtt"},
|
|
)
|
|
assert response.status_code == 201
|
|
listener_id = response.json()["id"]
|
|
|
|
preview = admin_client.get("/api/v1/broker/config/preview")
|
|
assert preview.status_code == 200
|
|
config_text = preview.json()["config_text"]
|
|
assert "listener 1883" in config_text
|
|
# Generated config must not use options deprecated in 2.1 (spec 3.1).
|
|
assert "password_file" not in config_text
|
|
assert "acl_file" not in config_text
|
|
assert "per_listener_settings" not in config_text
|
|
|
|
delete = admin_client.delete(f"/api/v1/broker/listeners/{listener_id}")
|
|
assert delete.status_code == 204
|
|
|
|
|
|
def test_listener_update_and_preview(admin_client):
|
|
listener_id = admin_client.post(
|
|
"/api/v1/broker/listeners",
|
|
json={"name": "Production MQTT", "port": 1883, "protocol": "mqtt"},
|
|
).json()["id"]
|
|
|
|
updated = admin_client.put(
|
|
f"/api/v1/broker/listeners/{listener_id}",
|
|
json={
|
|
"name": "Production MQTT",
|
|
"port": 1884,
|
|
"bind_address": "127.0.0.1",
|
|
"protocol": "websockets",
|
|
"allow_anonymous": True,
|
|
},
|
|
)
|
|
assert updated.status_code == 200, updated.text
|
|
body = updated.json()
|
|
assert body["port"] == 1884
|
|
assert body["bind_address"] == "127.0.0.1"
|
|
assert body["protocol"] == "websockets"
|
|
assert body["allow_anonymous"] is True
|
|
|
|
config_text = admin_client.get("/api/v1/broker/config/preview").json()[
|
|
"config_text"
|
|
]
|
|
assert "listener 1884 127.0.0.1" in config_text
|
|
assert "protocol websockets" in config_text
|
|
assert "listener_allow_anonymous true" in config_text
|
|
|
|
|
|
def test_audit_events_recorded(admin_client, db_session):
|
|
from sqlalchemy import select
|
|
|
|
from app.models import AuditEvent
|
|
|
|
admin_client.post("/api/v1/mqtt/clients", json={"username": "audited-1"})
|
|
actions = db_session.scalars(select(AuditEvent.action)).all()
|
|
assert "CLIENT_CREATED" in actions |