Advertisement
cooperlees

Untitled

Jun 6th, 2019
683
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 1.34 KB | None | 0 0
  1. #!/usr/bin/env python3
  2.  
  3. import asyncio
  4. import logging
  5. import os
  6. import ssl
  7. import sys
  8. import time
  9.  
  10. import aiohttp
  11.  
  12.  
  13. PROXY_PORT = 1443
  14. PROXY_HOSTNAME = "https://proxy.company.com"
  15. EXTERNAL_ENDPOINT = "https://www.google.com"
  16. CUSTOM_CA_BUNDLE = "/var/certs/ca.pem"
  17.  
  18.  
  19. logger = logging.getLogger(__name__)
  20. logger.setLevel(logging.INFO)
  21. logger.addHandler(logging.StreamHandler(sys.stdout))
  22.  
  23.  
  24. async def run_example():
  25.     cert = os.environ["TLS_CL_CERT_PATH"]
  26.     key = os.environ["TLS_CL_KEY_PATH"]
  27.  
  28.     if not thrift_cert or not thrift_key:
  29.         raise ValueError("Missing key TLS cert settings.")
  30.  
  31.     # For the example lets ensure HTTPS_PROXY is set
  32.     os.environ["HTTPS_PROXY"] = f"{PROXY_HOSTNAME}:{PROXY_PORT}"
  33.  
  34.     # Setup SSL Fun
  35.     ssl_ctx = ssl.create_default_context(cafile=CA_BUNDLE)
  36.     ssl_ctx.load_cert_chain(cert, key)
  37.     conn = aiohttp.TCPConnector(ssl=ssl_ctx)
  38.  
  39.     start_time = time.time()
  40.     # trust_env allows HTTP(s)_PROXY vars to work
  41.     async with aiohttp.ClientSession(connector=conn, trust_env=True) as session:
  42.         async with session.get(EXTERNAL_ENDPOINT) as response:
  43.             logger.info(
  44.                 "Received response with status code "
  45.                 + f"{response.status} in {time.time() - start_time}s"
  46.             )
  47.  
  48.  
  49. if __name__ == "__main__":
  50.     asyncio.run(run_example())
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement