feat: webhook event-model und shared-secret auth
This commit is contained in:
12
app/webhook/auth.py
Normal file
12
app/webhook/auth.py
Normal file
@@ -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")
|
||||
14
app/webhook/models.py
Normal file
14
app/webhook/models.py
Normal file
@@ -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
|
||||
31
tests/test_webhook.py
Normal file
31
tests/test_webhook.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user