lint.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420
  1. """
  2. WSGI Protocol Linter
  3. ====================
  4. This module provides a middleware that performs sanity checks on the
  5. behavior of the WSGI server and application. It checks that the
  6. :pep:`3333` WSGI spec is properly implemented. It also warns on some
  7. common HTTP errors such as non-empty responses for 304 status codes.
  8. .. autoclass:: LintMiddleware
  9. :copyright: 2007 Pallets
  10. :license: BSD-3-Clause
  11. """
  12. import typing as t
  13. from types import TracebackType
  14. from urllib.parse import urlparse
  15. from warnings import warn
  16. from ..datastructures import Headers
  17. from ..http import is_entity_header
  18. from ..wsgi import FileWrapper
  19. if t.TYPE_CHECKING:
  20. from _typeshed.wsgi import StartResponse
  21. from _typeshed.wsgi import WSGIApplication
  22. from _typeshed.wsgi import WSGIEnvironment
  23. class WSGIWarning(Warning):
  24. """Warning class for WSGI warnings."""
  25. class HTTPWarning(Warning):
  26. """Warning class for HTTP warnings."""
  27. def check_type(context: str, obj: object, need: t.Type = str) -> None:
  28. if type(obj) is not need:
  29. warn(
  30. f"{context!r} requires {need.__name__!r}, got {type(obj).__name__!r}.",
  31. WSGIWarning,
  32. stacklevel=3,
  33. )
  34. class InputStream:
  35. def __init__(self, stream: t.IO[bytes]) -> None:
  36. self._stream = stream
  37. def read(self, *args: t.Any) -> bytes:
  38. if len(args) == 0:
  39. warn(
  40. "WSGI does not guarantee an EOF marker on the input stream, thus making"
  41. " calls to 'wsgi.input.read()' unsafe. Conforming servers may never"
  42. " return from this call.",
  43. WSGIWarning,
  44. stacklevel=2,
  45. )
  46. elif len(args) != 1:
  47. warn(
  48. "Too many parameters passed to 'wsgi.input.read()'.",
  49. WSGIWarning,
  50. stacklevel=2,
  51. )
  52. return self._stream.read(*args)
  53. def readline(self, *args: t.Any) -> bytes:
  54. if len(args) == 0:
  55. warn(
  56. "Calls to 'wsgi.input.readline()' without arguments are unsafe. Use"
  57. " 'wsgi.input.read()' instead.",
  58. WSGIWarning,
  59. stacklevel=2,
  60. )
  61. elif len(args) == 1:
  62. warn(
  63. "'wsgi.input.readline()' was called with a size hint. WSGI does not"
  64. " support this, although it's available on all major servers.",
  65. WSGIWarning,
  66. stacklevel=2,
  67. )
  68. else:
  69. raise TypeError("Too many arguments passed to 'wsgi.input.readline()'.")
  70. return self._stream.readline(*args)
  71. def __iter__(self) -> t.Iterator[bytes]:
  72. try:
  73. return iter(self._stream)
  74. except TypeError:
  75. warn("'wsgi.input' is not iterable.", WSGIWarning, stacklevel=2)
  76. return iter(())
  77. def close(self) -> None:
  78. warn("The application closed the input stream!", WSGIWarning, stacklevel=2)
  79. self._stream.close()
  80. class ErrorStream:
  81. def __init__(self, stream: t.IO[str]) -> None:
  82. self._stream = stream
  83. def write(self, s: str) -> None:
  84. check_type("wsgi.error.write()", s, str)
  85. self._stream.write(s)
  86. def flush(self) -> None:
  87. self._stream.flush()
  88. def writelines(self, seq: t.Iterable[str]) -> None:
  89. for line in seq:
  90. self.write(line)
  91. def close(self) -> None:
  92. warn("The application closed the error stream!", WSGIWarning, stacklevel=2)
  93. self._stream.close()
  94. class GuardedWrite:
  95. def __init__(self, write: t.Callable[[bytes], object], chunks: t.List[int]) -> None:
  96. self._write = write
  97. self._chunks = chunks
  98. def __call__(self, s: bytes) -> None:
  99. check_type("write()", s, bytes)
  100. self._write(s)
  101. self._chunks.append(len(s))
  102. class GuardedIterator:
  103. def __init__(
  104. self,
  105. iterator: t.Iterable[bytes],
  106. headers_set: t.Tuple[int, Headers],
  107. chunks: t.List[int],
  108. ) -> None:
  109. self._iterator = iterator
  110. self._next = iter(iterator).__next__
  111. self.closed = False
  112. self.headers_set = headers_set
  113. self.chunks = chunks
  114. def __iter__(self) -> "GuardedIterator":
  115. return self
  116. def __next__(self) -> bytes:
  117. if self.closed:
  118. warn("Iterated over closed 'app_iter'.", WSGIWarning, stacklevel=2)
  119. rv = self._next()
  120. if not self.headers_set:
  121. warn(
  122. "The application returned before it started the response.",
  123. WSGIWarning,
  124. stacklevel=2,
  125. )
  126. check_type("application iterator items", rv, bytes)
  127. self.chunks.append(len(rv))
  128. return rv
  129. def close(self) -> None:
  130. self.closed = True
  131. if hasattr(self._iterator, "close"):
  132. self._iterator.close() # type: ignore
  133. if self.headers_set:
  134. status_code, headers = self.headers_set
  135. bytes_sent = sum(self.chunks)
  136. content_length = headers.get("content-length", type=int)
  137. if status_code == 304:
  138. for key, _value in headers:
  139. key = key.lower()
  140. if key not in ("expires", "content-location") and is_entity_header(
  141. key
  142. ):
  143. warn(
  144. f"Entity header {key!r} found in 304 response.", HTTPWarning
  145. )
  146. if bytes_sent:
  147. warn("304 responses must not have a body.", HTTPWarning)
  148. elif 100 <= status_code < 200 or status_code == 204:
  149. if content_length != 0:
  150. warn(
  151. f"{status_code} responses must have an empty content length.",
  152. HTTPWarning,
  153. )
  154. if bytes_sent:
  155. warn(f"{status_code} responses must not have a body.", HTTPWarning)
  156. elif content_length is not None and content_length != bytes_sent:
  157. warn(
  158. "Content-Length and the number of bytes sent to the"
  159. " client do not match.",
  160. WSGIWarning,
  161. )
  162. def __del__(self) -> None:
  163. if not self.closed:
  164. try:
  165. warn(
  166. "Iterator was garbage collected before it was closed.", WSGIWarning
  167. )
  168. except Exception:
  169. pass
  170. class LintMiddleware:
  171. """Warns about common errors in the WSGI and HTTP behavior of the
  172. server and wrapped application. Some of the issues it checks are:
  173. - invalid status codes
  174. - non-bytes sent to the WSGI server
  175. - strings returned from the WSGI application
  176. - non-empty conditional responses
  177. - unquoted etags
  178. - relative URLs in the Location header
  179. - unsafe calls to wsgi.input
  180. - unclosed iterators
  181. Error information is emitted using the :mod:`warnings` module.
  182. :param app: The WSGI application to wrap.
  183. .. code-block:: python
  184. from werkzeug.middleware.lint import LintMiddleware
  185. app = LintMiddleware(app)
  186. """
  187. def __init__(self, app: "WSGIApplication") -> None:
  188. self.app = app
  189. def check_environ(self, environ: "WSGIEnvironment") -> None:
  190. if type(environ) is not dict:
  191. warn(
  192. "WSGI environment is not a standard Python dict.",
  193. WSGIWarning,
  194. stacklevel=4,
  195. )
  196. for key in (
  197. "REQUEST_METHOD",
  198. "SERVER_NAME",
  199. "SERVER_PORT",
  200. "wsgi.version",
  201. "wsgi.input",
  202. "wsgi.errors",
  203. "wsgi.multithread",
  204. "wsgi.multiprocess",
  205. "wsgi.run_once",
  206. ):
  207. if key not in environ:
  208. warn(
  209. f"Required environment key {key!r} not found",
  210. WSGIWarning,
  211. stacklevel=3,
  212. )
  213. if environ["wsgi.version"] != (1, 0):
  214. warn("Environ is not a WSGI 1.0 environ.", WSGIWarning, stacklevel=3)
  215. script_name = environ.get("SCRIPT_NAME", "")
  216. path_info = environ.get("PATH_INFO", "")
  217. if script_name and script_name[0] != "/":
  218. warn(
  219. f"'SCRIPT_NAME' does not start with a slash: {script_name!r}",
  220. WSGIWarning,
  221. stacklevel=3,
  222. )
  223. if path_info and path_info[0] != "/":
  224. warn(
  225. f"'PATH_INFO' does not start with a slash: {path_info!r}",
  226. WSGIWarning,
  227. stacklevel=3,
  228. )
  229. def check_start_response(
  230. self,
  231. status: str,
  232. headers: t.List[t.Tuple[str, str]],
  233. exc_info: t.Optional[
  234. t.Tuple[t.Type[BaseException], BaseException, TracebackType]
  235. ],
  236. ) -> t.Tuple[int, Headers]:
  237. check_type("status", status, str)
  238. status_code_str = status.split(None, 1)[0]
  239. if len(status_code_str) != 3 or not status_code_str.isdecimal():
  240. warn("Status code must be three digits.", WSGIWarning, stacklevel=3)
  241. if len(status) < 4 or status[3] != " ":
  242. warn(
  243. f"Invalid value for status {status!r}. Valid status strings are three"
  244. " digits, a space and a status explanation.",
  245. WSGIWarning,
  246. stacklevel=3,
  247. )
  248. status_code = int(status_code_str)
  249. if status_code < 100:
  250. warn("Status code < 100 detected.", WSGIWarning, stacklevel=3)
  251. if type(headers) is not list:
  252. warn("Header list is not a list.", WSGIWarning, stacklevel=3)
  253. for item in headers:
  254. if type(item) is not tuple or len(item) != 2:
  255. warn("Header items must be 2-item tuples.", WSGIWarning, stacklevel=3)
  256. name, value = item
  257. if type(name) is not str or type(value) is not str:
  258. warn(
  259. "Header keys and values must be strings.", WSGIWarning, stacklevel=3
  260. )
  261. if name.lower() == "status":
  262. warn(
  263. "The status header is not supported due to"
  264. " conflicts with the CGI spec.",
  265. WSGIWarning,
  266. stacklevel=3,
  267. )
  268. if exc_info is not None and not isinstance(exc_info, tuple):
  269. warn("Invalid value for exc_info.", WSGIWarning, stacklevel=3)
  270. headers = Headers(headers)
  271. self.check_headers(headers)
  272. return status_code, headers
  273. def check_headers(self, headers: Headers) -> None:
  274. etag = headers.get("etag")
  275. if etag is not None:
  276. if etag.startswith(("W/", "w/")):
  277. if etag.startswith("w/"):
  278. warn(
  279. "Weak etag indicator should be upper case.",
  280. HTTPWarning,
  281. stacklevel=4,
  282. )
  283. etag = etag[2:]
  284. if not (etag[:1] == etag[-1:] == '"'):
  285. warn("Unquoted etag emitted.", HTTPWarning, stacklevel=4)
  286. location = headers.get("location")
  287. if location is not None:
  288. if not urlparse(location).netloc:
  289. warn(
  290. "Absolute URLs required for location header.",
  291. HTTPWarning,
  292. stacklevel=4,
  293. )
  294. def check_iterator(self, app_iter: t.Iterable[bytes]) -> None:
  295. if isinstance(app_iter, bytes):
  296. warn(
  297. "The application returned a bytestring. The response will send one"
  298. " character at a time to the client, which will kill performance."
  299. " Return a list or iterable instead.",
  300. WSGIWarning,
  301. stacklevel=3,
  302. )
  303. def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Iterable[bytes]:
  304. if len(args) != 2:
  305. warn("A WSGI app takes two arguments.", WSGIWarning, stacklevel=2)
  306. if kwargs:
  307. warn(
  308. "A WSGI app does not take keyword arguments.", WSGIWarning, stacklevel=2
  309. )
  310. environ: "WSGIEnvironment" = args[0]
  311. start_response: "StartResponse" = args[1]
  312. self.check_environ(environ)
  313. environ["wsgi.input"] = InputStream(environ["wsgi.input"])
  314. environ["wsgi.errors"] = ErrorStream(environ["wsgi.errors"])
  315. # Hook our own file wrapper in so that applications will always
  316. # iterate to the end and we can check the content length.
  317. environ["wsgi.file_wrapper"] = FileWrapper
  318. headers_set: t.List[t.Any] = []
  319. chunks: t.List[int] = []
  320. def checking_start_response(
  321. *args: t.Any, **kwargs: t.Any
  322. ) -> t.Callable[[bytes], None]:
  323. if len(args) not in {2, 3}:
  324. warn(
  325. f"Invalid number of arguments: {len(args)}, expected 2 or 3.",
  326. WSGIWarning,
  327. stacklevel=2,
  328. )
  329. if kwargs:
  330. warn("'start_response' does not take keyword arguments.", WSGIWarning)
  331. status: str = args[0]
  332. headers: t.List[t.Tuple[str, str]] = args[1]
  333. exc_info: t.Optional[
  334. t.Tuple[t.Type[BaseException], BaseException, TracebackType]
  335. ] = (args[2] if len(args) == 3 else None)
  336. headers_set[:] = self.check_start_response(status, headers, exc_info)
  337. return GuardedWrite(start_response(status, headers, exc_info), chunks)
  338. app_iter = self.app(environ, t.cast("StartResponse", checking_start_response))
  339. self.check_iterator(app_iter)
  340. return GuardedIterator(
  341. app_iter, t.cast(t.Tuple[int, Headers], headers_set), chunks
  342. )