施工(HEL-226): 实现登录门户与本机免密切换账号
用设备 Cookie 和授权表记住本机已验证账号,登录页按确认样图做成门户,不再把密码写进浏览器。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
co-authored by
Cursor
multica-agent
parent
8a5e78f022
commit
f0a1adf52f
@@ -77,15 +77,22 @@ class AccountService:
|
||||
access = self.access_supplier() or self.database.user_access(self.current_user_id) or {}
|
||||
return self.membership_for_access(access)
|
||||
|
||||
def register(self, username: str, password: str) -> dict[str, Any]:
|
||||
GRANT_SLIDE_DAYS = 30
|
||||
GRANT_HARD_DAYS = 180
|
||||
MAX_GRANTS_PER_DEVICE = 5
|
||||
SWITCH_REAUTH_MESSAGE = "该账号需重新验证"
|
||||
|
||||
def register(self, username: str, password: str, device_hash: str = "") -> dict[str, Any]:
|
||||
username = username.strip()
|
||||
self.validate_input(username, password)
|
||||
with self.auth_lock:
|
||||
salt, password_digest = hash_password(password)
|
||||
user = self.database.create_user(username, salt, password_digest)
|
||||
return self.create_session(user)
|
||||
result = self.create_session(user)
|
||||
self.remember_account(device_hash, int(user["id"]), fresh=True)
|
||||
return result
|
||||
|
||||
def login(self, username: str, password: str) -> dict[str, Any]:
|
||||
def login(self, username: str, password: str, device_hash: str = "") -> dict[str, Any]:
|
||||
username = username.strip()
|
||||
if not username or not password:
|
||||
raise ValueError("账号名和密码不能为空。")
|
||||
@@ -96,7 +103,9 @@ class AccountService:
|
||||
str(user.get("password_hash") or ""),
|
||||
):
|
||||
raise ValueError("账号名或密码不正确。")
|
||||
return self.create_session(user)
|
||||
result = self.create_session(user)
|
||||
self.remember_account(device_hash, int(user["id"]), fresh=True)
|
||||
return result
|
||||
|
||||
def change_password(self, current_password: str, new_password: str) -> None:
|
||||
current_password = str(current_password or "")
|
||||
@@ -112,6 +121,86 @@ class AccountService:
|
||||
salt, digest = hash_password(new_password)
|
||||
if not self.database.update_user_password(self.current_user_id, salt, digest):
|
||||
raise ValueError("账号不存在。")
|
||||
self.database.delete_switch_grants_for_user(self.current_user_id)
|
||||
|
||||
@staticmethod
|
||||
def _utc_now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
@classmethod
|
||||
def _iso(cls, value: datetime) -> str:
|
||||
return value.isoformat(timespec="seconds")
|
||||
|
||||
def remember_account(self, device_hash: str, user_id: int, *, fresh: bool = False) -> None:
|
||||
if not device_hash or user_id <= 0:
|
||||
return
|
||||
now = self._utc_now()
|
||||
now_text = self._iso(now)
|
||||
self.database.cleanup_expired_switch_grants(now_text)
|
||||
existing = None if fresh else self.database.get_switch_grant(device_hash, user_id)
|
||||
granted_at = parse_iso_datetime(existing["granted_at"]) if existing else now
|
||||
if granted_at is None:
|
||||
granted_at = now
|
||||
expires = min(
|
||||
now + timedelta(days=self.GRANT_SLIDE_DAYS),
|
||||
granted_at + timedelta(days=self.GRANT_HARD_DAYS),
|
||||
)
|
||||
if not existing:
|
||||
self.database.prune_switch_grants(device_hash, self.MAX_GRANTS_PER_DEVICE - 1)
|
||||
self.database.upsert_switch_grant(
|
||||
device_hash,
|
||||
user_id,
|
||||
self._iso(granted_at),
|
||||
now_text,
|
||||
self._iso(expires),
|
||||
)
|
||||
|
||||
def list_device_accounts(
|
||||
self, device_hash: str, current_user_id: int | None = None
|
||||
) -> dict[str, Any]:
|
||||
if not device_hash:
|
||||
return {"accounts": [], "current_user_id": current_user_id}
|
||||
now_text = self._iso(self._utc_now())
|
||||
self.database.cleanup_expired_switch_grants(now_text)
|
||||
accounts = []
|
||||
for row in self.database.list_switch_grants(device_hash, now_text):
|
||||
accounts.append(
|
||||
{
|
||||
"user_id": int(row["id"]),
|
||||
"username": str(row["username"]),
|
||||
"role": str(row.get("role") or "user"),
|
||||
"membership": self.membership_for_access(row),
|
||||
"last_used_at": str(row.get("last_used_at") or ""),
|
||||
}
|
||||
)
|
||||
return {"accounts": accounts, "current_user_id": current_user_id}
|
||||
|
||||
def switch_account(self, device_hash: str, user_id: int) -> dict[str, Any]:
|
||||
if not device_hash or user_id <= 0:
|
||||
raise PermissionError(self.SWITCH_REAUTH_MESSAGE)
|
||||
now = self._utc_now()
|
||||
now_text = self._iso(now)
|
||||
self.database.cleanup_expired_switch_grants(now_text)
|
||||
grant = self.database.get_switch_grant(device_hash, user_id)
|
||||
expires = parse_iso_datetime(grant.get("expires_at")) if grant else None
|
||||
if not grant or not expires or expires <= now:
|
||||
if grant:
|
||||
self.database.delete_switch_grant(device_hash, user_id)
|
||||
raise PermissionError(self.SWITCH_REAUTH_MESSAGE)
|
||||
user = self.database.user_access(user_id)
|
||||
if not user:
|
||||
raise PermissionError(self.SWITCH_REAUTH_MESSAGE)
|
||||
result = self.create_session(user)
|
||||
self.remember_account(device_hash, user_id)
|
||||
return result
|
||||
|
||||
def forget_account(self, device_hash: str, user_id: int) -> None:
|
||||
if device_hash and user_id > 0:
|
||||
self.database.delete_switch_grant(device_hash, user_id)
|
||||
|
||||
def revoke_current_device_grant(self, device_hash: str, user_id: int) -> None:
|
||||
if device_hash and user_id > 0:
|
||||
self.database.delete_switch_grant(device_hash, user_id)
|
||||
|
||||
def create_session(self, user: dict[str, Any]) -> dict[str, Any]:
|
||||
session_token = secrets.token_urlsafe(32)
|
||||
|
||||
Reference in New Issue
Block a user