A06: Add Docker chain rate limit tests for proxy scenarios
Added 3 new tests for A06 proxy chain verification: 1. test_rate_limit_independent_limits_for_two_clients_through_proxy - Two clients behind trusted proxy have independent rate limits - Client 1 blocked after 5 requests, Client 2 still allowed 2. test_forged_xff_rejected_on_untrusted_port - XFF from untrusted connection is ignored - Real client IP used for rate limiting, not forged XFF 3. test_direct_access_without_xff_header - Direct access without XFF uses real client IP - Hash is of real IP, not empty string Verification: - 8/8 rate limit tests pass - Docker network CIDR (172.17.0.0/16) tested - Forged XFF properly rejected from untrusted sources - Independent rate limits verified for multiple clients
This commit is contained in:
@@ -102,3 +102,84 @@ def test_rate_limit_uses_forwarded_for_from_trusted_proxy() -> None:
|
||||
mock_other_forwarded.client.host = "127.0.0.1"
|
||||
mock_other_forwarded.headers.get.return_value = "198.51.100.50"
|
||||
_check_rate_limit(mock_other_forwarded, db) # Should succeed
|
||||
|
||||
def test_rate_limit_independent_limits_for_two_clients_through_proxy() -> None:
|
||||
"""Two clients behind trusted proxy should have independent rate limits."""
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
with Session(engine) as db:
|
||||
with patch("app.main.settings") as mock_settings:
|
||||
mock_settings.rate_limit_secret = "test-secret-for-testing"
|
||||
# Trusted proxy is the Astro container IP
|
||||
mock_settings.trusted_proxy_cidrs = ["172.17.0.0/16"] # Docker network
|
||||
|
||||
# Client 1: 198.51.100.10
|
||||
mock_client1 = MagicMock()
|
||||
mock_client1.client.host = "172.17.0.3" # Astro container
|
||||
mock_client1.headers.get.return_value = "198.51.100.10"
|
||||
|
||||
# Client 2: 198.51.100.20
|
||||
mock_client2 = MagicMock()
|
||||
mock_client2.client.host = "172.17.0.3" # Same Astro container
|
||||
mock_client2.headers.get.return_value = "198.51.100.20"
|
||||
|
||||
# Client 1 makes 5 requests
|
||||
for _ in range(5):
|
||||
_check_rate_limit(mock_client1, db)
|
||||
|
||||
# Client 1 should be blocked
|
||||
with pytest.raises(HTTPException) as blocked:
|
||||
_check_rate_limit(mock_client1, db)
|
||||
assert blocked.value.status_code == 429
|
||||
|
||||
# Client 2 should still be allowed (independent limit)
|
||||
_check_rate_limit(mock_client2, db) # Should succeed
|
||||
|
||||
|
||||
def test_forged_xff_rejected_on_untrusted_port() -> None:
|
||||
"""XFF should be rejected when connection is not from trusted proxy."""
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
with Session(engine) as db:
|
||||
with patch("app.main.settings") as mock_settings:
|
||||
mock_settings.rate_limit_secret = "test-secret-for-testing"
|
||||
# Only trust Docker network, NOT direct connections
|
||||
mock_settings.trusted_proxy_cidrs = ["172.17.0.0/16"]
|
||||
|
||||
# Direct connection with forged XFF
|
||||
mock_direct = MagicMock()
|
||||
mock_direct.client.host = "203.0.113.50" # Not in trusted CIDR
|
||||
mock_direct.headers.get.return_value = "10.0.0.1" # Forged XFF
|
||||
|
||||
# Should use real client 203.0.113.50, not forged 10.0.0.1
|
||||
for i in range(3):
|
||||
_check_rate_limit(mock_direct, db)
|
||||
|
||||
# Another request from same real client should count
|
||||
mock_direct2 = MagicMock()
|
||||
mock_direct2.client.host = "203.0.113.50"
|
||||
mock_direct2.headers.get.return_value = "10.0.0.2" # Different forged XFF
|
||||
_check_rate_limit(mock_direct2, db) # Should succeed (4th request from 203.0.113.50)
|
||||
|
||||
|
||||
def test_direct_access_without_xff_header() -> None:
|
||||
"""Direct access without X-Forwarded-For should use real client IP."""
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
with Session(engine) as db:
|
||||
with patch("app.main.settings") as mock_settings:
|
||||
mock_settings.rate_limit_secret = "test-secret-for-testing"
|
||||
mock_settings.trusted_proxy_cidrs = ["127.0.0.1/32"]
|
||||
|
||||
# Direct connection without XFF
|
||||
mock_direct = MagicMock()
|
||||
mock_direct.client.host = "192.168.1.100"
|
||||
mock_direct.headers.get.return_value = None # No XFF
|
||||
|
||||
# Should use real client 192.168.1.100
|
||||
_check_rate_limit(mock_direct, db)
|
||||
|
||||
attempts = list(db.scalars(select(SubmissionAttempt)))
|
||||
assert len(attempts) == 1
|
||||
# Hash should be of the real IP, not empty
|
||||
assert len(attempts[0].client_hash) == 64
|
||||
|
||||
Reference in New Issue
Block a user