test/test_http.py

   1 #!/usr/bin/env python3
   2 # Allow direct execution
   3 import os
   4 import sys
   5 import unittest
   6 import http.server
   7
   8 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
   9
  10
  11 import ssl
  12 import threading
  13 import urllib.request
  14 from test.helper import http_server_port
  15
  16 from yt_dlp import YoutubeDL
  17
  18 TEST_DIR = os.path.dirname(os.path.abspath(__file__))
  19
  20
  21 class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler):
  22     def log_message(self, format, *args):
  23         pass
  24
  25     def do_GET(self):
  26         if self.path == '/video.html':
  27             self.send_response(200)
  28             self.send_header('Content-Type', 'text/html; charset=utf-8')
  29             self.end_headers()
  30             self.wfile.write(b'<html><video src="/vid.mp4" /></html>')
  31         elif self.path == '/vid.mp4':
  32             self.send_response(200)
  33             self.send_header('Content-Type', 'video/mp4')
  34             self.end_headers()
  35             self.wfile.write(b'\x00\x00\x00\x00\x20\x66\x74[video]')
  36         elif self.path == '/%E4%B8%AD%E6%96%87.html':
  37             self.send_response(200)
  38             self.send_header('Content-Type', 'text/html; charset=utf-8')
  39             self.end_headers()
  40             self.wfile.write(b'<html><video src="/vid.mp4" /></html>')
  41         else:
  42             assert False
  43
  44
  45 class FakeLogger:
  46     def debug(self, msg):
  47         pass
  48
  49     def warning(self, msg):
  50         pass
  51
  52     def error(self, msg):
  53         pass
  54
  55
  56 class TestHTTP(unittest.TestCase):
  57     def setUp(self):
  58         self.httpd = http.server.HTTPServer(
  59             ('127.0.0.1', 0), HTTPTestRequestHandler)
  60         self.port = http_server_port(self.httpd)
  61         self.server_thread = threading.Thread(target=self.httpd.serve_forever)
  62         self.server_thread.daemon = True
  63         self.server_thread.start()
  64
  65
  66 class TestHTTPS(unittest.TestCase):
  67     def setUp(self):
  68         certfn = os.path.join(TEST_DIR, 'testcert.pem')
  69         self.httpd = http.server.HTTPServer(
  70             ('127.0.0.1', 0), HTTPTestRequestHandler)
  71         sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
  72         sslctx.load_cert_chain(certfn, None)
  73         self.httpd.socket = sslctx.wrap_socket(self.httpd.socket, server_side=True)
  74         self.port = http_server_port(self.httpd)
  75         self.server_thread = threading.Thread(target=self.httpd.serve_forever)
  76         self.server_thread.daemon = True
  77         self.server_thread.start()
  78
  79     def test_nocheckcertificate(self):
  80         ydl = YoutubeDL({'logger': FakeLogger()})
  81         self.assertRaises(
  82             Exception,
  83             ydl.extract_info, 'https://127.0.0.1:%d/video.html' % self.port)
  84
  85         ydl = YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True})
  86         r = ydl.extract_info('https://127.0.0.1:%d/video.html' % self.port)
  87         self.assertEqual(r['entries'][0]['url'], 'https://127.0.0.1:%d/vid.mp4' % self.port)
  88
  89
  90 class TestClientCert(unittest.TestCase):
  91     def setUp(self):
  92         certfn = os.path.join(TEST_DIR, 'testcert.pem')
  93         self.certdir = os.path.join(TEST_DIR, 'testdata', 'certificate')
  94         cacertfn = os.path.join(self.certdir, 'ca.crt')
  95         self.httpd = http.server.HTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler)
  96         sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
  97         sslctx.verify_mode = ssl.CERT_REQUIRED
  98         sslctx.load_verify_locations(cafile=cacertfn)
  99         sslctx.load_cert_chain(certfn, None)
 100         self.httpd.socket = sslctx.wrap_socket(self.httpd.socket, server_side=True)
 101         self.port = http_server_port(self.httpd)
 102         self.server_thread = threading.Thread(target=self.httpd.serve_forever)
 103         self.server_thread.daemon = True
 104         self.server_thread.start()
 105
 106     def _run_test(self, **params):
 107         ydl = YoutubeDL({
 108             'logger': FakeLogger(),
 109             # Disable client-side validation of unacceptable self-signed testcert.pem
 110             # The test is of a check on the server side, so unaffected
 111             'nocheckcertificate': True,
 112             **params,
 113         })
 114         r = ydl.extract_info('https://127.0.0.1:%d/video.html' % self.port)
 115         self.assertEqual(r['entries'][0]['url'], 'https://127.0.0.1:%d/vid.mp4' % self.port)
 116
 117     def test_certificate_combined_nopass(self):
 118         self._run_test(client_certificate=os.path.join(self.certdir, 'clientwithkey.crt'))
 119
 120     def test_certificate_nocombined_nopass(self):
 121         self._run_test(client_certificate=os.path.join(self.certdir, 'client.crt'),
 122                        client_certificate_key=os.path.join(self.certdir, 'client.key'))
 123
 124     def test_certificate_combined_pass(self):
 125         self._run_test(client_certificate=os.path.join(self.certdir, 'clientwithencryptedkey.crt'),
 126                        client_certificate_password='foobar')
 127
 128     def test_certificate_nocombined_pass(self):
 129         self._run_test(client_certificate=os.path.join(self.certdir, 'client.crt'),
 130                        client_certificate_key=os.path.join(self.certdir, 'clientencrypted.key'),
 131                        client_certificate_password='foobar')
 132
 133
 134 def _build_proxy_handler(name):
 135     class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler):
 136         proxy_name = name
 137
 138         def log_message(self, format, *args):
 139             pass
 140
 141         def do_GET(self):
 142             self.send_response(200)
 143             self.send_header('Content-Type', 'text/plain; charset=utf-8')
 144             self.end_headers()
 145             self.wfile.write(f'{self.proxy_name}: {self.path}'.encode())
 146     return HTTPTestRequestHandler
 147
 148
 149 class TestProxy(unittest.TestCase):
 150     def setUp(self):
 151         self.proxy = http.server.HTTPServer(
 152             ('127.0.0.1', 0), _build_proxy_handler('normal'))
 153         self.port = http_server_port(self.proxy)
 154         self.proxy_thread = threading.Thread(target=self.proxy.serve_forever)
 155         self.proxy_thread.daemon = True
 156         self.proxy_thread.start()
 157
 158         self.geo_proxy = http.server.HTTPServer(
 159             ('127.0.0.1', 0), _build_proxy_handler('geo'))
 160         self.geo_port = http_server_port(self.geo_proxy)
 161         self.geo_proxy_thread = threading.Thread(target=self.geo_proxy.serve_forever)
 162         self.geo_proxy_thread.daemon = True
 163         self.geo_proxy_thread.start()
 164
 165     def test_proxy(self):
 166         geo_proxy = f'127.0.0.1:{self.geo_port}'
 167         ydl = YoutubeDL({
 168             'proxy': f'127.0.0.1:{self.port}',
 169             'geo_verification_proxy': geo_proxy,
 170         })
 171         url = 'http://foo.com/bar'
 172         response = ydl.urlopen(url).read().decode()
 173         self.assertEqual(response, f'normal: {url}')
 174
 175         req = urllib.request.Request(url)
 176         req.add_header('Ytdl-request-proxy', geo_proxy)
 177         response = ydl.urlopen(req).read().decode()
 178         self.assertEqual(response, f'geo: {url}')
 179
 180     def test_proxy_with_idn(self):
 181         ydl = YoutubeDL({
 182             'proxy': f'127.0.0.1:{self.port}',
 183         })
 184         url = 'http://中文.tw/'
 185         response = ydl.urlopen(url).read().decode()
 186         # b'xn--fiq228c' is '中文'.encode('idna')
 187         self.assertEqual(response, 'normal: http://xn--fiq228c.tw/')
 188
 189
 190 if __name__ == '__main__':
 191     unittest.main()