Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 55 additions & 45 deletions apps/users/views/login.py
Original file line number Diff line number Diff line change
@@ -1,58 +1,65 @@
# coding=utf-8
"""
@project: MaxKB
@Author:虎虎
@file: user.py
@date:2025/4/14 10:22
@desc:
@project: MaxKB
@Author:虎虎
@file: user.py
@date:2025/4/14 10:22
@desc:
Comment on lines +3 to +7
"""
from django.core.cache import cache
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
from rest_framework.request import Request
from rest_framework.views import APIView

from common import result
from common.auth import TokenAuth
from common.constants.cache_version import Cache_Version
from common.log.log import log
from common.utils.common import encryption
from django.core.cache import cache
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
from maxkb.const import CONFIG
from models_provider.api.model import DefaultModelResponse
from users.api.login import LoginAPI, CaptchaAPI
from users.serializers.login import LoginSerializer, CaptchaSerializer
from rest_framework.request import Request
from rest_framework.views import APIView
from users.api.login import CaptchaAPI, LoginAPI
from users.serializers.login import CaptchaSerializer, LoginSerializer


def _get_details(request):
path = request.path
body = request.data
query = request.query_params
return {
'path': path,
'body': {**body, 'password': encryption(body.get('password', ''))},
'query': query
}
return {"path": path, "body": {**body, "password": encryption(body.get("password", ""))}, "query": query}
Comment on lines 26 to +30


class LoginView(APIView):
@extend_schema(methods=['POST'],
description=_("Log in"),
summary=_("Log in"),
operation_id=_("Log in"), # type: ignore
tags=[_("User Management")], # type: ignore
request=LoginAPI.get_request(),
responses=LoginAPI.get_response())
@log(menu='User management', operate='Log in', get_user=lambda r: {'username': r.data.get('username', None)},
get_details=_get_details,
get_operation_object=lambda r, k: {'name': r.data.get('username')})
@extend_schema(
methods=["POST"],
description=_("Log in"),
summary=_("Log in"),
operation_id=_("Log in"), # type: ignore
tags=[_("User Management")], # type: ignore
request=LoginAPI.get_request(),
responses=LoginAPI.get_response(),
)
@log(
menu="User management",
operate="Log in",
get_user=lambda r: {"username": r.data.get("username", None)},
get_details=_get_details,
get_operation_object=lambda r, k: {"name": r.data.get("username")},
)
def post(self, request: Request):
token, f_token = LoginSerializer().login(request.data)
response = result.success(token)
from common.utils.logger import maxkb_logger

maxkb_logger.info(
f"LoginView.post: request.scheme={request.scheme}, request.is_secure()={request.is_secure()}, request.META['wsgi.url_scheme']={request.META.get('wsgi.url_scheme')}, request.META['HTTPS']={request.META.get('HTTPS')}"
)
Comment on lines +53 to +57
response.set_cookie(
'mk_file_auth',
"mk_file_auth",
value=f_token,
max_age=7 * 24 * 3600,
path=f'{CONFIG.get_admin_path()}',
path=f"{CONFIG.get_admin_path()}",
domain=None,
secure=request.scheme == "https",
httponly=True,
Expand All @@ -64,27 +71,30 @@ def post(self, request: Request):
class Logout(APIView):
authentication_classes = [TokenAuth]

@extend_schema(methods=['POST'],
summary=_("Sign out"),
description=_("Sign out"),
operation_id=_("Sign out"), # type: ignore
tags=[_("User Management")], # type: ignore
responses=DefaultModelResponse.get_response())
@log(menu='User management', operate='Sign out',
get_operation_object=lambda r, k: {'name': r.user.username})
@extend_schema(
methods=["POST"],
summary=_("Sign out"),
description=_("Sign out"),
operation_id=_("Sign out"), # type: ignore
tags=[_("User Management")], # type: ignore
responses=DefaultModelResponse.get_response(),
)
@log(menu="User management", operate="Sign out", get_operation_object=lambda r, k: {"name": r.user.username})
def post(self, request: Request):
version, get_key = Cache_Version.TOKEN.value
cache.delete(get_key(token=request.META.get('HTTP_AUTHORIZATION')[7:]), version=version)
cache.delete(get_key(token=request.META.get("HTTP_AUTHORIZATION")[7:]), version=version)
return result.success(True)


class CaptchaView(APIView):
@extend_schema(methods=['GET'],
summary=_("Get captcha"),
description=_("Get captcha"),
operation_id=_("Get captcha"), # type: ignore
tags=[_("User Management")], # type: ignore
responses=CaptchaAPI.get_response())
@extend_schema(
methods=["GET"],
summary=_("Get captcha"),
description=_("Get captcha"),
operation_id=_("Get captcha"), # type: ignore
tags=[_("User Management")], # type: ignore
responses=CaptchaAPI.get_response(),
)
def get(self, request: Request):
username = request.query_params.get('username', None)
username = request.query_params.get("username", None)
return result.success(CaptchaSerializer().generate(username))
Loading