feat(auth): query-string token for headerless webhook senders
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
"""``require_token`` accepts ``?token=`` only when no auth header is present."""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from starlette.requests import Request
|
||||
|
||||
from beaver_gateway.core.auth import TokenStore
|
||||
from beaver_gateway.frontends._auth import require_token
|
||||
|
||||
|
||||
def _request(query: str = "", headers: dict[str, str] | None = None) -> Request:
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "POST",
|
||||
"path": "/hooks/komodo",
|
||||
"query_string": query.encode(),
|
||||
"headers": [
|
||||
(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()
|
||||
],
|
||||
}
|
||||
return Request(scope)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runtime() -> SimpleNamespace:
|
||||
return SimpleNamespace(token_store=TokenStore(bootstrap={"komodo": "qs-secret"}))
|
||||
|
||||
|
||||
async def test_query_token_authorizes_a_headerless_webhook(runtime) -> None:
|
||||
name = await require_token(_request("token=qs-secret"), runtime, scope="api")
|
||||
assert name == "komodo"
|
||||
|
||||
|
||||
async def test_header_wins_over_query(runtime) -> None:
|
||||
bad_header = _request("token=qs-secret", {"Authorization": "Bearer nope"})
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await require_token(bad_header, runtime, scope="api")
|
||||
assert exc.value.status_code == 401
|
||||
|
||||
|
||||
async def test_missing_everything_is_401(runtime) -> None:
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await require_token(_request(), runtime, scope="api")
|
||||
assert exc.value.status_code == 401
|
||||
Reference in New Issue
Block a user