146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
|
|
from django.http import JsonResponse, HttpResponseBadRequest
|
|
from django.shortcuts import render, redirect
|
|
from django.views.decorators.http import require_http_methods
|
|
from django.views.decorators.csrf import csrf_protect
|
|
from django.conf import settings
|
|
|
|
from .es_client import get_user_by_username, _salt_for_username
|
|
|
|
|
|
@require_http_methods(["GET"])
|
|
def login_page(request):
|
|
return render(request, "accounts/login.html")
|
|
|
|
|
|
@require_http_methods(["POST"])
|
|
@csrf_protect
|
|
def challenge(request):
|
|
try:
|
|
payload = json.loads(request.body.decode("utf-8"))
|
|
except json.JSONDecodeError:
|
|
return HttpResponseBadRequest("Invalid JSON")
|
|
|
|
username = payload.get("username", "").strip()
|
|
if not username:
|
|
return HttpResponseBadRequest("Username required")
|
|
|
|
# Generate nonce and compute per-username salt
|
|
nonce = os.urandom(16)
|
|
salt = _salt_for_username(username)
|
|
|
|
# Persist challenge in session to prevent replay with mismatched user
|
|
request.session["challenge_nonce"] = base64.b64encode(nonce).decode("ascii")
|
|
request.session["challenge_username"] = username
|
|
|
|
return JsonResponse({
|
|
"nonce": base64.b64encode(nonce).decode("ascii"),
|
|
"salt": base64.b64encode(salt).decode("ascii"),
|
|
})
|
|
|
|
|
|
@require_http_methods(["POST"])
|
|
@csrf_protect
|
|
def login_submit(request):
|
|
try:
|
|
payload = json.loads(request.body.decode("utf-8"))
|
|
except json.JSONDecodeError:
|
|
return HttpResponseBadRequest("Invalid JSON")
|
|
|
|
username = payload.get("username", "").strip()
|
|
client_hmac_b64 = payload.get("hmac", "")
|
|
if not username or not client_hmac_b64:
|
|
return HttpResponseBadRequest("Missing fields")
|
|
|
|
# Validate challenge stored in session
|
|
session_username = request.session.get("challenge_username")
|
|
nonce_b64 = request.session.get("challenge_nonce")
|
|
if not session_username or not nonce_b64 or session_username != username:
|
|
return HttpResponseBadRequest("Challenge not found or mismatched user")
|
|
|
|
# Lookup user in ES (placeholder)
|
|
user = get_user_by_username(username)
|
|
if not user:
|
|
return JsonResponse({"ok": False, "message": "User not found"}, status=401)
|
|
|
|
# Server-side HMAC verification
|
|
try:
|
|
nonce = base64.b64decode(nonce_b64)
|
|
stored_derived_b64 = user.get("password", "")
|
|
stored_derived = base64.b64decode(stored_derived_b64)
|
|
# HMAC-SHA256: server computes with stored derived secret
|
|
import hmac, hashlib
|
|
server_hmac = hmac.new(stored_derived, nonce, hashlib.sha256).digest()
|
|
server_hmac_b64 = base64.b64encode(server_hmac).decode("ascii")
|
|
except Exception:
|
|
return HttpResponseBadRequest("Verification error")
|
|
|
|
if not hmac.compare_digest(server_hmac_b64, client_hmac_b64):
|
|
return JsonResponse({"ok": False, "message": "Invalid credentials"}, status=401)
|
|
|
|
# Successful login: rotate session key and set user session
|
|
try:
|
|
request.session.cycle_key()
|
|
except Exception:
|
|
pass
|
|
|
|
request.session["user_id"] = user["user_id"]
|
|
request.session["username"] = user["username"]
|
|
request.session["premission"] = user["premission"]
|
|
|
|
# Clear challenge to prevent reuse
|
|
for k in ("challenge_username", "challenge_nonce"):
|
|
if k in request.session:
|
|
del request.session[k]
|
|
|
|
return JsonResponse({
|
|
"ok": True,
|
|
"redirect_url": f"/main/home/?user_id={user['user_id']}",
|
|
})
|
|
|
|
|
|
@require_http_methods(["GET"])
|
|
def home(request):
|
|
# Minimal placeholder page per requirement
|
|
# Ensure user_id is passed via query and session contains id
|
|
user_id = request.GET.get("user_id")
|
|
session_user_id = request.session.get("user_id")
|
|
context = {
|
|
"user_id": user_id or session_user_id,
|
|
}
|
|
return render(request, "accounts/home.html", context)
|
|
|
|
|
|
@require_http_methods(["POST"])
|
|
@csrf_protect
|
|
def logout(request):
|
|
# Flush the session to clear all data and rotate the key
|
|
try:
|
|
request.session.flush()
|
|
except Exception:
|
|
pass
|
|
|
|
# Return a response that also deletes cookies client-side
|
|
resp = JsonResponse({"ok": True, "redirect_url": "/accounts/login/"})
|
|
try:
|
|
# Delete session cookie
|
|
resp.delete_cookie(
|
|
settings.SESSION_COOKIE_NAME,
|
|
path='/',
|
|
samesite=settings.SESSION_COOKIE_SAMESITE,
|
|
secure=settings.SESSION_COOKIE_SECURE,
|
|
)
|
|
# Optionally delete CSRF cookie to satisfy "清除cookie" 的要求
|
|
resp.delete_cookie(
|
|
settings.CSRF_COOKIE_NAME,
|
|
path='/',
|
|
samesite=settings.CSRF_COOKIE_SAMESITE,
|
|
secure=settings.CSRF_COOKIE_SECURE,
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return resp |