yumoqing/kboss
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of git.opencomputing.cn:yumoqing/kboss

Browse Source
This commit is contained in:
hrx 2026-04-29 09:36:49 +08:00
commit bab0ecce51
2 changed files with 195 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
b/user/loginUser.dspy
View File

@ -140,7 +140,7 @@ async def loginUser(ns):
ns['password'] = back_text.decode('utf-8')
type1 = 0
if ns.get('username') != 'admin':
if ns.get('username') != 'admin' and ns.get('username') != '开元云(北京)科技有限公司':
# 这里用户名密码登录也需要手机号，放开下面代码
# logincode = await sor.R('params',{'pname':'login_message','pvalue':'0','del_flg':'0'})
# if len(logincode) >= 1:
@ -153,6 +153,10 @@ async def loginUser(ns):
if type1 == 1:
# 手机号验证码登录
userreacs = await sor.R('users', {'mobile': ns.get('username')})
if not userreacs:
userreacs = await sor.R('users', {'username': ns.get('username')})
# 如果是微信扫码后绑定已有账号
if ns.get('wechat_openid'):
if userreacs:
@ -166,7 +170,7 @@ async def loginUser(ns):
else:
ns['password'] = password_encode(ns['password'])
ns['del_flg'] = '0'
userreacs = await sor.R('users', ns)
userreacs = await sor.R('users', {'username': ns.get('username'), 'password': ns.get('password')})
if len(userreacs) >= 1:
type += 1
await remember_user(userreacs[0]['id'], username=userreacs[0]['username'], userorgid=userreacs[0]['orgid'])

230
b/user/logintype.dspy
View File

