fix(bkpaas-auth): 修复 apigw_manager 的 JWT 中间件新增 tenant_id 参数导致的 JWT 认证失效问题 (TencentBlueKing#213)

Author: jiayuan929

sdks/bkpaas-auth/CHANGES.md

@@ -1,5 +1,8 @@
 # 版本历史
 
+## 3.1.1
+- fix: 修复 apigw-manager 4.0.1 版本 JWT 中间件新增 tenant_id 参数导致的 JWT 认证失效问题
+
 ## 3.1.0
 - feat: UniversalAuthBackend 支持获取用户租户身份信息
 

sdks/bkpaas-auth/bkpaas_auth/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-__version__ = "3.1.0"
+__version__ = "3.1.1"
 
 
 def get_user_by_user_id(user_id: str, username_only: bool = True):
@@ -18,7 +18,7 @@ def get_user_by_user_id(user_id: str, username_only: bool = True):
 
     # 多租户模式下, 暂时没有根据用户名获取用户详细信息的接口
     if conf.ENABLE_MULTI_TENANT_MODE:
-        raise ValueError('Multi-tenant mode only return username, please set username_only=True')
+        raise ValueError("Multi-tenant mode only return username, please set username_only=True")
 
     # Request third party service to get info other than username
     if provider_type == ProviderType.RTX:
@@ -30,4 +30,4 @@ def get_user_by_user_id(user_id: str, username_only: bool = True):
         user_info.provide(user)
         return user
     else:
-        raise ValueError('ProviderType is not supported yet!')
+        raise ValueError("ProviderType is not supported yet!")

sdks/bkpaas-auth/bkpaas_auth/backends.py

@@ -200,46 +200,55 @@ def configure_user(self, db_user, bk_user: User):
 
 
 class APIGatewayAuthBackend:
-    """This backend is to be used in conjunction with the ``ApiGatewayJWTUserMiddleware``
-    found in the middleware module of ``apigw_manager`` package.
+    """Authentication backend for API Gateway JWT validation.
 
+    This backend works with `ApiGatewayJWTUserMiddleware` from the
+    `apigw_manager` package to handle JWT-based authentication.
     """
 
-    def authenticate_with_signature_v3(self, request, gateway_name, bk_username, verified, **credentials):
-        """authenticate function with signature required by ApiGatewayJWTUserMiddleware in apigw_manager == '^3.0.0'"""
-        if not verified:
-            return self.make_anonymous_user(bk_username)
+    _TOKEN_EXPIRE_TIME = 86400  # 24 hours in seconds
 
+    def _create_authenticated_user(self, username: str, provider_type: ProviderType) -> User:
+        """Create a user object for authenticated requests."""
         return User(
-            token=LoginToken("any_token", expires_in=86400),
-            provider_type=self.get_provider_type(),
-            username=bk_username,
+            token=LoginToken("any_token", expires_in=self._TOKEN_EXPIRE_TIME),
+            provider_type=provider_type,
+            username=username,
         )
 
-    def authenticate_with_signature_v1(self, request, api_name, bk_username, verified, **credentials):
-        """authenticate function with signature required by ApiGatewayJWTUserMiddleware in apigw_manager == '^1.0.0'"""
-        if not verified:
-            return self.make_anonymous_user(bk_username)
+    def _authenticate_common(self, verified: bool, username: Optional[str]) -> Union[User, AnonymousUser]:
+        """Common authentication logic for all versions."""
+        if not verified or not username:
+            return self.make_anonymous_user(username)
 
-        return User(
-            token=LoginToken("any_token", expires_in=86400),
-            provider_type=self.get_provider_type(),
-            username=bk_username,
-        )
+        return self._create_authenticated_user(username=username, provider_type=self.get_provider_type())
+
+    def authenticate_with_signature_v3(
+        self, request: HttpRequest, gateway_name: str, bk_username: str, verified: bool, **credentials: Dict
+    ) -> Union[User, AnonymousUser]:
+        """authenticate function with signature required by ApiGatewayJWTUserMiddleware in apigw_manager == '^3.0.0'"""
+        return self._authenticate_common(verified, bk_username)
+
+    def authenticate_with_signature_v1(
+        self, request: HttpRequest, api_name: str, bk_username: str, verified: bool, **credentials: Dict
+    ) -> Union[User, AnonymousUser]:
+        """authenticate function with signature required by ApiGatewayJWTUserMiddleware in apigw_manager == '^1.0.0'"""
+        return self._authenticate_common(verified, bk_username)
 
     try:
         from apigw_manager.apigw.authentication import ApiGatewayJWTUserMiddleware
 
-        get_user_parameters = sorted(inspect.signature(ApiGatewayJWTUserMiddleware.get_user).parameters.keys())
-        v3_parameters = sorted(inspect.signature(authenticate_with_signature_v3).parameters.keys())
-        if get_user_parameters == v3_parameters:
-            authenticate = authenticate_with_signature_v3
-        else:
+        get_user_parameters = inspect.signature(ApiGatewayJWTUserMiddleware.get_user).parameters.keys()
+        # django 的 authenticate 方法会保证向后兼容参数，调用方新增参数不会影响用户认证（认证只用到了 verified、bk_username 这 2 个参数）
+        # apigw_manager 的 3.0.0 版本开始 将 api_name 修改为了 gateway_name，导致无法保证向后兼容，所以需要单独处理
+        # https://github.com/django/django/blob/stable/4.2.x/django/contrib/auth/__init__.py#L69
+        if "api_name" in get_user_parameters and "gateway_name" not in get_user_parameters:
             authenticate = authenticate_with_signature_v1  # type: ignore
+        else:
+            authenticate = authenticate_with_signature_v3  # type: ignore
         del get_user_parameters
-        del v3_parameters
     except ImportError:
-        authenticate = authenticate_with_signature_v1  # type: ignore
+        authenticate = authenticate_with_signature_v3  # type: ignore
 
     def get_user(self, user_id):
         raise NotImplementedError(

sdks/bkpaas-auth/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "bkpaas-auth"
-version = "3.1.0"
+version = "3.1.1"
 description = "User authentication django app for blueking internal projects"
 readme = "README.md"
 authors = ["blueking <blueking@tencent.com>"]

sdks/bkpaas-auth/setup.py

@@ -12,27 +12,27 @@
 
 import os.path
 
-readme = ''
+readme = ""
 here = os.path.abspath(os.path.dirname(__file__))
-readme_path = os.path.join(here, 'README.rst')
+readme_path = os.path.join(here, "README.rst")
 if os.path.exists(readme_path):
-    with open(readme_path, 'rb') as stream:
-        readme = stream.read().decode('utf8')
+    with open(readme_path, "rb") as stream:
+        readme = stream.read().decode("utf8")
 
 
 setup(
     long_description=readme,
-    name='bkpaas-auth',
-    version='3.1.0',
-    description='User authentication django app for blueking internal projects',
-    python_requires='<4.0,>=3.8',
-    author='blueking',
-    author_email='blueking@tencent.com',
-    license='Apach License 2.0',
-    packages=['bkpaas_auth', 'bkpaas_auth.core'],
+    name="bkpaas-auth",
+    version="3.1.1",
+    description="User authentication django app for blueking internal projects",
+    python_requires="<4.0,>=3.8",
+    author="blueking",
+    author_email="blueking@tencent.com",
+    license="Apach License 2.0",
+    packages=["bkpaas_auth", "bkpaas_auth.core"],
     package_dir={"": "."},
     package_data={},
-    install_requires=['django<5.0,>=4.2', 'requests', 'six'],
+    install_requires=["django<5.0,>=4.2", "requests", "six"],
     extras_require={
         "dev": [
             "flake8==3.*,>=3.8.4",

sdks/bkpaas-auth/tests/test_backends.py

@@ -9,9 +9,8 @@
 
 
 class TestUniversalAuthBackend:
-
     @override_settings(BKAUTH_ENABLE_MULTI_TENANT_MODE=False, BKAUTH_BACKEND_TYPE="bk_ticket")
-    @mock.patch('requests.Session.request')
+    @mock.patch("requests.Session.request")
     def test_authenticate_bk_ticket(self, mock_request, mocker):
         mock_request.return_value = mock_json_response({"msg": "", "data": {"username": "foo"}, "ret": 0})
 
@@ -26,7 +25,7 @@ def test_authenticate_bk_ticket(self, mock_request, mocker):
         assert getattr(user, "display_name") == "foo"
 
     @override_settings(BKAUTH_ENABLE_MULTI_TENANT_MODE=False, BKAUTH_BACKEND_TYPE="bk_token")
-    @mock.patch('requests.Session.request')
+    @mock.patch("requests.Session.request")
     def test_authenticate_bk_token(self, mock_request, mocker):
         mock_request.return_value = mock_json_response(
             {"result": True, "code": 0, "message": "", "data": {"bk_username": "bar"}}
@@ -47,7 +46,7 @@ def test_authenticate_bk_token(self, mock_request, mocker):
         BKAUTH_BACKEND_TYPE="bk_token",
         BKAUTH_USER_INFO_APIGW_URL="fake_url",
     )
-    @mock.patch('requests.Session.request')
+    @mock.patch("requests.Session.request")
     def test_authenticate_bk_token_for_tenant_mode(self, mock_request, mocker):
         mock_request.return_value = mock_raw_response(
             {
@@ -74,7 +73,7 @@ def test_authenticate_bk_token_for_tenant_mode(self, mock_request, mocker):
 
 
 class TestAPIGatewayAuthBackend:
-    @pytest.fixture
+    @pytest.fixture()
     def backend(self):
         return APIGatewayAuthBackend()
 
@@ -89,7 +88,7 @@ def test_get_provider_type(self, backend):
     def test_authenticate_not_verified(self, mocker, backend):
         user = backend.authenticate(
             request=mocker.MagicMock(),
-            api_name="test",
+            gateway_name="test",
             bk_username="admin",
             verified=False,
         )
@@ -101,11 +100,27 @@ def test_authenticate_not_verified(self, mocker, backend):
     def test_authenticate_verified(self, mocker, backend):
         user = backend.authenticate(
             request=mocker.MagicMock(),
-            api_name="test",
+            gateway_name="test",
             bk_username="admin",
             verified=True,
         )
 
         assert not user.is_anonymous
         assert user.is_authenticated
         assert user.username == "admin"
+
+    def test_authenticate_with_additional_params(self, mocker, backend):
+        """测试带多个额外参数的情况"""
+        user = backend.authenticate(
+            request=mocker.MagicMock(),
+            gateway_name="test",
+            bk_username="multi_param_user",
+            verified=True,
+            param1="value1",
+            param2=2,
+            param3={"key": "value"},
+        )
+
+        assert not user.is_anonymous
+        assert user.is_authenticated
+        assert user.username == "multi_param_user"