WEBDIENSTE/aaPanel
1
0
Fork 0
mirror of https://github.com/aaPanel/aaPanel.git synced 2026-08-17 21:25:47 +02:00
Code Issues Packages Projects Releases Wiki Activity

#Optimize APP login

Browse Source 
#Add XSS filtering in more places
This commit is contained in:
bt.cn
2021-10-16 14:48:55 +08:00
parent ed37b02fff
commit 8138179e68
11 changed files with 126 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
BTPanel/__init__.py
+20 -56
View File
@@ -14,8 +14,9 @@ import threading
import time
import re
import uuid
panel_path = '/www/server/panel'
if not os.name in ['nt']:
os.chdir('/www/server/panel')
os.chdir(panel_path)
if not 'class/' in sys.path:
sys.path.insert(0,'class/')
@@ -203,7 +204,7 @@ def error_404(e):
<head><title>404 Not Found</title></head>
<body>
<center><h1>404 Not Found</h1></center>
<hr><center>server</center>
<hr><center>nginx</center>
</body>
</html>'''
headers={
@@ -238,7 +239,7 @@ REQUEST_FORM: {request_form}
os_version = public.get_os_version()
)
result = public.readFile('/www/server/panel/BTPanel/templates/default/panel_error.html').format(error_title=error_info.split("\n")[-1],request_info = request_info,error_msg=error_info)
result = public.readFile(public.get_panel_path() + '/BTPanel/templates/default/panel_error.html').format(error_title=error_info.split("\n")[-1],request_info = request_info,error_msg=error_info)
return Resp(result,500)
# ===================================Flask HOOK========================#
@@ -1134,71 +1135,34 @@ def down(token=None,fname=None):
@app.route('/public',methods=method_all)
def panel_public():
#小程序控制接口
get = get_input()
try:
import panelWaf
panelWaf_data = panelWaf.panelWaf()
if panelWaf_data.is_sql(get.__dict__):return 'ERROR'
if panelWaf_data.is_xss(get.__dict__):return 'ERROR'
except:
pass
if len("{}".format(get.__dict__)) > 1024 * 32:
return 'ERROR'
get.client_ip = public.GetClientIp()
num_key = get.client_ip + '_wxapp'
if not public.get_error_num(num_key,10):
return public.returnMsg(False,'AUTH_FAILED')
if not hasattr(get,'name'): get.name = ''
if not hasattr(get,'fun'): return abort(404)
if not public.path_safe_check("%s/%s" % (get.name,get.fun)): return abort(404)
if get.fun in ['scan_login', 'login_qrcode', 'set_login', 'is_scan_ok', 'blind','static']:
if get.fun == 'static':
if not 'filename' in get: return abort(404)
if not public.path_safe_check("%s" % (get.filename)): return abort(404)
s_file = '/www/server/panel/BTPanel/static/' + get.filename
if s_file.find('..') != -1 or s_file.find('./') != -1: return abort(404)
if not os.path.exists(s_file): return abort(404)
return send_file(s_file, conditional=True, add_etags=True)
#检查是否验证过安全入口
if get.fun in ['login_qrcode','is_scan_ok']:
global admin_check_auth,admin_path,route_path,admin_path_file
if admin_path != '/bt' and os.path.exists(admin_path_file) and not 'admin_auth' in session:
return 'False'
if not public.get_error_num(num_key, 10):
return public.returnMsg(False, 'AUTH_FAILED')
if not hasattr(get, 'name'): get.name = ''
if not hasattr(get, 'fun'): return abort(403)
if not public.path_safe_check("%s/%s" % (get.name, get.fun)): return abort(403)
if get.fun in ['login_qrcode', 'is_scan_ok','set_login']:
# 检查是否验证过安全入口
global admin_check_auth, admin_path, route_path, admin_path_file
if admin_path != '/bt' and os.path.exists(admin_path_file) and not 'admin_auth' in session:
return abort(403)
#验证是否绑定了设备
if not get.fun in ['blind']:
if not public.check_app('app'):return public.returnMsg(False,'UNBOUND_USER')
if not public.check_app('app'):return public.returnMsg(False,'UNBOUND_USER')
import wxapp
pluwx = wxapp.wxapp()
checks = pluwx._check(get)
if type(checks) != bool or not checks:
public.set_error_num(num_key)
return public.getJson(checks),json_header
data = public.getJson(eval('pluwx.'+get.fun+'(get)'))
return data,json_header
return public.getJson(checks), json_header
data = public.getJson(eval('pluwx.' + get.fun + '(get)'))
return data, json_header
else:
return abort(404)
if get.name != 'app': return abort(404)
if not public.check_app('wxapp'): return public.returnMsg(False, 'UNBOUND_USER')
import panelPlugin
plu = panelPlugin.panelPlugin()
get.s = '_check'
checks = plu.a(get)
if type(checks) != bool or not checks:
public.set_error_num(num_key)
return public.getJson(checks),json_header
get.s = get.fun
comm.setSession()
comm.init()
comm.checkWebType()
comm.GetOS()
result = plu.a(get)
#session.clear()
public.set_error_num(num_key,True)
return public.getJson(result),json_header
@app.route('/favicon.ico',methods=method_get)
def send_favicon():
class/config.py
+4 -4
View File
@@ -328,15 +328,15 @@ class config:
isReWeb = True
if get.webname != session['title']:
session['title'] = get.webname
public.SetConfigValue('title',get.webname)
session['title'] = public.xssencode(get.webname)
public.SetConfigValue('title',public.xssencode(get.webname))
limitip = public.readFile('data/limitip.conf')
if get.limitip != limitip:
public.writeFile('data/limitip.conf',get.limitip)
cache.set('limit_ip',[])
public.writeFile('data/domain.conf',get.domain.strip())
public.writeFile('data/domain.conf',public.xssencode2(get.domain).strip())
public.writeFile('data/iplist.txt',get.address)
@@ -812,7 +812,7 @@ class config:
isAdd = public.M('panel').where('title=? OR url=?',(get.title,get.url)).count()
if isAdd: return public.returnMsg(False,'PANEL_SSL_ADD_EXISTS')
import time,json
isRe = public.M('panel').add('title,url,username,password,click,addtime',(get.title,get.url,get.username,get.password,0,int(time.time())))
isRe = public.M('panel').add('title,url,username,password,click,addtime',(public.xssencode2(get.title),public.xssencode2(get.url),public.xssencode2(get.username),get.password,0,int(time.time())))
if isRe: return public.returnMsg(True,'ADD_SUCCESS')
return public.returnMsg(False,'ADD_ERROR')
class/crontab.py
+2 -2
View File
@@ -207,13 +207,13 @@ class crontab:
self.CrondReload()
columns = 'name,type,where1,where_hour,where_minute,echo,addtime,\
status,save,backupTo,sType,sName,sBody,urladdress'
values = (get['name'],get['type'],get['where1'],get['hour'],
values = (public.xssencode(get['name']),get['type'],get['where1'],get['hour'],
get['minute'],cronName,time.strftime('%Y-%m-%d %X',time.localtime()),
1,get['save'],get['backupTo'],get['sType'],get['sName'],get['sBody'],
get['urladdress'])
if "save_local" in get:
columns += ",save_local,notice,notice_channel"
values = (get['name'],get['type'],get['where1'],get['hour'],
values = (public.xssencode(get['name']),get['type'],get['where1'],get['hour'],
get['minute'],cronName,time.strftime('%Y-%m-%d %X',time.localtime()),
1,get['save'],get['backupTo'],get['sType'],get['sName'],get['sBody'],
get['urladdress'], get["save_local"], get['notice'], get['notice_channel'])
class/data.py
+1
View File
@@ -25,6 +25,7 @@ class data:
'''
def setPs(self,get):
id = get.id
get.ps = public.xssencode(get.ps)
if public.M(get.table).where("id=?",(id,)).setField('ps',get.ps):
return public.returnMsg(True,'EDIT_SUCCESS')
return public.returnMsg(False,'EDIT_ERROR')
class/database.py
+5
View File
@@ -76,6 +76,7 @@ class database(datatool.datatools):
self.__CreateUsers(data_name,username,password,address,ssl)
if get['ps'] == '': get['ps']=public.getMsg('INPUT_PS')
get['ps'] = public.xssencode(get['ps'])
addTime = time.strftime('%Y-%m-%d %X',time.localtime())
pid = 0
@@ -687,6 +688,10 @@ SetLink
ps = public.getMsg('INPUT_PS')
if value[0] == 'test':
ps = public.getMsg('DATABASE_TEST')
# XSS filter
if not re.match("^[\w+\.-]+$",value[0]):continue
addTime = time.strftime('%Y-%m-%d %X',time.localtime())
if sql.table('databases').add('name,username,password,accept,ps,addtime',(value[0],value[0],'',host,ps,addTime)): n +=1
class/files.py
+1 -1
View File
@@ -540,7 +540,7 @@ session.save_handler = files'''.format(path, sess_path, sess_path)
'''
filename = args.filename.strip()
ps_type = int(args.ps_type)
ps_body = args.ps_body
ps_body = public.xssencode(args.ps_body)
ps_path = '/www/server/panel/data/files_ps'
if not os.path.exists(ps_path):
os.makedirs(ps_path,384)
class/firewalls.py
+1 -1
View File
@@ -138,7 +138,7 @@ class firewalls:
import time
port = get.port
ps = get.ps
ps = public.xssencode(get.ps)
is_exists = public.M('firewall').where("port=? or port=?",(port,src_port)).count()
if is_exists: return public.returnMsg(False,'FIREWALL_PORT_EXISTS')
notudps = ['80','443','8888','888','39000:40000','21','22']
class/ftp.py
+1 -1
View File
@@ -35,7 +35,7 @@ class ftp:
public.ExecShell('chown www.www ' + get.path)
public.ExecShell(self.__runPath + '/pure-pw useradd ' + username + ' -u www -d ' + get.path + '<<EOF \n' + password + '\n' + password + '\nEOF')
self.FtpReload()
ps=get['ps']
ps=public.xssencode(get['ps'])
if get['ps']=='': ps= public.getMsg('INPUT_PS');
addtime=time.strftime('%Y-%m-%d %X',time.localtime())
class/panelSite.py
+3 -10
View File
@@ -596,6 +596,7 @@ set $bt_safe_open "{}/:/tmp/";'''.format(self.sitePath)
if not result: return public.returnMsg(False, 'SITE_ADD_ERR_WRITE')
ps = get.ps
ps = public.xssencode(ps)
# 添加放行端口
if self.sitePort != '80':
import firewalls
@@ -3856,16 +3857,8 @@ RewriteRule ^%s(.*)$ http://%s/$1 [P,E=Proxy-Host:%s]
# proxysite1 = re.search(rep,get.proxysite).group(1)
ng_proxy = '''
#PROXY-START%s
location ~* \.(gif|png|jpg|css|js|woff|woff2)$
{
proxy_pass %s;
proxy_set_header Host %s;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header REMOTE-HOST $remote_addr;
expires 12h;
}
location %s
location ^~ %s
{
proxy_pass %s;
proxy_set_header Host %s;
class/public.py
+61 -53
View File
@@ -47,7 +47,7 @@ def HttpGet(url,timeout = 6,headers = {}):
"""
if is_local(): return False
import http_requests
res = http_requests.get(url,timeout=timeout,headers = headers)
res = http_requests.get(url,timeout=timeout,headers = headers,verify=False)
if res.status_code == 0:
if headers: return False
s_body = res.text
@@ -346,36 +346,36 @@ def writeFile(filename,s_body,mode='w+'):
def WriteLog(type,logMsg,args=(),not_web = False):
#写日志
#try:
import time,db,json
username = 'system'
uid = 1
tmp_msg = ''
if not not_web:
try:
from BTPanel import session
if 'username' in session:
username = session['username']
uid = session['uid']
if session.get('debug') == 1: return
except:
pass
global _LAN_LOG
if not _LAN_LOG:
_LAN_LOG = json.loads(ReadFile('BTPanel/static/language/' + GetLanguage() + '/log.json'))
keys = _LAN_LOG.keys()
if logMsg in keys:
logMsg = _LAN_LOG[logMsg]
for i in range(len(args)):
rep = '{'+str(i+1)+'}'
logMsg = logMsg.replace(rep,args[i])
if type in keys: type = _LAN_LOG[type]
sql = db.Sql()
mDate = time.strftime('%Y-%m-%d %X',time.localtime())
data = (uid,username,type,logMsg + tmp_msg,mDate)
result = sql.table('logs').add('uid,username,type,log,addtime',data)
#except:
#pass
try:
import time,db,json
username = 'system'
uid = 1
tmp_msg = ''
if not not_web:
try:
from BTPanel import session
if 'username' in session:
username = session['username']
uid = session['uid']
if session.get('debug') == 1: return
except:
pass
global _LAN_LOG
if not _LAN_LOG:
_LAN_LOG = json.loads(ReadFile('BTPanel/static/language/' + GetLanguage() + '/log.json'))
keys = _LAN_LOG.keys()
if logMsg in keys:
logMsg = _LAN_LOG[logMsg]
for i in range(len(args)):
rep = '{'+str(i+1)+'}'
logMsg = logMsg.replace(rep,args[i])
if type in keys: type = _LAN_LOG[type]
sql = db.Sql()
mDate = time.strftime('%Y-%m-%d %X',time.localtime())
data = (uid,username,type,xssencode2(logMsg + tmp_msg),mDate)
result = sql.table('logs').add('uid,username,type,log,addtime',data)
except:
pass
def GetLanguage():
'''
@@ -835,7 +835,7 @@ def downloadFile(url,filename):
if sys.version_info[0] == 2:
import requests
headers = {'User-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.135 Safari/537.36'}
r = requests.get(url, headers=headers)
r = requests.get(url, headers=headers,verify=False)
with open(filename,"wb") as f:
f.write(r.content)
else:
@@ -1413,6 +1413,12 @@ def xssencode(text):
text2 = cgi.escape(str_convert, quote=True)
return text2
#xss 防御
def xssencode2(text):
import cgi
text2 = cgi.escape(text, quote=True)
return text2
# 取缓存
def cache_get(key):
from BTPanel import cache
@@ -1710,30 +1716,32 @@ def de_crypt(key,strings):
#获取IP限制列表
def get_limit_ip():
iplong_list = []
ip_file = 'data/limitip.conf'
if not os.path.exists(ip_file): return iplong_list
try:
ip_file = 'data/limitip.conf'
if not os.path.exists(ip_file): return iplong_list
from BTPanel import cache
ikey = 'limit_ip'
iplong_list = cache.get(ikey)
if iplong_list: return iplong_list
from BTPanel import cache
ikey = 'limit_ip'
iplong_list = cache.get(ikey)
if iplong_list: return iplong_list
iplong_list = []
iplist = ReadFile(ip_file)
if not iplist:return iplong_list
iplist = iplist.strip()
for limit_ip in iplist.split(','):
if not limit_ip: continue
limit_ip = limit_ip.split('-')
iplong = {}
iplong['min'] = ip2long(limit_ip[0])
if len(limit_ip) > 1:
iplong['max'] = ip2long(limit_ip[1])
else:
iplong['max'] = iplong['min']
iplong_list.append(iplong)
iplong_list = []
iplist = ReadFile(ip_file)
if not iplist:return iplong_list
iplist = iplist.strip()
for limit_ip in iplist.split(','):
if not limit_ip: continue
limit_ip = limit_ip.split('-')
iplong = {}
iplong['min'] = ip2long(limit_ip[0])
if len(limit_ip) > 1:
iplong['max'] = ip2long(limit_ip[1])
else:
iplong['max'] = iplong['min']
iplong_list.append(iplong)
cache.set(ikey,iplong_list,3600)
cache.set(ikey,iplong_list,3600)
except:pass
return iplong_list
class/wxapp.py
+27 -177
View File
@@ -11,31 +11,30 @@ import sys
if not 'class/' in sys.path:
sys.path.insert(0,'class/')
import public
import db
import json
import time
import binascii
import base64
from BTPanel import session,cache,request
class ScanLogin(object):
# 扫码登录面板
def scan_login(self, get):
# 用于小程序
data = public.GetRandomString(48) + ':' + str(time.time())
public.writeFile(self.app_path+"login.pl", data)
return public.returnMsg(True, 'SCAN_QRCORE_SUCCESS_LOGGING_IN')
class wxapp():
def __init__(self):
self.app_path = '/www/server/panel/data/'
self.app_path_p = '/www/server/panel/plugin/app/'
def _check(self, get):
if get['fun'] in ['set_login', 'is_scan_ok', 'login_qrcode']:
return True
return public.returnMsg(False, 'UNAUTHORIZED')
# 验证是否扫码成功
def is_scan_ok(self, get):
if os.path.exists(self.app_path+"login.pl"):
key, init_time = public.readFile(
self.app_path+'login.pl').split(':')
if time.time() - float(init_time) < 60:
return public.returnMsg(True, key)
session_id = public.get_session_id()
if cache.get(session_id) == 'True':
return public.returnMsg(True, 'Scan QRCORE successfully')
if os.path.exists(self.app_path+"app_login_check.pl"):
key, init_time = public.readFile(self.app_path+'app_login_check.pl').split(':')
if time.time() - float(init_time) > 180:
return public.returnMsg(False, 'QRCORE_EXPIRE')
session_id = public.get_session_id()
if cache.get(session_id) == 'True':
return public.returnMsg(True, 'Scan QRCORE successfully')
return public.returnMsg(False, '')
# 返回二维码地址
@@ -48,57 +47,31 @@ class ScanLogin(object):
cache.set(public.get_session_id(),tid,360)
return public.returnMsg(True, qrcode_str)
#生成request_token
def set_request_token(self):
session['request_token_head'] = public.GetRandomString(48)
# 设置登录状态
def set_login(self, get):
session_id = public.get_session_id()
if cache.get(session_id) == 'True':
return self.check_app_login(get)
# if os.path.exists(self.app_path+"login.pl"):
# data = public.readFile(self.app_path+'login.pl')
# public.ExecShell('rm ' + self.app_path+"login.pl")
# secret_key, init_time = data.split(':')
# if time.time() - float(init_time) < 60 and get['secret_key'] == secret_key:
# sql = db.Sql()
# userInfo = sql.table('users').where(
# "id=?", (1,)).field('id,username,password').find()
# session['login'] = True
# session['username'] = userInfo['username']
# cache.delete('panelNum')
# cache.delete('dologin')
# public.WriteLog('TYPE_LOGIN', 'LOGIN_SUCCESS',
# ('微信扫码登录', public.GetClientIp()+ ":" + str(request.environ.get('REMOTE_PORT'))))
# login_type = 'data/app_login.pl'
# self.set_request_token()
# import config
# config.config().reload_session()
# public.writeFile(login_type,'True')
# public.login_send_body("微信小程序",userInfo['username'],public.GetClientIp(),str(request.environ.get('REMOTE_PORT')))
# return public.returnMsg(True, '登录成功')
return public.returnMsg(False, '登录失败')
return public.returnMsg(False, 'Login failed 1')
#验证APP是否登录成功
def check_app_login(self,get):
#判断是否存在绑定
btapp_info = json.loads(public.readFile('/www/server/panel/config/api.json'))
if not btapp_info:return public.returnMsg(False,'Unbound')
if not btapp_info['open']:return public.returnMsg(False,'API not open')
if not btapp_info['open']:return public.returnMsg(False,'API is not turned on')
if not 'apps' in btapp_info:return public.returnMsg(False,'Unbound phone')
if not btapp_info['apps']:return public.returnMsg(False,'Unbound phone')
try:
session_id=public.get_session_id()
if not os.path.exists(self.app_path+'app_login_check.pl'):return public.returnMsg(False,'Wait for the app to scan the code and log in 1')
if not os.path.exists(self.app_path+'app_login_check.pl'):return public.returnMsg(False,'Waiting for APP scan code login 1')
data = public.readFile(self.app_path+'app_login_check.pl')
public.ExecShell('rm ' + self.app_path+"app_login_check.pl")
secret_key, init_time = data.split(':')
if len(session_id)!=64:return public.returnMsg(False,'Wait for the app to scan the code and log in 2')
if len(session_id)!=64:return public.returnMsg(False,'Waiting for APP scan code login 2')
if len(secret_key)!=64:return public.returnMsg(False,'Waiting for APP scan code login 2')
if time.time() - float(init_time) < 180 and session_id != secret_key:
return public.returnMsg(False,'Wait for the app to scan the code and log in')
return public.returnMsg(False,'Waiting for APP scan code login')
cache.delete(session_id)
userInfo = public.M('users').where("id=?",(1,)).field('id,username').find()
session['login'] = True
@@ -116,130 +89,7 @@ class ScanLogin(object):
public.login_send_body("aaPanel Mobile",userInfo['username'],public.GetClientIp(),str(request.environ.get('REMOTE_PORT')))
return public.returnMsg(True,'login successful!')
except:
return public.returnMsg(False, 'login fail')
class SelfModule():
'''
只能在面板执行的模块
不允许外部访问
'''
def __init__(self):
self.user_info_file = self.app_path + "user.json"
if not os.path.exists(self.user_info_file):
public.ExecShell("echo '{}' > " + self.user_info_file)
try:
self.user_info = json.loads(public.readFile(self.user_info_file))
except: public.ExecShell("echo '{}' > " + self.user_info_file)
user_info_file_app = self.app_path_p + "user.json"
if os.path.exists(user_info_file_app):
try:
user_info_app = json.loads(public.readFile(user_info_file_app))
for userId in user_info_app.keys():
if userId in self.user_info: continue;
self.user_info[userId] = user_info_app[userId];
except:pass
def blind_qrcode(self, get):
'''
生成绑定二维码
'''
panel_addr = public.getPanelAddr()
token = public.GetRandomString(32)
data = '%s:%s' % (token, int(time.time()))
public.writeFile(self.app_path + 'token.pl',data)
public.writeFile(self.app_path_p + 'token.pl',data)
qrcode_str = 'https://app.bt.cn/app.html?panel_url=' + \
panel_addr+'&panel_token=' + token + '?blind'
return public.returnMsg(True, qrcode_str)
def blind_del(self, get):
# 删除绑定
del self.user_info[get['uid']]
public.writeFile(self.app_path+"user.json", json.dumps(self.user_info))
public.writeFile(self.app_path_p + "user.json", json.dumps(self.user_info))
return public.returnMsg(True, 'DEL_SUCCESS')
def get_user_info(self, get):
if session['version'] < '5.8.6':
return public.returnMsg(False, 'PANEL_TOO_LOW')
data = {}
if not get: data = []
for k in self.user_info.keys():
v = self.user_info[k]
if get:
del(v['token'])
data[k] = v
else:
data.append(v['nickName'])
if not get:
data = ','.join(data);
if not data: data = public.GetMsg("NOT_BIND_WECHAT");
return public.returnMsg(True, data)
def blind_result(self, get):
return not os.path.exists(self.app_path + "token.pl")
class wxapp(SelfModule, ScanLogin):
def __init__(self):
self.app_path = '/www/server/panel/data/'
self.app_path_p = '/www/server/panel/plugin/app/'
SelfModule.__init__(self)
def _check(self, get):
token_data = public.readFile(self.app_path + 'token.pl')
if not token_data: token_data = public.readFile(self.app_path_p + 'token.pl')
if hasattr(SelfModule, get['fun']):
return False
elif get['fun'] in ['set_login', 'is_scan_ok', 'login_qrcode']:
return True
elif get['fun'] == 'blind':
if not token_data:
return public.returnMsg(False, 'QRCORE_EXPIRE',("1",))
token_data = token_data.replace('\n', '')
password, expiration_time = token_data.split(':')
# return True
if time.time() - int(expiration_time) > 8*60:
return public.returnMsg(False, 'QRCORE_EXPIRE',("2",))
elif get['panel_token'] != password:
return public.returnMsg(False, 'SK_NOT_INCORRECT')
return True
# else:
# # 是否在白名单ip sgin 是否正确
# if hasattr(get, 'uid') and hasattr(get, 'sgin') and hasattr(get, 'fun') and get['uid'] in self.user_info.keys():
# encryption_str = self.user_info[get['uid']]['token']+get['fun']+get['uid']
# if sys.version_info[0] == 3:
# if type(encryption_str) == str:
# encryption_str = encryption_str.encode()
# if get['sgin'] == public.md5(binascii.hexlify(base64.b64encode(encryption_str))):
# if public.GetClientIp() in ['47.52.194.186']:
# return public.returnMsg(False, '未授权')
return public.returnMsg(False, 'UNAUTHORIZED')
# 用户绑定
def blind(self, get):
# 用于小程序
self.user_info[get['uid']] = {
"avatarUrl": get['avatarUrl'],
"nickName": get['nickName'],
"token": get['token']
}
public.writeFile(self.app_path+"user.json", json.dumps(self.user_info))
public.writeFile(self.app_path_p + "user.json", json.dumps(self.user_info))
public.ExecShell("rm -rf %stoken.pl" % self.app_path)
public.ExecShell("rm -rf %stoken.pl" % self.app_path_p)
return public.returnMsg(True, 'BIND_SUCCESS')
def get_safe_log(self):
get = {
'page': 1,
'count': 10
}
print(get['page'] - 1) * get['count'], get['count']
data = public.M('logs').limit('%s, %s' % (
(get['page'] - 1) * get['count'], get['count'])).select()
return data
return public.returnMsg(False, 'Login failed 2')
#生成request_token
def set_request_token(self):
session['request_token_head'] = public.GetRandomString(48)