feat(bkpaas_auth): 不再对用户 ID 加密

sdks/bkpaas-auth/CHANGES.md

@@ -1,5 +1,8 @@
 # 版本历史
 
+## 3.0.1
+- 不再对用户 ID 加密
+
 ## 3.0.0
 - BreakChange: 不再支持 Python 3.6，3.7
 - BreakChange: Django 版本要求 >=4.2，<5.0

sdks/bkpaas-auth/bkpaas_auth/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-__version__ = "3.0.0"
+__version__ = "3.0.1"
 
 
 def get_user_by_user_id(user_id: str, username_only: bool = True):
@@ -24,4 +24,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/core/encoder.py

@@ -4,39 +4,47 @@
 
 from six import ensure_binary, ensure_text
 
-from bkpaas_auth.core.constants import ProviderType
 from bkpaas_auth.core.algorithms import ARC4
+from bkpaas_auth.core.constants import ProviderType
 
 
 class BluekingUserIdEncoder:
     """Generator for blueking user id"""
 
-    # It is not a real secret key, just for encoding the username
-    secret_key = 'jdvoqu3o4'
+    # Used to parse existing encrypted user IDs. New user IDs are no longer encrypted.
+    # It is not a real secret key, just for encoding the username.
+    secret_key = "jdvoqu3o4"
+    # Flag indicating that the user ID is unencrypted.
+    unencrypted_user_flag = "$$"
 
     def encode(self, provider_type: Union[int, ProviderType], username: Union[str, bytes]):
-        """Generate a hex string used as blueking user id
+        """Encode the provider type and username into a Blueking user ID.
 
         :param provider_type: See constants.ProviderType
-        :param username: User uin or user rtx username
-        :returns: A hex string
+        :param username: The username format includes: uin (QQ login), username, uuid (latest user management rules)
+        :return: A hex string representing the Blueking user ID.
         """
         id_prefix = ProviderType(provider_type).get_id_prefix()
-
-        encoded = ARC4(ensure_binary(self.secret_key)).encrypt(ensure_binary(username))
-        return id_prefix + ensure_text(binascii.hexlify(encoded))
+        return self.unencrypted_user_flag + id_prefix + username
 
     def decode(self, user_id: Union[str, bytes]) -> Tuple[int, str]:
-        """Decode a given bk_user_id to the combination of (provider_type, username)
+        """Decode a given Blueking user ID into its constituent provider type and username.
 
         :param str user_id: Blueking user id
         :returns: (provider_type, username)
         """
-        _provider_type, username = user_id[:2], user_id[2:]
-        provider_type = int(_provider_type)
+        # Check if the user ID is unencrypted
+        if user_id.startswith(self.unencrypted_user_flag):
+            user_info = user_id[2:]
+            _provider_type, decoded_username = user_info[:2], user_info[2:]
+        else:
+            # Handle encrypted user ID
+            _provider_type, encrypted_username = user_id[:2], user_id[2:]
+            decoded = ARC4(ensure_binary(self.secret_key)).decrypt(binascii.unhexlify(encrypted_username))
+            decoded_username = ensure_text(decoded)
 
-        decoded = ARC4(ensure_binary(self.secret_key)).decrypt(binascii.unhexlify(username))
-        return provider_type, ensure_text(decoded)
+        provider_type = int(_provider_type)
+        return provider_type, decoded_username
 
 
 user_id_encoder = BluekingUserIdEncoder()

sdks/bkpaas-auth/pyproject.toml

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

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.0.0',
-    description='User authentication django app for blueking internal projects',
-    python_requires='<4.0,>=3.8',
-    author='blueking',
-    author_email='[email protected]',
-    license='Apach License 2.0',
-    packages=['bkpaas_auth', 'bkpaas_auth.core'],
+    name="bkpaas-auth",
+    version="3.0.1",
+    description="User authentication django app for blueking internal projects",
+    python_requires="<4.0,>=3.8",
+    author="blueking",
+    author_email="[email protected]",
+    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/conftest.py

