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.
Connection: close
and closes the connection immediately after the data is transferred, I still would not get the "peercert". I had to patch requests.packages.urllib3.connection.HTTPSConnection.connect
too to first call the original function and then immediately get the self.sock.getpeercert()
and then pass it up the chain. –
Lumbye 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://mcmap.net/q/48362/-how-to-get-response-ssl-certificate-from-requests-in-python
What follows is a series of patching the low level libraries in requests.
"""
"""
https://mcmap.net/q/48362/-how-to-get-response-ssl-certificate-from-requests-in-python
"""
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://mcmap.net/q/48362/-how-to-get-response-ssl-certificate-from-requests-in-python
"""
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)
import requests
import json
import ssl
with requests.get("https://www.google.com", stream=True) as response:
certificate_info_raw = response.raw.connection.sock.getpeercert(True)
pem_cert = ssl.DER_cert_to_PEM_cert(certificate_info_raw)
print(pem_cert)
certificate_info = response.raw.connection.sock.getpeercert()
print(json.dumps(certificate_info, indent=4))
The Output is:
-----BEGIN CERTIFICATE-----
MIIEhjCCA26gAwIBAgIQApqj0oLcEXwKFFjW6rz/AjANBgkqhkiG9w0BAQsFADBG
MQswCQYDVQQGEwJVUzEiMCAGA1UEChMZR29vZ2xlIFRydXN0IFNlcnZpY2VzIExM
QzETMBEGA1UEAxMKR1RTIENBIDFDMzAeFw0yMzA0MDMwODI1MDdaFw0yMzA2MjYw
ODI1MDZaMBkxFzAVBgNVBAMTDnd3dy5nb29nbGUuY29tMFkwEwYHKoZIzj0CAQYI
KoZIzj0DAQcDQgAEWsob6/KiuiRzjGTzkJLpOPM3ESkXWFkNmqM9WWyWgb+EmnXq
ITDSQWy4NjIVuk9srxOw4OOW2QqpPdgG21e6PqOCAmYwggJiMA4GA1UdDwEB/wQE
AwIHgDATBgNVHSUEDDAKBggrBgEFBQcDATAMBgNVHRMBAf8EAjAAMB0GA1UdDgQW
BBSKd+L1AHQBG1iw8oWwIivemR4I1zAfBgNVHSMEGDAWgBSKdH+vhc3ulc09nNDi
RhTzcTUdJzBqBggrBgEFBQcBAQReMFwwJwYIKwYBBQUHMAGGG2h0dHA6Ly9vY3Nw
LnBraS5nb29nL2d0czFjMzAxBggrBgEFBQcwAoYlaHR0cDovL3BraS5nb29nL3Jl
cG8vY2VydHMvZ3RzMWMzLmRlcjAZBgNVHREEEjAQgg53d3cuZ29vZ2xlLmNvbTAh
BgNVHSAEGjAYMAgGBmeBDAECATAMBgorBgEEAdZ5AgUDMDwGA1UdHwQ1MDMwMaAv
oC2GK2h0dHA6Ly9jcmxzLnBraS5nb29nL2d0czFjMy9RcUZ4Ymk5TTQ4Yy5jcmww
ggEDBgorBgEEAdZ5AgQCBIH0BIHxAO8AdQC3Pvsk35xNunXyOcW6WPRsXfxCz3qf
NcSeHQmBJe20mQAAAYdGbz2aAAAEAwBGMEQCIG7nw2oMgD6XqIecHM8dwFz2h4j9
uhJe9pKnZkdEFvj9AiADhJz3LXHohaTPi1GdLUORvdGquHrNQ6EpGvaxWRDoVQB2
AK33vvp8/xDIi509nB4+GGq0Zyldz7EMJMqFhjTr3IKKAAABh0ZvPbwAAAQDAEcw
RQIgRJcXnT5QCU9tYSqK+r407UgoS7k3E0AFXHmHDJOQGJYCIQCEnHxWxFqh/heK
OqvYhHy1v7cLZ5mywQ9hGKJt1sBJRTANBgkqhkiG9w0BAQsFAAOCAQEAAbKQe92N
mhqCTUgL8ssqz7wa2jHYTotcCq7CwFu3Iy/IeAKomzowFpYHtDqdbJPDh8qTMxnp
f0Z3cLNYRzlIl6rSOMRG7Ij3xv8E0jGO+US6QIpdoNSKEMUwVAXEyD2/gkQYDcFv
q1p2GgikEN6dL7ohXPr5MxB211tPwEoC6uI0zwSfZa0m/ZsvRESbtN88975GnrRz
YPPLSJB+nV7d9BuG2Xt9BPGydEyQeDslgm8QX4kPyGYKkNJMM3I7ZdgIbpe7EJpZ
eHaEpdY8TzSadL2jCokCS0hmwzhuqqsycDIkDyIKQKxTexemB8pf2Sw3UAe7cU9Q
lW5FOc9ifDnEMw==
-----END CERTIFICATE-----
{
"subject": [
[
[
"commonName",
"www.google.com"
]
]
],
"issuer": [
[
[
"countryName",
"US"
]
],
[
[
"organizationName",
"Google Trust Services LLC"
]
],
[
[
"commonName",
"GTS CA 1C3"
]
]
],
"version": 3,
"serialNumber": "029AA3D282DC117C0A1458D6EABCFF02",
"notBefore": "Apr 3 08:25:07 2023 GMT",
"notAfter": "Jun 26 08:25:06 2023 GMT",
"subjectAltName": [
[
"DNS",
"www.google.com"
]
],
"OCSP": [
"http://ocsp.pki.goog/gts1c3"
],
"caIssuers": [
"http:/../Playground/"
],
"crlDistributionPoints": [
"http:/../Playground/"
]
}
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())
Connection: close
. Or if you've sent out other requests (especially if you're using this in an async or threaded context). Or if urllib3
just feels like it for no obvious reason. –
Hydrolysis 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
Just do this:
import requests
with requests.get("https://www.bilibili.com", stream=True) as response:
certificate_info = response.raw.connection.sock.getpeercert()
subject = dict(x[0] for x in certificate_info['subject'])
issuer = dict(x[0] for x in certificate_info['issuer'])
print("commonName:", subject['commonName'])
print("issuer:", issuer['commonName'])
Then the output is :
commonName: *.bilibili.com
issuer: GlobalSign RSA OV SSL CA 2018
Wish to help you.
AttributeError: 'NoneType' object has no attribute 'getpeercert'
. What am I missing? –
Genitals I learned a lot from the above answers.
In order to solve the problem of connection: close, I have combined the answers of Mark Amenry and Yin Lei to come up with the following solution.
May it help.
# from Connection get SSL certificate patch
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.getpeercert(binary_form=True)
except AttributeError as e:
pass
HTTPSConnection.connect = new_HTTPSConnection_connect
And then
with requests.get(url, stream=True) as r:
certificate_info = r.raw.connection.peer_certificate
print(certificate_info)
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'>
© 2022 - 2024 — McMap. All rights reserved.