{"record":{"id":"8ea35fbd6392e1cd","repo":"aio-libs/aiohttp","slug":"forbidden-control-character-detected-in-headers-p","errorCode":null,"errorMessage":"Forbidden control character detected in headers. Potential header injection attack.","messagePattern":"Forbidden control character detected in headers\\. Potential header injection attack\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"aiohttp/http_writer.py","lineNumber":374,"sourceCode":"\n        The intended use is to write\n\n          await w.write(data)\n          await w.drain()\n        \"\"\"\n        protocol = self._protocol\n        if protocol.transport is not None and protocol._paused:\n            await protocol._drain_helper()\n\n\n# https://www.rfc-editor.org/info/rfc9110/#section-5.5-5\n# https://www.rfc-editor.org/info/rfc9112/#section-4-3\n_FORBIDDEN_HEADER_CHARS_RE = re.compile(r\"[\\x00-\\x08\\x0a-\\x1f\\x7f]\")\n\n\ndef _safe_header(string: str) -> str:\n    if _FORBIDDEN_HEADER_CHARS_RE.search(string) is not None:\n        raise ValueError(\n            \"Forbidden control character detected in headers. \"\n            \"Potential header injection attack.\"\n        )\n    return string\n\n\ndef _py_serialize_headers(status_line: str, headers: \"CIMultiDict[str]\") -> bytes:\n    _safe_header(status_line)\n    headers_gen = (_safe_header(k) + \": \" + _safe_header(v) for k, v in headers.items())\n    line = status_line + \"\\r\\n\" + \"\\r\\n\".join(headers_gen) + \"\\r\\n\\r\\n\"\n    return line.encode(\"utf-8\")\n\n\n_serialize_headers = _py_serialize_headers\n\ntry:\n    import aiohttp._http_writer as _http_writer  # type: ignore[import-not-found]\n","sourceCodeStart":356,"sourceCodeEnd":392,"githubUrl":"https://github.com/aio-libs/aiohttp/blob/d041d4d0fd48c3f0832084d33be16cf1c4835f85/aiohttp/http_writer.py#L356-L392","documentation":"Raised as a plain ValueError by _safe_header when serializing headers/status-line if any forbidden control character (bytes 0x00-0x08, 0x0a-0x1f, 0x7f) is found. The regex _FORBIDDEN_HEADER_CHARS_RE enforces RFC 9110 §5.5 / RFC 9112 §4.3, preventing header-injection (CRLF injection / response splitting) attacks. Applied to both the status line and every header name/value.","triggerScenarios":"Application code sets a header name or value (or a status line) containing a raw CR/LF/NUL or other control byte, e.g. including '\\r\\n' inside a cookie/Location/User-Agent value. _py_serialize_headers runs _safe_header on each component and raises before any bytes hit the wire.","commonSituations":"User-controlled input placed into a header without sanitization (reflected XSS / injection vector); a Location header built from a query param containing newlines; logging/correlation IDs containing control chars; a server-side header value computed from untrusted data; tests with literal newlines in headers.","solutions":["Sanitize any user-controlled value before placing it in a header: strip/reject CR, LF, NUL, and other control bytes.","Use URL-encoding for arbitrary data passed in headers.","Validate header values against _FORBIDDEN_HEADER_CHARS_RE (or [\\x20-\\x7e] plus tab) before setting them.","For redirects, build Location from a safelist or validated URL, never raw input.","Keep the security check in _safe_header enabled; do not bypass it."],"exampleFix":"# before\r\nimport re\r\nasync def handler(request):\r\n    name = request.query.get('name', '')\r\n    return web.Response(headers={'X-Name': name})  # CRLF in name -> ValueError\r\n\r\n# after\r\nimport re\r\n_FORBIDDEN = re.compile(r'[\\x00-\\x08\\x0a-\\x1f\\x7f]')\r\nasync def handler(request):\r\n    name = _FORBIDDEN.sub('', request.query.get('name', ''))\r\n    return web.Response(headers={'X-Name': name})","handlingStrategy":"validation","validationCode":"import re\n_FORBIDDEN_HEADER_CHARS_RE = re.compile(r'[\\x00-\\x08\\x0a-\\x1f\\x7f]')\n\ndef safe_header_value(value: str) -> str | None:\n    if _FORBIDDEN_HEADER_CHARS_RE.search(value) is None:\n        return value\n    return None  # or: return _FORBIDDEN_HEADER_CHARS_RE.sub('', value)\n\ndef validate_headers(headers: dict[str, str]) -> dict[str, str]:\n    out = {}\n    for k, v in headers.items():\n        if _FORBIDDEN_HEADER_CHARS_RE.search(k) is not None:\n            raise ValueError(f'Forbidden chars in header name {k!r}')\n        cleaned = safe_header_value(v)\n        if cleaned is None:\n            raise ValueError(f'Forbidden chars in header value for {k!r}')\n        out[k] = cleaned\n    return out","typeGuard":null,"tryCatchPattern":"try:\n    resp = web.Response(headers={'X-Name': user_input})\n    return resp\nexcept ValueError:\n    # ValueError from _safe_header during serialization\n    return web.Response(status=400, text='Invalid header input')","preventionTips":["Sanitize all user-controlled input before placing it in headers (strip CR/LF/NUL/control).","URL-encode arbitrary data passed via headers.","Build redirect Location from validated/safelisted URLs only.","Never bypass _safe_header; treat the ValueError as a real attack signal."],"tags":["http-writer","header-injection","security","crlf-injection","validation"],"backgroundTag":null,"analyzedSha":"d041d4d0fd48c3f0832084d33be16cf1c4835f85","analyzedAt":"2026-08-11T20:44:15.550Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}