🚧 WIP
This commit is contained in:
@@ -63,6 +63,45 @@ def test_disable_enable_delete_client(admin_client):
|
||||
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"})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user