Created
September 4, 2021 00:26
-
-
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""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