Skip to content

Instantly share code, notes, and snippets.

@mzpqnxow
Created September 4, 2021 00:26
Show Gist options
  • Save mzpqnxow/002b0b639a1072ce57af2a721f4542cc to your computer and use it in GitHub Desktop.
Save mzpqnxow/002b0b639a1072ce57af2a721f4542cc to your computer and use it in GitHub Desktop.
Provide a PEM blob as the verify / CA value to a requests Session instead of a file
"""An extension of requests.Session that provides an interface for specifying a PEM blob
This was created as a workaround for https://github.com/psf/requests/issues/4032
This is really only useful if you have a sprawling application and/or many sessions
and you don't want to handle the NamedTemporaryFile solution outside of the Session
yourself
It's surprising at first that requests.Session doesn't provide the ability to specify
a StringIO (or other file stream like object) but when you see the OpenSSL interface
that is available, it makes sense
An alternative approach, managing the NamedTemporaryFile in the application, is below
the class
This is not intended to work on Python2, but it might
- AG
LICENSE
-------
Do whatever you want with this, EXCEPT use it
Use of this snippet in any way is expressly forbidden under the laws of Thermodynamics
"""
import contextlib
import tempfile
import requests
class VerifyBlobSession(requests.Session):
"""Pythonic (if not clunky) approach to working around https://github.com/psf/requests/issues/4032
The `verify_blob` kwarg can be a PEM blob or list of PEM blobs to act as the trusted certificate bundle
The session takes care of the temporary file backing, including the cleanup, via use of tempfile.NamedTemporaryFile
"""
def __init__(self, verify_blob=None):
super(VerifyBlobSession, self).__init__()
if verify_blob is not None:
self._temp_cafile = tempfile.NamedTemporaryFile(mode='w')
fd = self._temp_cafile.file
if isinstance(verify_blob, str):
# Assume iterable of str or bytes, write each to temp file
fd.write(verify_blob)
else:
fd.writelines([blob + '\n' for blob in verify_blob])
fd.close()
self.verify = self._temp_cafile.name
def __del__(self):
if hasattr(self, '_temp_cafile'):
# Explicitly clean up the temporary file
self._temp_cafile.close()
ca_pem_blob = """-----BEGIN CERTIFICATE-----
...
YOUR PEM HERE
...
-----END CERTIFICATE----"""
#
# Alternately ...
# ca_pem_blob = [
# """
# -----BEGING CERTIFICATE-----
# ...
# -----END CERTIFICATE-----""",
# """
# -----BEGIN CERTIFICATE-----
# ...
# -----END CERTIFICATE-----""",
# ]
#
#
# First Solution: Using VerifyBlobSession class
#
session = VerifyBlobSession(verify_blob=ca_pem_blob)
session.get('https://some.site.private')
#
# Alternate Solution: Using a helper contextmanager to reduce cruft and manage the temporary file in your app
#
@contextlib.contextmanager
def write_temp(buf, encoding='utf-8'):
"""Create a temporary file with specified contents, ensuring it gets cleaned up properly"""
with tempfile.NamedTemporaryFile() as fp:
filename, fd = fp.name, fp.file
if isinstance(buf, str):
buf = buf.encode(encoding)
fd.write(buf)
fd.flush()
yield filename
session = requests.Session()
print('Managing creation of temporary CA file ourselves...')
# write_temp() eliminates a bit of clunk and clutter, useful if you're a neat-freak :>
with write_temp(ca_pem) as cafile:
print(f'Using temporary CA file @ {cafile}')
session.get('https://some.site.private', verify=cafile)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment