40 lines
1.5 KiB
Python
40 lines
1.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from yuxi.channel.domain.port.external.channel_request_verifier_port import (
|
||
|
|
VerifyResult,
|
||
|
|
)
|
||
|
|
from yuxi.channel.domain.port.external.authentication_port import (
|
||
|
|
AuthenticationPort,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class WebRequestVerifier:
|
||
|
|
def __init__(self, auth_service: AuthenticationPort | None = None, *, allow_anonymous: bool = False):
|
||
|
|
self._auth_service = auth_service
|
||
|
|
self._allow_anonymous = allow_anonymous
|
||
|
|
|
||
|
|
@property
|
||
|
|
def channel_type(self) -> str:
|
||
|
|
return "web"
|
||
|
|
|
||
|
|
@property
|
||
|
|
def enabled(self) -> bool:
|
||
|
|
return self._auth_service is not None
|
||
|
|
|
||
|
|
async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult:
|
||
|
|
auth_header = headers.get("authorization", "")
|
||
|
|
if not auth_header:
|
||
|
|
if self._allow_anonymous:
|
||
|
|
return VerifyResult(passed=True, method="web_none", reason="no auth header, anonymous allowed")
|
||
|
|
if not self._auth_service:
|
||
|
|
return VerifyResult(passed=True, method="web_none", reason="no auth_service configured")
|
||
|
|
return VerifyResult(passed=False, method="web_bearer", reason="no credentials configured")
|
||
|
|
|
||
|
|
if not self._auth_service:
|
||
|
|
return VerifyResult(passed=True, method="web_none", reason="no auth_service configured")
|
||
|
|
|
||
|
|
passed, reason = await self._auth_service.authenticate(auth_header)
|
||
|
|
if passed:
|
||
|
|
return VerifyResult(passed=True, method="web_bearer")
|
||
|
|
return VerifyResult(passed=False, method="web_bearer", reason=reason)
|