(self, scope: Scope, receive: Receive, send: Send)
| 29 | self.www_redirect = www_redirect |
| 30 | |
| 31 | async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: |
| 32 | if self.allow_any or scope["type"] not in ( |
| 33 | "http", |
| 34 | "websocket", |
| 35 | ): # pragma: no cover |
| 36 | await self.app(scope, receive, send) |
| 37 | return |
| 38 | |
| 39 | headers = Headers(scope=scope) |
| 40 | host = headers.get("host", "").split(":")[0] |
| 41 | is_valid_host = False |
| 42 | found_www_redirect = False |
| 43 | for pattern in self.allowed_hosts: |
| 44 | if host == pattern or (pattern.startswith("*") and host.endswith(pattern[1:])): |
| 45 | is_valid_host = True |
| 46 | break |
| 47 | elif "www." + host == pattern: |
| 48 | found_www_redirect = True |
| 49 | |
| 50 | if is_valid_host: |
| 51 | await self.app(scope, receive, send) |
| 52 | else: |
| 53 | response: Response |
| 54 | if found_www_redirect and self.www_redirect: |
| 55 | url = URL(scope=scope) |
| 56 | redirect_url = url.replace(netloc="www." + url.netloc) |
| 57 | response = RedirectResponse(url=str(redirect_url)) |
| 58 | else: |
| 59 | response = PlainTextResponse("Invalid host header", status_code=400) |
| 60 | await response(scope, receive, send) |
nothing calls this directly
no test coverage detected