yt_dlp/networking/_helper.py

   1 from __future__ import annotations
   2
   3 import contextlib
   4 import ssl
   5 import sys
   6 import urllib.parse
   7
   8 from ..dependencies import certifi
   9 from ..socks import ProxyType
  10 from ..utils import YoutubeDLError
  11
  12
  13 def ssl_load_certs(context: ssl.SSLContext, use_certifi=True):
  14     if certifi and use_certifi:
  15         context.load_verify_locations(cafile=certifi.where())
  16     else:
  17         try:
  18             context.load_default_certs()
  19         # Work around the issue in load_default_certs when there are bad certificates. See:
  20         # https://github.com/yt-dlp/yt-dlp/issues/1060,
  21         # https://bugs.python.org/issue35665, https://bugs.python.org/issue45312
  22         except ssl.SSLError:
  23             # enum_certificates is not present in mingw python. See https://github.com/yt-dlp/yt-dlp/issues/1151
  24             if sys.platform == 'win32' and hasattr(ssl, 'enum_certificates'):
  25                 for storename in ('CA', 'ROOT'):
  26                     _ssl_load_windows_store_certs(context, storename)
  27             context.set_default_verify_paths()
  28
  29
  30 def _ssl_load_windows_store_certs(ssl_context, storename):
  31     # Code adapted from _load_windows_store_certs in https://github.com/python/cpython/blob/main/Lib/ssl.py
  32     try:
  33         certs = [cert for cert, encoding, trust in ssl.enum_certificates(storename)
  34                  if encoding == 'x509_asn' and (
  35                      trust is True or ssl.Purpose.SERVER_AUTH.oid in trust)]
  36     except PermissionError:
  37         return
  38     for cert in certs:
  39         with contextlib.suppress(ssl.SSLError):
  40             ssl_context.load_verify_locations(cadata=cert)
  41
  42
  43 def make_socks_proxy_opts(socks_proxy):
  44     url_components = urllib.parse.urlparse(socks_proxy)
  45     if url_components.scheme.lower() == 'socks5':
  46         socks_type = ProxyType.SOCKS5
  47     elif url_components.scheme.lower() in ('socks', 'socks4'):
  48         socks_type = ProxyType.SOCKS4
  49     elif url_components.scheme.lower() == 'socks4a':
  50         socks_type = ProxyType.SOCKS4A
  51
  52     def unquote_if_non_empty(s):
  53         if not s:
  54             return s
  55         return urllib.parse.unquote_plus(s)
  56     return {
  57         'proxytype': socks_type,
  58         'addr': url_components.hostname,
  59         'port': url_components.port or 1080,
  60         'rdns': True,
  61         'username': unquote_if_non_empty(url_components.username),
  62         'password': unquote_if_non_empty(url_components.password),
  63     }
  64
  65
  66 def get_redirect_method(method, status):
  67     """Unified redirect method handling"""
  68
  69     # A 303 must either use GET or HEAD for subsequent request
  70     # https://datatracker.ietf.org/doc/html/rfc7231#section-6.4.4
  71     if status == 303 and method != 'HEAD':
  72         method = 'GET'
  73     # 301 and 302 redirects are commonly turned into a GET from a POST
  74     # for subsequent requests by browsers, so we'll do the same.
  75     # https://datatracker.ietf.org/doc/html/rfc7231#section-6.4.2
  76     # https://datatracker.ietf.org/doc/html/rfc7231#section-6.4.3
  77     if status in (301, 302) and method == 'POST':
  78         method = 'GET'
  79     return method
  80
  81
  82 def make_ssl_context(
  83     verify=True,
  84     client_certificate=None,
  85     client_certificate_key=None,
  86     client_certificate_password=None,
  87     legacy_support=False,
  88     use_certifi=True,
  89 ):
  90     context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
  91     context.check_hostname = verify
  92     context.verify_mode = ssl.CERT_REQUIRED if verify else ssl.CERT_NONE
  93
  94     # Some servers may reject requests if ALPN extension is not sent. See:
  95     # https://github.com/python/cpython/issues/85140
  96     # https://github.com/yt-dlp/yt-dlp/issues/3878
  97     with contextlib.suppress(NotImplementedError):
  98         context.set_alpn_protocols(['http/1.1'])
  99     if verify:
 100         ssl_load_certs(context, use_certifi)
 101
 102     if legacy_support:
 103         context.options |= 4  # SSL_OP_LEGACY_SERVER_CONNECT
 104         context.set_ciphers('DEFAULT')  # compat
 105
 106     elif ssl.OPENSSL_VERSION_INFO >= (1, 1, 1) and not ssl.OPENSSL_VERSION.startswith('LibreSSL'):
 107         # Use the default SSL ciphers and minimum TLS version settings from Python 3.10 [1].
 108         # This is to ensure consistent behavior across Python versions and libraries, and help avoid fingerprinting
 109         # in some situations [2][3].
 110         # Python 3.10 only supports OpenSSL 1.1.1+ [4]. Because this change is likely
 111         # untested on older versions, we only apply this to OpenSSL 1.1.1+ to be safe.
 112         # LibreSSL is excluded until further investigation due to cipher support issues [5][6].
 113         # 1. https://github.com/python/cpython/commit/e983252b516edb15d4338b0a47631b59ef1e2536
 114         # 2. https://github.com/yt-dlp/yt-dlp/issues/4627
 115         # 3. https://github.com/yt-dlp/yt-dlp/pull/5294
 116         # 4. https://peps.python.org/pep-0644/
 117         # 5. https://peps.python.org/pep-0644/#libressl-support
 118         # 6. https://github.com/yt-dlp/yt-dlp/commit/5b9f253fa0aee996cf1ed30185d4b502e00609c4#commitcomment-89054368
 119         context.set_ciphers(
 120             '@SECLEVEL=2:ECDH+AESGCM:ECDH+CHACHA20:ECDH+AES:DHE+AES:!aNULL:!eNULL:!aDSS:!SHA1:!AESCCM')
 121         context.minimum_version = ssl.TLSVersion.TLSv1_2
 122
 123     if client_certificate:
 124         try:
 125             context.load_cert_chain(
 126                 client_certificate, keyfile=client_certificate_key,
 127                 password=client_certificate_password)
 128         except ssl.SSLError:
 129             raise YoutubeDLError('Unable to load client certificate')
 130
 131     return context
 132
 133
 134 def add_accept_encoding_header(headers, supported_encodings):
 135     if supported_encodings and 'Accept-Encoding' not in headers:
 136         headers['Accept-Encoding'] = ', '.join(supported_encodings)
 137
 138     elif 'Accept-Encoding' not in headers:
 139         headers['Accept-Encoding'] = 'identity'