diff --git a/app/webhook/auth.py b/app/webhook/auth.py new file mode 100644 index 0000000..2829c07 --- /dev/null +++ b/app/webhook/auth.py @@ -0,0 +1,12 @@ +import hmac + +from fastapi import HTTPException + + +def verify_secret(provided: str | None, expected: str) -> None: + """Constant-time comparison of the shared secret. + + Raises HTTPException(401) on mismatch or missing header. + """ + if provided is None or not hmac.compare_digest(provided, expected): + raise HTTPException(status_code=401, detail="invalid or missing secret") diff --git a/app/webhook/models.py b/app/webhook/models.py new file mode 100644 index 0000000..1865c62 --- /dev/null +++ b/app/webhook/models.py @@ -0,0 +1,14 @@ +from enum import Enum +from pydantic import BaseModel + + +class EventType(str, Enum): + CREATED = "created" + UPDATED = "updated" + DELETED = "deleted" + + +class NextcloudEvent(BaseModel): + event_type: EventType + file_path: str + file_name: str diff --git a/tests/test_webhook.py b/tests/test_webhook.py new file mode 100644 index 0000000..9ef9212 --- /dev/null +++ b/tests/test_webhook.py @@ -0,0 +1,31 @@ +import pytest +from fastapi import HTTPException + +from app.webhook.models import NextcloudEvent, EventType +from app.webhook.auth import verify_secret + + +def test_event_parses_created(): + evt = NextcloudEvent(event_type="created", file_path="a/b.pdf", file_name="b.pdf") + assert evt.event_type == EventType.CREATED + + +def test_event_invalid_type_raises(): + with pytest.raises(Exception): + NextcloudEvent(event_type="exploded", file_path="a", file_name="a") + + +def test_verify_secret_pass(): + verify_secret(provided="abc", expected="abc") # no exception + + +def test_verify_secret_fail(): + with pytest.raises(HTTPException) as exc_info: + verify_secret(provided="wrong", expected="abc") + assert exc_info.value.status_code == 401 + + +def test_verify_secret_missing_fail(): + with pytest.raises(HTTPException) as exc_info: + verify_secret(provided=None, expected="abc") + assert exc_info.value.status_code == 401