186 lines
4.8 KiB
Python
186 lines
4.8 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from proxy_pool.plugins.builtin.checkers.http_anonymity import (
|
|
HttpAnonymityChecker,
|
|
classify_anonymity,
|
|
)
|
|
from proxy_pool.plugins.protocols import CheckContext
|
|
|
|
|
|
@pytest.fixture
|
|
def checker():
|
|
return HttpAnonymityChecker(
|
|
judge_url="http://httpbin.org/get",
|
|
timeout=10.0,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def context():
|
|
return CheckContext(
|
|
started_at=datetime.now(),
|
|
http_client=None,
|
|
tcp_latency_ms=50.0,
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
|
|
class TestClassifyAnonymity:
|
|
def test_elite_no_revealing_headers(self):
|
|
headers = {
|
|
"Host": "httpbin.org",
|
|
"Accept": "*/*",
|
|
"User-Agent": "python-httpx/0.27",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
assert level == "elite"
|
|
assert leaked == []
|
|
|
|
def test_transparent_real_ip_in_forwarded_for(self):
|
|
headers = {
|
|
"Host": "httpbin.org",
|
|
"X-Forwarded-For": "198.51.100.1",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
assert level == "transparent"
|
|
assert "x-forwarded-for" in leaked
|
|
|
|
def test_transparent_real_ip_in_chained_header(self):
|
|
headers = {
|
|
"X-Forwarded-For": "198.51.100.1, 10.0.0.1",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
assert level == "transparent"
|
|
|
|
def test_anonymous_via_header_present(self):
|
|
headers = {
|
|
"Via": "1.1 proxy.example.com",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
assert level == "anonymous"
|
|
assert "via" in leaked
|
|
|
|
def test_anonymous_proxy_headers_without_real_ip(self):
|
|
headers = {
|
|
"X-Forwarded-For": "10.0.0.1",
|
|
"Via": "1.1 squid",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
assert level == "anonymous"
|
|
assert "x-forwarded-for" in leaked
|
|
assert "via" in leaked
|
|
|
|
def test_multiple_revealing_headers(self):
|
|
headers = {
|
|
"X-Forwarded-For": "198.51.100.1",
|
|
"Via": "1.1 proxy",
|
|
"X-Real-Ip": "198.51.100.1",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
assert level == "transparent"
|
|
assert len(leaked) == 3
|
|
|
|
def test_no_real_ip_known_falls_back_to_header_presence(self):
|
|
headers = {
|
|
"X-Forwarded-For": "10.0.0.1",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip=None,
|
|
)
|
|
|
|
assert level == "anonymous"
|
|
assert "x-forwarded-for" in leaked
|
|
|
|
def test_case_insensitive_header_matching(self):
|
|
headers = {
|
|
"X-FORWARDED-FOR": "198.51.100.1",
|
|
"VIA": "1.1 proxy",
|
|
}
|
|
|
|
level, leaked = classify_anonymity(
|
|
headers=headers,
|
|
origin_ip="203.0.113.1",
|
|
proxy_ip="203.0.113.1",
|
|
real_ip="198.51.100.1",
|
|
)
|
|
|
|
assert level == "transparent"
|
|
assert len(leaked) == 2
|
|
|
|
|
|
class TestHttpAnonymityChecker:
|
|
def test_stage_and_priority(self, checker):
|
|
assert checker.stage == 2
|
|
assert checker.priority == 0
|
|
|
|
def test_does_not_skip_any_protocol(self, checker):
|
|
assert checker.should_skip("http") is False
|
|
assert checker.should_skip("socks5") is False
|
|
|
|
async def test_fails_on_unreachable_proxy(self, checker, context):
|
|
result = await checker.check(
|
|
proxy_ip="192.0.2.1",
|
|
proxy_port=1,
|
|
proxy_protocol="http",
|
|
context=context,
|
|
)
|
|
|
|
assert result.passed is False
|
|
|
|
async def test_fails_on_invalid_socks_proxy(self, checker, context):
|
|
result = await checker.check(
|
|
proxy_ip="192.0.2.1",
|
|
proxy_port=1,
|
|
proxy_protocol="socks5",
|
|
context=context,
|
|
)
|
|
|
|
assert result.passed is False |