Add TLS client authentication

jellyfin_apiclient_python/api.py

@@ -758,7 +758,7 @@ def send_request(self, url, path, method="get", timeout=None, headers=None, data
 
         return request_method(url, **request_settings)
 
-    def login(self, server_url, username, password=""):
+    def login(self, server_url, username, password="", session=None):
         path = "Users/AuthenticateByName"
         authData = {
                     "username": username,
@@ -771,7 +771,7 @@ def login(self, server_url, username, password=""):
         try:
             LOG.info("Trying to login to %s/%s as %s" % (server_url, path, username))
             response = self.send_request(server_url, path, method="post", headers=headers,
-                                         data=json.dumps(authData), timeout=(5, 30))
+                                         data=json.dumps(authData), timeout=(5, 30), session=session)
 
             if response.status_code == 200:
                 return response.json()
@@ -786,23 +786,23 @@ def login(self, server_url, username, password=""):
 
         return {}
 
-    def validate_authentication_token(self, server):
+    def validate_authentication_token(self, server, session=None):
         headers = self.get_default_headers()
         comma = "," if "app.device_name" in self.config.data else ""
         headers["Authorization"] += f"{comma} Token=\"{server['AccessToken']}\""
 
-        response = self.send_request(server['address'], "system/info", headers=headers)
+        response = self.send_request(server['address'], "system/info", headers=headers, session=session)
         return response.json() if response.status_code == 200 else {}
 
-    def get_public_info(self, server_address):
-        response = self.send_request(server_address, "system/info/public")
+    def get_public_info(self, server_address, session=None):
+        response = self.send_request(server_address, "system/info/public", session=session)
         return response.json() if response.status_code == 200 else {}
 
-    def check_redirect(self, server_address):
+    def check_redirect(self, server_address, session=None):
         ''' Checks if the server is redirecting traffic to a new URL and
         returns the URL the server prefers to use
         '''
-        response = self.send_request(server_address, "system/info/public")
+        response = self.send_request(server_address, "system/info/public", session=session)
         url = response.url.replace('/system/info/public', '')
         return url
 

jellyfin_apiclient_python/connection_manager.py

@@ -8,6 +8,7 @@
 from datetime import datetime
 from operator import itemgetter
 
+import requests
 import urllib3
 
 from .credentials import Credentials
@@ -40,6 +41,16 @@ def __init__(self, client):
         self.config = client.config
         self.credentials = Credentials()
         self.API = API(HTTP(client))
+
+        self.session = None
+
+    def create_session_with_client_auth(self):
+        if self.config.data['auth.tls_client_cert'] and self.config.data['auth.tls_client_key']:
+            self.session = requests.Session()
+            self.session.cert = (self.config.data['auth.tls_client_cert'], self.config.data['auth.tls_client_key'])
+
+            if self.config.data['auth.tls_server_ca']:
+                self.session.verify = self.config.data['auth.tls_server_ca']
 
     def clear_data(self):
 
@@ -108,7 +119,7 @@ def login(self, server_url, username, password=None, clear=None, options=None):
         if options is not None:
             LOG.warn("The options option on login() has no effect.")
 
-        data = self.API.login(server_url, username, password) # returns empty dict on failure
+        data = self.API.login(server_url, username, password, self.session) # returns empty dict on failure
 
         if not data:
             LOG.info("Failed to login as `"+username+"`")
@@ -153,7 +164,7 @@ def connect_to_address(self, address, options={}):
         address = self._normalize_address(address)
 
         try:
-            response_url = self.API.check_redirect(address)
+            response_url = self.API.check_redirect(address, self.session)
             if address != response_url:
                 address = response_url
             LOG.info("connect_to_address %s succeeded", address)
@@ -176,7 +187,7 @@ def connect_to_server(self, server, options={}):
         LOG.info("begin connect_to_server")
 
         try:
-            result = self.API.get_public_info(server.get('address'))
+            result = self.API.get_public_info(server.get('address'), self.session)
 
             if not result:
                 LOG.error("Failed to connect to server: %s" % server.get('address'))
@@ -335,7 +346,7 @@ def _after_connect_validated(self, server, credentials, system_info, verify_auth
             self.config.data['auth.token'] = server.pop('AccessToken', None)
 
         elif verify_authentication and server.get('AccessToken'):
-            system_info = self.API.validate_authentication_token(server)
+            system_info = self.API.validate_authentication_token(server, self.session)
             if system_info:
 
                 self._update_server_info(server, system_info)

jellyfin_apiclient_python/http.py

@@ -29,10 +29,18 @@ def __init__(self, client):
         self.config = client.config
 
     def start_session(self):
-
         self.session = requests.Session()
 
         max_retries = self.config.data['http.max_retries']
+
+        # Configure the session for tls client authentication
+        if self.client.config.data['auth.tls_client_cert'] and self.client.config.data['auth.tls_client_key']:
+            self.session.cert = (self.client.config.data['auth.tls_client_cert'], 
+                                 self.client.config.data['auth.tls_client_key'])
+
+            if self.client.config.data['auth.tls_server_ca']:
+                self.session.verify = self.client.config.data['auth.tls_server_ca']
+
         self.session.mount("http://", requests.adapters.HTTPAdapter(max_retries=max_retries))
         self.session.mount("https://", requests.adapters.HTTPAdapter(max_retries=max_retries))
 

jellyfin_apiclient_python/ws_client.py

@@ -56,6 +56,16 @@ def run(self):
 
         LOG.info("Websocket url: %s", wsc_url)
 
+        # Configure SSL context for client authentication
+        ssl_context = None
+        if self.client.config.data['auth.tls_client_cert'] and self.client.config.data['auth.tls_client_key']:
+            ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+            ssl_context.load_default_certs()
+            ssl_context.load_cert_chain(self.client.config.data['auth.tls_client_cert'], self.client.config.data['auth.tls_client_key'])
+
+            if self.client.config.data['auth.tls_server_ca']:
+                ssl_context.load_verify_locations(self.client.config.data['auth.tls_server_ca'])
+
         self.wsc = websocket.WebSocketApp(wsc_url,
                                           on_message=lambda ws, message: self.on_message(ws, message),
                                           on_error=lambda ws, error: self.on_error(ws, error))
@@ -74,7 +84,7 @@ def run(self):
                         ping_interval=10, sslopt={"cert_reqs": ssl.CERT_NONE}
                     )
                 else:
-                    self.wsc.run_forever(ping_interval=10, sslopt={"ca_certs": certifi.where()})
+                    self.wsc.run_forever(ping_interval=10, sslopt={"context": ssl_context, "ca_certs": certifi.where()})
             else:
                 self.wsc.run_forever(ping_interval=10)