Client certificate support and example

Author: eyazrgeotab

examples/basic/sample.py

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+
+import os
+import getpass
+from mygeotab import api
+
+
+def main():
+    # Get credentials
+    username = input("Username: ")              # [email protected]
+    password = getpass.getpass("Password: ")    # password
+    database = input("Database: ")              # demo
+    server = input("Server: ")                  # my.geotab.com
+    port = input("Port: ")
+
+    # Get path to private key/certificate from environment variables
+    key_path = os.environ["MYGEOTAB_CERTIFICATE_KEY"]
+    cert_path = os.environ["MYGEOTAB_CERTIFICATE_CER"]
+
+    if not key_path or not cert_path:
+        print("MyGeotab key or certificate not provided")
+        exit(1)
+
+    cert = (cert_path, key_path)
+    session = api.API(username=username, password=password, database=database, server=server, cert=cert)
+
+    try:
+        session.authenticate()
+    except api.MyGeotabException as exception:
+        print(f"AUTH ERROR: {exception}")
+        exit(1)
+
+    # Create new session using authenticated session to the target server and port
+    if ':' not in session.credentials.server:
+        credentials = session.credentials
+        credentials.server = f"{credentials.server}:{port}"
+        session = api.API.from_credentials(credentials)
+
+    # Send requests
+    # ...
+
+
+if __name__ == "__main__":
+    main()

mygeotab/api.py

@@ -39,6 +39,7 @@ def __init__(
         server="my.geotab.com",
         timeout=DEFAULT_TIMEOUT,
         proxies=None,
+        cert=None
     ):
         """Initialize the MyGeotab API object with credentials.
 
@@ -56,6 +57,8 @@ def __init__(
         :type timeout: float or None
         :param proxies: The proxies dictionary to apply to the request.
         :type proxies: dict or None
+        :param cert: The path to client certificate. A single path to .pem file or a Tuple (.cer file, .key file).
+        :type cert: str or Tuple or None
         :raise Exception: Raises an Exception if a username, or one of the session_id or password is not provided.
         """
         if username is None:
@@ -68,6 +71,7 @@ def __init__(
         self.timeout = timeout
         self._proxies = proxies
         self.__reauthorize_count = 0
+        self._cert = cert
 
     @property
     def _server(self):
@@ -105,7 +109,12 @@ def call(self, method, **parameters):
 
         try:
             result = _query(
-                self._server, method, params, self.timeout, verify_ssl=self._is_verify_ssl, proxies=self._proxies
+                self._server,
+                method, params,
+                self.timeout,
+                verify_ssl=self._is_verify_ssl,
+                proxies=self._proxies,
+                cert=self._cert
             )
             if result is not None:
                 self.__reauthorize_count = 0
@@ -219,6 +228,7 @@ def authenticate(self):
                 self.timeout,
                 verify_ssl=self._is_verify_ssl,
                 proxies=self._proxies,
+                cert=self._cert
             )
             if result:
                 if "path" not in result and self.credentials.session_id:
@@ -307,7 +317,7 @@ def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
         )
 
 
-def _query(server, method, parameters, timeout=DEFAULT_TIMEOUT, verify_ssl=True, proxies=None):
+def _query(server, method, parameters, timeout=DEFAULT_TIMEOUT, verify_ssl=True, proxies=None, cert=None):
     """Formats and performs the query against the API.
 
     :param server: The MyGeotab server.
@@ -332,6 +342,8 @@ def _query(server, method, parameters, timeout=DEFAULT_TIMEOUT, verify_ssl=True,
     headers = get_headers()
     with requests.Session() as session:
         session.mount("https://", GeotabHTTPAdapter())
+        if cert:
+            session.cert = cert
         try:
             response = session.post(
                 api_endpoint,

tests/test_api_call.py

@@ -11,6 +11,9 @@
 PASSWORD = os.environ.get("MYGEOTAB_PASSWORD")
 DATABASE = os.environ.get("MYGEOTAB_DATABASE")
 SERVER = os.environ.get("MYGEOTAB_SERVER")
+CER_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_CER")
+KEY_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_KEY")
+PEM_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_PEM")
 TRAILER_NAME = "mygeotab-python test trailer"
 
 FAKE_USERNAME = "fakeusername"
@@ -21,8 +24,13 @@
 
 @pytest.fixture(scope="session")
 def populated_api():
+    cert = None
+    if CER_FILE and KEY_FILE:
+        cert = (CER_FILE, KEY_FILE)
+    elif PEM_FILE:
+        cert = PEM_FILE
     if USERNAME and PASSWORD:
-        session = api.API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER)
+        session = api.API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER, cert=cert)
         try:
             session.authenticate()
         except api.MyGeotabException as exception: