177 lines
7.8 KiB
Python
177 lines
7.8 KiB
Python
"""账号登录、会话和密码管理。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import secrets
|
|
import time
|
|
from typing import Any
|
|
|
|
from group_system.config import SESSION_DAYS, TEACHER_USER
|
|
from group_system.security import hash_password, verify_password
|
|
|
|
|
|
class AccountStoreMixin:
|
|
|
|
def create_session(self, role: str, username: str, student_id: int | None = None) -> str:
|
|
token = secrets.token_urlsafe(32)
|
|
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
|
now = int(time.time())
|
|
with self.lock, self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO sessions(token_hash,role,username,student_id,created_at,expires_at) VALUES(?,?,?,?,?,?)",
|
|
(token_hash, role, username, student_id, now, now + SESSION_DAYS * 86400),
|
|
)
|
|
self.connection.execute(
|
|
"DELETE FROM sessions WHERE expires_at < ?", (now,)
|
|
)
|
|
return token
|
|
|
|
def get_session(self, token: str | None) -> dict[str, Any] | None:
|
|
if not token:
|
|
return None
|
|
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
|
now = int(time.time())
|
|
with self.lock:
|
|
row = self.connection.execute(
|
|
"""SELECT s.token_hash,s.role,s.username,s.student_id,s.expires_at,
|
|
st.class_name,st.student_no,st.name,st.must_change_password
|
|
FROM sessions s LEFT JOIN students st ON st.id=s.student_id
|
|
WHERE s.token_hash=? AND s.expires_at>?""",
|
|
(token_hash, now),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return dict(row)
|
|
|
|
def destroy_session(self, token: str | None) -> None:
|
|
if not token:
|
|
return
|
|
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
|
with self.lock, self.connection:
|
|
self.connection.execute("DELETE FROM sessions WHERE token_hash=?", (token_hash,))
|
|
|
|
def login_student(self, student_no: str, password: str) -> tuple[dict[str, Any] | None, str]:
|
|
student_no = student_no.strip()
|
|
active = self.active_class()
|
|
with self.lock:
|
|
student = self.connection.execute(
|
|
"SELECT * FROM students WHERE student_no=?", (student_no,)
|
|
).fetchone()
|
|
if student is None or not verify_password(password, student["password_hash"]):
|
|
return None, "学号或密码不正确"
|
|
if student["class_name"] != active:
|
|
return None, f"当前仅开放 {active} 班登录，{student['class_name']} 班暂不能登录"
|
|
token = self.create_session("student", student["student_no"], student["id"])
|
|
return {
|
|
"token": token,
|
|
"student": {
|
|
"id": student["id"],
|
|
"name": student["name"],
|
|
"student_no": student["student_no"],
|
|
"class_name": student["class_name"],
|
|
"must_change_password": bool(student["must_change_password"]),
|
|
},
|
|
}, "登录成功"
|
|
|
|
def login_teacher(self, username: str, password: str) -> tuple[dict[str, Any] | None, str]:
|
|
with self.lock:
|
|
expected_user = self.get_setting("teacher_username") or TEACHER_USER
|
|
expected_password = self.get_setting("teacher_password")
|
|
if username.strip() != expected_user or not verify_password(password, expected_password):
|
|
return None, "教师账号或密码不正确"
|
|
token = self.create_session("teacher", expected_user, None)
|
|
return {"token": token, "teacher": {"username": expected_user}}, "登录成功"
|
|
|
|
def change_teacher_password(
|
|
self, session: dict[str, Any], old_password: str, new_password: str
|
|
) -> tuple[bool, str]:
|
|
if session.get("role") != "teacher":
|
|
return False, "教师会话无效"
|
|
old_password = old_password or ""
|
|
new_password = new_password or ""
|
|
if len(new_password) < 8:
|
|
return False, "新密码至少 8 位"
|
|
if new_password == old_password:
|
|
return False, "新密码不能与原密码相同"
|
|
with self.lock:
|
|
expected_user = self.get_setting("teacher_username") or TEACHER_USER
|
|
if session.get("username") != expected_user:
|
|
return False, "教师账号已变更，请重新登录"
|
|
stored_password = self.get_setting("teacher_password")
|
|
if not verify_password(old_password, stored_password):
|
|
return False, "原密码不正确"
|
|
self.set_setting("teacher_password", hash_password(new_password))
|
|
return True, "教师密码修改成功"
|
|
|
|
def change_student_password(
|
|
self, session: dict[str, Any], old_password: str, new_password: str
|
|
) -> tuple[bool, str]:
|
|
student_id = session.get("student_id")
|
|
if not student_id:
|
|
return False, "学生会话无效"
|
|
old_password = old_password or ""
|
|
new_password = new_password or ""
|
|
if len(new_password) < 8:
|
|
return False, "新密码至少 8 位"
|
|
if new_password == old_password:
|
|
return False, "新密码不能与原密码相同"
|
|
with self.lock:
|
|
student = self.connection.execute(
|
|
"SELECT * FROM students WHERE id=?", (student_id,)
|
|
).fetchone()
|
|
if student is None or not verify_password(old_password, student["password_hash"]):
|
|
return False, "原密码不正确"
|
|
if new_password == student["student_no"]:
|
|
return False, "新密码不能设置为学号"
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"UPDATE students SET password_hash=?, must_change_password=0 WHERE id=?",
|
|
(hash_password(new_password), student_id),
|
|
)
|
|
token_hash = hashlib.sha256(session["_token"].encode()).hexdigest()
|
|
self.connection.execute(
|
|
"DELETE FROM sessions WHERE student_id=? AND token_hash<>?",
|
|
(student_id, token_hash),
|
|
)
|
|
return True, "密码修改成功"
|
|
|
|
def teacher_set_student_password(
|
|
self, student_id: int, new_password: str
|
|
) -> tuple[bool, str]:
|
|
"""教师后台重置指定学生的登录密码。"""
|
|
new_password = new_password or ""
|
|
if len(new_password) < 8:
|
|
return False, "新密码至少 8 位"
|
|
with self.lock, self.connection:
|
|
student = self.connection.execute(
|
|
"SELECT student_no FROM students WHERE id=?", (student_id,)
|
|
).fetchone()
|
|
if student is None:
|
|
return False, "学生不存在"
|
|
if new_password == student["student_no"]:
|
|
return False, "新密码不能设置为学号"
|
|
self.connection.execute(
|
|
"UPDATE students SET password_hash=?, must_change_password=0 WHERE id=?",
|
|
(hash_password(new_password), student_id),
|
|
)
|
|
self.connection.execute(
|
|
"DELETE FROM sessions WHERE student_id=?", (student_id,)
|
|
)
|
|
return True, f"已修改 {student['student_no']} 的密码"
|
|
|
|
def get_student(self, student_id: int) -> dict[str, Any] | None:
|
|
with self.lock:
|
|
row = self.connection.execute("SELECT * FROM students WHERE id=?", (student_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def student_membership(self, student_id: int) -> dict[str, Any] | None:
|
|
with self.lock:
|
|
row = self.connection.execute(
|
|
"""SELECT m.group_id,g.name,g.class_name,g.leader_student_id
|
|
FROM group_members m JOIN groups g ON g.id=m.group_id
|
|
WHERE m.student_id=?""",
|
|
(student_id,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|