mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-09-17 23:17:40 +02:00
jose
4936a99e17
2.The page is displayed when the adjustment panel is wrong 3.App store add common plug-in display 4.Adjust the sessionid name to a non-fixed name 5.Panel CSRF defense mechanism covers panel websocket communication 6.Adjust the font size of the panel list 7.Adjust the panel pop-up window (add a close button to cancel the automatic closing time) 8.Optimize the front/back end of the file manager 9.Add the entrance of the MailServer Rspamd 10.Refactor the debug module 11.Other known bug fixes
333 lines
16 KiB
Python
333 lines
16 KiB
Python
#coding: utf-8
|
|
# +-------------------------------------------------------------------
|
|
# | 宝塔Linux面板
|
|
# +-------------------------------------------------------------------
|
|
# | Copyright (c) 2015-2099 宝塔软件(http:#bt.cn) All rights reserved.
|
|
# +-------------------------------------------------------------------
|
|
# | Author: hwliang <hwl@bt.cn>
|
|
# +-------------------------------------------------------------------
|
|
|
|
import public,os,sys,db,time,json,re
|
|
from BTPanel import session,cache,json_header
|
|
from flask import request,redirect,g
|
|
|
|
class userlogin:
|
|
|
|
def request_post(self,post):
|
|
if not hasattr(post, 'username') or not hasattr(post, 'password'):
|
|
return public.returnJson(False,'LOGIN_USER_EMPTY'),json_header
|
|
|
|
self.error_num(False)
|
|
if self.limit_address('?') < 1: return public.returnJson(False,'LOGIN_ERR_LIMIT'),json_header
|
|
post.username = post.username.strip()
|
|
|
|
public.chdck_salt()
|
|
sql = db.Sql()
|
|
user_list = sql.table('users').field('id,username,password,salt').select()
|
|
userInfo = None
|
|
for u_info in user_list:
|
|
if public.md5(u_info['username']) == post.username:
|
|
userInfo = u_info
|
|
if 'code' in session:
|
|
if session['code'] and not 'is_verify_password' in session:
|
|
if not hasattr(post, 'code'): return public.returnJson(False,'Verification code can not be empty!'),json_header
|
|
if not public.checkCode(post.code):
|
|
public.WriteLog('TYPE_LOGIN','LOGIN_ERR_CODE',('****','****',public.GetClientIp()))
|
|
return public.returnJson(False,'CODE_ERR'),json_header
|
|
try:
|
|
if not userInfo['salt']:
|
|
public.chdck_salt()
|
|
userInfo = sql.table('users').where('id=?',(userInfo['id'],)).field('id,username,password,salt').find()
|
|
|
|
password = public.md5(post.password.strip() + userInfo['salt'])
|
|
if public.md5(userInfo['username']) != post.username or userInfo['password'] != password:
|
|
public.WriteLog('TYPE_LOGIN','LOGIN_ERR_PASS',('****','******',public.GetClientIp()))
|
|
num = self.limit_address('+')
|
|
return public.returnJson(False,'LOGIN_USER_ERR',(str(num),)),json_header
|
|
_key_file = "/www/server/panel/data/two_step_auth.txt"
|
|
#登陆告警
|
|
public.login_send_body("Userinfo",userInfo['username'],public.GetClientIp(),str(request.environ.get('REMOTE_PORT')))
|
|
if hasattr(post,'vcode'):
|
|
if self.limit_address('?',v="vcode") < 1: return public.returnJson(False,'You have failed verification many times, forbidden for 10 minutes'),json_header
|
|
import pyotp
|
|
secret_key = public.readFile(_key_file)
|
|
if not secret_key:
|
|
return public.returnJson(False, "Did not find the key, please close Google verification on the command line and trun on again"),json_header
|
|
t = pyotp.TOTP(secret_key)
|
|
result = t.verify(post.vcode)
|
|
if not result:
|
|
if public.sync_date(): result = t.verify(post.vcode)
|
|
if not result:
|
|
num = self.limit_address('++',v="vcode")
|
|
return public.returnJson(False, 'Invalid Verification code. You have [{}] times left to try!'.format(num)), json_header
|
|
now = int(time.time())
|
|
public.writeFile("/www/server/panel/data/dont_vcode_ip.txt",json.dumps({"client_ip":public.GetClientIp(),"add_time":now}))
|
|
self.limit_address('--',v="vcode")
|
|
self.set_cdn_host(post)
|
|
return self._set_login_session(userInfo)
|
|
|
|
acc_client_ip = self.check_two_step_auth()
|
|
|
|
if not os.path.exists(_key_file) or acc_client_ip:
|
|
self.set_cdn_host(post)
|
|
return self._set_login_session(userInfo)
|
|
self.limit_address('-')
|
|
session['is_verify_password'] = True
|
|
return "1"
|
|
except Exception as ex:
|
|
stringEx = str(ex)
|
|
if stringEx.find('unsupported') != -1 or stringEx.find('-1') != -1:
|
|
public.ExecShell("rm -f /tmp/sess_*")
|
|
public.ExecShell("rm -f /www/wwwlogs/*log")
|
|
public.ServiceReload()
|
|
return public.returnJson(False,'USER_INODE_ERR'),json_header
|
|
public.WriteLog('TYPE_LOGIN','LOGIN_ERR_PASS',('****','******',public.GetClientIp()))
|
|
num = self.limit_address('+')
|
|
return public.returnJson(False,'LOGIN_USER_ERR',(str(num),)),json_header
|
|
|
|
def request_tmp(self,get):
|
|
try:
|
|
if not hasattr(get,'tmp_token'): return public.returnJson(False,'INIT_ARGS_ERR'),json_header
|
|
if len(get.tmp_token) == 48:
|
|
return self.request_temp(get)
|
|
if len(get.tmp_token) != 64: return public.returnJson(False,'INIT_ARGS_ERR'),json_header
|
|
if not re.match(r"^\w+$",get.tmp_token):return public.returnJson(False,'INIT_ARGS_ERR'),json_header
|
|
save_path = '/www/server/panel/config/api.json'
|
|
data = json.loads(public.ReadFile(save_path))
|
|
if not 'tmp_token' in data or not 'tmp_time' in data: return public.returnJson(False,'VERIFICATION_FAILED'),json_header
|
|
if (time.time() - data['tmp_time']) > 120: return public.returnJson(False,'EXPIRED_TOKEN'),json_header
|
|
if get.tmp_token != data['tmp_token']: return public.returnJson(False,'INIT_TOKEN_ERR'),json_header
|
|
userInfo = public.M('users').where("id=?",(1,)).field('id,username').find()
|
|
session['login'] = True
|
|
session['username'] = userInfo['username']
|
|
session['tmp_login'] = True
|
|
session['uid'] = userInfo['id']
|
|
public.WriteLog('TYPE_LOGIN','LOGIN_SUCCESS',(userInfo['username'],public.GetClientIp()+ ":" + str(request.environ.get('REMOTE_PORT'))))
|
|
self.limit_address('-')
|
|
cache.delete('panelNum')
|
|
cache.delete('dologin')
|
|
session['session_timeout'] = time.time() + public.get_session_timeout()
|
|
del(data['tmp_token'])
|
|
del(data['tmp_time'])
|
|
public.writeFile(save_path,json.dumps(data))
|
|
self.set_request_token()
|
|
self.login_token()
|
|
self.set_cdn_host(get)
|
|
return redirect('/')
|
|
except:
|
|
return public.returnJson(False,'Login failed,' + public.get_error_info()),json_header
|
|
|
|
|
|
def request_temp(self,get):
|
|
try:
|
|
if len(get.__dict__.keys()) > 2: return public.getMsg('INIT_ARGS_ERR')
|
|
if not hasattr(get,'tmp_token'): return public.getMsg('INIT_ARGS_ERR')
|
|
if len(get.tmp_token) != 48: return public.getMsg('INIT_ARGS_ERR')
|
|
if not re.match(r"^\w+$",get.tmp_token):return public.getMsg('INIT_ARGS_ERR')
|
|
skey = public.GetClientIp() + '_temp_login'
|
|
if not public.get_error_num(skey,10): return public.getMsg('AUTH_FAILED')
|
|
s_time = int(time.time())
|
|
data = public.M('temp_login').where('state=? and expire>?',(0,s_time)).field('id,token,salt,expire').find()
|
|
if not data:
|
|
public.set_error_num(skey)
|
|
return public.getMsg('VERIFICATION_FAILED')
|
|
if not isinstance(data,dict):
|
|
public.set_error_num(skey)
|
|
return public.getMsg('VERIFICATION_FAILED')
|
|
r_token = public.md5(get.tmp_token + data['salt'])
|
|
if r_token != data['token']:
|
|
public.set_error_num(skey)
|
|
return public.getMsg('VERIFICATION_FAILED')
|
|
public.set_error_num(skey,True)
|
|
userInfo = public.M('users').where("id=?",(1,)).field('id,username').find()
|
|
session['login'] = True
|
|
session['username'] = public.getMsg('TEMPORARY_ID',(data['id'],))
|
|
session['tmp_login'] = True
|
|
session['tmp_login_id'] = str(data['id'])
|
|
session['tmp_login_expire'] = time.time() + 3600
|
|
session['uid'] = data['id']
|
|
sess_path = 'data/session'
|
|
if not os.path.exists(sess_path):
|
|
os.makedirs(sess_path,384)
|
|
public.writeFile(sess_path + '/' + str(data['id']),'')
|
|
login_addr = public.GetClientIp()+ ":" + str(request.environ.get('REMOTE_PORT'))
|
|
public.WriteLog('TYPE_LOGIN','LOGIN_SUCCESS',(userInfo['username'],login_addr))
|
|
public.M('temp_login').where('id=?',(data['id'],)).update({"login_time":s_time,'state':1,'login_addr':login_addr})
|
|
self.limit_address('-')
|
|
cache.delete('panelNum')
|
|
cache.delete('dologin')
|
|
session['session_timeout'] = time.time() + public.get_session_timeout()
|
|
self.set_request_token()
|
|
self.login_token()
|
|
self.set_cdn_host(get)
|
|
public.login_send_body("Temporary authorization",userInfo['username'],public.GetClientIp(),str(request.environ.get('REMOTE_PORT')))
|
|
return redirect('/')
|
|
except:
|
|
return public.getMsg('LOGIN_FAIL')
|
|
|
|
|
|
def login_token(self):
|
|
import config
|
|
config.config().reload_session()
|
|
|
|
def request_get(self,get):
|
|
#if os.path.exists('/www/server/panel/install.pl'): raise redirect('/install');
|
|
if not 'title' in session: session['title'] = public.getMsg('NAME')
|
|
domain = public.readFile('data/domain.conf')
|
|
|
|
if domain:
|
|
if(public.GetHost().lower() != domain.strip().lower()):
|
|
errorStr = public.ReadFile('./BTPanel/templates/' + public.GetConfigValue('template') + '/error2.html')
|
|
try:
|
|
errorStr = errorStr.format(public.getMsg('PAGE_ERR_TITLE'),public.getMsg('PAGE_ERR_DOMAIN_H1'),public.getMsg('PAGE_ERR_DOMAIN_P1'),public.getMsg('PAGE_ERR_DOMAIN_P2'),public.getMsg('PAGE_ERR_DOMAIN_P3'),public.getMsg('NAME'),public.getMsg('PAGE_ERR_HELP'))
|
|
except IndexError:pass
|
|
return errorStr
|
|
if os.path.exists('data/limitip.conf'):
|
|
iplist = public.readFile('data/limitip.conf')
|
|
if iplist:
|
|
iplist = iplist.strip()
|
|
if not public.GetClientIp() in iplist.split(','):
|
|
errorStr = public.ReadFile('./BTPanel/templates/' + public.GetConfigValue('template') + '/error2.html')
|
|
try:
|
|
errorStr = errorStr.format(public.getMsg('PAGE_ERR_TITLE'),public.getMsg('PAGE_ERR_IP_H1'),public.getMsg('PAGE_ERR_IP_P1',(public.GetClientIp(),)),public.getMsg('PAGE_ERR_IP_P2'),public.getMsg('PAGE_ERR_IP_P3'),public.getMsg('NAME'),public.getMsg('PAGE_ERR_HELP'))
|
|
except IndexError:pass
|
|
return errorStr
|
|
|
|
if 'login' in session:
|
|
if session['login'] == True:
|
|
return redirect('/')
|
|
|
|
if not 'code' in session:
|
|
session['code'] = False
|
|
self.error_num(False)
|
|
|
|
#生成request_token
|
|
def set_request_token(self):
|
|
session['request_token_head'] = public.GetRandomString(48)
|
|
|
|
def set_cdn_host(self,get):
|
|
try:
|
|
if not 'cdn_url' in get: return True
|
|
plugin_path = 'plugin/static_cdn'
|
|
if not os.path.exists(plugin_path): return True
|
|
cdn_url = public.get_cdn_url()
|
|
if not cdn_url or cdn_url == get.cdn_url: return True
|
|
public.set_cdn_url(get.cdn_url)
|
|
except:
|
|
return False
|
|
|
|
#防暴破
|
|
def error_num(self,s = True):
|
|
nKey = 'panelNum'
|
|
num = cache.get(nKey)
|
|
if not num:
|
|
cache.set(nKey,1)
|
|
num = 1
|
|
if s: cache.inc(nKey,1)
|
|
if num > 6: session['code'] = True
|
|
|
|
#IP限制
|
|
def limit_address(self,type,v=""):
|
|
import time
|
|
clientIp = public.GetClientIp()
|
|
numKey = 'limitIpNum_' + v + clientIp
|
|
limit = 6
|
|
outTime = 600
|
|
try:
|
|
#初始化
|
|
num1 = cache.get(numKey)
|
|
if not num1:
|
|
cache.set(numKey,1,outTime)
|
|
num1 = 1
|
|
|
|
#计数
|
|
if type == '+':
|
|
cache.inc(numKey,1)
|
|
self.error_num()
|
|
session['code'] = True
|
|
return limit - (num1+1)
|
|
|
|
#计数验证器
|
|
if type == '++':
|
|
cache.inc(numKey,1)
|
|
self.error_num()
|
|
session['code'] = False
|
|
return limit - (num1+1)
|
|
|
|
#清空
|
|
if type == '-':
|
|
cache.delete(numKey)
|
|
session['code'] = False
|
|
return 1
|
|
|
|
#清空验证器
|
|
if type == '--':
|
|
cache.delete(numKey)
|
|
session['code'] = False
|
|
return 1
|
|
return limit - num1
|
|
except:
|
|
return limit
|
|
|
|
# 登录成功设置session
|
|
def _set_login_session(self,userInfo):
|
|
try:
|
|
session['login'] = True
|
|
session['username'] = userInfo['username']
|
|
session['uid'] = userInfo['id']
|
|
session['login_user_agent'] = public.md5(request.headers.get('User-Agent',''))
|
|
public.WriteLog('TYPE_LOGIN','LOGIN_SUCCESS',(userInfo['username'],public.GetClientIp()+ ":" + str(request.environ.get('REMOTE_PORT'))))
|
|
self.limit_address('-')
|
|
cache.delete('panelNum')
|
|
cache.delete('dologin')
|
|
session['session_timeout'] = time.time() + public.get_session_timeout()
|
|
self.set_request_token()
|
|
self.login_token()
|
|
login_type = 'data/app_login.pl'
|
|
if os.path.exists(login_type):
|
|
os.remove(login_type)
|
|
return public.returnJson(True,'LOGIN_SUCCESS'),json_header
|
|
except Exception as ex:
|
|
stringEx = str(ex)
|
|
if stringEx.find('unsupported') != -1 or stringEx.find('-1') != -1:
|
|
public.ExecShell("rm -f /tmp/sess_*")
|
|
public.ExecShell("rm -f /www/wwwlogs/*log")
|
|
public.ServiceReload()
|
|
return public.returnJson(False,'USER_INODE_ERR'),json_header
|
|
public.WriteLog('TYPE_LOGIN','LOGIN_ERR_PASS',('****','******',public.GetClientIp()))
|
|
num = self.limit_address('+')
|
|
return public.returnJson(False,'LOGIN_USER_ERR',(str(num),)),json_header
|
|
|
|
|
|
# 检查是否需要进行二次验证
|
|
def check_two_step_auth(self):
|
|
dont_vcode_ip_info = public.readFile("/www/server/panel/data/dont_vcode_ip.txt")
|
|
acc_client_ip = False
|
|
if dont_vcode_ip_info:
|
|
dont_vcode_ip_info = json.loads(dont_vcode_ip_info)
|
|
ip = dont_vcode_ip_info["client_ip"] == public.GetClientIp()
|
|
now = int(time.time())
|
|
v_time = now - int(dont_vcode_ip_info["add_time"])
|
|
if ip and v_time < 86400:
|
|
acc_client_ip = True
|
|
return acc_client_ip
|
|
|
|
# 清理多余SESSION数据
|
|
def clear_session(self):
|
|
try:
|
|
session_file = '/dev/shm/session.db'
|
|
if not os.path.exists(session_file): return False
|
|
s_size = os.path.getsize(session_file)
|
|
if s_size < 1024 * 512: return False
|
|
if s_size > 1024 * 1024 * 10:
|
|
from BTPanel import sdb
|
|
if os.path.exists(session_file): os.remove(session_file)
|
|
sdb.create_all()
|
|
if not os.path.exists(session_file):
|
|
public.writeFile('/www/server/panel/data/reload.pl','True')
|
|
return False
|
|
return True
|
|
except:
|
|
return False
|
|
|