Trying to get the SSL certificate from a response in requests.
What is a good way to do this?
Trying to get the SSL certificate from a response in requests.
What is a good way to do this?
requests deliberately wraps up low-level stuff like this. Normally, the only thing you want to do is to verify that the certs are valid. To do that, just pass verify=True. If you want to use a non-standard cacert bundle, you can pass that too. For example:
resp = requests.get('https://example.com', verify=True, cert=['/path/to/my/ca.crt'])
Also, requests is primarily a set of wrappers around other libraries, mostly urllib3 and the stdlib's http.client (or, for 2.x, httplib) and ssl.
Sometimes, the answer is just to get at the lower-level objects (e.g., resp.raw is the urllib3.response.HTTPResponse), but in many cases that's impossible.
And this is one of those cases. The only objects that ever see the certs are an http.client.HTTPSConnection (or a urllib3.connectionpool.VerifiedHTTPSConnection, but that's just a subclass of the former) and an ssl.SSLSocket, and neither of those exist anymore by the time the request returns. (As the name connectionpool implies, the HTTPSConnection object is stored in a pool, and may be reused as soon as it's done; the SSLSocket is a member of the HTTPSConnection.)
So, you need to patch things so you can copy the data up the chain. It may be as simple as this:
HTTPResponse = requests.packages.urllib3.response.HTTPResponse
orig_HTTPResponse__init__ = HTTPResponse.__init__
def new_HTTPResponse__init__(self, *args, **kwargs):
orig_HTTPResponse__init__(self, *args, **kwargs)
try:
self.peercert = self._connection.sock.getpeercert()
except AttributeError:
pass
HTTPResponse.__init__ = new_HTTPResponse__init__
HTTPAdapter = requests.adapters.HTTPAdapter
orig_HTTPAdapter_build_response = HTTPAdapter.build_response
def new_HTTPAdapter_build_response(self, request, resp):
response = orig_HTTPAdapter_build_response(self, request, resp)
try:
response.peercert = resp.peercert
except AttributeError:
pass
return response
HTTPAdapter.build_response = new_HTTPAdapter_build_response
That's untested, so no guarantees; you may need to patch more than that.
Also, subclassing and overriding would probably be cleaner than monkeypatching (especially since HTTPAdapter was designed to be subclassed).
Or, even better, forking urllib3 and requests, modifying your fork, and (if you think this is legitimately useful) submitting pull requests upstream.
Anyway, now, from your code, you can do this:
resp.peercert
This will give you a dict with 'subject' and 'subjectAltName' keys, as returned by pyopenssl.WrappedSocket.getpeercert. If you instead want more information about the cert, try Christophe Vandeplas's variant of this answer that lets you get an OpenSSL.crypto.X509 object. If you want to get the entire peer certificate chain, see GoldenStake's answer.
Of course you may also want to pass along all the information necessary to verify the cert, but that's even easier, because it already passes through the top level.
To start, abarnert's answer is very complete. While chasing the proposed connection-close issue of Kalkran I actually discovered that the peercert didn't contain detailed information about the SSL Certificate.
I dug deeper in the connection and socket info and extracted the self.sock.connection.get_peer_certificate() function which contains great functions like:
get_subject() for CNget_notAfter() and get_notBefore() for expiration datesget_serial_number() and get_signature_algorithm() for crypto related technical details Note that these are only available if you have pyopenssl installed on your system. Under the hood, urllib3 uses pyopenssl if it's available and the standard library's ssl module otherwise. The self.sock.connection attribute shown below only exists if self.sock is a urllib3.contrib.pyopenssl.WrappedSocket, not if it's a ssl.SSLSocket. You can install pyopenssl with pip install pyopenssl.
Once that's done, the code becomes:
import requests
HTTPResponse = requests.packages.urllib3.response.HTTPResponse
orig_HTTPResponse__init__ = HTTPResponse.__init__
def new_HTTPResponse__init__(self, *args, **kwargs):
orig_HTTPResponse__init__(self, *args, **kwargs)
try:
self.peer_certificate = self._connection.peer_certificate
except AttributeError:
pass
HTTPResponse.__init__ = new_HTTPResponse__init__
HTTPAdapter = requests.adapters.HTTPAdapter
orig_HTTPAdapter_build_response = HTTPAdapter.build_response
def new_HTTPAdapter_build_response(self, request, resp):
response = orig_HTTPAdapter_build_response(self, request, resp)
try:
response.peer_certificate = resp.peer_certificate
except AttributeError:
pass
return response
HTTPAdapter.build_response = new_HTTPAdapter_build_response
HTTPSConnection = requests.packages.urllib3.connection.HTTPSConnection
orig_HTTPSConnection_connect = HTTPSConnection.connect
def new_HTTPSConnection_connect(self):
orig_HTTPSConnection_connect(self)
try:
self.peer_certificate = self.sock.connection.get_peer_certificate()
except AttributeError:
pass
HTTPSConnection.connect = new_HTTPSConnection_connect
You will be able to access the result easily:
r = requests.get('https://yourdomain.tld', timeout=0.1)
print('Expires on: {}'.format(r.peer_certificate.get_notAfter()))
print(dir(r.peer_certificate))
If, like me, you want to ignore SSL Certificate warnings just add the following in the top of the file and do not SSL verify:
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
r = requests.get('https://yourdomain.tld', timeout=0.1, verify=False)
print(dir(r.peer_certificate))
Thanks for everyone's awesome answers.
It helped me over engineer an answer to this question:
How to add a custom CA Root certificate to the CA Store used by Python in Windows?
Please take a look at Cert Human: SSL Certificates for Humans for an impressive rewrite of my https://github.com/neozenith/get-ca-py project by lifehackjim.
I have archived the original repository now.
#! /usr/bin/env python
# -*- coding: utf-8 -*-
"""
Get Certificates from a request and dump them.
"""
import argparse
import sys
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
"""
Inspired by the answers from this Stackoverflow question:
https://stackoverflow.com/questions/16903528/how-to-get-response-ssl-certificate-from-requests-in-python
What follows is a series of patching the low level libraries in requests.
"""
"""
https://stackoverflow.com/a/47931103/622276
"""
sock_requests = requests.packages.urllib3.contrib.pyopenssl.WrappedSocket
def new_getpeercertchain(self, *args, **kwargs):
x509 = self.connection.get_peer_cert_chain()
return x509
sock_requests.getpeercertchain = new_getpeercertchain
"""
https://stackoverflow.com/a/16904808/622276
"""
HTTPResponse = requests.packages.urllib3.response.HTTPResponse
orig_HTTPResponse__init__ = HTTPResponse.__init__
def new_HTTPResponse__init__(self, *args, **kwargs):
orig_HTTPResponse__init__(self, *args, **kwargs)
try:
self.peercertchain = self._connection.sock.getpeercertchain()
except AttributeError:
pass
HTTPResponse.__init__ = new_HTTPResponse__init__
HTTPAdapter = requests.adapters.HTTPAdapter
orig_HTTPAdapter_build_response = HTTPAdapter.build_response
def new_HTTPAdapter_build_response(self, request, resp):
response = orig_HTTPAdapter_build_response(self, request, resp)
try:
response.peercertchain = resp.peercertchain
except AttributeError:
pass
return response
HTTPAdapter.build_response = new_HTTPAdapter_build_response
"""
Attempt to wrap in a somewhat usable CLI
"""
def cli(args):
parser = argparse.ArgumentParser(description="Request any URL and dump the certificate chain")
parser.add_argument("url", metavar="URL", type=str, nargs=1, help="Valid https URL to be handled by requests")
verify_parser = parser.add_mutually_exclusive_group(required=False)
verify_parser.add_argument("--verify", dest="verify", action="store_true", help="Explicitly set SSL verification")
verify_parser.add_argument(
"--no-verify", dest="verify", action="store_false", help="Explicitly disable SSL verification"
)
parser.set_defaults(verify=True)
return vars(parser.parse_args(args))
def dump_pem(cert, outfile="ca-chain.crt"):
"""Use the CN to dump certificate to PEM format"""
PyOpenSSL = requests.packages.urllib3.contrib.pyopenssl
pem_data = PyOpenSSL.OpenSSL.crypto.dump_certificate(PyOpenSSL.OpenSSL.crypto.FILETYPE_PEM, cert)
issuer = cert.get_issuer().get_components()
print(pem_data.decode("utf-8"))
with open(outfile, "a") as output:
for part in issuer:
output.write(part[0].decode("utf-8"))
output.write("=")
output.write(part[1].decode("utf-8"))
output.write(",\t")
output.write("\n")
output.write(pem_data.decode("utf-8"))
if __name__ == "__main__":
cli_args = cli(sys.argv[1:])
url = cli_args["url"][0]
req = requests.get(url, verify=cli_args["verify"])
for cert in req.peercertchain:
dump_pem(cert)
This, although not pretty at all, works:
import requests
req = requests.get('https://httpbin.org')
pool = req.connection.poolmanager.connection_from_url('https://httpbin.org')
conn = pool.pool.get()
# get() removes it from the pool, so put it back in
pool.pool.put(conn)
print(conn.sock.getpeercert())
To start, abarnert's answer is very complete
But I would like to add, that in the case you're looking for the peer cert chain, you would need to patch yet another piece of code
import requests
sock_requests = requests.packages.urllib3.contrib.pyopenssl.WrappedSocket
def new_getpeercertchain(self,*args, **kwargs):
x509 = self.connection.get_peer_cert_chain()
return x509
sock_requests.getpeercertchain = new_getpeercertchain
after that you can call it in a very similiar manner as the accepted answer
HTTPResponse = requests.packages.urllib3.response.HTTPResponse
orig_HTTPResponse__init__ = HTTPResponse.__init__
def new_HTTPResponse__init__(self, *args, **kwargs):
orig_HTTPResponse__init__(self, *args, **kwargs)
try:
self.peercertchain = self._connection.sock.getpeercertchain()
except AttributeError:
pass
HTTPResponse.__init__ = new_HTTPResponse__init__
HTTPAdapter = requests.adapters.HTTPAdapter
orig_HTTPAdapter_build_response = HTTPAdapter.build_response
def new_HTTPAdapter_build_response(self, request, resp):
response = orig_HTTPAdapter_build_response(self, request, resp)
try:
response.peercertchain = resp.peercertchain
except AttributeError:
pass
return response
HTTPAdapter.build_response = new_HTTPAdapter_build_response
you will get resp.peercertchain which contains a tuple of OpenSSL.crypto.X509 objects
For retrieving the details of a certificate such as CN and expiry date the following script adapted from this example works well. It also avoids some errors I got which I assume were due to incorrect/incompatible versions of requests and urllib3: "AttributeError: 'SSLSocket' object has no attribute 'connection'" and "AttributeError: 'VerifiedHTTPSConnection' object has no attribute 'peer_certificate'"
from OpenSSL.SSL import Connection, Context, SSLv3_METHOD, TLSv1_2_METHOD
from datetime import datetime, time
import socket
host = 'www.google.com'
try:
try:
ssl_connection_setting = Context(SSLv3_METHOD)
except ValueError:
ssl_connection_setting = Context(TLSv1_2_METHOD)
ssl_connection_setting.set_timeout(5)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((host, 443))
c = Connection(ssl_connection_setting, s)
c.set_tlsext_host_name(str.encode(host))
c.set_connect_state()
c.do_handshake()
cert = c.get_peer_certificate()
print("Is Expired: ", cert.has_expired())
print("Issuer: ", cert.get_issuer())
subject_list = cert.get_subject().get_components()
cert_byte_arr_decoded = {}
for item in subject_list:
cert_byte_arr_decoded.update({item[0].decode('utf-8'): item[1].decode('utf-8')})
print(cert_byte_arr_decoded)
if len(cert_byte_arr_decoded) > 0:
print("Subject: ", cert_byte_arr_decoded)
if cert_byte_arr_decoded["CN"]:
print("Common Name: ", cert_byte_arr_decoded["CN"])
end_date = datetime.strptime(str(cert.get_notAfter().decode('utf-8')), "%Y%m%d%H%M%SZ")
print("Not After (UTC Time): ", end_date)
diff = end_date - datetime.now()
print('Summary: "{}" SSL certificate expires on {} i.e. {} days.'.format(host, end_date, diff.days))
c.shutdown()
s.close()
except:
print("Connection to {} failed.".format(host))
This script requires Python 3 and pyOpenSSL.
Cleaner(-ish) solution, based on previous very good answers !
"""
Subclassing HTTP / requests to get peer_certificate back from lower levels
"""
from typing import Optional, Mapping, Any
from http.client import HTTPSConnection
from requests.adapters import HTTPAdapter, DEFAULT_POOLBLOCK
from urllib3.poolmanager import PoolManager,key_fn_by_scheme
from urllib3.connectionpool import HTTPSConnectionPool,HTTPConnectionPool
from urllib3.connection import HTTPSConnection,HTTPConnection
from urllib3.response import HTTPResponse as URLLIB3_HTTPResponse
#force urllib3 to use pyopenssl
import urllib3.contrib.pyopenssl
urllib3.contrib.pyopenssl.inject_into_urllib3()
class HTTPSConnection_withcert(HTTPSConnection):
def __init__(self, *args, **kw):
self.peer_certificate = None
super().__init__(*args, **kw)
def connect(self):
res = super().connect()
self.peer_certificate = self.sock.connection.get_peer_certificate()
return res
class HTTPResponse_withcert(URLLIB3_HTTPResponse):
def __init__(self, *args, **kwargs):
self.peer_certificate = None
res = super().__init__( *args, **kwargs)
self.peer_certificate = self._connection.peer_certificate
return res
class HTTPSConnectionPool_withcert(HTTPSConnectionPool):
ConnectionCls = HTTPSConnection_withcert
ResponseCls = HTTPResponse_withcert
class PoolManager_withcert(PoolManager):
def __init__(
self,
num_pools: int = 10,
headers: Optional[Mapping[str, str]] = None,
**connection_pool_kw: Any,
) -> None:
super().__init__(num_pools,headers,**connection_pool_kw)
self.pool_classes_by_scheme = {"http": HTTPConnectionPool, "https": HTTPSConnectionPool_withcert}
self.key_fn_by_scheme = key_fn_by_scheme.copy()
class HTTPAdapter_withcert(HTTPAdapter):
_clsHTTPResponse = HTTPResponse_withcert
def build_response(self, request, resp):
response = super().build_response( request, resp)
response.peer_certificate = resp.peer_certificate
return response
def init_poolmanager(self, connections, maxsize, block=DEFAULT_POOLBLOCK, **pool_kwargs):
#do not call super() to not initialize PoolManager twice
# save these values for pickling
self._pool_connections = connections
self._pool_maxsize = maxsize
self._pool_block = block
self.poolmanager = PoolManager_withcert(num_pools=connections,
maxsize=maxsize,
block=block,
strict=True,
**pool_kwargs)
class Session_withcert(Session):
def __init__(self):
super().__init__()
self.mount('https://', HTTPAdapter_withcert())
ss= Session_withcert()
resp=ss.get("https://www.google.fr")
resp.peer_certificate.get_subject()
print(resp.peer_certificate.get_subject())
which will output:
<X509Name object '/CN=*.google.fr'>