@ -1,3 +1,98 @@
async def handle_login_failed(user_name):
from datetime import datetime, timedelta
db = DBPools()
async with db.sqlorContext('kboss') as sor:
try:
current_time = datetime.now()
# 1. 查询该用户当前的失败记录
record = await sor.sqlExe("SELECT first_failed_time, failed_count FROM login_fail_log WHERE user_name = '%s'" % user_name, {})
if not record:
# === 第一次失败 ===
await sor.sqlExe("""
INSERT INTO login_fail_log (user_name, first_failed_time, failed_count)
VALUES ('%s', '%s', 1)
""" % (user_name, current_time), {})
return # 正常失败，不需要等待
first_failed_time = record[0]['first_failed_time']
failed_count = record[0]['failed_count']
first_failed_time = datetime.strptime(first_failed_time, '%Y-%m-%d %H:%M:%S')
# 2. 判断是否还在10分钟窗口内
time_diff = (current_time - first_failed_time).total_seconds() / 60 # 分钟
if time_diff <= 10:
# 还在10分钟窗口内，失败次数+1
new_count = failed_count + 1
if new_count >= 3:
# === 第三次及以上失败：触发锁定 ===
# 更新失败次数，并记录锁定开始时间
await sor.sqlExe("""
UPDATE login_fail_log
SET failed_count = 3, last_failed_time = '%s', lock_until = '%s'
WHERE user_name = '%s'
""" % (current_time, current_time + timedelta(minutes=20), user_name), {})
print("十分钟内登录失败三次，请等待20分钟后再试")
else:
# 第二次失败
await sor.sqlExe("""
UPDATE login_fail_log
SET failed_count = '%s', last_failed_time = '%s'
WHERE user_name = '%s'
""" % (new_count, current_time, user_name), {})
# 可选：返回还剩几次机会
if new_count == 2:
print("登录失败，还剩1次机会（10分钟内）")
else:
# === 已经超过10分钟窗口，重置计数器 ===
await sor.sqlExe("""
UPDATE login_fail_log
SET first_failed_time = '%s', failed_count = 1, last_failed_time = '%s', lock_until = NULL
WHERE user_name = '%s'
""" % (current_time, current_time, user_name), {})
except Exception as e:
print("exception:", user_name, e)
return {'status': False, 'msg': '登录操作失败', 'error': str(e)}
async def check_login_allowed(user_name):
from datetime import datetime, timedelta
db = DBPools()
async with db.sqlorContext('kboss') as sor:
try:
record = await sor.sqlExe("SELECT lock_until, first_failed_time, failed_count FROM login_fail_log WHERE user_name = '%s'" % user_name, {})
current_time = datetime.now()
if not record:
return {'status': True} # 允许登录
record = record[0]
if record.get('lock_until'):
record['lock_until'] = datetime.strptime(record['lock_until'], '%Y-%m-%d %H:%M:%S')
# 检查是否处于锁定状态
if record.get('lock_until') and record['lock_until'] > current_time:
wait_minutes = (record['lock_until'] - current_time).total_seconds() // 60
msg = f"十分钟内登录失败三次，请等待{int(wait_minutes)}分钟后再试"
return {
'status': False,
'msg': msg
}
# 如果锁定时间已过，但还在检查窗口，也可以重置（可选）
return {'status': True} # 允许登录
except Exception as e:
print("exception:", user_name, e)
return {'status': False, 'msg': '登录操作失败, %s' % str(e)}
async def logintype(ns):
"""
1、判断用户是否为主级（如果在reseller没要找到数据，证明就是主级）
@ -43,57 +138,110 @@ async def logintype(ns):
db = DBPools()
async with db.sqlorContext('kboss') as sor:
type = 0
if ns.get('codeid'):
type += 1
domain_name = ns.get('domain_name')
if domain_name in ['www.opencomputing.cn', 'dev.opencomputing.cn', 'localhost:9527'] and ns.get('username') not in ['开元云(北京)科技有限公司', 'admin', 'kyy_root']:
# 登录失败次数限制
login_allowed = await check_login_allowed(ns.get('username'))
if not login_allowed.get('status'):
return {'status': False, 'msg': login_allowed.get('msg')}
if not ns.get('mobile'):
return {
'status': False,
'msg': '请输入手机号'
}
real_mobile_li = await sor.R('users', {'username': ns['username']})
if not real_mobile_li:
return {
'status': False,
'msg': '用户名有误'
}
real_mobile = real_mobile_li[0]['mobile'] if real_mobile_li else None
if not real_mobile:
return {
'status': False,
'msg': '未查询到匹配得手机号码'
}
if real_mobile and ns['mobile'] != real_mobile:
return {
'status': False,
'msg': '您的手机号与用户名不匹配'
}
if not ns.get('codeid'):
return {
'status': False,
'msg': 'codeid不存在, 请输入验证码'
}
code = await sor.R('validatecode', {'id': ns.get('codeid'), 'vcode': ns.get('vcode')})
if len(code) < 1:
await handle_login_failed(ns.get('username'))
return {'status': False, 'msg': '验证码不正确'}
if type == 1:
# 手机号登录
users = await sor.R('users', {'mobile': ns.get('username')})
else:
password = password_encode(ns['password'])
users = await sor.R('users', {'username': ns.get('username'), 'password': password})
if len(users) < 1:
return {"status": False,'msg':'用户名或密码错误'}
elif ns.get('username') == "admin":
if len(users) < 1:
await handle_login_failed(ns.get('username'))
return {"status": False,'msg':'用户名或密码错误'}
return {'status': True}
else:
reseller = await sor.R('reseller', {'domain_name': ns.get('domain_name')})
# 查到用户的所在机构
user_org = await sor.R('organization', {'id': users[0]['orgid']})
#0代表用户为主 1代表用户为子
user_type = 0
# 子页面判断用户
if len(reseller) >= 1:
if reseller[0]['orgid'] == user_org[0]['parentid']:
user_type += 1
if reseller[0]['orgid'] == users[0]['orgid']:
user_type += 1
# 主页面判断用户
type = 0
if ns.get('codeid'):
type += 1
code = await sor.R('validatecode', {'id': ns.get('codeid'), 'vcode': ns.get('vcode')})
if len(code) < 1:
return {'status': False, 'msg': '验证码不正确'}
if type == 1:
# 手机号登录
users = await sor.R('users', {'mobile': ns.get('username')})
else:
resellers = await sor.R('reseller', {'orgid': user_org[0]['parentid']})
if len(resellers) >= 1:
user_type += 1
if users[0]['user_reseller'] == '1':
user_type += 1
# 证明是主级页面
if len(reseller) < 1:
# 用户为主
if user_type == 0:
return {'status': True}
password = password_encode(ns['password'])
users = await sor.R('users', {'username': ns.get('username'), 'password': password})
if len(users) < 1:
await handle_login_failed(ns.get('username'))
return {"status": False,'msg':'用户名或密码错误'}
elif ns.get('username') == "admin":
return {'status': True}
else:
reseller = await sor.R('reseller', {'domain_name': ns.get('domain_name')})
# 查到用户的所在机构
user_org = await sor.R('organization', {'id': users[0]['orgid']})
#0代表用户为主 1代表用户为子
user_type = 0
# 子页面判断用户
if len(reseller) >= 1:
if reseller[0]['orgid'] == user_org[0]['parentid']:
user_type += 1
if reseller[0]['orgid'] == users[0]['orgid']:
user_type += 1
# 主页面判断用户
else:
resellers = await sor.R('reseller', {'orgid': user_org[0]['parentid']})
if len(resellers) >= 1:
user_type += 1
if users[0]['user_reseller'] == '1':
user_type += 1
# 证明是主级页面
if len(reseller) < 1:
# 用户为主
if user_type == 0:
return {'status': True}
else:
# 用户为子
return {'status': False,'msg':'用户名或密码错误'}
# 子级页面
else:
# 用户为主
if user_type == 0:
return {'status': False,'msg':'用户名或密码错误'}
# 用户为子
return {'status': False,'msg':'用户名或密码错误'}
# 子级页面
else:
# 用户为主
if user_type == 0:
return {'status': False,'msg':'用户名或密码错误'}
# 用户为子
else:
return {'status': True}
else:
return {'status': True}
ret = await logintype(params_kw)
return ret