(self)
| 52 | _http_client.HTTPSConnection.__init__(self, *args, **kwargs) |
| 53 | |
| 54 | def connect(self): |
| 55 | def create_sock(): |
| 56 | sock = socket.create_connection((self.host, self.port), self.timeout) |
| 57 | if getattr(self, "_tunnel_host", None): |
| 58 | self.sock = sock |
| 59 | self._tunnel() |
| 60 | return sock |
| 61 | |
| 62 | success = False |
| 63 | |
| 64 | # Reference(s): https://docs.python.org/2/library/ssl.html#ssl.SSLContext |
| 65 | # https://www.mnot.net/blog/2014/12/27/python_2_and_tls_sni |
| 66 | if hasattr(ssl, "SSLContext"): |
| 67 | for protocol in (_ for _ in _protocols if _ >= ssl.PROTOCOL_TLSv1): |
| 68 | try: |
| 69 | sock = create_sock() |
| 70 | if protocol not in _contexts: |
| 71 | _contexts[protocol] = ssl.SSLContext(protocol) |
| 72 | |
| 73 | # Disable certificate and hostname validation enabled by default with PROTOCOL_TLS_CLIENT |
| 74 | _contexts[protocol].check_hostname = False |
| 75 | _contexts[protocol].verify_mode = ssl.CERT_NONE |
| 76 | |
| 77 | if getattr(self, "cert_file", None) and getattr(self, "key_file", None): |
| 78 | _contexts[protocol].load_cert_chain(certfile=self.cert_file, keyfile=self.key_file) |
| 79 | try: |
| 80 | # Reference(s): https://askubuntu.com/a/1263098 |
| 81 | # https://askubuntu.com/a/1250807 |
| 82 | # https://git.zknt.org/mirror/bazarr/commit/7f05f932ffb84ba8b9e5630b2adc34dbd77e2b4a?style=split&whitespace=show-all&show-outdated= |
| 83 | _contexts[protocol].set_ciphers("ALL@SECLEVEL=0") |
| 84 | except (ssl.SSLError, AttributeError): |
| 85 | pass |
| 86 | result = _contexts[protocol].wrap_socket(sock, do_handshake_on_connect=True, server_hostname=self.host if re.search(r"\A[\d.]+\Z", self.host or "") is None else None) |
| 87 | if result: |
| 88 | success = True |
| 89 | self.sock = result |
| 90 | _protocols.remove(protocol) |
| 91 | _protocols.insert(0, protocol) |
| 92 | break |
| 93 | else: |
| 94 | sock.close() |
| 95 | except (ssl.SSLError, socket.error, _http_client.BadStatusLine, AttributeError) as ex: |
| 96 | self._tunnel_host = None |
| 97 | logger.debug("SSL connection error occurred for '%s' ('%s')" % (_lut[protocol], getSafeExString(ex))) |
| 98 | |
| 99 | elif hasattr(ssl, "wrap_socket"): |
| 100 | for protocol in _protocols: |
| 101 | try: |
| 102 | sock = create_sock() |
| 103 | _ = ssl.wrap_socket(sock, keyfile=getattr(self, "key_file"), certfile=getattr(self, "cert_file"), ssl_version=protocol) |
| 104 | if _: |
| 105 | success = True |
| 106 | self.sock = _ |
| 107 | _protocols.remove(protocol) |
| 108 | _protocols.insert(0, protocol) |
| 109 | break |
| 110 | else: |
| 111 | sock.close() |
no test coverage detected