Restrict reason (#12209)

Co-authored-by: Dhiral Vyas <dhiral@users.noreply.github.com>

Author: Dreamsorcerer

aiohttp/_http_writer.pyx

@@ -131,7 +131,7 @@ def _serialize_headers(str status_line, headers):
     _init_writer(&writer, buf)
 
     try:
-        if _write_str(&writer, status_line) < 0:
+        if _write_str_raise_on_nlcr(&writer, status_line) < 0:
             raise
         if _write_byte(&writer, b'\r') < 0:
             raise

aiohttp/http_writer.py

@@ -373,6 +373,7 @@ def _safe_header(string: str) -> str:
 
 
 def _py_serialize_headers(status_line: str, headers: "CIMultiDict[str]") -> bytes:
+    _safe_header(status_line)
     headers_gen = (_safe_header(k) + ": " + _safe_header(v) for k, v in headers.items())
     line = status_line + "\r\n" + "\r\n".join(headers_gen) + "\r\n\r\n"
     return line.encode("utf-8")

aiohttp/web_exceptions.py

@@ -99,8 +99,8 @@ def __init__(
     ) -> None:
         if reason is None:
             reason = self.default_reason
-        elif "\n" in reason:
-            raise ValueError("Reason cannot contain \\n")
+        elif "\r" in reason or "\n" in reason:
+            raise ValueError("Reason cannot contain \\r or \\n")
 
         if text is None:
             if not self.empty_body:

aiohttp/web_response.py

@@ -157,8 +157,8 @@ def _set_status(self, status: int, reason: str | None) -> None:
         self._status = status
         if reason is None:
             reason = REASON_PHRASES.get(self._status, "")
-        elif "\n" in reason:
-            raise ValueError("Reason cannot contain \\n")
+        elif "\r" in reason or "\n" in reason:
+            raise ValueError("Reason cannot contain \\r or \\n")
         self._reason = reason
 
     @property

tests/test_web_exceptions.py

@@ -185,9 +185,17 @@ def test_ctor_all(self) -> None:
         assert resp.status == 200
 
     def test_multiline_reason(self) -> None:
-        with pytest.raises(ValueError, match=r"Reason cannot contain \\n"):
+        with pytest.raises(ValueError, match=r"Reason cannot contain"):
             web.HTTPOk(reason="Bad\r\nInjected-header: foo")
 
+    def test_reason_with_cr(self) -> None:
+        with pytest.raises(ValueError, match=r"Reason cannot contain"):
+            web.HTTPOk(reason="OK\rSet-Cookie: evil=1")
+
+    def test_reason_with_lf(self) -> None:
+        with pytest.raises(ValueError, match=r"Reason cannot contain"):
+            web.HTTPOk(reason="OK\nSet-Cookie: evil=1")
+
     def test_pickle(self) -> None:
         resp = web.HTTPOk(
             headers={"X-Custom": "value"},

tests/test_web_response.py

@@ -934,6 +934,27 @@ def test_set_status_with_empty_reason() -> None:
     assert resp.reason == ""
 
 
+def test_set_status_reason_with_cr() -> None:
+    resp = web.StreamResponse()
+
+    with pytest.raises(ValueError, match="Reason cannot contain"):
+        resp.set_status(200, "OK\rSet-Cookie: evil=1")
+
+
+def test_set_status_reason_with_lf() -> None:
+    resp = web.StreamResponse()
+
+    with pytest.raises(ValueError, match="Reason cannot contain"):
+        resp.set_status(200, "OK\nSet-Cookie: evil=1")
+
+
+def test_set_status_reason_with_crlf() -> None:
+    resp = web.StreamResponse()
+
+    with pytest.raises(ValueError, match="Reason cannot contain"):
+        resp.set_status(200, "OK\r\nSet-Cookie: evil=1")
+
+
 async def test_start_force_close() -> None:
     req = make_request("GET", "/")
     resp = web.StreamResponse()
@@ -1236,7 +1257,7 @@ async def test_render_with_body(buf: bytearray, writer: AbstractStreamWriter) ->
 
 
 async def test_multiline_reason(buf: bytearray, writer: AbstractStreamWriter) -> None:
-    with pytest.raises(ValueError, match=r"Reason cannot contain \\n"):
+    with pytest.raises(ValueError, match=r"Reason cannot contain \\r or \\n"):
         web.Response(reason="Bad\r\nInjected-header: foo")