Skip to content

Forwarded Does not Parse IPv6 Addresses Enclosed in Square Brackets #558

@topnotcher

Description

@topnotcher

Per RFC7239 (non-normative), IPv6 addresses in the Forwarded header should be enclosed in square brackets:

Also, note that an IPv6 address is always enclosed in square brackets.

(See also section 7.4: https://datatracker.ietf.org/doc/html/rfc7239#section-7.4)

Any attempt to pass an IPv6 address in square brackets in a Forwarded header results in a ValueError:

Example: for="[2001:0db8:85a3:0000:0000:8a2e:0370:7334]";proto=https;host=example.com;by=127.0.0.1

Exception:

|Traceback (most recent call last):
  File "aiohttp_remotes/forwarded.py", line 71, in middleware
    ips.append(ip_address(for_))
               ~~~~~~~~~~^^^^^^
  File "/usr/lib/python3.13/ipaddress.py", line 54, in ip_address
    raise ValueError(f'{address!r} does not appear to be an IPv4 or IPv6 address')
ValueError: '[2001:0db8:85a3:0000:0000:8a2e:0370:7334]' does not appear to be an IPv4 or IPv6 address

This seems to work for me:

diff --git a/aiohttp_remotes/forwarded.py b/aiohttp_remotes/forwarded.py
index 3589a1c..a0631c9 100644
--- a/aiohttp_remotes/forwarded.py
+++ b/aiohttp_remotes/forwarded.py
@@ -8,6 +8,32 @@ from .exceptions import IncorrectForwardedCount, RemoteError
 from .utils import TrustedOrig, parse_trusted_list, remote_ip
 
 
+def parse_forwarded_ip(value: str) -> str:
+    try:
+        # Handle a bare IP address. IPv6 addresses should be enclosed in square
+        # brackets, but we try to parse as just an address first for backwards
+        # compatibility.
+        return ip_address(value)
+
+    except ValueError as err:
+        # Correctly formatted IPv6 address
+        if value[0] == "[":
+            end_idx = value.find("]")
+
+            if end_idx == -1:
+                raise ValueError(f"Invalid IPv6 address: {value!r}")
+
+            return ip_address(value[1:end_idx])
+
+        # Not an IPv6 address, so must be IPv4 with a port.
+        elif ":" in value:
+            ip, _port = value.split(":", 1)
+            return ip_address(ip)
+
+        else:
+            raise err
+
+
 class ForwardedRelaxed(ABC):
     def __init__(self, num: int = 1) -> None:
         self._num = num
@@ -63,12 +89,12 @@ class ForwardedStrict(ABC):
 
             assert request.transport is not None
             peer_ip, *_ = request.transport.get_extra_info("peername")
-            ips = [ip_address(peer_ip)]
+            ips = [parse_forwarded_ip(peer_ip)]
 
             for elem in reversed(request.forwarded):
                 for_ = elem.get("for")
                 if for_:
-                    ips.append(ip_address(for_))
+                    ips.append(parse_forwarded_ip(for_))
                 proto = elem.get("proto")
                 if proto is not None:
                     overrides["scheme"] = proto

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions