mirror of
https://github.com/yt-dlp/yt-dlp
synced 2025-07-31 09:34:57 -05:00
[networking] Fix various socks proxy bugs (#8065)
- Fixed support for IPv6 socks proxies - Fixed support for IPv6 over socks5 - Fixed --source-address not being obeyed for socks4 and socks5 - Fixed socks4a when the destination address is an IPv4 address Closes https://github.com/yt-dlp/yt-dlp/issues/7959 Fixes https://github.com/ytdl-org/youtube-dl/issues/15368 Authored by: coletdjnz Co-authored-by: Simon Sawicki <accounts@grub4k.xyz> Co-authored-by: bashonly <bashonly@bashonly.com>
This commit is contained in:
@@ -281,17 +281,13 @@ class TestSocks4Proxy:
|
||||
rh, proxies={'all': f'socks4://user:@{server_address}'})
|
||||
assert response['version'] == 4
|
||||
|
||||
@pytest.mark.parametrize('handler,ctx', [
|
||||
pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
|
||||
reason='socks4a implementation currently broken when destination is not a domain name'))
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_socks4a_ipv4_target(self, handler, ctx):
|
||||
with ctx.socks_server(Socks4ProxyHandler) as server_address:
|
||||
with handler(proxies={'all': f'socks4a://{server_address}'}) as rh:
|
||||
response = ctx.socks_info_request(rh, target_domain='127.0.0.1')
|
||||
assert response['version'] == 4
|
||||
assert response['ipv4_address'] == '127.0.0.1'
|
||||
assert response['domain_address'] is None
|
||||
assert (response['ipv4_address'] == '127.0.0.1') != (response['domain_address'] == '127.0.0.1')
|
||||
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_socks4a_domain_target(self, handler, ctx):
|
||||
@@ -302,10 +298,7 @@ class TestSocks4Proxy:
|
||||
assert response['ipv4_address'] is None
|
||||
assert response['domain_address'] == 'localhost'
|
||||
|
||||
@pytest.mark.parametrize('handler,ctx', [
|
||||
pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
|
||||
reason='source_address is not yet supported for socks4 proxies'))
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_ipv4_client_source_address(self, handler, ctx):
|
||||
with ctx.socks_server(Socks4ProxyHandler) as server_address:
|
||||
source_address = f'127.0.0.{random.randint(5, 255)}'
|
||||
@@ -327,10 +320,7 @@ class TestSocks4Proxy:
|
||||
with pytest.raises(ProxyError):
|
||||
ctx.socks_info_request(rh)
|
||||
|
||||
@pytest.mark.parametrize('handler,ctx', [
|
||||
pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
|
||||
reason='IPv6 socks4 proxies are not yet supported'))
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_ipv6_socks4_proxy(self, handler, ctx):
|
||||
with ctx.socks_server(Socks4ProxyHandler, bind_ip='::1') as server_address:
|
||||
with handler(proxies={'all': f'socks4://{server_address}'}) as rh:
|
||||
@@ -342,7 +332,7 @@ class TestSocks4Proxy:
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_timeout(self, handler, ctx):
|
||||
with ctx.socks_server(Socks4ProxyHandler, sleep=2) as server_address:
|
||||
with handler(proxies={'all': f'socks4://{server_address}'}, timeout=1) as rh:
|
||||
with handler(proxies={'all': f'socks4://{server_address}'}, timeout=0.5) as rh:
|
||||
with pytest.raises(TransportError):
|
||||
ctx.socks_info_request(rh)
|
||||
|
||||
@@ -383,7 +373,7 @@ class TestSocks5Proxy:
|
||||
with ctx.socks_server(Socks5ProxyHandler) as server_address:
|
||||
with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
|
||||
response = ctx.socks_info_request(rh, target_domain='localhost')
|
||||
assert response['ipv4_address'] == '127.0.0.1'
|
||||
assert (response['ipv4_address'] == '127.0.0.1') != (response['ipv6_address'] == '::1')
|
||||
assert response['version'] == 5
|
||||
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
@@ -404,22 +394,15 @@ class TestSocks5Proxy:
|
||||
assert response['domain_address'] is None
|
||||
assert response['version'] == 5
|
||||
|
||||
@pytest.mark.parametrize('handler,ctx', [
|
||||
pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
|
||||
reason='IPv6 destination addresses are not yet supported'))
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_socks5_ipv6_destination(self, handler, ctx):
|
||||
with ctx.socks_server(Socks5ProxyHandler) as server_address:
|
||||
with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
|
||||
response = ctx.socks_info_request(rh, target_domain='[::1]')
|
||||
assert response['ipv6_address'] == '::1'
|
||||
assert response['port'] == 80
|
||||
assert response['version'] == 5
|
||||
|
||||
@pytest.mark.parametrize('handler,ctx', [
|
||||
pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
|
||||
reason='IPv6 socks5 proxies are not yet supported'))
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_ipv6_socks5_proxy(self, handler, ctx):
|
||||
with ctx.socks_server(Socks5ProxyHandler, bind_ip='::1') as server_address:
|
||||
with handler(proxies={'all': f'socks5://{server_address}'}) as rh:
|
||||
@@ -430,10 +413,7 @@ class TestSocks5Proxy:
|
||||
|
||||
# XXX: is there any feasible way of testing IPv6 source addresses?
|
||||
# Same would go for non-proxy source_address test...
|
||||
@pytest.mark.parametrize('handler,ctx', [
|
||||
pytest.param('Urllib', 'http', marks=pytest.mark.xfail(
|
||||
reason='source_address is not yet supported for socks5 proxies'))
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize('handler,ctx', [('Urllib', 'http')], indirect=True)
|
||||
def test_ipv4_client_source_address(self, handler, ctx):
|
||||
with ctx.socks_server(Socks5ProxyHandler) as server_address:
|
||||
source_address = f'127.0.0.{random.randint(5, 255)}'
|
||||
|
Reference in New Issue
Block a user