Merge pull request #3287 from InfinityPacer/feature/security

app/modules/emby/__init__.py

@@ -66,16 +66,26 @@ class EmbyModule(_ModuleBase, _MediaServerBase[Emby]):
                 logger.info(f"Emby服务器 {name} 连接断开，尝试重连 ...")
                 server.reconnect()
 
-    def user_authenticate(self, credentials: AuthCredentials) -> Optional[AuthCredentials]:
+    def user_authenticate(self, credentials: AuthCredentials, service_name: Optional[str] = None) \
+            -> Optional[AuthCredentials]:
         """
         使用Emby用户辅助完成用户认证
         :param credentials: 认证数据
+        :param service_name: 指定要认证的媒体服务器名称，若为 None 则认证所有服务
         :return: 认证数据
         """
         # Emby认证
         if not credentials or credentials.grant_type != "password":
             return None
-        for name, server in self.get_instances().items():
+        # 确定要认证的服务器列表
+        if service_name:
+            # 如果指定了服务名，获取该服务实例
+            servers = [(service_name, server)] if (server := self.get_instance(service_name)) else []
+        else:
+            # 如果没有指定服务名，遍历所有服务
+            servers = self.get_instances().items()
+        # 遍历要认证的服务器
+        for name, server in servers:
             # 触发认证拦截事件
             intercept_event = eventmanager.send_event(
                 etype=ChainEventType.AuthIntercept,

app/modules/jellyfin/__init__.py

@@ -66,16 +66,26 @@ class JellyfinModule(_ModuleBase, _MediaServerBase[Jellyfin]):
                 return False, f"无法连接Jellyfin服务器：{name}"
         return True, ""
 
-    def user_authenticate(self, credentials: AuthCredentials) -> Optional[AuthCredentials]:
+    def user_authenticate(self, credentials: AuthCredentials, service_name: Optional[str] = None) \
+            -> Optional[AuthCredentials]:
         """
         使用Jellyfin用户辅助完成用户认证
         :param credentials: 认证数据
+        :param service_name: 指定要认证的媒体服务器名称，若为 None 则认证所有服务
         :return: 认证数据
         """
         # Jellyfin认证
         if not credentials or credentials.grant_type != "password":
             return None
-        for name, server in self.get_instances().items():
+        # 确定要认证的服务器列表
+        if service_name:
+            # 如果指定了服务名，获取该服务实例
+            servers = [(service_name, server)] if (server := self.get_instance(service_name)) else []
+        else:
+            # 如果没有指定服务名，遍历所有服务
+            servers = self.get_instances().items()
+        # 遍历要认证的服务器
+        for name, server in servers:
             # 触发认证拦截事件
             intercept_event = eventmanager.send_event(
                 etype=ChainEventType.AuthIntercept,

app/modules/plex/__init__.py

@@ -66,16 +66,26 @@ class PlexModule(_ModuleBase, _MediaServerBase[Plex]):
                 logger.info(f"Plex {name} 服务器连接断开，尝试重连 ...")
                 server.reconnect()
 
-    def user_authenticate(self, credentials: AuthCredentials) -> Optional[AuthCredentials]:
+    def user_authenticate(self, credentials: AuthCredentials, service_name: Optional[str] = None) \
+            -> Optional[AuthCredentials]:
         """
         使用Plex用户辅助完成用户认证
         :param credentials: 认证数据
+        :param service_name: 指定要认证的媒体服务器名称，若为 None 则认证所有服务
         :return: 认证数据
         """
         # Plex认证
         if not credentials or credentials.grant_type != "password":
             return None
-        for name, server in self.get_instances().items():
+        # 确定要认证的服务器列表
+        if service_name:
+            # 如果指定了服务名，获取该服务实例
+            servers = [(service_name, server)] if (server := self.get_instance(service_name)) else []
+        else:
+            # 如果没有指定服务名，遍历所有服务
+            servers = self.get_instances().items()
+        # 遍历要认证的服务器
+        for name, server in servers:
             # 触发认证拦截事件
             intercept_event = eventmanager.send_event(
                 etype=ChainEventType.AuthIntercept,