100 lines
4.0 KiB
Python
100 lines
4.0 KiB
Python
from gvsdsdk.auth.cookie import ELT_COOKIE_NAME
|
||
from gvsdsdk.models import UserRole, Role
|
||
from gvsdsdk.model_utils import UUIDObj
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ELTAuthMiddleware:
|
||
def __init__(self, get_response):
|
||
self.get_response = get_response
|
||
|
||
def __call__(self, request):
|
||
elt_token = request.COOKIES.get(ELT_COOKIE_NAME)
|
||
if elt_token:
|
||
request.elt_token = elt_token
|
||
response = self.get_response(request)
|
||
return response
|
||
|
||
|
||
class CurrentRoleMiddleware:
|
||
"""解析 CurrentRoleUUID Cookie，设置 request.current_role 和 request.current_company_uuid
|
||
|
||
request.current_role : gvsdsdk.models.Role 实例（或 None）
|
||
request.current_company_uuid : 通过 UserRole.CompanyUUID 获取当前角色所属公司UUID
|
||
|
||
注意：此中间件在 DRF 认证之前运行，因此需要自行从 ELT token 解析用户。
|
||
"""
|
||
|
||
def __init__(self, get_response):
|
||
self.get_response = get_response
|
||
|
||
def _get_user_from_elt(self, request):
|
||
"""从 ELT token cookie 中解析用户（不依赖 DRF 认证）"""
|
||
import uuid as _uuid
|
||
import datetime as _datetime
|
||
elt_token = request.COOKIES.get(ELT_COOKIE_NAME)
|
||
if not elt_token:
|
||
return None
|
||
try:
|
||
token_uuid = _uuid.UUID(elt_token)
|
||
except (ValueError, AttributeError):
|
||
return None
|
||
try:
|
||
from gvsdsdk.models import EphemeralToken
|
||
elt = EphemeralToken.objects.get(ELTUUID=token_uuid, ELTStatus=1)
|
||
except EphemeralToken.DoesNotExist:
|
||
return None
|
||
now_utc = _datetime.datetime.now(_datetime.timezone.utc)
|
||
if elt.ELTEndTime.replace(tzinfo=None) < now_utc.replace(tzinfo=None):
|
||
return None
|
||
try:
|
||
from gvsds.apps.core.models import User
|
||
return User.objects.get(UserUUID=elt.ELTUserUUID)
|
||
except User.DoesNotExist:
|
||
return None
|
||
|
||
def __call__(self, request):
|
||
request.current_role = None
|
||
request.current_company_uuid = None
|
||
|
||
# 优先使用已认证的 request.user，否则自行从 ELT token 解析
|
||
user = None
|
||
if hasattr(request, 'user') and request.user and request.user.is_authenticated:
|
||
user = request.user
|
||
else:
|
||
user = self._get_user_from_elt(request)
|
||
|
||
if not user:
|
||
return self.get_response(request)
|
||
|
||
role_uuid_str = request.COOKIES.get('CurrentRoleUUID', '')
|
||
logger.info('[CurrentRoleMiddleware] user=%s, CurrentRoleUUID cookie=%s', user, role_uuid_str or '(empty)')
|
||
if role_uuid_str:
|
||
try:
|
||
role_uuid_bytes = UUIDObj(role_uuid_str).bytes
|
||
user_uuid = user.UserUUID
|
||
logger.info('[CurrentRoleMiddleware] querying UserRole: UserUUID=%s, RoleUUID=%s',
|
||
user_uuid.hex() if hasattr(user_uuid, 'hex') else user_uuid,
|
||
role_uuid_bytes.hex())
|
||
user_role = UserRole.objects.filter(
|
||
UserUUID=user.UserUUID,
|
||
RoleUUID=role_uuid_bytes
|
||
).first()
|
||
logger.info('[CurrentRoleMiddleware] user_role found=%s', user_role is not None)
|
||
if user_role:
|
||
role = Role.objects.filter(RoleUUID=role_uuid_bytes).first()
|
||
if role:
|
||
request.current_role = role
|
||
if user_role.CompanyUUID:
|
||
request.current_company_uuid = user_role.CompanyUUID
|
||
logger.info('[CurrentRoleMiddleware] set current_company_uuid=%s',
|
||
user_role.CompanyUUID.hex() if hasattr(user_role.CompanyUUID, 'hex') else user_role.CompanyUUID)
|
||
else:
|
||
logger.warning('[CurrentRoleMiddleware] user_role.CompanyUUID is None')
|
||
except Exception as e:
|
||
logger.error('[CurrentRoleMiddleware] error: %s', e, exc_info=True)
|
||
|
||
return self.get_response(request)
|