Coverage for src/lexigram/web/middleware/host.py: 26%

1"""Host-header validation middleware. 

2 

3Rejects requests whose ``Host`` header does not match the configured 

4``SecurityConfig.allowed_hosts`` allowlist. Host-header poisoning enables 

5cache poisoning, password-reset link injection, and DNS-rebinding style 

6attacks — validation must fail closed. 

7 

8An empty allowlist rejects every request, so the middleware is only wired by 

9``MiddlewareSetup`` when ``allowed_hosts`` is non-empty (the production 

10validator requires a non-empty list; in non-production deployments host 

11validation is opt-in via ``SecurityConfig.allowed_hosts``). 

12""" 

13 

14from __future__ import annotations 

15 

16from typing import TYPE_CHECKING 

17 

18if TYPE_CHECKING: 

19 from starlette.types import ASGIApp, Receive, Scope, Send 

20 

21 

22class HostValidationMiddleware: 

23 """ASGI middleware enforcing an explicit host allowlist. 

24 

25 Args: 

26 app: The ASGI application to wrap. 

27 allowed_hosts: Hostnames permitted (compared lower-cased, port 

28 stripped). An empty list rejects all hosts (fail-closed). 

29 """ 

30 

31 def __init__(self, app: ASGIApp, allowed_hosts: list[str]) -> None: 

32 self._app = app 

33 self._allowed_hosts = {host.lower() for host in allowed_hosts} 

34 

35 @staticmethod 

36 def _extract_host(scope: Scope) -> str | None: 

37 """Return the lower-cased hostname (port stripped) or ``None``.""" 

38 host: str | None = None 

39 for name, value in scope.get("headers", []): 

40 if name.lower() == b"host": 

41 host = value.decode("latin-1").strip().lower() 

42 break 

43 if host is None: 

44 return None 

45 if host.startswith("["): # IPv6 literal 

46 if "]" in host: 

47 return host[1 : host.index("]")].lower() 

48 return None 

49 if ":" in host: 

50 candidate = host.rsplit(":", 1) 

51 if candidate[1].isdigit(): 

52 return candidate[0] 

53 return host 

54 

55 async def __call__( 

56 self, 

57 scope: Scope, 

58 receive: Receive, 

59 send: Send, 

60 ) -> None: 

61 """Validate the ``Host`` header before calling the wrapped app. 

62 

63 Args: 

64 scope: The ASGI scope dictionary. 

65 receive: The ASGI receive callable. 

66 send: The ASGI send callable. 

67 """ 

68 if scope.get("type") != "http": 

69 await self._app(scope, receive, send) 

70 return 

71 

72 host = self._extract_host(scope) 

73 if host is None or host not in self._allowed_hosts: 

74 await self._reject(send) 

75 return 

76 

77 await self._app(scope, receive, send) 

78 

79 @staticmethod 

80 async def _reject(send: Send) -> None: 

81 """Send a plain 400 response.""" 

82 body = b"400 Bad Request: invalid Host header" 

83 await send( 

84 { 

85 "type": "http.response.start", 

86 "status": 400, 

87 "headers": [ 

88 (b"content-type", b"text/plain; charset=utf-8"), 

89 (b"content-length", str(len(body)).encode()), 

90 ], 

91 } 

92 ) 

93 await send({"type": "http.response.body", "body": body}) 

94 

95 

96__all__ = ["HostValidationMiddleware"]