Coverage for src/lexigram/web/middleware/sanitization.py: 38%

1"""Input sanitization middleware. 

2 

3Strips dangerous characters and patterns from query parameters and path 

4parameters before they reach application code. It does **not** inspect 

5request bodies — callers should validate/sanitize body payloads at the 

6service layer. 

7 

8Usage:: 

9 

10 from lexigram.web.middleware.sanitization import InputSanitizationMiddleware 

11 

12 app.add_middleware(InputSanitizationMiddleware) 

13 

14Or via the WebProvider middleware stack by adding it to ``extra_middleware``. 

15""" 

16 

17from __future__ import annotations 

18 

19import re 

20from typing import Any 

21import urllib.parse 

22 

23from lexigram.logging import get_logger 

24 

25logger = get_logger(__name__) 

26 

27# Patterns that are typically dangerous in query/path parameters. 

28_NULL_BYTES_RE = re.compile(r"\x00") 

29 

30# Very basic XSS / script-injection check: tags and javascript: pseudo-URIs. 

31_SCRIPT_TAG_RE = re.compile(r"<\s*script", re.IGNORECASE) 

32_JS_URI_RE = re.compile(r"javascript\s*:", re.IGNORECASE) 

33 

34 

35def _sanitize_value(value: str) -> str: 

36 """Remove null bytes and flag obvious script-injection patterns. 

37 

38 Args: 

39 value: Raw string value from a query or path parameter. 

40 

41 Returns: 

42 Sanitized string with null bytes stripped. Returns an empty 

43 string when an obvious injection pattern is detected and the 

44 request is not forwarded with that parameter intact. 

45 """ 

46 # Strip null bytes unconditionally — they are never legitimate in URLs. 

47 value = _NULL_BYTES_RE.sub("", value) 

48 

49 # Drop the entire value for obvious script injection. 

50 if _SCRIPT_TAG_RE.search(value) or _JS_URI_RE.search(value): 

51 logger.warning( 

52 "security.input_sanitization.blocked", 

53 reason="script_injection", 

54 value_prefix=value[:50], 

55 ) 

56 return "" 

57 

58 return value 

59 

60 

61class InputSanitizationMiddleware: 

62 """ASGI middleware that sanitizes query string parameters. 

63 

64 Strips null bytes and rejects obvious script-injection patterns from 

65 query parameters before they reach request handlers. Path parameters 

66 are not modified because they have been parsed and validated by the 

67 router before they hit middleware. 

68 

69 This is a **defense-in-depth** measure, not a substitute for 

70 validation at the service layer. 

71 

72 Args: 

73 app: The ASGI application to wrap. 

74 sanitize_query_params: Whether to sanitize query string values 

75 (default: ``True``). 

76 """ 

77 

78 def __init__( 

79 self, 

80 app: Any, 

81 sanitize_query_params: bool = True, 

82 ) -> None: 

83 self.app = app 

84 self.sanitize_query_params = sanitize_query_params 

85 

86 async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: 

87 if scope["type"] != "http" or not self.sanitize_query_params: 

88 await self.app(scope, receive, send) 

89 return 

90 

91 raw_query: bytes = scope.get("query_string", b"") 

92 if raw_query: 

93 sanitized_scope = dict(scope) 

94 sanitized_scope["query_string"] = self._sanitize_query(raw_query) 

95 await self.app(sanitized_scope, receive, send) 

96 else: 

97 await self.app(scope, receive, send) 

98 

99 def _sanitize_query(self, raw: bytes) -> bytes: 

100 """Sanitize the raw query string bytes. 

101 

102 Args: 

103 raw: Raw ``QUERY_STRING`` bytes from the ASGI scope. 

104 

105 Returns: 

106 Sanitized query string bytes with dangerous values replaced. 

107 """ 

108 try: 

109 decoded = raw.decode("utf-8", errors="replace") 

110 except (UnicodeDecodeError, AttributeError): 

111 return raw 

112 

113 params = urllib.parse.parse_qsl(decoded, keep_blank_values=True) 

114 sanitized = [(k, _sanitize_value(v)) for k, v in params] 

115 return urllib.parse.urlencode(sanitized).encode("utf-8")