summaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/test_networking.py82
-rw-r--r--test/test_utils.py5
-rw-r--r--test/test_websockets.py58
3 files changed, 144 insertions, 1 deletions
diff --git a/test/test_networking.py b/test/test_networking.py
index af3ece3..826f11a 100644
--- a/test/test_networking.py
+++ b/test/test_networking.py
@@ -265,6 +265,11 @@ class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler):
self.end_headers()
self.wfile.write(payload)
self.finish()
+ elif self.path == '/get_cookie':
+ self.send_response(200)
+ self.send_header('Set-Cookie', 'test=ytdlp; path=/')
+ self.end_headers()
+ self.finish()
else:
self._status(404)
@@ -338,6 +343,52 @@ class TestHTTPRequestHandler(TestRequestHandlerBase):
validate_and_send(rh, Request(f'https://127.0.0.1:{https_port}/headers'))
assert not issubclass(exc_info.type, CertificateVerifyError)
+ @pytest.mark.skip_handler('CurlCFFI', 'legacy_ssl ignored by CurlCFFI')
+ def test_legacy_ssl_extension(self, handler):
+ # HTTPS server with old ciphers
+ # XXX: is there a better way to test this than to create a new server?
+ https_httpd = http.server.ThreadingHTTPServer(
+ ('127.0.0.1', 0), HTTPTestRequestHandler)
+ sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ sslctx.maximum_version = ssl.TLSVersion.TLSv1_2
+ sslctx.set_ciphers('SHA1:AESCCM:aDSS:eNULL:aNULL')
+ sslctx.load_cert_chain(os.path.join(TEST_DIR, 'testcert.pem'), None)
+ https_httpd.socket = sslctx.wrap_socket(https_httpd.socket, server_side=True)
+ https_port = http_server_port(https_httpd)
+ https_server_thread = threading.Thread(target=https_httpd.serve_forever)
+ https_server_thread.daemon = True
+ https_server_thread.start()
+
+ with handler(verify=False) as rh:
+ res = validate_and_send(rh, Request(f'https://127.0.0.1:{https_port}/headers', extensions={'legacy_ssl': True}))
+ assert res.status == 200
+ res.close()
+
+ # Ensure only applies to request extension
+ with pytest.raises(SSLError):
+ validate_and_send(rh, Request(f'https://127.0.0.1:{https_port}/headers'))
+
+ @pytest.mark.skip_handler('CurlCFFI', 'legacy_ssl ignored by CurlCFFI')
+ def test_legacy_ssl_support(self, handler):
+ # HTTPS server with old ciphers
+ # XXX: is there a better way to test this than to create a new server?
+ https_httpd = http.server.ThreadingHTTPServer(
+ ('127.0.0.1', 0), HTTPTestRequestHandler)
+ sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ sslctx.maximum_version = ssl.TLSVersion.TLSv1_2
+ sslctx.set_ciphers('SHA1:AESCCM:aDSS:eNULL:aNULL')
+ sslctx.load_cert_chain(os.path.join(TEST_DIR, 'testcert.pem'), None)
+ https_httpd.socket = sslctx.wrap_socket(https_httpd.socket, server_side=True)
+ https_port = http_server_port(https_httpd)
+ https_server_thread = threading.Thread(target=https_httpd.serve_forever)
+ https_server_thread.daemon = True
+ https_server_thread.start()
+
+ with handler(verify=False, legacy_ssl_support=True) as rh:
+ res = validate_and_send(rh, Request(f'https://127.0.0.1:{https_port}/headers'))
+ assert res.status == 200
+ res.close()
+
def test_percent_encode(self, handler):
with handler() as rh:
# Unicode characters should be encoded with uppercase percent-encoding
@@ -490,6 +541,24 @@ class TestHTTPRequestHandler(TestRequestHandlerBase):
rh, Request(f'http://127.0.0.1:{self.http_port}/headers', extensions={'cookiejar': cookiejar})).read()
assert b'cookie: test=ytdlp' in data.lower()
+ def test_cookie_sync_only_cookiejar(self, handler):
+ # Ensure that cookies are ONLY being handled by the cookiejar
+ with handler() as rh:
+ validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/get_cookie', extensions={'cookiejar': YoutubeDLCookieJar()}))
+ data = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/headers', extensions={'cookiejar': YoutubeDLCookieJar()})).read()
+ assert b'cookie: test=ytdlp' not in data.lower()
+
+ def test_cookie_sync_delete_cookie(self, handler):
+ # Ensure that cookies are ONLY being handled by the cookiejar
+ cookiejar = YoutubeDLCookieJar()
+ with handler(cookiejar=cookiejar) as rh:
+ validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/get_cookie'))
+ data = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/headers')).read()
+ assert b'cookie: test=ytdlp' in data.lower()
+ cookiejar.clear_session_cookies()
+ data = validate_and_send(rh, Request(f'http://127.0.0.1:{self.http_port}/headers')).read()
+ assert b'cookie: test=ytdlp' not in data.lower()
+
def test_headers(self, handler):
with handler(headers=HTTPHeaderDict({'test1': 'test', 'test2': 'test2'})) as rh:
@@ -914,7 +983,6 @@ class TestRequestsRequestHandler(TestRequestHandlerBase):
class TestCurlCFFIRequestHandler(TestRequestHandlerBase):
@pytest.mark.parametrize('params,extensions', [
- ({}, {'impersonate': ImpersonateTarget('chrome')}),
({'impersonate': ImpersonateTarget('chrome', '110')}, {}),
({'impersonate': ImpersonateTarget('chrome', '99')}, {'impersonate': ImpersonateTarget('chrome', '110')}),
])
@@ -1200,6 +1268,9 @@ class TestRequestHandlerValidation:
({'timeout': 1}, False),
({'timeout': 'notatimeout'}, AssertionError),
({'unsupported': 'value'}, UnsupportedRequest),
+ ({'legacy_ssl': False}, False),
+ ({'legacy_ssl': True}, False),
+ ({'legacy_ssl': 'notabool'}, AssertionError),
]),
('Requests', 'http', [
({'cookiejar': 'notacookiejar'}, AssertionError),
@@ -1207,6 +1278,9 @@ class TestRequestHandlerValidation:
({'timeout': 1}, False),
({'timeout': 'notatimeout'}, AssertionError),
({'unsupported': 'value'}, UnsupportedRequest),
+ ({'legacy_ssl': False}, False),
+ ({'legacy_ssl': True}, False),
+ ({'legacy_ssl': 'notabool'}, AssertionError),
]),
('CurlCFFI', 'http', [
({'cookiejar': 'notacookiejar'}, AssertionError),
@@ -1220,6 +1294,9 @@ class TestRequestHandlerValidation:
({'impersonate': ImpersonateTarget(None, None, None, None)}, False),
({'impersonate': ImpersonateTarget()}, False),
({'impersonate': 'chrome'}, AssertionError),
+ ({'legacy_ssl': False}, False),
+ ({'legacy_ssl': True}, False),
+ ({'legacy_ssl': 'notabool'}, AssertionError),
]),
(NoCheckRH, 'http', [
({'cookiejar': 'notacookiejar'}, False),
@@ -1228,6 +1305,9 @@ class TestRequestHandlerValidation:
('Websockets', 'ws', [
({'cookiejar': YoutubeDLCookieJar()}, False),
({'timeout': 2}, False),
+ ({'legacy_ssl': False}, False),
+ ({'legacy_ssl': True}, False),
+ ({'legacy_ssl': 'notabool'}, AssertionError),
]),
]
diff --git a/test/test_utils.py b/test/test_utils.py
index 3ff1f8b..e82f551 100644
--- a/test/test_utils.py
+++ b/test/test_utils.py
@@ -929,6 +929,11 @@ class TestUtil(unittest.TestCase):
'acodec': 'none',
'dynamic_range': 'DV',
})
+ self.assertEqual(parse_codecs('fLaC'), {
+ 'vcodec': 'none',
+ 'acodec': 'flac',
+ 'dynamic_range': None,
+ })
self.assertEqual(parse_codecs('theora, vorbis'), {
'vcodec': 'theora',
'acodec': 'vorbis',
diff --git a/test/test_websockets.py b/test/test_websockets.py
index 5f101ab..43f20ac 100644
--- a/test/test_websockets.py
+++ b/test/test_websockets.py
@@ -61,6 +61,10 @@ def process_request(self, request):
return websockets.http11.Response(
status.value, status.phrase, websockets.datastructures.Headers([('Location', '/')]), b'')
return self.protocol.reject(status.value, status.phrase)
+ elif request.path.startswith('/get_cookie'):
+ response = self.protocol.accept(request)
+ response.headers['Set-Cookie'] = 'test=ytdlp'
+ return response
return self.protocol.accept(request)
@@ -102,6 +106,15 @@ def create_mtls_wss_websocket_server():
return create_websocket_server(ssl_context=sslctx)
+def create_legacy_wss_websocket_server():
+ certfn = os.path.join(TEST_DIR, 'testcert.pem')
+ sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ sslctx.maximum_version = ssl.TLSVersion.TLSv1_2
+ sslctx.set_ciphers('SHA1:AESCCM:aDSS:eNULL:aNULL')
+ sslctx.load_cert_chain(certfn, None)
+ return create_websocket_server(ssl_context=sslctx)
+
+
def ws_validate_and_send(rh, req):
rh.validate(req)
max_tries = 3
@@ -132,6 +145,9 @@ class TestWebsSocketRequestHandlerConformance:
cls.mtls_wss_thread, cls.mtls_wss_port = create_mtls_wss_websocket_server()
cls.mtls_wss_base_url = f'wss://127.0.0.1:{cls.mtls_wss_port}'
+ cls.legacy_wss_thread, cls.legacy_wss_port = create_legacy_wss_websocket_server()
+ cls.legacy_wss_host = f'wss://127.0.0.1:{cls.legacy_wss_port}'
+
def test_basic_websockets(self, handler):
with handler() as rh:
ws = ws_validate_and_send(rh, Request(self.ws_base_url))
@@ -166,6 +182,22 @@ class TestWebsSocketRequestHandlerConformance:
ws_validate_and_send(rh, Request(self.bad_wss_host))
assert not issubclass(exc_info.type, CertificateVerifyError)
+ def test_legacy_ssl_extension(self, handler):
+ with handler(verify=False) as rh:
+ ws = ws_validate_and_send(rh, Request(self.legacy_wss_host, extensions={'legacy_ssl': True}))
+ assert ws.status == 101
+ ws.close()
+
+ # Ensure only applies to request extension
+ with pytest.raises(SSLError):
+ ws_validate_and_send(rh, Request(self.legacy_wss_host))
+
+ def test_legacy_ssl_support(self, handler):
+ with handler(verify=False, legacy_ssl_support=True) as rh:
+ ws = ws_validate_and_send(rh, Request(self.legacy_wss_host))
+ assert ws.status == 101
+ ws.close()
+
@pytest.mark.parametrize('path,expected', [
# Unicode characters should be encoded with uppercase percent-encoding
('/中文', '/%E4%B8%AD%E6%96%87'),
@@ -248,6 +280,32 @@ class TestWebsSocketRequestHandlerConformance:
assert json.loads(ws.recv())['cookie'] == 'test=ytdlp'
ws.close()
+ @pytest.mark.skip_handler('Websockets', 'Set-Cookie not supported by websockets')
+ def test_cookie_sync_only_cookiejar(self, handler):
+ # Ensure that cookies are ONLY being handled by the cookiejar
+ with handler() as rh:
+ ws_validate_and_send(rh, Request(f'{self.ws_base_url}/get_cookie', extensions={'cookiejar': YoutubeDLCookieJar()}))
+ ws = ws_validate_and_send(rh, Request(self.ws_base_url, extensions={'cookiejar': YoutubeDLCookieJar()}))
+ ws.send('headers')
+ assert 'cookie' not in json.loads(ws.recv())
+ ws.close()
+
+ @pytest.mark.skip_handler('Websockets', 'Set-Cookie not supported by websockets')
+ def test_cookie_sync_delete_cookie(self, handler):
+ # Ensure that cookies are ONLY being handled by the cookiejar
+ cookiejar = YoutubeDLCookieJar()
+ with handler(verbose=True, cookiejar=cookiejar) as rh:
+ ws_validate_and_send(rh, Request(f'{self.ws_base_url}/get_cookie'))
+ ws = ws_validate_and_send(rh, Request(self.ws_base_url))
+ ws.send('headers')
+ assert json.loads(ws.recv())['cookie'] == 'test=ytdlp'
+ ws.close()
+ cookiejar.clear_session_cookies()
+ ws = ws_validate_and_send(rh, Request(self.ws_base_url))
+ ws.send('headers')
+ assert 'cookie' not in json.loads(ws.recv())
+ ws.close()
+
def test_source_address(self, handler):
source_address = f'127.0.0.{random.randint(5, 255)}'
verify_address_availability(source_address)
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-21 23:11:34 +0000