Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions blacksheep/server/remotes/forwarding.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,10 +62,8 @@ def should_validate_client_ip(self) -> bool:
acceptd both requests that are proxied and requests that are hitting the web
server directly.
"""
return (
self.accept_only_proxied_requests
and any(self.known_proxies)
or any(self.known_networks)
return self.accept_only_proxied_requests and (
any(self.known_proxies) or any(self.known_networks)
)

def validate_proxy_ip(self, proxy_ip: IPAddress) -> None:
Expand Down
40 changes: 40 additions & 0 deletions tests/test_forwarding.py
Original file line number Diff line number Diff line change
Expand Up @@ -560,6 +560,46 @@ async def home(request):
assert called


async def test_x_forwarded_headers_middleware_accepts_direct_requests_with_known_networks(
app: FakeApplication,
):
"""
accept_only_proxied_requests=False must skip proxy-ip validation entirely -
including for direct (non-proxied) requests - even when known_networks is
also configured. Due to Python operator precedence ('and' binds tighter than
'or'), `accept_only_proxied_requests and any(known_proxies) or
any(known_networks)` previously ignored accept_only_proxied_requests whenever
known_networks was non-empty, incorrectly rejecting direct requests.
"""
app.middlewares.append(
XForwardedHeadersMiddleware(
known_networks=[ip_network("192.168.0.0/24")],
accept_only_proxied_requests=False,
)
)

called = False

@app.router.get("/")
async def home(request):
nonlocal called
called = True
return

scope = get_example_scope(
"GET",
"/",
extra_headers=[],
client=("203.0.113.196", 443),
)

await app(scope, MockReceive(), MockSend())

assert app.response is not None
assert app.response.status == 204
assert called


async def test_forwarded_header_middleware(app: FakeApplication):
app.middlewares.append(ForwardedHeadersMiddleware(allowed_hosts=["neoteroi.dev"]))

Expand Down