Add option to authentication using private key

The TransIP API also allows the creation of an access token by using a private key. By allowing the user to specify either an access token or private key when initialising a new transip.TransIP client, the client could dynamically generate a new access token when a private key is used.

Fixes #14

Signed-off-by: Roald Nefs <info@roaldnefs.com>

Author: roaldnefs

setup.py

@@ -29,7 +29,7 @@ def get_long_description() -> str:
     url="https://github.com/roaldnefs/python-transip",
     packages=find_packages(),
     include_package_data=True,
-    install_requires=["requests>=2.25.1"],
+    install_requires=["cryptography>=3.3.1", "requests>=2.25.1"],
     python_requires=">=3.6.12",
     entry_points={},
     classifiers=[

transip/__init__.py

@@ -3,27 +3,29 @@
 # Copyright (C) 2020, 2021 Roald Nefs <info@roaldnefs.com>
 #
 # This file is part of python-transip.
-
+#
 # python-transip is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Lesser General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
-
+#
 # python-transip is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU Lesser General Public License for more details.
-
+#
 # You should have received a copy of the GNU Lesser General Public License
 # along with python-transip.  If not, see <https://www.gnu.org/licenses/>.
 """Wrapper for the TransIP API."""
 
 import importlib
 import requests
+import base64
 
 from typing import Dict, Optional, Any, Type
 
 from transip.exceptions import TransIPHTTPError, TransIPParsingError
+from transip.utils import generate_message_signature, generate_nonce
 
 
 __title__ = "python-transip"
@@ -38,29 +40,41 @@ class TransIP:
     """Represents a TransIP server connection.
 
     Args:
+        login (str): The TransIP username
         api_version (str): TransIP API version to use
         access_token (str): The TransIP API access token
+        private_key (str): The content of the private key for accessing the
+            TransIP API 
+        private_key_file (str): Path to the private key for accessing the
+            TransIP API
     """
 
     def __init__(
         self,
+        login: str = None,
         api_version: str = "6",
-        access_token: Optional[str] = None
+        access_token: Optional[str] = None,
+        private_key: Optional[str] = None,
+        private_key_file: Optional[str] = None,
     ) -> None:
-
         self._api_version: str = api_version
         self._url: str = f"https://api.transip.nl/v{api_version}"
-        self._access_token: Optional[str] = access_token
 
         # Headers to use when making a request to TransIP
         self.headers: Dict[str, str] = {
-            "User-Agent": f"{__title__}/{__version__}",
-            "Authorization": f"Bearer {access_token}"
+            "User-Agent": f"{__title__}/{__version__}"
         }
 
         # Initialize a session object for making requests
         self.session: requests.Session = requests.Session()
 
+        # Set authentication information
+        self._login: str = login
+        self._access_token: Optional[str] = access_token
+        self._private_key: Optional[str] = private_key
+        self._private_key_file: Optional[str] = private_key_file
+        self._set_auth_info()
+
         # Dynamically import the services for the specified API version
         objects = importlib.import_module(f"transip.v{api_version}.objects")
 
@@ -91,14 +105,133 @@ def _get_headers(
     def _build_url(self, path: str) -> str:
         return f"{self._url}{path}"
 
-    def _send(
+    def _request_access_token(self) -> str:
+        """
+        Request an access token using the supplied private key.
+
+        Returns:
+            str: The access token to use for authorization.
+
+        Raises:
+            TransIPParsingError: If the requested access token couldn't be
+                extracted from the API response.
+        """
+
+        url: str = self._build_url('/auth')
+        payload: Dict[str, Any] = {
+            "login": self._login,
+            # The TransIP API requires that the length of the nonce is between
+            # 6 and 32 characters
+            "nonce": generate_nonce(32),
+            # TODO(roaldnefs): Allow the creation of read-only access tokens
+            "read_only": False,
+            # TODO(roaldnefs): Allow the expiration time of the access token
+            # to be overwritten
+            # "expiration_time": "30 minutes",
+            # TODO(roaldnefs): Allow a custom label to be specified when
+            # generating a new access token
+            # "label": "python-transip",
+            # TODO(roaldnefs): Allow the access token to only be use from
+            # whitelisted IP-addresses
+            "global_key": False
+        }
+
+        headers: Dict[str, str] = self.headers.copy()
+        request: requests.Request = requests.Request("POST", url, headers=headers, json=payload)
+        prepped: requests.PreparedRequest = self.session.prepare_request(request)
+
+        # Get the prepped body for signature generation
+        body: str = prepped.body
+
+        # Generate a signature if the request body
+        signature: str = generate_message_signature(body, self._private_key)
+
+        # Add 'Signature' header to the prepared request
+        prepped.headers["Signature"] = signature
+
+        response: requests.Response = self.session.send(prepped)
+        data = self._validate_response(response)
+
+        # Attempt to extract the access token from the result
+        try:
+            return data['token']
+        except (AttributeError, KeyError) as exc:
+            raise TransIPParsingError(
+                "Failed to extract access token from the API response"
+            ) from exc
+
+    def _read_private_key(self) -> str:
+        """Read the private key from file.
+
+        Returns:
+            str: The private key content
+        
+        Raises:
+            RuntimeError: If the private key file doesn't exist
+        """
+        if os.path.exists(self._private_key_file):
+            try:
+                with open(self._private_key_file) as keyfile:
+                    return keyfile.read()
+            except IOError as exc:
+                raise RuntimeError("The private key couldn't be read") from exc
+        else:
+            raise RuntimeError("The private key doesn't exist")
+
+    def _set_auth_info(self) -> None:
+        """
+        Set authentication information based upon the defined attributes.
+        
+        Raises:
+            ValueError: If the required attributes are not defined.
+        """
+        if not self._access_token and not self._private_key and not self._private_key_file:
+            raise ValueError(
+                "At least one of access_token, private_key and "
+                "private_key_file should be defined"
+            )
+        if self._access_token and self._private_key:
+            raise ValueError(
+                "Only one of access_token and private_key should be defined"
+            )
+        if self._access_token and self._private_key_file:
+            raise ValueError(
+                "Only one of access_token and private_key_file should be "
+                "defined"
+            )
+        if self._private_key and self._private_key_file:
+            raise ValueError(
+                "Only one of private_key and private_key_file should be "
+                "defined"
+            )
+        if self._private_key and not self._login:
+            raise ValueError(
+                "Both private_key and login should be defined"
+            )
+        if self._private_key_file and not self._login:
+            raise ValueError(
+                "Both private_key_file and login should be defined"
+            )
+
+        # Read the private key from file
+        if self._private_key_file:
+            self._private_key = self._read_private_key()
+
+        # Use the private key to request a new access token
+        if self._private_key:
+            self._access_token = self._request_access_token()
+
+        # Set the 'Authorization' header
+        self.headers["Authorization"] = f"Bearer {self._access_token}"
+
+    def request(
         self,
         method: str,
         path: str,
         data: Optional[Any] = None,
         json: Optional[Any] = None,
         params: Optional[Dict[str, Any]] = None
-    ) -> requests.Response:
+    ) -> Any:
         """Make an HTTP request to the TransIP API.
 
         Args:
@@ -109,10 +242,11 @@ def _send(
             params (dict): URL parameters to append to the URL
 
         Returns:
-            A requests response object.
+            Returns the json-encoded content of a response, if any.
 
         Raises:
             TransIPHTTPError: When the return code of the request is not 2xx
+            TransIPParsingError: When the content couldn't be parsed as JSON
         """
         url: str = self._build_url(path)
 
@@ -131,9 +265,25 @@ def _send(
             request
         )
         response: requests.Response = self.session.send(prepped)
+        return self._validate_response(response)
 
+    def _validate_response(self, response: requests.Response) -> Any:
+        """
+        Validate the API response.
+        
+        Raises:
+            TransIPHTTPError: When the return code of the request is not 2xx
+            TransIPParsingError: When the content couldn't be parsed as JSON
+        """
         if 200 <= response.status_code < 300:
-            return response
+            if response.text:
+                try:
+                    return response.json()
+                except Exception:
+                    raise TransIPParsingError(
+                        message="Failed to parse the API response as JSON"
+                    )
+            return None
 
         error_message = str(response.content)
         try:
@@ -148,42 +298,6 @@ def _send(
             response_code=response.status_code
         )
 
-    def request(
-        self,
-        method: str,
-        path: str,
-        data: Optional[Any] = None,
-        json: Optional[Any] = None,
-        params: Optional[Dict[str, Any]] = None
-    ) -> Any:
-        """Make an HTTP request to the TransIP API.
-
-        Args:
-            method (str): HTTP method to use
-            path (str): The path to append to the API URL
-            data (dict): The body to attach to the request
-            json (dict): The json body to attach to the request
-            params (dict): URL parameters to append to the URL
-
-        Returns:
-            Returns the json-encoded content of a response, if any.
-
-        Raises:
-            TransIPHTTPError: When the return code of the request is not 2xx
-        """
-        response: requests.Response = self._send(
-            method, path, data=data, json=json, params=params
-        )
-
-        if response.text:
-            try:
-                return response.json()
-            except Exception:
-                raise TransIPParsingError(
-                    message="Failed to parse the API response as JSON"
-                )
-        return None
-
     def get(
         self,
         path: str,

transip/utils.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2021 Roald Nefs <info@roaldnefs.com>
+#
+# This file is part of python-transip.
+#
+# python-transip is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# python-transip is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with python-transip.  If not, see <https://www.gnu.org/licenses/>.
+
+import base64
+import os
+import secrets
+import string
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
+from cryptography.hazmat.primitives.hashes import SHA512
+from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15
+
+
+def load_rsa_private_key(key: str) -> RSAPrivateKey:
+    """
+    Convert the private key string to RSAPrivateKey object.
+    
+    Returns:
+        RSAPrivateKey: The private RSA key.
+    """
+    # Convert the key string to bytes
+    if isinstance(key, str):
+        key: bytes = key.encode()
+    
+    return serialization.load_pem_private_key(
+        key, password=None, backend=default_backend()
+    )
+
+
+def generate_message_signature(message: str, private_key: str) -> str:
+    """Return the BASE64 encoded SHA514 signature of a message.
+    
+    Args:
+        message (str): The message to sign.
+        private_key (str): The private key content used to sign the message.
+    
+    Returns:
+        str: The BASE64 encoded SHA514 signature of a message.
+    """
+    # Convert the message string to bytes
+    if isinstance(message, str):
+        message: bytes = message.encode()
+
+    # Convert the private key content to a RSAPrivateKey object
+    if isinstance(private_key, str):
+        private_key: RSAPrivateKey = load_rsa_private_key(private_key)
+
+    # Sign the message using the RSAPrivateKey object
+    signature: str = private_key.sign(message, PKCS1v15(), SHA512())
+
+    # Return the BASE64 encoded SHA512 signature
+    return base64.b64encode(signature)
+
+
+def generate_nonce(length: int) -> str:
+    """
+    Generate a nonce.
+
+    Args:
+        length (int): The number of characters to return.
+
+    Returns:
+        str: The nonce of specified characters.
+    """
+    alphabet = string.ascii_letters + string.digits
+    return ''.join(secrets.choice(alphabet) for i in range(length))