When developing applications or integrations with aiohttp, developers must assume full responsibility for boundary enforcement, request validation, resource limits, and credential safety. While aiohttp provides robust parsing and built-in connection primitives, it does not implicitly validate authorization, sanitize untrusted file paths, or prevent resource exhaustion from large payloads. Security-sensitive surfaces including request headers, cookies, routing paths, file uploads, and proxy configurations must fail closed when invalid or unauthorized data is encountered.
Essential implementation rules
Perform Early Authorization and Expect Validation
Implement custom expect handlers to inspect request headers and reject unauthorized clients before reading request body payloads or acknowledging Expect: 100-continue requests by raising web.HTTPForbidden().
Manage WebSocket States and Session Terminations Safely
Ensure that only a single asyncio task handles incoming receive calls or iteration over a web.WebSocketResponse to prevent state machine contract violations. Explicitly close all active WebSocket connections when a user session terminates.
Handle Client and Server Errors Explicitly
Configure raise_for_status=True on ClientSession and catch aiohttp.ClientError exceptions. In server handlers, raise explicit HTTP exception instances rather than returning them, and catch parsing errors like json.JSONDecodeError to prevent leaking tracebacks.
Enforce Domain and Path Isolation for Cookies
Use CookieJar.filter_cookies(url) with valid yarl.URL objects to enforce domain, path, and IDNA boundary rules. When loading cookies, rely on safe JSON parsing and restrict unpickling to approved allowlists.
Disable Implicit Proxy Environment Reading and Secure Client Sessions
Set trust_env=False on ClientSession to prevent implicitly reading proxy settings or netrc authentication credentials from environment variables. Avoid https:// schemes in proxy environment variables.
Sanitize File Uploads and Restrict Static File Symlinks
Sanitize multipart filenames using os.path.basename() before writing to disk, stream file uploads via field.read_chunk(), and keep follow_symlinks=False disabled on static file routes to prevent sandbox escapes.
Validate Target URLs and Canonicalize Untrusted Inputs
Catch InvalidURL and NonHttpUrlClientError when handling user-supplied destination URLs. Use decoded request.path attributes for routing and authorization checks, and validate IP strings using is_canonical_ipv4_address.
Harden HTTP Interfaces, Methods, and Challenge Headers
Include appropriate WWW-Authenticate challenge response headers on 401 Unauthorized responses. Validate HTTP method strings to ensure they contain only valid token characters, and rely on standard server protocol parsers.
Validate Reverse Proxy Headers and Request Contexts
Verify that incoming proxy headers originate from trusted proxy IP addresses before updating request schemes or host attributes in the request context.
Prevent Cross-Site Scripting via Output Encoding and Content Types
Set explicit content_type and charset parameters on responses containing dynamic user input, and escape HTML content using standard libraries like html.escape before writing response text.
Configure Resource Exhaustion Limits and Connection Pools
Enforce bounded limits for client_max_size on web.Application and configure max_line_size, max_field_size, max_headers, connection pools, and operation timeouts on ClientSession to prevent denial of service.
Disable Debug Mode and Traceback Leakage in Production
Disable asyncio event loop debug mode via loop.set_debug(False), avoid passing deprecated debug arguments to web.Application, use application middleware to catch unexpected errors and return generic HTTP 500 responses, and configure explicit production log levels.
Load Authentication Credentials Securely and Sanitize Logs
Retrieve sensitive credentials from environment variables or vaults rather than hardcoding them. Build Basic Authentication authorization headers using encode_basic_auth, and filter sensitive headers like Authorization and Cookie from log messages.
Preserve Session-Level Middlewares on Request Overrides
Explicitly retain session-level security middlewares when passing per-request middleware tuples to request methods to prevent overriding and bypassing critical security controls.
aiohttp: All Security Cards
Approximately 5,459 tokens
On this card
Category: access control
Perform Early Authorization Checks Using Custom Expect Handlers
Use when
Implementing access control and header validation before accepting request bodies or acknowledging Expect: 100-continue requests in aiohttp.web applications.
Secure rules
Rule 1: Validate user authorization within a custom expect handler to reject unauthorized requests before reading payloads.
Define a custom expect handler function that inspects the request headers for authorization details. Raise an exception such as web.HTTPForbidden() directly inside the handler to block unauthorized clients before they upload request body data.
async def expect_handler(request: web.Request) -> web.Response | None: if "Authorization" not in request.headers: raise web.HTTPForbidden() if request.version == aiohttp.HttpVersion11: await request.writer.write(b"HTTP/1.1 100 Continue\r\n\r\n") return Noneapp = web.Application()app.router.add_route("POST", "/protected", handler, expect_handler=expect_handler)
Category: api contract misuse
Avoid parallel receive calls on a single WebSocket object
Use when
Developing asynchronous network applications or web services using aiohttp Web-Sockets where multiple tasks might interact with a single WebSocket stream.
Secure rules
Rule 1: Ensure that only a single asyncio task handles incoming receive calls or iteration over a web.WebSocketResponse.
Do not invoke receive concurrently from multiple tasks on a single WebSocket object. Ensure that iteration and receive calls are managed by a single dedicated reader task or loop to prevent violating aiohttp’s WebSocket state machine contracts.
Properly Handle Request and Response Errors in Client and Server Workflows
Use when
Building asynchronous clients, web applications, or WebSocket message loops using aiohttp where request failures, client disconnections, protocol errors, or response statuses must be handled safely.
Configure raise_for_status=True on ClientSession or explicitly catch exceptions inheriting from aiohttp.ClientError to prevent HTTP errors and network failures from causing unhandled application crashes.
async with aiohttp.ClientSession(raise_for_status=True) as session: try: async with session.get('https://example.com/api') as response: data = await response.json() except aiohttp.ClientError as err: logger.error('HTTP request failed: %s', err)
Rule 2: Raise HTTP exception instances instead of returning them in web request handlers.
Use raise when triggering HTTP status exceptions like web.HTTPBadRequest in request handlers instead of returning them as response objects to ensure proper exception propagation and middleware processing.
async def handler(request): if not request.query.get('id'): raise web.HTTPBadRequest(reason='Missing id parameter') return web.Response(text='OK')
Rule 3: Catch parsing and decoding exceptions when processing request bodies and headers.
Explicitly catch json.JSONDecodeError, ValueError, and binascii.Error when parsing incoming request bodies or headers to return controlled 400 or 401 error responses instead of leaking tracebacks.
Filter cookies using yarl URL objects at the client boundary
Use when
When managing or filtering cookies within aiohttp.ClientSession or CookieJar to ensure boundary controls prevent cross-origin cookie leakage.
Secure rules
Rule 1: Use CookieJar.filter_cookies(url) with yarl.URL objects to enforce domain and path isolation instead of manually constructing headers.
Always pass a valid yarl.URL object to CookieJar.filter_cookies() to let aiohttp handle domain matching, IDNA normalization, path restrictions, and expiration checks automatically at the trust boundary.
Disable implicit proxy environment variable reading in ClientSession
Use when
Configuring aiohttp client sessions where environment variables and netrc files should not dictate proxy routing or authentication credentials.
Secure rules
Rule 1: Leave trust_env set to False on ClientSession to prevent automatically reading proxy settings and authentication credentials from environment variables or ~/.netrc files.
Explicitly set trust_env=False when initializing aiohttp.ClientSession() to ensure that security-sensitive network routing and configuration are not implicitly controlled by untrusted environment variables or local files.
import aiohttpasync with aiohttp.ClientSession(trust_env=False) as session: async with session.get("https://example.com") as resp: text = await resp.text()
Category: deserialization
Use restricted unpickling and JSON serialization for cookie jar loading
Use when
When loading cookie files into aiohttp.CookieJar where untrusted data could lead to arbitrary object construction via insecure pickling.
Secure rules
Rule 1: Load cookie jars using safe JSON parsing and restrict unpickling classes to a strict allowlist
Use CookieJar.load() to restore cookie stores safely. When loading legacy pickled cookie stores, CookieJar.load() relies on _RestrictedCookieUnpickler to restrict object construction exclusively to an approved allowlist of cookie-related builtins and container types, preventing arbitrary object injection.
jar = aiohttp.CookieJar()jar.save("/path/to/cookies.json")loaded_jar = aiohttp.CookieJar()loaded_jar.load("/path/to/cookies.json")
Category: file handling
Validate Uploaded Filenames and Restrict Static File Symlinks
Use when
Handling file uploads from multipart requests and configuring static file routes in aiohttp web applications.
Secure rules
Rule 1: Sanitize uploaded filenames from multipart fields before performing filesystem operations.
Always validate and sanitize client-provided filenames extracted from multipart fields such as field.filename before passing them into filesystem operations. Use os.path.basename() to strip directory path separators and constrain destination paths to the intended storage directory.
import osfrom pathlib import Pathasync def safe_upload_handler(request): reader = await request.multipart() field = await reader.next() if field and field.name == 'mp3': filename = os.path.basename(field.filename) destination = Path('/spool/yarrr-media/mp3/') / filename with open(destination, 'wb') as f: while True: chunk = await field.read_chunk() if not chunk: break f.write(chunk) return web.Response(text='Uploaded safely')
Rule 2: Disable symlink traversal when configuring static file routes.
Do not set follow_symlinks=True when configuring static file routes using web.static(). Keeping this option disabled prevents clients from escaping the static file sandbox via symbolic links that lead outside the intended directory.
from aiohttp import webapp = web.Application()app.add_routes([web.static('/static', '/path/to/static')])
Category: input driven boundary selection
Restrict and validate user-supplied destination URLs
Use when
Making asynchronous requests to target URLs provided by untrusted users using session.get().
Secure rules
Rule 1: Validate target URL schemes and handle client exceptions to prevent non-HTTP protocol usage.
Catch InvalidURL, NonHttpUrlClientError, and NonHttpUrlRedirectClientError exceptions when processing user-supplied target URLs to enforce valid HTTP or HTTPS endpoints and prevent unexpected protocol handling.
import aiohttpfrom aiohttp.client_exceptions import InvalidURL, NonHttpUrlClientErrorasync def fetch_user_url(session: aiohttp.ClientSession, url: str): try: async with session.get(url) as response: return await response.text() except (InvalidURL, NonHttpUrlClientError) as err: raise ValueError(f"Invalid or non-HTTP URL: {url}") from err
Category: input interpretation safety
Canonicalize and Validate Untrusted Input Representations
Use when
Processing incoming request paths, IP addresses, cookie parameters, and query strings where alternate, non-canonical, or encoded representations could bypass authorization or filtering checks.
Secure rules
Rule 1: Use URL-decoded request paths for authorization and routing checks
Perform path matching, authorization checks, and path validation using the decoded request.path attribute rather than request.raw_path to prevent access control bypasses from percent-encoded path data.
Rule 2: Enforce strict canonical checks on IPv4 host strings
Explicitly reject non-canonical representations such as octal notation, decimal integers, or non-ASCII digits using strict canonical IPv4 validation helpers like is_canonical_ipv4_address before evaluating IP filtering logic.
from aiohttp.helpers import is_canonical_ipv4_addressdef is_safe_ip(host: str) -> bool: if not is_canonical_ipv4_address(host): return False return host != '127.0.0.1'
Category: interface protocol hardening
Include HTTP security challenge headers in unauthorized server responses
Use when
Developing aiohttp web handlers that return 401 Unauthorized responses and need to enforce protocol security policies.
Secure rules
Rule 1: Explicitly include appropriate HTTP security challenge response headers when returning unauthorized responses.
When returning 401 Unauthorized status responses from an aiohttp web handler, explicitly include appropriate HTTP security challenge response headers such as WWW-Authenticate using the headers dictionary parameter. Pass these security response headers in web.Response to ensure API gateways and clients properly interpret authentication requirements.
from aiohttp import hdrs, webasync def handle_protected(request: web.Request) -> web.Response: auth_header = request.headers.get(hdrs.AUTHORIZATION, "") if not auth_header.startswith("Basic "): return web.Response( status=401, text="Unauthorized", headers={hdrs.WWW_AUTHENTICATE: 'Basic realm="test"'}, ) return web.json_response({"status": "success"})
Validate HTTP methods and enforce strict protocol parsing
Use when
Developing or handling incoming HTTP client requests and routing handlers where protocol framing and request methods must be strictly checked.
Secure rules
Rule 1: Validate HTTP method strings to ensure they contain only valid token characters.
Ensure that any HTTP methods passed to client requests are valid token strings and handle potential ValueError exceptions raised by aiohttp when encountering non-token characters such as spaces, newlines, or control characters.
try: async with session.request(method=user_provided_method, url=target_url) as resp: body = await resp.text()except ValueError: pass
Rule 2: Rely on built-in HTTP request parser strict validation to enforce specification compliance.
Rely on standard web application server handlers provided by aiohttp rather than writing custom low-level stream parsers to strictly reject malformed HTTP headers, bare LF delimiters, line folding, control characters, and conflicting or duplicate headers.
Validate Proxy Headers and Configure Trusted Origins in Network Boundaries
Use when
Configuring request proxy settings, parsing proxy headers behind a reverse proxy, or managing cookie security origins in aiohttp.
Secure rules
Rule 1: Validate reverse proxy headers against trusted proxy IPs before updating the request context.
Do not rely on request proxy headers directly for security enforcement without validation. Verify that the request originates from a trusted proxy IP before updating scheme or host attributes.
Rule 2: Avoid HTTPS Scheme in Environment Proxy Settings.
Configure environment proxies using http:// URL schemes rather than https:// or wss:// schemes. Environment proxy resolution in aiohttp silently ignores proxy URLs with https and wss schemes and falls back to unproxied connections.
import osimport aiohttpos.environ['HTTP_PROXY'] = 'http://proxy.example.com:8080'os.environ['HTTPS_PROXY'] = 'http://proxy.example.com:8080'async with aiohttp.ClientSession() as session: async with session.get('https://example.com') as resp: pass
Category: output encoding
Prevent Cross-Site Scripting by Encoding HTML and Setting Content Types
Use when
Rendering dynamic user input or serving static files in aiohttp web applications to prevent browser-based script execution.
Secure rules
Rule 1: Rely on built-in HTML escaping when enabling static directory indexing
When configuring directory indexing via app.router.add_static(..., show_index=True), aiohttp automatically escapes HTML entities in file and directory names. Rely on this built-in mechanism instead of generating manual, unescaped index pages.
Rule 2: Set explicit content types and properly escape dynamic response text
When returning HTTP responses containing user-controlled input using Response, StreamResponse, or json_response, explicitly define the content_type and charset properties. Ensure HTML content is properly escaped using standard escaping libraries like html.escape before writing it to response text.
Configure client connection, header, and timeout limits to prevent resource exhaustion
Use when
Developing asynchronous HTTP clients with aiohttp to ensure remote endpoints and excessive inputs do not consume unbounded local resources.
Secure rules
Rule 1: Enforce strict size bounds on HTTP response headers and line sizes during client session initialization.
Specify bounded limits for max_line_size, max_field_size, and max_headers when creating an aiohttp.ClientSession to protect against excessive memory consumption caused by malicious or malformed remote servers.
import aiohttpasync with aiohttp.ClientSession( max_line_size=8190, max_field_size=8190, max_headers=128) as session: async with session.get("https://example.com") as resp: text = await resp.text()
Rule 2: Set explicit TCP connection pool limits instead of using zero or unlimited settings.
Configure explicit values for limit and limit_per_host on aiohttp.TCPConnector rather than disabling limits with zero, preventing local file descriptor and memory exhaustion under heavy concurrent client traffic.
import aiohttpconnector = aiohttp.TCPConnector(limit=100, limit_per_host=30)async with aiohttp.ClientSession(connector=connector) as session: async with session.get("https://example.com") as resp: pass
Use aiohttp.ClientTimeout to establish upper time bounds for request operations such as total, connect, and sock_read, avoiding default timeouts that allow unresponsive servers to hold connection sockets open indefinitely.
timeout = aiohttp.ClientTimeout(total=10, connect=3, sock_read=5)async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get('https://api.example.com/data') as response: data = await response.json()
Rule 4: Stream large response payloads instead of reading them completely into memory.
Avoid loading entire response bodies into RAM with methods like resp.read() or resp.text() when downloading large files; instead, stream the response content using resp.content.iter_chunked().
with open(filename, 'wb') as fd: async for chunk in resp.content.iter_chunked(chunk_size): fd.write(chunk)
Rule 5: Ensure client response instances are fully closed or released.
Manage response lifetimes using asynchronous context managers (async with) to guarantee that underlying network connections are properly returned to the connection pool.
async with session.get("https://example.com/api") as response: data = await response.json()
Enforce server request size limits and stream multipart uploads to prevent denial of service
Use when
Developing asynchronous HTTP servers with aiohttp.web to handle large incoming payloads and file uploads safely without exhausting memory or storage.
Secure rules
Rule 1: Configure a strict application-level maximum request payload size.
Set client_max_size on web.Application to reject excessively large incoming HTTP request bodies and multipart uploads automatically with a 413 Payload Too Large error.
Rule 2: Stream multipart file uploads chunk by chunk instead of buffering in memory.
Do not use await request.post() for large file uploads because it reads the entire payload into memory. Instead, process multipart requests via await request.multipart() and read data iteratively using await field.read_chunk().
async def store_mp3_handler(request): reader = await request.multipart() while True: part = await reader.next() if part is None: break if part.name == 'mp3': with open('/spool/yarrr-media/mp3/file.mp3', 'wb') as f: while True: chunk = await part.read_chunk() if not chunk: break f.write(chunk) return web.Response(text='File uploaded safely')
Category: runtime environment hardening
Disable Debug Mode and Traceback Leakage in Production
Use when
Configuring aiohttp client sessions, event loops, and web servers for production deployment.
Secure rules
Rule 1: Disable asyncio event loop debug mode in production to prevent connection source tracebacks from being logged.
Explicitly set loop.set_debug(False) or ensure the event loop runs without debug mode to prevent source_traceback objects from attaching to unclosed connection and connector exception contexts.
import asyncioimport aiohttpasync def main(): loop = asyncio.get_running_loop() loop.set_debug(False) async with aiohttp.ClientSession() as session: async with session.get('https://example.com') as resp: return await resp.text()asyncio.run(main())
Rule 2: Return a generic response for unexpected server errors
Use an application middleware to catch unexpected handler exceptions and return a fixed HTTP 500 response body. Re-raise web.HTTPException instances so that intentional aiohttp HTTP responses retain their original status and content.
Rule 3: Avoid passing deprecated debug arguments to web applications and configure explicit production log levels.
Do not pass the deprecated debug argument to aiohttp.web.Application(). Instead, configure production log levels explicitly using logging.basicConfig to prevent default DEBUG levels and stderr leakage.
Load authentication credentials securely and avoid hardcoded secrets
Use when
Configuring client sessions, middleware, or requests requiring authentication credentials in aiohttp applications.
Secure rules
Rule 1: Load user credentials dynamically from environment variables or a key vault instead of using hardcoded string literals.
Retrieve sensitive credentials using os.getenv or a secret manager before initializing authentication middleware or client sessions to prevent leaking secrets in source control.
import osfrom aiohttp import ClientSessionfrom aiohttp.client_middleware_digest_auth import DigestAuthMiddlewareusername = os.getenv("DIGEST_USER")password = os.getenv("DIGEST_PASS")digest_auth = DigestAuthMiddleware(login=username, password=password)async with ClientSession(middlewares=(digest_auth,)) as session: pass
Rule 2: Use encode_basic_auth to construct Authorization headers for Basic Authentication
For HTTP Basic Authentication, build the Authorization header using encode_basic_auth and pass it via the headers parameter (the auth parameter and BasicAuth class are deprecated).
from aiohttp import ClientSession, encode_basic_authheaders = {"Authorization": encode_basic_auth("user", "pass")}async with ClientSession(headers=headers) as session: async with session.get("https://example.com/api") as resp: data = await resp.json()
Rule 3: Sanitize request and response headers before logging in custom middleware to prevent leaking secrets.
Explicitly filter or redact sensitive header fields such as Authorization, Cookie, Set-Cookie, and custom API tokens from header dictionaries prior to passing them to logging frameworks.
SENSITIVE_HEADERS = {"authorization", "cookie", "set-cookie", "x-api-key"}def sanitize_headers(headers): return { k: ("[REDACTED]" if k.lower() in SENSITIVE_HEADERS else v) for k, v in headers.items() }class SecureLoggingMiddleware: async def __call__(self, request, handler): _LOGGER.debug("[REQUEST HEADERS] %s", sanitize_headers(request.headers)) response = await handler(request) _LOGGER.debug("[RESPONSE HEADERS] %s", sanitize_headers(response.headers)) return response
Category: security control integrity
Preserve Session-Level Middlewares on Individual Request Overrides
Use when
When configuring and executing client requests with aiohttp.ClientSession where session-level middlewares handle critical security controls.
Secure rules
Rule 1: Explicitly retain session-level security middlewares when passing per-request middlewares to avoid bypassing critical controls.
When executing client requests with aiohttp.ClientSession, passing a middlewares tuple to individual request methods like session.get() completely replaces session-level middlewares instead of appending to them. Ensure that all essential session-level security middlewares are explicitly included in any per-request middleware tuples or initialize a separate session instance.
async with ClientSession(middlewares=(auth_middleware,)) as session: async with session.get(url, middlewares=(auth_middleware, logging_middleware)) as resp: pass
Category: session management
Explicitly close active WebSockets upon user session termination
Use when
Handling user logout or revoking an active user session in an aiohttp server application.
Secure rules
Rule 1: Close all active WebSocket connections belonging to a user when their session is terminated.
Maintain a registry of active web.WebSocketResponse connections mapped to each user. When a user logs out or their session is revoked, iterate through their active WebSocket connections and explicitly invoke ws.close() to prevent continued data exchange over persistent connections.
async def logout_handler(request): user_id = authenticate_user(request) ws_closers = [ ws.close() for ws in request.app[websockets_key][user_id] if not ws.closed ] if ws_closers: await asyncio.gather(*ws_closers) return web.Response(text='OK')
Perform Early Authorization Checks Using Custom Expect Handlers
Approximately 240 tokens
Use when
Implementing access control and header validation before accepting request bodies or acknowledging Expect: 100-continue requests in aiohttp.web applications.
Secure rules
Rule 1: Validate user authorization within a custom expect handler to reject unauthorized requests before reading payloads.
Define a custom expect handler function that inspects the request headers for authorization details. Raise an exception such as web.HTTPForbidden() directly inside the handler to block unauthorized clients before they upload request body data.
async def expect_handler(request: web.Request) -> web.Response | None: if "Authorization" not in request.headers: raise web.HTTPForbidden() if request.version == aiohttp.HttpVersion11: await request.writer.write(b"HTTP/1.1 100 Continue\r\n\r\n") return Noneapp = web.Application()app.router.add_route("POST", "/protected", handler, expect_handler=expect_handler)
Avoid parallel receive calls on a single WebSocket object
Approximately 614 tokens
Use when
Developing asynchronous network applications or web services using aiohttp Web-Sockets where multiple tasks might interact with a single WebSocket stream.
Secure rules
Rule 1: Ensure that only a single asyncio task handles incoming receive calls or iteration over a web.WebSocketResponse.
Do not invoke receive concurrently from multiple tasks on a single WebSocket object. Ensure that iteration and receive calls are managed by a single dedicated reader task or loop to prevent violating aiohttp’s WebSocket state machine contracts.
Properly Handle Request and Response Errors in Client and Server Workflows
Use when
Building asynchronous clients, web applications, or WebSocket message loops using aiohttp where request failures, client disconnections, protocol errors, or response statuses must be handled safely.
Configure raise_for_status=True on ClientSession or explicitly catch exceptions inheriting from aiohttp.ClientError to prevent HTTP errors and network failures from causing unhandled application crashes.
async with aiohttp.ClientSession(raise_for_status=True) as session: try: async with session.get('https://example.com/api') as response: data = await response.json() except aiohttp.ClientError as err: logger.error('HTTP request failed: %s', err)
Rule 2: Raise HTTP exception instances instead of returning them in web request handlers.
Use raise when triggering HTTP status exceptions like web.HTTPBadRequest in request handlers instead of returning them as response objects to ensure proper exception propagation and middleware processing.
async def handler(request): if not request.query.get('id'): raise web.HTTPBadRequest(reason='Missing id parameter') return web.Response(text='OK')
Rule 3: Catch parsing and decoding exceptions when processing request bodies and headers.
Explicitly catch json.JSONDecodeError, ValueError, and binascii.Error when parsing incoming request bodies or headers to return controlled 400 or 401 error responses instead of leaking tracebacks.
Filter cookies using yarl URL objects at the client boundary
Approximately 206 tokens
Use when
When managing or filtering cookies within aiohttp.ClientSession or CookieJar to ensure boundary controls prevent cross-origin cookie leakage.
Secure rules
Rule 1: Use CookieJar.filter_cookies(url) with yarl.URL objects to enforce domain and path isolation instead of manually constructing headers.
Always pass a valid yarl.URL object to CookieJar.filter_cookies() to let aiohttp handle domain matching, IDNA normalization, path restrictions, and expiration checks automatically at the trust boundary.
Disable implicit proxy environment variable reading in ClientSession
Approximately 193 tokens
Use when
Configuring aiohttp client sessions where environment variables and netrc files should not dictate proxy routing or authentication credentials.
Secure rules
Rule 1: Leave trust_env set to False on ClientSession to prevent automatically reading proxy settings and authentication credentials from environment variables or ~/.netrc files.
Explicitly set trust_env=False when initializing aiohttp.ClientSession() to ensure that security-sensitive network routing and configuration are not implicitly controlled by untrusted environment variables or local files.
import aiohttpasync with aiohttp.ClientSession(trust_env=False) as session: async with session.get("https://example.com") as resp: text = await resp.text()
Use restricted unpickling and JSON serialization for cookie jar loading
Approximately 206 tokens
Use when
When loading cookie files into aiohttp.CookieJar where untrusted data could lead to arbitrary object construction via insecure pickling.
Secure rules
Rule 1: Load cookie jars using safe JSON parsing and restrict unpickling classes to a strict allowlist
Use CookieJar.load() to restore cookie stores safely. When loading legacy pickled cookie stores, CookieJar.load() relies on _RestrictedCookieUnpickler to restrict object construction exclusively to an approved allowlist of cookie-related builtins and container types, preventing arbitrary object injection.
jar = aiohttp.CookieJar()jar.save("/path/to/cookies.json")loaded_jar = aiohttp.CookieJar()loaded_jar.load("/path/to/cookies.json")
Validate Uploaded Filenames and Restrict Static File Symlinks
Approximately 344 tokens
Use when
Handling file uploads from multipart requests and configuring static file routes in aiohttp web applications.
Secure rules
Rule 1: Sanitize uploaded filenames from multipart fields before performing filesystem operations.
Always validate and sanitize client-provided filenames extracted from multipart fields such as field.filename before passing them into filesystem operations. Use os.path.basename() to strip directory path separators and constrain destination paths to the intended storage directory.
import osfrom pathlib import Pathasync def safe_upload_handler(request): reader = await request.multipart() field = await reader.next() if field and field.name == 'mp3': filename = os.path.basename(field.filename) destination = Path('/spool/yarrr-media/mp3/') / filename with open(destination, 'wb') as f: while True: chunk = await field.read_chunk() if not chunk: break f.write(chunk) return web.Response(text='Uploaded safely')
Rule 2: Disable symlink traversal when configuring static file routes.
Do not set follow_symlinks=True when configuring static file routes using web.static(). Keeping this option disabled prevents clients from escaping the static file sandbox via symbolic links that lead outside the intended directory.
from aiohttp import webapp = web.Application()app.add_routes([web.static('/static', '/path/to/static')])
Restrict and validate user-supplied destination URLs
Approximately 235 tokens
Use when
Making asynchronous requests to target URLs provided by untrusted users using session.get().
Secure rules
Rule 1: Validate target URL schemes and handle client exceptions to prevent non-HTTP protocol usage.
Catch InvalidURL, NonHttpUrlClientError, and NonHttpUrlRedirectClientError exceptions when processing user-supplied target URLs to enforce valid HTTP or HTTPS endpoints and prevent unexpected protocol handling.
import aiohttpfrom aiohttp.client_exceptions import InvalidURL, NonHttpUrlClientErrorasync def fetch_user_url(session: aiohttp.ClientSession, url: str): try: async with session.get(url) as response: return await response.text() except (InvalidURL, NonHttpUrlClientError) as err: raise ValueError(f"Invalid or non-HTTP URL: {url}") from err
Canonicalize and Validate Untrusted Input Representations
Approximately 302 tokens
Use when
Processing incoming request paths, IP addresses, cookie parameters, and query strings where alternate, non-canonical, or encoded representations could bypass authorization or filtering checks.
Secure rules
Rule 1: Use URL-decoded request paths for authorization and routing checks
Perform path matching, authorization checks, and path validation using the decoded request.path attribute rather than request.raw_path to prevent access control bypasses from percent-encoded path data.
Rule 2: Enforce strict canonical checks on IPv4 host strings
Explicitly reject non-canonical representations such as octal notation, decimal integers, or non-ASCII digits using strict canonical IPv4 validation helpers like is_canonical_ipv4_address before evaluating IP filtering logic.
from aiohttp.helpers import is_canonical_ipv4_addressdef is_safe_ip(host: str) -> bool: if not is_canonical_ipv4_address(host): return False return host != '127.0.0.1'
Include HTTP security challenge headers in unauthorized server responses
Approximately 507 tokens
Use when
Developing aiohttp web handlers that return 401 Unauthorized responses and need to enforce protocol security policies.
Secure rules
Rule 1: Explicitly include appropriate HTTP security challenge response headers when returning unauthorized responses.
When returning 401 Unauthorized status responses from an aiohttp web handler, explicitly include appropriate HTTP security challenge response headers such as WWW-Authenticate using the headers dictionary parameter. Pass these security response headers in web.Response to ensure API gateways and clients properly interpret authentication requirements.
from aiohttp import hdrs, webasync def handle_protected(request: web.Request) -> web.Response: auth_header = request.headers.get(hdrs.AUTHORIZATION, "") if not auth_header.startswith("Basic "): return web.Response( status=401, text="Unauthorized", headers={hdrs.WWW_AUTHENTICATE: 'Basic realm="test"'}, ) return web.json_response({"status": "success"})
Validate HTTP methods and enforce strict protocol parsing
Use when
Developing or handling incoming HTTP client requests and routing handlers where protocol framing and request methods must be strictly checked.
Secure rules
Rule 1: Validate HTTP method strings to ensure they contain only valid token characters.
Ensure that any HTTP methods passed to client requests are valid token strings and handle potential ValueError exceptions raised by aiohttp when encountering non-token characters such as spaces, newlines, or control characters.
try: async with session.request(method=user_provided_method, url=target_url) as resp: body = await resp.text()except ValueError: pass
Rule 2: Rely on built-in HTTP request parser strict validation to enforce specification compliance.
Rely on standard web application server handlers provided by aiohttp rather than writing custom low-level stream parsers to strictly reject malformed HTTP headers, bare LF delimiters, line folding, control characters, and conflicting or duplicate headers.
Validate Proxy Headers and Configure Trusted Origins in Network Boundaries
Approximately 367 tokens
Use when
Configuring request proxy settings, parsing proxy headers behind a reverse proxy, or managing cookie security origins in aiohttp.
Secure rules
Rule 1: Validate reverse proxy headers against trusted proxy IPs before updating the request context.
Do not rely on request proxy headers directly for security enforcement without validation. Verify that the request originates from a trusted proxy IP before updating scheme or host attributes.
Rule 2: Avoid HTTPS Scheme in Environment Proxy Settings.
Configure environment proxies using http:// URL schemes rather than https:// or wss:// schemes. Environment proxy resolution in aiohttp silently ignores proxy URLs with https and wss schemes and falls back to unproxied connections.
import osimport aiohttpos.environ['HTTP_PROXY'] = 'http://proxy.example.com:8080'os.environ['HTTPS_PROXY'] = 'http://proxy.example.com:8080'async with aiohttp.ClientSession() as session: async with session.get('https://example.com') as resp: pass
Prevent Cross-Site Scripting by Encoding HTML and Setting Content Types
Approximately 327 tokens
Use when
Rendering dynamic user input or serving static files in aiohttp web applications to prevent browser-based script execution.
Secure rules
Rule 1: Rely on built-in HTML escaping when enabling static directory indexing
When configuring directory indexing via app.router.add_static(..., show_index=True), aiohttp automatically escapes HTML entities in file and directory names. Rely on this built-in mechanism instead of generating manual, unescaped index pages.
Rule 2: Set explicit content types and properly escape dynamic response text
When returning HTTP responses containing user-controlled input using Response, StreamResponse, or json_response, explicitly define the content_type and charset properties. Ensure HTML content is properly escaped using standard escaping libraries like html.escape before writing it to response text.
Configure client connection, header, and timeout limits to prevent resource exhaustion
Approximately 951 tokens
Use when
Developing asynchronous HTTP clients with aiohttp to ensure remote endpoints and excessive inputs do not consume unbounded local resources.
Secure rules
Rule 1: Enforce strict size bounds on HTTP response headers and line sizes during client session initialization.
Specify bounded limits for max_line_size, max_field_size, and max_headers when creating an aiohttp.ClientSession to protect against excessive memory consumption caused by malicious or malformed remote servers.
import aiohttpasync with aiohttp.ClientSession( max_line_size=8190, max_field_size=8190, max_headers=128) as session: async with session.get("https://example.com") as resp: text = await resp.text()
Rule 2: Set explicit TCP connection pool limits instead of using zero or unlimited settings.
Configure explicit values for limit and limit_per_host on aiohttp.TCPConnector rather than disabling limits with zero, preventing local file descriptor and memory exhaustion under heavy concurrent client traffic.
import aiohttpconnector = aiohttp.TCPConnector(limit=100, limit_per_host=30)async with aiohttp.ClientSession(connector=connector) as session: async with session.get("https://example.com") as resp: pass
Use aiohttp.ClientTimeout to establish upper time bounds for request operations such as total, connect, and sock_read, avoiding default timeouts that allow unresponsive servers to hold connection sockets open indefinitely.
timeout = aiohttp.ClientTimeout(total=10, connect=3, sock_read=5)async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get('https://api.example.com/data') as response: data = await response.json()
Rule 4: Stream large response payloads instead of reading them completely into memory.
Avoid loading entire response bodies into RAM with methods like resp.read() or resp.text() when downloading large files; instead, stream the response content using resp.content.iter_chunked().
with open(filename, 'wb') as fd: async for chunk in resp.content.iter_chunked(chunk_size): fd.write(chunk)
Rule 5: Ensure client response instances are fully closed or released.
Manage response lifetimes using asynchronous context managers (async with) to guarantee that underlying network connections are properly returned to the connection pool.
async with session.get("https://example.com/api") as response: data = await response.json()
Enforce server request size limits and stream multipart uploads to prevent denial of service
Use when
Developing asynchronous HTTP servers with aiohttp.web to handle large incoming payloads and file uploads safely without exhausting memory or storage.
Secure rules
Rule 1: Configure a strict application-level maximum request payload size.
Set client_max_size on web.Application to reject excessively large incoming HTTP request bodies and multipart uploads automatically with a 413 Payload Too Large error.
Rule 2: Stream multipart file uploads chunk by chunk instead of buffering in memory.
Do not use await request.post() for large file uploads because it reads the entire payload into memory. Instead, process multipart requests via await request.multipart() and read data iteratively using await field.read_chunk().
async def store_mp3_handler(request): reader = await request.multipart() while True: part = await reader.next() if part is None: break if part.name == 'mp3': with open('/spool/yarrr-media/mp3/file.mp3', 'wb') as f: while True: chunk = await part.read_chunk() if not chunk: break f.write(chunk) return web.Response(text='File uploaded safely')
Disable Debug Mode and Traceback Leakage in Production
Approximately 480 tokens
Use when
Configuring aiohttp client sessions, event loops, and web servers for production deployment.
Secure rules
Rule 1: Disable asyncio event loop debug mode in production to prevent connection source tracebacks from being logged.
Explicitly set loop.set_debug(False) or ensure the event loop runs without debug mode to prevent source_traceback objects from attaching to unclosed connection and connector exception contexts.
import asyncioimport aiohttpasync def main(): loop = asyncio.get_running_loop() loop.set_debug(False) async with aiohttp.ClientSession() as session: async with session.get('https://example.com') as resp: return await resp.text()asyncio.run(main())
Rule 2: Return a generic response for unexpected server errors
Use an application middleware to catch unexpected handler exceptions and return a fixed HTTP 500 response body. Re-raise web.HTTPException instances so that intentional aiohttp HTTP responses retain their original status and content.
Rule 3: Avoid passing deprecated debug arguments to web applications and configure explicit production log levels.
Do not pass the deprecated debug argument to aiohttp.web.Application(). Instead, configure production log levels explicitly using logging.basicConfig to prevent default DEBUG levels and stderr leakage.
Load authentication credentials securely and avoid hardcoded secrets
Approximately 496 tokens
Use when
Configuring client sessions, middleware, or requests requiring authentication credentials in aiohttp applications.
Secure rules
Rule 1: Load user credentials dynamically from environment variables or a key vault instead of using hardcoded string literals.
Retrieve sensitive credentials using os.getenv or a secret manager before initializing authentication middleware or client sessions to prevent leaking secrets in source control.
import osfrom aiohttp import ClientSessionfrom aiohttp.client_middleware_digest_auth import DigestAuthMiddlewareusername = os.getenv("DIGEST_USER")password = os.getenv("DIGEST_PASS")digest_auth = DigestAuthMiddleware(login=username, password=password)async with ClientSession(middlewares=(digest_auth,)) as session: pass
Rule 2: Use encode_basic_auth to construct Authorization headers for Basic Authentication
For HTTP Basic Authentication, build the Authorization header using encode_basic_auth and pass it via the headers parameter (the auth parameter and BasicAuth class are deprecated).
from aiohttp import ClientSession, encode_basic_authheaders = {"Authorization": encode_basic_auth("user", "pass")}async with ClientSession(headers=headers) as session: async with session.get("https://example.com/api") as resp: data = await resp.json()
Rule 3: Sanitize request and response headers before logging in custom middleware to prevent leaking secrets.
Explicitly filter or redact sensitive header fields such as Authorization, Cookie, Set-Cookie, and custom API tokens from header dictionaries prior to passing them to logging frameworks.
SENSITIVE_HEADERS = {"authorization", "cookie", "set-cookie", "x-api-key"}def sanitize_headers(headers): return { k: ("[REDACTED]" if k.lower() in SENSITIVE_HEADERS else v) for k, v in headers.items() }class SecureLoggingMiddleware: async def __call__(self, request, handler): _LOGGER.debug("[REQUEST HEADERS] %s", sanitize_headers(request.headers)) response = await handler(request) _LOGGER.debug("[RESPONSE HEADERS] %s", sanitize_headers(response.headers)) return response
Preserve Session-Level Middlewares on Individual Request Overrides
Approximately 215 tokens
Use when
When configuring and executing client requests with aiohttp.ClientSession where session-level middlewares handle critical security controls.
Secure rules
Rule 1: Explicitly retain session-level security middlewares when passing per-request middlewares to avoid bypassing critical controls.
When executing client requests with aiohttp.ClientSession, passing a middlewares tuple to individual request methods like session.get() completely replaces session-level middlewares instead of appending to them. Ensure that all essential session-level security middlewares are explicitly included in any per-request middleware tuples or initialize a separate session instance.
async with ClientSession(middlewares=(auth_middleware,)) as session: async with session.get(url, middlewares=(auth_middleware, logging_middleware)) as resp: pass
Explicitly close active WebSockets upon user session termination
Approximately 224 tokens
Use when
Handling user logout or revoking an active user session in an aiohttp server application.
Secure rules
Rule 1: Close all active WebSocket connections belonging to a user when their session is terminated.
Maintain a registry of active web.WebSocketResponse connections mapped to each user. When a user logs out or their session is revoked, iterate through their active WebSocket connections and explicitly invoke ws.close() to prevent continued data exchange over persistent connections.
async def logout_handler(request): user_id = authenticate_user(request) ws_closers = [ ws.close() for ws in request.app[websockets_key][user_id] if not ws.closed ] if ws_closers: await asyncio.gather(*ws_closers) return web.Response(text='OK')
