rbac/wwwroot/user/up_login.dspy

debug(f'{params_kw=}')
ns = {
	"username":params_kw.username,
	"password":password_encode(params_kw.password)
}

async def _audit_login(sor, user_id, username, action, result='ok'):
	"""登录审计：检查系统是否有审计模块，有则写入。"""
	try:
		from app_audit import audit_log
	except ImportError:
		return
	try:
		ip = request.get('client_ip', '') if hasattr(request, 'get') else ''
		await audit_log(sor, user_id, username, action, result=result, client_ip=ip)
	except Exception:
		pass

info(f'{ns=}')
db = DBPools()
dbname = get_module_dbname('rbac')
async with db.sqlorContext(dbname) as sor:
	# Get user record with lockout fields (standard SQL)
	r = await sor.sqlExe('select * from users where username=${username}$', ns.copy())
	if len(r) == 0:
		await _audit_login(sor, None, ns['username'], 'login_fail', 'fail')
		return {
			"status": "error",
			"data": {
				"message": "user name or password error"
			},
			"widgettype":"Error",
			"options":{
				"timeout":3,
				"title":"Login Error",
				"message":"user name or password error"
			}
		}
	user = r[0]
	
	# Lockout check in Python (DB-agnostic)
	fail_count = getattr(user, 'login_fail_count', 0) or 0
	last_fail = getattr(user, 'last_login_fail', None)
	if fail_count >= 3 and last_fail:
		try:
			stored = datetime.datetime.strptime(str(last_fail), '%Y-%m-%d %H:%M:%S')
			elapsed = (datetime.datetime.now() - stored).total_seconds()
			if elapsed < 300:
				await _audit_login(sor, user.id, ns['username'], 'login_fail', 'locked')
				return {
				"status": "error",
				"data": {
					"message": "Account locked due to too many failed login attempts. Please try again in 5 minutes."
				},
					"widgettype":"Error",
					"options":{
						"timeout":5,
						"title":"Account Locked",
						"message":"Account locked due to too many failed login attempts. Please try again in 5 minutes."
					}
				}
			else:
				# Lockout expired, reset in background (best-effort)
				await sor.sqlExe("""
					UPDATE users SET login_fail_count = 0, last_login_fail = NULL
					WHERE id = ${id}$
				""", {'id': user.id})
		except (ValueError, TypeError):
			pass
	
	# Verify password
	r = await sor.sqlExe('select * from users where username=${username}$ and password=${password}$', ns.copy())
	if len(r) == 0:
		# 5分钟窗口：超过5分钟前的失败次数默认为0（原子 SQL）
		now_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
		stale_before = (datetime.datetime.now() - datetime.timedelta(seconds=300)).strftime('%Y-%m-%d %H:%M:%S')
		await sor.sqlExe("""
			UPDATE users 
			SET login_fail_count = CASE WHEN last_login_fail IS NULL OR last_login_fail < ${stale}$ THEN 1 ELSE login_fail_count + 1 END,
			    last_login_fail = ${now}$
			WHERE id = ${id}$
		""", {'id': user.id, 'now': now_str, 'stale': stale_before})
		await _audit_login(sor, user.id, ns['username'], 'login_fail', 'fail')
		
		new_fail_count = fail_count + 1
		if new_fail_count >= 3:
			msg = "Too many failed attempts. Account locked for 5 minutes."
		else:
			msg = f"user name or password error ({3 - new_fail_count} attempts remaining)"
		return {
			"status": "error",
			"data": {
				"message": msg
			},
			"widgettype":"Error",
			"options":{
				"timeout":3,
				"title":"Login Error",
				"message": msg
			}
		}
	# Success -- atomic reset
	now_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
	await sor.sqlExe("""
		UPDATE users 
		SET login_fail_count = 0, last_login_fail = NULL,
		    last_login = ${now}$
		WHERE id = ${id}$
	""", {'id': user.id, 'now': now_str})
	await remember_user(r[0].id, username=r[0].username, userorgid=r[0].orgid)
	await _audit_login(sor, r[0].id, r[0].username, 'login', 'ok')
	login_source = params_kw.get('login_source', 'manual')
	msg = {
		"status": "ok",
		"data": { "user": r[0]},
		"widgettype":"Message",
		"options":{
			"timeout":3,
			"anchor": "cc",
			"auto_open":True,
			"title":"Login",
			"message":f"{r[0].username} Welcome back"
		},
		"binds":[
		{
			"wid":"self",
			"event":"opened",
			"actiontype":"script",
			"target":"window.login_window",
			"script":"this.destroy()"
		}
		]
	}
	if login_source == 'auto':
		# 非用户主动点击登录(如401自动弹窗)：不刷新页面，标记不刷新
		msg["options"]["refresh"] = False
	else:
		# 用户主动点击登录：登录成功后刷新页面(跳转/index.ui)
		msg["binds"].insert(0, {
			"wid":"self",
			"event":"dismissed",
			"actiontype":"urlwidget",
			"target":"window",
			"options":{
				"url":entire_url('/index.ui')
			}
		})
	return msg
return {
			"status": "error",
			"data": {
				"message": "system error"
			},

	"widgettype":"Error",
	"options":{
		"timeout":3,
		"title":"Login Error",
		"message":"system error"
	}
}