93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from backup_tool.db.models import IdempotencyRecord
|
|
from sqlalchemy import select
|
|
|
|
PASSWORD = "correct horse battery staple"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_catalog_subscription_and_write_only_webhook_secret(app_client) -> None:
|
|
client, _ = app_client
|
|
setup = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
|
|
assert setup.status_code == 201
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
catalog = await client.get("/api/v2/notifications/event-catalog")
|
|
assert catalog.status_code == 200
|
|
assert catalog.json()["event_schema_version"] == 1
|
|
assert "execution.queued" in catalog.json()["events"]
|
|
created = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "webhook",
|
|
"event_filters": ["execution.*"],
|
|
"destination": {"url": "https://hooks.example.test/backup"},
|
|
"signing_secret": "not-returned-webhook-secret",
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert created.status_code == 201
|
|
assert "signing_secret" not in created.text
|
|
assert "not-returned-webhook-secret" not in created.text
|
|
assert created.headers["ETag"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rotation_idempotency_never_persists_secret_verifier(app_client) -> None:
|
|
client, _ = app_client
|
|
setup = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
|
|
assert setup.status_code == 201
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
created = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "webhook",
|
|
"event_filters": ["execution.*"],
|
|
"destination": {"url": "https://hooks.example.test/backup"},
|
|
"signing_secret": "first-signing-secret",
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert created.status_code == 201
|
|
route = f"/api/v2/notifications/subscriptions/{created.json()['id']}/signing-keys/rotate"
|
|
first = await client.post(
|
|
route,
|
|
json={"secret": "rotation-secret-one", "overlap_seconds": 60},
|
|
headers={"X-CSRF-Token": csrf, "Idempotency-Key": "rotation-one"},
|
|
)
|
|
replay = await client.post(
|
|
route,
|
|
json={"secret": "rotation-secret-two", "overlap_seconds": 60},
|
|
headers={"X-CSRF-Token": csrf, "Idempotency-Key": "rotation-one"},
|
|
)
|
|
assert first.status_code == replay.status_code == 200
|
|
assert first.json() == replay.json()
|
|
app = client._transport.app
|
|
async with app.state.sessions() as db:
|
|
record = await db.scalar(
|
|
select(IdempotencyRecord).where(
|
|
IdempotencyRecord.operation == "rotate_notification_signing_key"
|
|
)
|
|
)
|
|
assert record is not None
|
|
assert "rotation-secret" not in record.request_digest
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notification_rejects_empty_or_unknown_filters(app_client) -> None:
|
|
client, _ = app_client
|
|
setup = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
|
|
assert setup.status_code == 201
|
|
response = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "email",
|
|
"event_filters": ["unknown.event"],
|
|
"destination": {"recipients": ["operator@example.test"]},
|
|
},
|
|
headers={"X-CSRF-Token": client.cookies["backup_tool_csrf"]},
|
|
)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "validation_failed"
|