You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

352 lines
11 KiB

#coding: utf-8
# +-------------------------------------------------------------------
# | 宝塔Linux面板
# +-------------------------------------------------------------------
# | Copyright (c) 2015-2099 宝塔软件(http://bt.cn) All rights reserved.
# +-------------------------------------------------------------------
# | Author: hwliang <hwl@bt.cn>
# +-------------------------------------------------------------------
#+--------------------------------------------------------------------
#| 插件和模块加载器
#+--------------------------------------------------------------------
import public,os,sys,json
def plugin_run(plugin_name,def_name,args):
'''
@name 执行插件方法
@param plugin_name<string> 插件名称
@param def_name<string> 方法名称
@param args<dict_obj> 参数对像
@return mixed
'''
if not plugin_name or not def_name: return public.returnMsg(False,'插件名称和插件方法名称不能为空!')
# 获取插件目录
plugin_path = public.get_plugin_path(plugin_name)
is_php = os.path.exists(os.path.join(plugin_path,'index.php'))
# 检查插件目录是否合法
if is_php:
plugin_file = os.path.join(plugin_path,'index.php')
else:
plugin_file = os.path.join(plugin_path, plugin_name + '_main.py')
if not public.path_safe_check(plugin_file): return public.returnMsg(False,'插件路径不合法')
# 检查插件入口文件是否存在
if not os.path.exists(plugin_file): return public.returnMsg(False,'指定插件入口文件不存在')
# 添加插件目录到系统路径
public.sys_path_append(plugin_path)
if not is_php:
# 引用插件入口文件
_name = "{}_main".format(plugin_name)
plugin_main = __import__(_name)
# 检查类名是否符合规范
if not hasattr(plugin_main,_name):
return public.returnMsg(False,'指定插件入口文件不符合规范')
try:
if sys.version_info[0] == 2:
reload(plugin_main)
else:
from imp import reload
reload(plugin_main)
except:
pass
# 实例化插件类
plugin_obj = getattr(plugin_main,_name)()
# 检查方法是否存在
if not hasattr(plugin_obj,def_name):
return public.returnMsg(False,'在[%s]插件中找不到[%s]方法' % (plugin_name,def_name))
if 'plugin_get_object' in args and args.plugin_get_object == 1:
return getattr(plugin_obj, def_name)
# 执行方法
return getattr(plugin_obj,def_name)(args)
else:
if 'plugin_get_object' in args and args.plugin_get_object == 1:
return None
import panelPHP
args.s = def_name
args.name = plugin_name
return panelPHP.panelPHP(plugin_name).exec_php_script(args)
def get_module_list():
'''
@name 获取模块列表
@return list
'''
module_list = []
class_path = public.get_class_path()
for name in os.listdir(class_path):
path = os.path.join(class_path,name)
# 过滤无效文件
if not name or name.endswith('.py') or name[0] == '.' or not name.endswith('Model') or os.path.isfile(path):continue
module_list.append(name)
return module_list
def module_run(module_name,def_name,args):
'''
@name 执行模块方法
@param module_name<string> 模块名称
@param def_name<string> 方法名称
@param args<dict_obj> 参数对像
@return mixed
'''
if not module_name or not def_name: return public.returnMsg(False,'模块名称和模块方法名称不能为空!')
model_index = args.get('model_index',None)
class_path = public.get_class_path()
panel_path = public.get_panel_path()
module_file = None
if 'model_index' in args:
# 新模块目录
if model_index in ['mod']:
module_file = os.path.join(panel_path,'mod','project',module_name + 'Mod.py')
elif model_index:
# 旧模块目录
module_file = os.path.join(class_path,model_index+"Model",module_name + 'Model.py')
else:
module_file = os.path.join(class_path,"projectModel",module_name + 'Model.py')
else:
# 如果没指定模块名称,则遍历所有模块目录
module_list = get_module_list()
for name in module_list:
module_file = os.path.join(class_path,name,module_name + 'Model.py')
if os.path.exists(module_file): break
# 判断模块入口文件是否存在
if not os.path.exists(module_file):
return public.returnMsg(False,'模块[%s]不存在' % module_name)
# 判断模块路径是否合法
if not public.path_safe_check(module_file):
return public.returnMsg(False,'模块路径不合法')
def_object = public.get_script_object(module_file)
if not def_object: return public.returnMsg(False,'模块[%s]不存在' % module_file)
# 模块实例化并返回方法对象
try:
run_object = getattr(def_object.main(),def_name,None)
except:
return public.returnMsg(False,'模块[%s]入口实例化失败' % module_file)
if not run_object: return public.returnMsg(False,'在[%s]模块中找不到[%s]方法' % (module_file,def_name))
if 'module_get_object' in args and args.module_get_object == 1:
return run_object
# 执行方法
result = run_object(args)
return result
def get_plugin_list(upgrade_force = False):
'''
@name 获取插件列表
@param upgrade_force<bool> 是否强制重新获取列表
@return dict
'''
api_root_url = 'https://api.bt.cn'
api_url = api_root_url+ '/panel/get_plugin_list'
panel_path = public.get_panel_path()
data_path = os.path.join(panel_path,'data')
if not os.path.exists(data_path):
os.makedirs(data_path,384)
plugin_list = {}
plugin_list_file = os.path.join(data_path,'plugin_list.json')
if os.path.exists(plugin_list_file) and not upgrade_force:
plugin_list_body = public.readFile(plugin_list_file)
try:
plugin_list = json.loads(plugin_list_body)
except:
plugin_list = {}
if not os.path.exists(plugin_list_file) or upgrade_force or not plugin_list:
try:
res = public.HttpGet(api_url)
except Exception as ex:
raise public.error_conn_cloud(str(ex))
if not res: raise Exception(False,'云端插件列表获取失败')
plugin_list = json.loads(res)
if type(plugin_list)!=dict or 'list' not in plugin_list:
if type(plugin_list)==str:
raise Exception(plugin_list)
else:
raise Exception('云端插件列表获取失败')
public.writeFile(plugin_list_file,json.dumps(plugin_list))
return plugin_list
def start_total():
'''
@name 启动统计服务
@return dict
'''
pass
def get_soft_list(args):
'''
@name 获取软件列表
@param args<dict_obj> 参数对像
@return dict
'''
pass
def db_encrypt(data):
'''
@name 数据库加密
@param args<dict_obj> 参数对像
@return dict
'''
try:
key = __get_db_sgin()
iv = __get_db_iv()
str_arr = data.split('\n')
res_str = ''
for data in str_arr:
if not data: continue
res_str += __aes_encrypt(data, key, iv)
except:
res_str = data
result = {
'status' : True,
'msg' : res_str
}
return result
def db_decrypt(data):
'''
@name 数据库解密
@param args<dict_obj> 参数对像
@return dict
'''
try:
key = __get_db_sgin()
iv = __get_db_iv()
str_arr = data.split('\n')
res_str = ''
for data in str_arr:
if not data: continue
res_str += __aes_decrypt(data, key, iv)
except:
res_str = data
result = {
'status' : True,
'msg' : res_str
}
return result
def __get_db_sgin():
keystr = '3gP7+k_7lSNg3$+Fj!PKW+6$KYgHtw#R'
key = ''
for i in range(31):
if i & 1 == 0:
key += keystr[i]
return key
def __get_db_iv():
div_file = "{}/data/div.pl".format(public.get_panel_path())
if not os.path.exists(div_file):
str = public.GetRandomString(16)
str = __aes_encrypt_module(str)
div = public.get_div(str)
public.WriteFile(div_file, div)
if os.path.exists(div_file):
div = public.ReadFile(div_file)
div = __aes_decrypt_module(div)
else:
keystr = '4jHCpBOFzL4*piTn^-4IHBhj-OL!fGlB'
div = ''
for i in range(31):
if i & 1 == 0:
div += keystr[i]
return div
def __aes_encrypt_module(data):
key = 'Z2B87NEAS2BkxTrh'
iv = 'WwadH66EGWpeeTT6'
return __aes_encrypt(data, key, iv)
def __aes_decrypt_module(data):
key = 'Z2B87NEAS2BkxTrh'
iv = 'WwadH66EGWpeeTT6'
return __aes_decrypt(data, key, iv)
def __aes_decrypt(data, key, iv):
from Crypto.Cipher import AES
import base64
encodebytes = base64.decodebytes(data.encode('utf-8'))
aes = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))
de_text = aes.decrypt(encodebytes)
unpad = lambda s: s[0:-s[-1]]
de_text = unpad(de_text)
return de_text.decode('utf-8')
def __aes_encrypt(data, key, iv):
from Crypto.Cipher import AES
import base64
data = (lambda s: s + (16 - len(s) % 16) * chr(16 - len(s) % 16).encode('utf-8'))(data.encode('utf-8'))
aes = AES.new(key.encode('utf8'), AES.MODE_CBC, iv.encode('utf8'))
encryptedbytes = aes.encrypt(data)
en_text = base64.b64encode(encryptedbytes)
return en_text.decode('utf-8')
def plugin_end():
'''
@name 插件到期处理
@return dict
'''
pass
def daemon_task():
'''
@name 后台任务守护
@return dict
'''
pass
def daemon_panel():
'''
@name 面板守护
@return dict
'''
pass
def flush_auth_key():
'''
@name 刷新授权密钥
@return dict
'''
pass
def get_auth_state():
'''
@name 获取授权状态
@return 返回:0.免费版 1.专业版 2.企业版 -1.获取失败
'''
try:
softList = get_plugin_list()
if softList['ltd'] > -1:
return 2
elif softList['pro'] > -1:
return 1
else:
return 0
except:
return -1