import json from importlib import import_module from jwcrypto import jwk, jwt from datetime import datetime, timedelta from django.conf import settings from django.utils.module_loading import import_string from django.utils.crypto import constant_time_compare from django.utils.translation import gettext as _ from django.utils import dateformat, timezone from django.contrib.auth import ( get_user_model, BACKEND_SESSION_KEY, SESSION_KEY, HASH_SESSION_KEY ) from rest_framework.response import Response from rest_framework.permissions import IsAdminUser from pingcap_sso.models import User from pingcap_sso.api.base import Endpoint from pingcap_sso.api.permissions import MethodPermission from pingcap_sso.forms.sso import SSOSessionDetailForm, SSOSyncTokenForm from pingcap_sso.forms.admin_edit_user_profile import SSOUserUploadFormSubmitForm from pingcap_sso.lib.sso import get_cookie_sync_token_instance, SUBSYSTEM_ASKTUG_COM from pingcap_sso.lib.user import user_profile_for_marketing class SessionDetailEndpoint(Endpoint): permission_classes = (IsAdminUser, MethodPermission) perms_map = { 'POST': ['pingcap_sso.sso_get_session_detail'] } throttle_classes = () def post(self, request, *args, **kwargs): form = SSOSessionDetailForm(data=request.data) if not form.is_valid(): return Response({ 'detail': _('params error'), 'errors': form.errors }, status=400) user = self._get_user_by_sid(form.cleaned_data['ssid']) if not user: return Response({'detail': _('user not found')}, status=404) return Response({ 'detail': 'success', 'data': { "openid": user.openid, "username": user.username, "phone": user.phone or None, "email": user.email or None, "email_verified": user.email_verified, "avatar_url": user.avatar_url, "name": user.name or None, "address": user.address or None, "bio": user.bio, "company_name": user.company_name, "date_of_birth": user.date_of_birth.strftime('%Y-%m-%d') if user.date_of_birth else None, "job_title_code": user.job_title_code, "job_title_text": user.job_title_text } }) def _get_user_by_sid(self, sid): engine = import_module(settings.SESSION_ENGINE) self.SessionStore = engine.SessionStore session = self.SessionStore(sid) user = None try: user_id = get_user_model()._meta.pk.to_python(session[SESSION_KEY]) backend_path = session[BACKEND_SESSION_KEY] except KeyError: pass else: if backend_path in settings.AUTHENTICATION_BACKENDS: backend = import_string(backend_path)() user = backend.get_user(user_id) # Verify the session if hasattr(user, 'get_session_auth_hash'): session_hash = session.get(HASH_SESSION_KEY) session_hash_verified = session_hash and constant_time_compare( session_hash, user.get_session_auth_hash() ) if not session_hash_verified: session.flush() user = None return user class SessionInfoEndpoint(Endpoint): permission_classes = (IsAdminUser, ) throttle_classes = () def get(self, request, *args, **kwargs): user = self._get_user_by_sid(request.COOKIES.get(settings.SESSION_COOKIE_NAME)) if not user: return Response({'detail': _('user not found')}, status=404) id_token = self.generate_id_token(user) return Response({'id_token': id_token}) def generate_id_token(self, user): key = jwk.JWK.from_pem(settings.JWT_PRIVATE_KEY.encode("utf8")) expiration_time = timezone.now() + timedelta(seconds=60) claims = { 'iss': 'pingcap sso', 'sub': user.openid, 'exp': int(dateformat.format(expiration_time, "U")), 'iat': int(dateformat.format(datetime.utcnow(), "U")), 'auth_time': int(dateformat.format(user.last_login, "U")), 'username': user.username, 'phone': user.phone, 'email': user.email, 'fake_email': user.fake_email, 'avatar_url': user.avatar_url } jwt_token = jwt.JWT( header=json.dumps({"alg": "RS256", 'typ': 'jwt'}, default=str), claims=json.dumps(claims, default=str) ) jwt_token.make_signed_token(key) return jwt_token.serialize() def _get_user_by_sid(self, sid): engine = import_module(settings.SESSION_ENGINE) self.SessionStore = engine.SessionStore session = self.SessionStore(sid) user = None try: user_id = get_user_model()._meta.pk.to_python(session[SESSION_KEY]) backend_path = session[BACKEND_SESSION_KEY] except KeyError: pass else: if backend_path in settings.AUTHENTICATION_BACKENDS: backend = import_string(backend_path)() user = backend.get_user(user_id) # Verify the session if hasattr(user, 'get_session_auth_hash'): session_hash = session.get(HASH_SESSION_KEY) session_hash_verified = session_hash and constant_time_compare( session_hash, user.get_session_auth_hash() ) if not session_hash_verified: session.flush() user = None return user class DiscourseSyncTokenEndpoint(Endpoint): permission_classes = (IsAdminUser, MethodPermission) perms_map = { 'POST': ['pingcap_sso.sso_get_session_detail'] } throttle_classes = () def post(self, request, *args, **kwargs): form = SSOSyncTokenForm(data=request.data) if not form.is_valid(): return Response({ 'detail': _('please provide a valid sync_token'), }, status=400) token_instance = get_cookie_sync_token_instance( form.cleaned_data['sync_token'], SUBSYSTEM_ASKTUG_COM ) if not token_instance: return Response({ 'detail': _('please provide a valid sync_token'), }, status=400) user = token_instance.user user_email = (user.email or user.fake_email).lower() return Response({ 'detail': 'success', 'data': { 'external_id': user.openid, 'username': user.username, 'email': user_email, 'name': user.name or '', 'bio': user.bio or '', 'ssid': token_instance.session_id, 'redirect_to': token_instance.redirect_to } }) class SSOUserMarketingProfileEndpoint(Endpoint): """ 使用用户的 openid 获取用户可以用来营销的信息, 营销信息可能没有经过验证 目前会被表单系统调用来预填充用户信息 """ permission_classes = (IsAdminUser, MethodPermission) perms_map = { 'GET': ['pingcap_sso.view_user_profile'] } throttle_classes = () def get(self, request, user_open_id, *args, **kwargs): try: user = User.objects.get(openid=user_open_id) except User.DoesNotExist: return Response({'detail': _('user not found')}, status=404) return Response({ 'detail': 'success', 'data': user_profile_for_marketing(user) }) class SSOUserUploadFormSubmitEndpoint(Endpoint): """ 使用用户的 openid 提交表单数据，表单数据会被保存到 UserProfileCollections 中 """ permission_classes = (IsAdminUser, MethodPermission) perms_map = { 'POST': ['pingcap_sso.view_user_profile'] } throttle_classes = () def post(self, request, user_open_id, *args, **kwargs): try: user = User.objects.get(openid=user_open_id) except User.DoesNotExist: return Response({'detail': _('user not found')}, status=404) form = SSOUserUploadFormSubmitForm(user=user, data=request.data) if not form.is_valid(): return Response({ 'detail': _('params error'), 'errors': form.errors }, status=400) form.save() return Response({'detail': _('success')})