1 from __future__
import annotations
7 from ._helper
import InstanceStoreMixin
, select_proxy
15 from .exceptions
import (
16 CertificateVerifyError
,
23 from .impersonate
import ImpersonateRequestHandler
, ImpersonateTarget
24 from ..dependencies
import curl_cffi
25 from ..utils
import int_or_none
28 raise ImportError('curl_cffi is not installed')
30 curl_cffi_version
= tuple(int_or_none(x
, default
=0) for x
in curl_cffi
.__version
__.split('.'))
32 if curl_cffi_version
!= (0, 5, 10):
33 curl_cffi
._yt
_dlp
__version
= f
'{curl_cffi.__version__} (unsupported)'
34 raise ImportError('Only curl_cffi 0.5.10 is supported')
36 import curl_cffi
.requests
37 from curl_cffi
.const
import CurlECode
, CurlOpt
40 class CurlCFFIResponseReader(io
.IOBase
):
41 def __init__(self
, response
: curl_cffi
.requests
.Response
):
42 self
._response
= response
43 self
._iterator
= response
.iter_content()
50 def read(self
, size
=None):
51 exception_raised
= True
53 while self
._iterator
and (size
is None or len(self
._buffer
) < size
):
54 chunk
= next(self
._iterator
, None)
59 self
.bytes_read
+= len(chunk
)
62 size
= len(self
._buffer
)
63 data
= self
._buffer
[:size
]
64 self
._buffer
= self
._buffer
[size
:]
66 # "free" the curl instance if the response is fully read.
67 # curl_cffi doesn't do this automatically and only allows one open response per thread
68 if not self
._iterator
and not self
._buffer
:
70 exception_raised
= False
78 self
._response
.close()
83 class CurlCFFIResponseAdapter(Response
):
84 fp
: CurlCFFIResponseReader
86 def __init__(self
, response
: curl_cffi
.requests
.Response
):
88 fp
=CurlCFFIResponseReader(response
),
89 headers
=response
.headers
,
91 status
=response
.status_code
)
93 def read(self
, amt
=None):
95 return self
.fp
.read(amt
)
96 except curl_cffi
.requests
.errors
.RequestsError
as e
:
97 if e
.code
== CurlECode
.PARTIAL_FILE
:
98 content_length
= int_or_none(e
.response
.headers
.get('Content-Length'))
100 partial
=self
.fp
.bytes_read
,
101 expected
=content_length
- self
.fp
.bytes_read
if content_length
is not None else None,
103 raise TransportError(cause
=e
) from e
107 class CurlCFFIRH(ImpersonateRequestHandler
, InstanceStoreMixin
):
108 RH_NAME
= 'curl_cffi'
109 _SUPPORTED_URL_SCHEMES
= ('http', 'https')
110 _SUPPORTED_FEATURES
= (Features
.NO_PROXY
, Features
.ALL_PROXY
)
111 _SUPPORTED_PROXY_SCHEMES
= ('http', 'https', 'socks4', 'socks4a', 'socks5', 'socks5h')
112 _SUPPORTED_IMPERSONATE_TARGET_MAP
= {
113 ImpersonateTarget('chrome', '110', 'windows', '10'): curl_cffi
.requests
.BrowserType
.chrome110
,
114 ImpersonateTarget('chrome', '107', 'windows', '10'): curl_cffi
.requests
.BrowserType
.chrome107
,
115 ImpersonateTarget('chrome', '104', 'windows', '10'): curl_cffi
.requests
.BrowserType
.chrome104
,
116 ImpersonateTarget('chrome', '101', 'windows', '10'): curl_cffi
.requests
.BrowserType
.chrome101
,
117 ImpersonateTarget('chrome', '100', 'windows', '10'): curl_cffi
.requests
.BrowserType
.chrome100
,
118 ImpersonateTarget('chrome', '99', 'windows', '10'): curl_cffi
.requests
.BrowserType
.chrome99
,
119 ImpersonateTarget('edge', '101', 'windows', '10'): curl_cffi
.requests
.BrowserType
.edge101
,
120 ImpersonateTarget('edge', '99', 'windows', '10'): curl_cffi
.requests
.BrowserType
.edge99
,
121 ImpersonateTarget('safari', '15.5', 'macos', '12'): curl_cffi
.requests
.BrowserType
.safari15_5
,
122 ImpersonateTarget('safari', '15.3', 'macos', '11'): curl_cffi
.requests
.BrowserType
.safari15_3
,
123 ImpersonateTarget('chrome', '99', 'android', '12'): curl_cffi
.requests
.BrowserType
.chrome99_android
,
126 def _create_instance(self
, cookiejar
=None):
127 return curl_cffi
.requests
.Session(cookies
=cookiejar
)
129 def _check_extensions(self
, extensions
):
130 super()._check
_extensions
(extensions
)
131 extensions
.pop('impersonate', None)
132 extensions
.pop('cookiejar', None)
133 extensions
.pop('timeout', None)
135 def send(self
, request
: Request
) -> Response
:
136 target
= self
._get
_request
_target
(request
)
138 response
= super().send(request
)
139 except HTTPError
as e
:
140 e
.response
.extensions
['impersonate'] = target
142 response
.extensions
['impersonate'] = target
145 def _send(self
, request
: Request
):
146 max_redirects_exceeded
= False
147 session
: curl_cffi
.requests
.Session
= self
._get
_instance
(
148 cookiejar
=self
._get
_cookiejar
(request
) if 'cookie' not in request
.headers
else None)
151 session
.curl
.setopt(CurlOpt
.VERBOSE
, 1)
153 proxies
= self
._get
_proxies
(request
)
155 session
.curl
.setopt(CurlOpt
.NOPROXY
, proxies
['no'])
156 proxies
.pop('no', None)
158 # curl doesn't support per protocol proxies, so we select the one that matches the request protocol
159 proxy
= select_proxy(request
.url
, proxies
=proxies
)
161 session
.curl
.setopt(CurlOpt
.PROXY
, proxy
)
162 scheme
= urllib
.parse
.urlparse(request
.url
).scheme
.lower()
164 # Enable HTTP CONNECT for HTTPS urls.
165 # Don't use CONNECT for http for compatibility with urllib behaviour.
166 # See: https://curl.se/libcurl/c/CURLOPT_HTTPPROXYTUNNEL.html
167 session
.curl
.setopt(CurlOpt
.HTTPPROXYTUNNEL
, 1)
169 headers
= self
._get
_impersonate
_headers
(request
)
171 if self
._client
_cert
:
172 session
.curl
.setopt(CurlOpt
.SSLCERT
, self
._client
_cert
['client_certificate'])
173 client_certificate_key
= self
._client
_cert
.get('client_certificate_key')
174 client_certificate_password
= self
._client
_cert
.get('client_certificate_password')
175 if client_certificate_key
:
176 session
.curl
.setopt(CurlOpt
.SSLKEY
, client_certificate_key
)
177 if client_certificate_password
:
178 session
.curl
.setopt(CurlOpt
.KEYPASSWD
, client_certificate_password
)
180 timeout
= self
._calculate
_timeout
(request
)
182 # set CURLOPT_LOW_SPEED_LIMIT and CURLOPT_LOW_SPEED_TIME to act as a read timeout. [1]
183 # curl_cffi does not currently do this. [2]
184 # Note: CURLOPT_LOW_SPEED_TIME is in seconds, so we need to round up to the nearest second. [3]
185 # [1] https://unix.stackexchange.com/a/305311
186 # [2] https://github.com/yifeikong/curl_cffi/issues/156
187 # [3] https://curl.se/libcurl/c/CURLOPT_LOW_SPEED_TIME.html
188 session
.curl
.setopt(CurlOpt
.LOW_SPEED_LIMIT
, 1) # 1 byte per second
189 session
.curl
.setopt(CurlOpt
.LOW_SPEED_TIME
, math
.ceil(timeout
))
192 curl_response
= session
.request(
193 method
=request
.method
,
200 impersonate
=self
._SUPPORTED
_IMPERSONATE
_TARGET
_MAP
.get(
201 self
._get
_request
_target
(request
)),
202 interface
=self
.source_address
,
205 except curl_cffi
.requests
.errors
.RequestsError
as e
:
206 if e
.code
== CurlECode
.PEER_FAILED_VERIFICATION
:
207 raise CertificateVerifyError(cause
=e
) from e
209 elif e
.code
== CurlECode
.SSL_CONNECT_ERROR
:
210 raise SSLError(cause
=e
) from e
212 elif e
.code
== CurlECode
.TOO_MANY_REDIRECTS
:
213 max_redirects_exceeded
= True
214 curl_response
= e
.response
216 elif e
.code
== CurlECode
.PROXY
:
217 raise ProxyError(cause
=e
) from e
219 raise TransportError(cause
=e
) from e
221 response
= CurlCFFIResponseAdapter(curl_response
)
223 if not 200 <= response
.status
< 300:
224 raise HTTPError(response
, redirect_loop
=max_redirects_exceeded
)
229 @register_preference(CurlCFFIRH
)
230 def curl_cffi_preference(rh
, request
):