Add client certificate support

tests/test_client.py

@@ -44,3 +44,32 @@ def test_async_uploader(self):
 
         self.assertIsInstance(async_uploader, AsyncUploader)
         self.assertEqual(async_uploader.client, self.client)
+
+class TusClientTestWithClientCertificate(unittest.TestCase):
+    def setUp(self):
+        self.client_cert=("/tmp/client.crt.pem")
+        self.client = client.TusClient('http://tusd.tusdemo.net/files/',
+                                       headers={'foo': 'bar'},
+                                       client_cert=self.client_cert)
+
+    @responses.activate
+    def test_uploader(self):
+        url = 'http://tusd.tusdemo.net/files/15acd89eabdf5738ffc'
+        responses.add(responses.HEAD, url,
+                      adding_headers={"upload-offset": "0"})
+        uploader = self.client.uploader('./LICENSE', url=url)
+
+        self.assertIsInstance(uploader, Uploader)
+        self.assertEqual(uploader.client, self.client)
+        self.assertEqual(uploader.client_cert, self.client_cert)
+
+    @responses.activate
+    def test_async_uploader(self):
+        url = 'http://tusd.tusdemo.net/files/15acd89eabdf5738ffc'
+        responses.add(responses.HEAD, url,
+                      adding_headers={"upload-offset": "0"})
+        async_uploader = self.client.async_uploader('./LICENSE', url=url)
+
+        self.assertIsInstance(async_uploader, AsyncUploader)
+        self.assertEqual(async_uploader.client, self.client)
+        self.assertEqual(async_uploader.client_cert, self.client_cert)

tusclient/client.py

@@ -1,4 +1,4 @@
-from typing import Dict, Optional
+from typing import Dict, Optional, Tuple, Union
 
 from tusclient.uploader import Uploader, AsyncUploader
 
@@ -16,15 +16,21 @@ class TusClient:
             along with every request made by the cleint to the server. This may be used to set
             authentication headers. These headers should not include headers required by tus
             protocol. If not set this defaults to an empty dictionary.
-
+        - client_cert (str|tuple[str,str]):
+            Path of PEM encoded client certitifacate and optionally path to PEM encoded
+            key file. The PEM encoded key of the certificate can either be included in the
+            certificate itself or be provided in a seperate file.
+            Only unencrypted keys are supported!
     :Constructor Args:
         - url (str)
         - headers (Optiional[dict])
+        - client_cert (Optional[str | Tuple[str, str]])
     """
 
-    def __init__(self, url: str, headers: Optional[Dict[str, str]] = None):
+    def __init__(self, url: str, headers: Optional[Dict[str, str]] = None, client_cert: Optional[Union[str, Tuple[str, str]]] = None):
         self.url = url
         self.headers = headers or {}
+        self.client_cert = client_cert
 
     def set_headers(self, headers: Dict[str, str]):
         """
@@ -49,9 +55,11 @@ def uploader(self, *args, **kwargs) -> Uploader:
         :Args:
             see tusclient.uploader.Uploader for required and optional arguments.
         """
-        kwargs["client"] = self
+        kwargs['client'] = self
+        kwargs['client_cert'] = self.client_cert
         return Uploader(*args, **kwargs)
 
     def async_uploader(self, *args, **kwargs) -> AsyncUploader:
-        kwargs["client"] = self
+        kwargs['client'] = self
+        kwargs['client_cert'] = self.client_cert
         return AsyncUploader(*args, **kwargs)

tusclient/request.py

@@ -46,6 +46,7 @@ def __init__(self, uploader):
         self.verify_tls_cert = bool(uploader.verify_tls_cert)
         self.file = uploader.get_file_stream()
         self.file.seek(uploader.offset)
+        self.client_cert = uploader.client_cert
 
         self._request_headers = {
             "upload-offset": str(uploader.offset),
@@ -79,12 +80,10 @@ def perform(self):
         try:
             chunk = self.file.read(self._content_length)
             self.add_checksum(chunk)
-            resp = requests.patch(
-                self._url,
-                data=chunk,
-                headers=self._request_headers,
-                verify=self.verify_tls_cert,
-            )
+            resp = requests.patch(self._url, data=chunk,
+                                  headers=self._request_headers,
+                                  verify=self.verify_tls_cert,
+                                  cert=self.client_cert,)
             self.status_code = resp.status_code
             self.response_content = resp.content
             self.response_headers = {k.lower(): v for k, v in resp.headers.items()}

tusclient/uploader/baseuploader.py

@@ -1,4 +1,4 @@
-from typing import Optional, IO, Dict, TYPE_CHECKING
+from typing import Optional, IO, Dict, Tuple, TYPE_CHECKING, Union
 import os
 import re
 from base64 import b64encode
@@ -114,6 +114,7 @@ def __init__(
         url_storage: Optional[Storage] = None,
         fingerprinter: Optional[interface.Fingerprint] = None,
         upload_checksum=False,
+        client_cert: Optional[Tuple[str, str]] = None,
     ):
         if file_path is None and file_stream is None:
             raise ValueError("Either 'file_path' or 'file_stream' cannot be None.")
@@ -131,6 +132,7 @@ def __init__(
         self.file_stream = file_stream
         self.stop_at = self.get_file_size()
         self.client = client
+        self.client_cert = client_cert
         self.metadata = metadata or {}
         self.metadata_encoding = metadata_encoding
         self.store_url = store_url
@@ -186,7 +188,7 @@ def get_offset(self):
         http request to the tus server to retrieve the offset.
         """
         resp = requests.head(
-            self.url, headers=self.get_headers(), verify=self.verify_tls_cert
+            self.url, headers=self.get_headers(), verify=self.verify_tls_cert, cert=self.client_cert
         )
         offset = resp.headers.get("upload-offset")
         if offset is None:

tusclient/uploader/uploader.py

@@ -70,6 +70,7 @@ def create_url(self):
             self.client.url,
             headers=self.get_url_creation_headers(),
             verify=self.verify_tls_cert,
+            cert=self.client_cert,
         )
         url = resp.headers.get("location")
         if url is None: