Partial fix of ProxyMiddlware where the request comes from the proxy itself

tests/middleware/test_proxy_headers.py

@@ -15,26 +15,30 @@ async def app(scope, receive, send):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
- ("trusted_hosts", "response_text"),
+ ("trusted_hosts", "x_forwarded_for", "response_text"),
  [
  # always trust
- ("*", "Remote: https://1.2.3.4:0"),
+ ("*", "1.2.3.4", "Remote: https://1.2.3.4:0"),
  # trusted proxy
- ("127.0.0.1", "Remote: https://1.2.3.4:0"),
- (["127.0.0.1"], "Remote: https://1.2.3.4:0"),
+ ("127.0.0.1", "1.2.3.4", "Remote: https://1.2.3.4:0"),
+ (["127.0.0.1"], "1.2.3.4", "Remote: https://1.2.3.4:0"),
  # trusted proxy list
- (["127.0.0.1", "10.0.0.1"], "Remote: https://1.2.3.4:0"),
- ("127.0.0.1, 10.0.0.1", "Remote: https://1.2.3.4:0"),
+ (["127.0.0.1", "10.0.0.1"], "1.2.3.4", "Remote: https://1.2.3.4:0"),
+ ("127.0.0.1, 10.0.0.1", "1.2.3.4", "Remote: https://1.2.3.4:0"),
  # request from untrusted proxy
- ("192.168.0.1", "Remote: http://127.0.0.1:123"),
+ ("192.168.0.1", "1.2.3.4", "Remote: http://127.0.0.1:123"),
+ # https://github.com/encode/uvicorn/issues/1068
+ ("127.0.0.1", "127.0.0.1", "Remote: https://127.0.0.1:0"),
  ],
 )
-async def test_proxy_headers_trusted_hosts(trusted_hosts, response_text):
+async def test_proxy_headers_trusted_hosts(
+ trusted_hosts, x_forwarded_for, response_text
+):
  app_with_middleware = ProxyHeadersMiddleware(app, trusted_hosts=trusted_hosts)
  async with httpx.AsyncClient(
  app=app_with_middleware, base_url="http://testserver"
  ) as client:
- headers = {"X-Forwarded-Proto": "https", "X-Forwarded-For": "1.2.3.4"}
+ headers = {"X-Forwarded-Proto": "https", "X-Forwarded-For": x_forwarded_for}
  response = await client.get("/", headers=headers)
 
  assert response.status_code == 200

uvicorn/middleware/proxy_headers.py

@@ -30,6 +30,12 @@ def get_trusted_client_host(
  if host not in self.trusted_hosts:
  return host
 
+ # https://github.com/encode/uvicorn/issues/1068#issuecomment-855371576
+ # returns the next ip in list after the last trusted or the last trusted
+ # in case it was the last host in the x-forwarded-for list
+ if host == x_forwarded_for_hosts[-1]:
+ return host
+
  async def __call__(self, scope, receive, send):
  if scope["type"] in ("http", "websocket"):
  client_addr = scope.get("client")