refactor: implements tls support for certificate based auth

adds initial support using ssl_context for certificate based auth
also adds testsuite to use the ssl context from env varables, see
docs for relevant how to use

pending: add test suite to run without certificates

refs #66

Author: devraj

gallagher/cc/__init__.py

@@ -29,6 +29,12 @@
 # to obtain an API key
 api_key: str = None
 
+# Certificate file to be used for authentication
+file_tls_certificate: str = None
+
+# Private key file to be used for authentication
+file_private_key: str = None
+
 # By default the base API is set to the Australian Gateway
 # Override this with the US gateway or a local DNS/IP address
 api_base: str = URL.CLOUD_GATEWAY_AU

gallagher/cc/core.py

@@ -26,6 +26,7 @@
 
 from http import HTTPStatus  # Provides constants for HTTP status codes
 
+import ssl
 import httpx
 
 from . import proxy as proxy_address
@@ -232,6 +233,23 @@ async def get_config(cls) -> EndpointConfig:
         provide additional configuration options.
         """
         raise NotImplementedError("get_config method not implemented")
+    
+    @classmethod
+    def _ssl_context(cls):
+        """Returns the SSL context for the endpoint
+
+        This method can be overridden by the child class to
+        provide additional SSL context options.
+        """
+        from . import file_tls_certificate, file_private_key
+
+        if not file_tls_certificate:
+            """TLS certificate is required for SSL context"""
+            return None
+        
+        context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
+        context.load_cert_chain(file_tls_certificate, file_private_key)
+        return context
 
     @classmethod
     async def _discover(cls):
@@ -269,7 +287,10 @@ async def _discover(cls):
         # be called as part of the bootstrapping process
         from . import api_base
 
-        async with httpx.AsyncClient(proxy=proxy_address) as _httpx_async:
+        async with httpx.AsyncClient(
+                proxy=proxy_address,
+                verify=cls._ssl_context(),
+            ) as _httpx_async:
             # Don't use the _get wrapper here, we need to get the raw response
             response = await _httpx_async.get(
                 api_base,
@@ -490,7 +511,11 @@ async def follow(
         # Initial url is set to endpoint_follow
         url = f"{cls.__config__.endpoint_follow.href}"
 
-        async with httpx.AsyncClient(proxy=proxy_address) as _httpx_async:
+        async with httpx.AsyncClient(
+                proxy=proxy_address,
+                verify=cls._ssl_context(),
+            ) as _httpx_async:
+
             while event.is_set():
                 try:
                     response = await _httpx_async.get(
@@ -546,7 +571,10 @@ async def _get(
         :param str url: URL to fetch the data from
         :param AppBaseModel response_class: DTO to be used for list requests
         """
-        async with httpx.AsyncClient(proxy=proxy_address) as _httpx_async:
+        async with httpx.AsyncClient(
+                proxy=proxy_address,
+                verify=cls._ssl_context(),
+            ) as _httpx_async:
 
             try:
 
@@ -594,7 +622,10 @@ async def _post(
         The behaviour is very similar to the _get method, except
         parsing and sending out a body as part of the request. 
         """
-        async with httpx.AsyncClient(proxy=proxy_address) as _httpx_async:
+        async with httpx.AsyncClient(
+                proxy=proxy_address,
+                verify=cls._ssl_context(),
+            ) as _httpx_async:
 
             try:
 

tests/__init__.py

@@ -12,6 +12,7 @@
   TODO: check if setup and teardown can be turned into async
 
 """
+import tempfile
 
 
 def setup_module(module):
@@ -25,9 +26,33 @@ def setup_module(module):
 
     api_key = os.environ.get("GACC_API_KEY")
 
+    # Read these from the environment variables, if they exists
+    # they will be written to temporary files
+    certificate_anomaly = os.environ.get("CERTIFICATE_ANOMALY")
+    private_key_anomaly = os.environ.get("PRIVATE_KEY_ANOMALY")
+
+    # Create temporary files to store the certificate and private key
+    temp_file_certificate = tempfile.NamedTemporaryFile(
+      suffix=".crt",
+      delete=False
+    )
+    temp_file_private_key = tempfile.NamedTemporaryFile(
+      suffix=".key",
+      delete=False
+    )
+
+    # Write the certificate and private key to temporary files
+    if certificate_anomaly and temp_file_certificate:
+      temp_file_certificate.write(certificate_anomaly.encode('utf-8'))
+
+    if private_key_anomaly and temp_file_private_key:
+      temp_file_private_key.write(private_key_anomaly.encode('utf-8'))
+
     from gallagher import cc
 
     cc.api_key = api_key
+    cc.file_tls_certificate = temp_file_certificate.name
+    cc.file_private_key = temp_file_private_key.name
 
 
 def teardown_module(module):