@@ -12,7 +12,7 @@ def setup_root_logger(level):
 
     handler = logging.StreamHandler(sys.stdout)
     handler.setLevel(level)
-    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
     handler.setFormatter(formatter)
     root.addHandler(handler)
 
@@ -23,36 +23,36 @@ def pytest_configure():
     settings.configure(
         DEBUG_PROPAGATE_EXCEPTIONS=True,
         DATABASES={
-            'default': {
-                'ENGINE': 'django.db.backends.sqlite3',
-                'NAME': ':memory:',
+            "default": {
+                "ENGINE": "django.db.backends.sqlite3",
+                "NAME": ":memory:",
             }
         },
-        SECRET_KEY='not very secret in tests',
+        SECRET_KEY="not very secret in tests",
         TEMPLATES=[
             {
-                'BACKEND': 'django.template.backends.django.DjangoTemplates',
-                'APP_DIRS': True,
+                "BACKEND": "django.template.backends.django.DjangoTemplates",
+                "APP_DIRS": True,
             },
         ],
         USE_TZ=True,
-        TIME_ZONE='Asia/Shanghai',
+        TIME_ZONE="Asia/Shanghai",
         MIDDLEWARE=[],
         MIDDLEWARE_CLASSES=[],
         INSTALLED_APPS=(
-            'django.contrib.auth',
-            'django.contrib.contenttypes',
-            'django.contrib.sessions',
-            'tests',
+            "django.contrib.auth",
+            "django.contrib.contenttypes",
+            "django.contrib.sessions",
+            "tests",
         ),
         # bkauth settings
-        BKAUTH_BACKEND_TYPE='bk_token',
-        BKAUTH_TOKEN_APP_CODE='mock_app_code',
-        BKAUTH_TOKEN_SECRET_KEY='mock_app_key',
-        BKAUTH_TOKEN_GRANT_ENDPOINT='',
-        USER_ID='0221dbef87cd',
-        USER_NAME='user1',
-        USER_NICKNAME='user1中文名',
+        BKAUTH_BACKEND_TYPE="bk_token",
+        BKAUTH_TOKEN_APP_CODE="mock_app_code",
+        BKAUTH_TOKEN_SECRET_KEY="mock_app_key",
+        BKAUTH_TOKEN_GRANT_ENDPOINT="",
+        USER_ID="$$02user1",
+        USER_NAME="user1",
+        USER_NICKNAME="user1中文名",
         AUTHENTICATION_BACKENDS=["bkpaas_auth.backends.UniversalAuthBackend"],
     )
 

sdks/bkpaas-auth/tests/core/test_bk_token.py

@@ -14,14 +14,16 @@
 
 @pytest.fixture(autouse=True)
 def use_bk_token_settings(settings):
-    settings.BKAUTH_BACKEND_TYPE = 'bk_token'
-    settings.USER_ID = '0335cce79c92'
-    settings.USER_NAME = 'admin'
-    settings.USER_NICKNAME = 'admin的中文名'
+    settings.BKAUTH_BACKEND_TYPE = "bk_token"
+    settings.USER_ID = "$$03admin"
+    # 按老的规则加密后的用户ID
+    settings.ENCODE_USER_ID = "0335cce79c92"
+    settings.USER_NAME = "admin"
+    settings.USER_NICKNAME = "admin的中文名"
 
 
 class TestMisc:
-    @mock.patch('requests.Session.request')
+    @mock.patch("requests.Session.request")
     def test_get_user_by_user_id(self, mocked_request):
         mocked_request.return_value = mock_json_response(
             {
@@ -36,19 +38,19 @@ def test_get_user_by_user_id(self, mocked_request):
                 "request_id": "ed3c8f75-d956-4dcd-b5d0-7bcd6c2e386a",
             }
         )
-        user = get_user_by_user_id(settings.USER_ID, username_only=True)
+        user = get_user_by_user_id(settings.ENCODE_USER_ID, username_only=True)
         assert user.username == settings.USER_NAME
         assert not user.nickname
 
-        user = get_user_by_user_id(settings.USER_ID, username_only=False)
+        user = get_user_by_user_id(settings.ENCODE_USER_ID, username_only=False)
         assert user.username == settings.USER_NAME
         assert user.nickname == settings.USER_NICKNAME
 
 
 class TestUser:
-    @mock.patch('requests.Session.request')
+    @mock.patch("requests.Session.request")
     def test_user_info(self, mocked_request):
-        token = LoginToken('token', expires_in=86400)
+        token = LoginToken("token", expires_in=86400)
         mocked_request.return_value = mock_json_response(
             {
                 "message": "",
@@ -69,17 +71,17 @@ def test_user_info(self, mocked_request):
         user = User(token)
         user_info.provide(user)
         assert user.bkpaas_user_id == settings.USER_ID
-        assert user.email == ''
+        assert user.email == ""
         assert user.username == settings.USER_NAME
         assert user.chinese_name == settings.USER_NICKNAME
 
 
 class TestToken:
-    @mock.patch('requests.Session.request')
+    @mock.patch("requests.Session.request")
     def test_create_user_from_token(self, mocked_request):
-        token = LoginToken('token3', expires_in=3600)
+        token = LoginToken("token3", expires_in=3600)
 
-        token.user_info = BkUserInfo(bk_username=settings.USER_NAME, chname='', email='', phone='')
+        token.user_info = BkUserInfo(bk_username=settings.USER_NAME, chname="", email="", phone="")
         user = create_user_from_token(token)
         assert user.provider_type == ProviderType.BK
         assert user.username == settings.USER_NAME

sdks/bkpaas-auth/tests/core/test_encoder.py

@@ -7,28 +7,36 @@
 
 class TestUserIdGeneration:
     @pytest.mark.parametrize(
-        "input, expected_encoded",
+        ("input", "expected_encoded"),
         [
-            ((1, "3"), "0167"),
-            ((ProviderType.UIN, "3"), "0167"),
-            ((2, "r"), "0226"),
-            ((ProviderType.RTX, "r"), "0226"),
-            ((1, "blueking"), "0136c4ff90974290a7"),
-            ((9, "admin"), "0935cce79c92"),
+            ((1, "3"), "$$013"),
+            ((ProviderType.UIN, "3"), "$$013"),
+            ((2, "r"), "$$02r"),
+            ((ProviderType.RTX, "r"), "$$02r"),
+            ((1, "blueking"), "$$01blueking"),
+            ((9, "admin"), "$$09admin"),
         ],
     )
     def test_encode(self, input, expected_encoded):
         assert user_id_encoder.encode(*input) == expected_encoded
 
     @pytest.mark.parametrize(
-        "input, expected_decoded",
+        ("input", "expected_encoded"),
         [
             ("0167", (1, "3")),
             ("0167", (ProviderType.UIN, "3")),
             ("0226", (2, "r")),
             ("0226", (ProviderType.RTX, "r")),
             ("0136c4ff90974290a7", (1, "blueking")),
             ("0935cce79c92", (9, "admin")),
+            # 新的规则生成的用户信息也能正常解密
+            ("$$01a47dea7cb11a11ef89645254009504dc", (1, "a47dea7cb11a11ef89645254009504dc")),
+            ("$$013", (1, "3")),
+            ("$$013", (ProviderType.UIN, "3")),
+            ("$$02r", (2, "r")),
+            ("$$02r", (ProviderType.RTX, "r")),
+            ("$$01blueking", (1, "blueking")),
+            ("$$09admin", (9, "admin")),
         ],
     )
     def test_decode(self, input, expected_decoded):