原创 | python代码审计-osroom

def json_to_pyseq(tjson):    """    json to python sequencer    :param json:    :return:    """    if tjson in [None, "None"]:        return None    elif not isinstance(tjson, (list, dict, tuple)) and tjson != "":        if isinstance(tjson, (str, bytes)) and tjson[0] not in ["{", "[", "("]:            return tjson        elif isinstance(tjson, (int, float)):            return tjson        try:            tjson = json.loads(tjson)        except BaseException:            tjson = eval(tjson)        else:            if isinstance(tjson, str):                tjson = eval(tjson)    return tjson

def audit_rule_delete():    ids = json_to_pyseq(request.argget.all('ids', []))    if not isinstance(ids, list):        ids = json.loads(ids)    for i, tid in enumerate(ids):        ids[i] = ObjectId(tid)    r = mdbs["sys"].db.audit_rules.delete_many({"_id": {"$in": ids}})    if r.deleted_count > 0:        data = {"msg": gettext("Delete the success,{}").format(            r.deleted_count), "msg_type": "s", "custom_status": 204}    else:        data = {            "msg": gettext("Delete failed"),            "msg_type": "w",            "custom_status": 400}    return data

def fileup_base_64(uploaded_files, file_name=None, prefix=""):    """     文件以base64编码上传上传    :param uploaded_files: 数组    :param bucket_var: 保存typroa图像服务器空间名的变量名, 如AVA_B    :param file_name:    :return:    """    if not uploaded_files:        return None    keys = []    for file_base in uploaded_files:        if file_base:            # data:image/jpeg            file_format = file_base.split(";")[0].split("/")[-1]            imgdata = base64.b64decode(file_base.split(",")[-1])            if file_name:                filename = '{}.{}'.format(file_name, file_format)            else:                filename = '{}_{}.{}'.format(                    time_to_utcdate(                        time_stamp=time.time(),                        tformat="%Y%m%d%H%M%S"),                    uuid1(),                    file_format)

def upload_plugin():    """    插件上传    :return:    """    file = request.files["upfile"]    file_name = os.path.splitext(file.filename)         #('123','.zip')    filename = os.path.splitext(file.filename)[0]         #123    extension = file_name[1]                           #.zip    if not extension.strip(".").lower() in ["zip"]:        data = {"msg": gettext("File format error, please upload zip archive"),                "msg_type": "w", "custom_status": 401}        return data    if not os.path.exists(PLUG_IN_FOLDER):            #osroom/apps/plugins        os.makedirs(PLUG_IN_FOLDER)    fpath = os.path.join(PLUG_IN_FOLDER, filename)   ##osroom/apps/plugins/123    if os.path.isdir(fpath) or os.path.exists(fpath):        if mdbs["sys"].db.plugin.find_one(                {"plugin_name": filename, "is_deleted": {"$in": [0, False]}}):            # 如果插件没有准备删除标志            data = {"msg": gettext("The same name plugin already exists"),                    "msg_type": "w", "custom_status": 403}            return data        else:            # 否则清除旧的插件            shutil.rmtree(fpath)            mdbs["sys"].db.plugin.update_one({"plugin_name": filename}, {                                         "$set": {"is_deleted": 0}})    # 保存主题    save_file = os.path.join("{}/{}".format(PLUG_IN_FOLDER, file.filename))     ##osroom/apps/plugins/123.zip    file.save(save_file)

def p_sign_in(        username,        password,        code_url_obj,        code,        remember_me,        use_jwt_auth=0):    """    用户登录函数    :param adm:    :return:    """    data = {}    if current_user.is_authenticated and username in [current_user.username,                                                      current_user.email,                                                      current_user.mphone_num]:        data['msg'] = gettext("Is logged in")        data["msg_type"] = "s"        data["custom_status"] = 201        data['to_url'] = request.argget.all(            'next') or get_config("login_manager", "LOGIN_IN_TO")        return data

var result = osrHttp("PUT","/api/sign-in", d);      result.then(function (r) {             if(r.data.msg_type=="s"){                 window.location.href = r.data/to_url;             }else if(r.data.open_img_verif_code){                 get_imgcode();             }      }).catch(function (r) {         if(r.data.open_img_verif_code){             get_imgcode();         }      });

def get_static_file_content():    """    获取静态文件内容, 如html文件    :return:    """    filename = request.argget.all('filename', "index").strip("/")    file_path = request.argget.all('file_path', "").strip("/")    theme_name = request.argget.all("theme_name")    s, r = arg_verify([(gettext("theme name"), theme_name)], required=True)    if not s:        return r    path = os.path.join(        THEME_TEMPLATE_FOLDER, theme_name)    file = "{}/{}/{}".format(path, file_path, filename)    if not os.path.exists(file) or THEME_TEMPLATE_FOLDER not in file:        data = {"msg": gettext("File not found,'{}'").format(file),                "msg_type": "w", "custom_status": 404}    else:        with open(file) as wf:            content = wf.read()        data = {            "content": content,            "file_relative_path": file_path.replace(                path,                "").strip("/")}    return data