78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""``require_token`` accepts ``?token=`` only when no auth header is present."""
|
|
|
|
import logging
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from starlette.requests import Request
|
|
|
|
from beaver_gateway.security.auth import TokenStore
|
|
from beaver_gateway.frontends.bearer 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
|
|
|
|
|
|
async def test_bootstrap_entry_can_carry_a_scope() -> None:
|
|
raw = "admin:secret-a,komodo:hook-value:api"
|
|
store = TokenStore(
|
|
bootstrap=TokenStore.parse_bootstrap(raw),
|
|
bootstrap_scopes=TokenStore.parse_bootstrap_scopes(raw),
|
|
)
|
|
admin = await store.verify("secret-a")
|
|
hook = await store.verify("hook-value")
|
|
assert admin is not None and admin.scope == "*"
|
|
assert hook is not None and hook.scope == "api" and hook.name == "komodo"
|
|
assert not hook.allows("admin") and hook.allows("api")
|
|
|
|
|
|
def test_access_log_filter_masks_query_tokens() -> None:
|
|
from beaver_gateway.security.redact import RedactFilter
|
|
|
|
record = logging.LogRecord(
|
|
"uvicorn.access",
|
|
logging.INFO,
|
|
__file__,
|
|
1,
|
|
'%s - "%s %s HTTP/%s" %d',
|
|
("1.2.3.4:1", "POST", "/hooks/komodo?token=s3cret&x=1", "1.1", 202),
|
|
None,
|
|
)
|
|
assert RedactFilter().filter(record)
|
|
assert "s3cret" not in record.getMessage()
|
|
assert "/hooks/komodo?token=<…>&x=1" in record.getMessage()
|