feat: add UserTimezoneMiddleware (#255)

Author: xogas

sdks/bkpaas-auth/README.md

@@ -152,3 +152,45 @@ MIDDLEWARE += [
 ```
 
 设置之后，通过 JWT 透传的用户态会验证后，会写入到 `request.user` 中。注意，配置了不认证用户的网关资源透传的请求，会生成一个有对应用户名的匿名用户对象（`is_authenticated` 为 `False`）。
+
+### UserTimezoneMiddleware 用户时区中间件
+
+`UserTimezoneMiddleware` 是一个用于根据用户配置的时区自动设置 Django 时区的中间件。
+
+#### 功能特性
+
+- 自动从用户对象的 `time_zone` 属性读取时区配置
+- 时区配置无效时自动回退到默认时区 `settings.TIME_ZONE`
+- 响应返回时自动重置时区，避免线程复用导致的时区污染
+
+#### 使用说明
+
+1. **配置要求**：
+   - 用户对象必须包含 `time_zone` 属性（字符串类型）
+   - 时区名称必须符合 IANA 时区数据库标准（如 "Asia/Shanghai"）
+
+2. **中间件顺序**：
+   - 必须放在所有用户认证中间件之后
+   - 建议放在 `CookieLoginMiddleware` 之后
+
+3. **执行逻辑**：
+   - 未登录用户跳过时区设置
+   - 读取用户 `time_zone` 属性并验证有效性
+   - 时区无效时记录警告日志并回退到默认时区
+   - 响应处理完成后自动重置时区
+
+#### 日志输出示例
+
+```bash
+# 时区激活成功
+DEBUG: Activated timezone 'Asia/Shanghai' for user 'test_user'
+
+# 时区无效警告
+WARNING: Invalid time_zone 'Invalid/Timezone' for user 'test_user', fallback to default. Error: ...
+```
+
+#### 注意事项
+
+- 确保用户管理系统正确设置 `time_zone` 属性
+- 时区名称必须为有效的 IANA 时区标识符
+- 中间件依赖 Django 的时区功能，确保 `USE_TZ = True`

sdks/bkpaas-auth/bkpaas_auth/middlewares.py

@@ -4,12 +4,14 @@
 import pickle
 import time
 from typing import Dict
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from django.conf import settings
 from django.contrib import auth
 from django.http import HttpRequest, HttpResponse
 from django.utils.deprecation import MiddlewareMixin
 from django.utils.encoding import force_str
+from django.utils import timezone as dj_timezone
 
 from bkpaas_auth.backends import UniversalAuthBackend
 from bkpaas_auth.core.constants import ACCESS_PERMISSION_DENIED_CODE
@@ -94,3 +96,51 @@ def authenticate_and_login(self, request: HttpRequest, credentials: Dict[str, st
         # rotation.
         if getattr(request, "user", None) != user:
             auth.login(request, user)
+
+
+class UserTimezoneMiddleware(MiddlewareMixin):
+    """按用户的时区属性激活 Django 时区。
+
+    该中间件从用户管理系统获取用户时区信息并激活，使所有时间相关的序列化输出
+    都使用用户所在时区的偏移量。
+
+    执行逻辑:
+    1. 未登录用户跳过处理
+    2. 从 request.user 读取 time_zone 属性
+    3. 若时区字段缺失或非法，回退到默认时区 settings.TIME_ZONE
+    4. 在响应返回时重置时区，避免线程复用导致的时区污染
+
+    NOTE: 必须放在所有用户认证中间件之后
+    """
+
+    def process_request(self, request):
+        # Ignore request without user attribute or anonymous user
+        if not hasattr(request, "user") or not request.user.is_authenticated:
+            return
+
+        user = request.user
+        tz_name = getattr(user, "time_zone", None)
+
+        # Try to activate user's timezone if it's a non-empty string
+        if tz_name and isinstance(tz_name, str):
+            try:
+                user_tz = ZoneInfo(tz_name)
+                dj_timezone.activate(user_tz)
+            except ZoneInfoNotFoundError as e:
+                logger.warning(
+                    "Invalid time_zone '%s' for user '%s', fallback to default. Error: %s",
+                    tz_name,
+                    user.username,
+                    str(e),
+                )
+            else:
+                logger.debug("Activated timezone '%s' for user '%s'", tz_name, user.username)
+                return
+
+        # Fallback to default timezone when time_zone is empty or invalid
+        dj_timezone.activate(dj_timezone.get_default_timezone())
+
+    def process_response(self, request, response):
+        """重置时区"""
+        dj_timezone.deactivate()
+        return response

sdks/bkpaas-auth/tests/test_middlewares.py

@@ -12,12 +12,13 @@
 from django.contrib.sessions.middleware import SessionMiddleware
 from django.http import HttpRequest
 from django.test.utils import override_settings
+from django.utils import timezone as dj_timezone
 
 from bkpaas_auth.backends import UniversalAuthBackend
 from bkpaas_auth.core.constants import ACCESS_PERMISSION_DENIED_CODE, ProviderType
 from bkpaas_auth.core.exceptions import AccessPermissionDenied
 from bkpaas_auth.core.token import LoginToken
-from bkpaas_auth.middlewares import CookieLoginMiddleware, auth
+from bkpaas_auth.middlewares import CookieLoginMiddleware, auth, UserTimezoneMiddleware
 from bkpaas_auth.models import User
 from tests.utils import generate_random_string
 
@@ -182,3 +183,69 @@ def test_auth(self, db, bk_token, dj_request):
         assert isinstance(user, UserModel)
         assert not isinstance(user, User)
         assert isinstance(user.pk, int)
+
+
+class TestUserTimezoneMiddleware:
+    """Test cases for UserTimezoneMiddleware"""
+
+    @pytest.fixture
+    def middleware(self):
+        return UserTimezoneMiddleware(MagicMock())
+
+    @pytest.fixture
+    def authenticated_user(self):
+        """Create a mock authenticated user"""
+        user = MagicMock()
+        user.is_authenticated = True
+        return user
+
+    @pytest.fixture(autouse=True)
+    def setup_timezone(self):
+        """Reset timezone before and after each test to avoid pollution"""
+        with override_settings(TIME_ZONE="UTC"):
+            dj_timezone.deactivate()
+            yield
+            dj_timezone.deactivate()
+
+    def test_skip_request_without_user_attr(self, rf, middleware):
+        """Test that requests without user attribute don't change timezone"""
+        middleware.process_request(rf.get("/"))
+        assert dj_timezone.get_current_timezone_name() == "UTC"
+
+    def test_skip_anonymous_user(self, rf, middleware):
+        """Test that anonymous users don't change timezone"""
+        request = rf.get("/")
+        request.user = AnonymousUser()
+        middleware.process_request(request)
+        assert dj_timezone.get_current_timezone_name() == "UTC"
+
+    def test_activate_valid_user_timezone(self, rf, middleware, authenticated_user):
+        """Test that valid user timezone is actually activated"""
+        request = rf.get("/")
+        authenticated_user.time_zone = "America/New_York"
+        request.user = authenticated_user
+        middleware.process_request(request)
+        assert dj_timezone.get_current_timezone_name() == "America/New_York"
+
+    @pytest.mark.parametrize(
+        ("time_zone_value", "has_attr"),
+        [
+            ("Invalid/Timezone", True),
+            ("", True),
+            (None, True),
+            (None, False),
+            (123, True),
+        ],
+        ids=["invalid_timezone", "empty_string", "none_value", "no_attr", "non_string_type"],
+    )
+    @override_settings(TIME_ZONE="Asia/Shanghai")
+    def test_fallback_to_default_timezone(self, rf, middleware, authenticated_user, time_zone_value, has_attr):
+        """Test fallback to default timezone for various edge cases"""
+        request = rf.get("/")
+        if has_attr:
+            authenticated_user.time_zone = time_zone_value
+        else:
+            del authenticated_user.time_zone
+        request.user = authenticated_user
+        middleware.process_request(request)
+        assert dj_timezone.get_current_timezone_name() == "Asia/Shanghai"