bwadmin/pear-admin-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化验证与过滤

Browse Source
This commit is contained in:
不胜舟
2022-12-28 00:21:20 +08:00
parent 656548eb64
commit 123f62f9d5
8 changed files with 108 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
applications/common/admin_log.py
+5 -5
View File
@@ -1,6 +1,6 @@
from flask_login import current_user from flask_login import current_user
from applications.common.utils.validate import xss_escape from applications.common.utils.validate import str_escape
from applications.extensions import db from applications.extensions import db
from applications.models import AdminLog from applications.models import AdminLog
@@ -10,8 +10,8 @@ def login_log(request, uid, is_access):
'method': request.method, 'method': request.method,
'url': request.path, 'url': request.path,
'ip': request.remote_addr, 'ip': request.remote_addr,
'user_agent': xss_escape(request.headers.get('User-Agent')), 'user_agent': str_escape(request.headers.get('User-Agent')),
'desc': xss_escape(request.form.get('username')), 'desc': str_escape(request.form.get('username')),
'uid': uid, 'uid': uid,
'success': int(is_access) 'success': int(is_access)
@@ -36,8 +36,8 @@ def admin_log(request, is_access):
'method': request.method, 'method': request.method,
'url': request.path, 'url': request.path,
'ip': request.remote_addr, 'ip': request.remote_addr,
'user_agent': xss_escape(request.headers.get('User-Agent')), 'user_agent': str_escape(request.headers.get('User-Agent')),
'desc': xss_escape(str(dict(request.values))), 'desc': str_escape(str(dict(request.values))),
'uid': current_user.id, 'uid': current_user.id,
'success': int(is_access) 'success': int(is_access)
applications/common/utils/validate.py
+4 -6
View File
@@ -1,13 +1,11 @@
# xss过滤 # xss过滤
from flask import abort, make_response, jsonify from flask import abort, make_response, jsonify, escape
def xss_escape(s: str): def str_escape(s):
if s is None: if not s:
return None return None
else: return str(escape(s))
return s.replace("&", "&amp;").replace(">", "&gt;").replace("<", "&lt;").replace("'", "&#39;").replace('"',
"&#34;")
def check_data(schema, data): def check_data(schema, data):
applications/view/admin/admin_log.py
+2
View File
@@ -13,6 +13,8 @@ admin_log = Blueprint('adminLog', __name__, url_prefix='/admin/log')
@admin_log.get('/') @admin_log.get('/')
@authorize("admin:log:main") @authorize("admin:log:main")
def index(): def index():
from markupsafe import escape
escape
return render_template('admin/admin_log/main.html') return render_template('admin/admin_log/main.html')
applications/view/admin/dict.py
+22 -22
View File
@@ -4,7 +4,7 @@ from applications.common import curd
from applications.common.helper import ModelFilter from applications.common.helper import ModelFilter
from applications.common.utils.http import table_api, success_api, fail_api from applications.common.utils.http import table_api, success_api, fail_api
from applications.common.utils.rights import authorize from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape from applications.common.utils.validate import str_escape
from applications.extensions import db from applications.extensions import db
from applications.models import DictType, DictData from applications.models import DictType, DictData
from applications.schemas import DictTypeOutSchema, DictDataOutSchema from applications.schemas import DictTypeOutSchema, DictDataOutSchema
@@ -23,7 +23,7 @@ def main():
@authorize("admin:dict:main", log=True) @authorize("admin:dict:main", log=True)
def dict_type_data(): def dict_type_data():
# 获取请求参数 # 获取请求参数
type_name = xss_escape(request.args.get('typeName', type=str)) type_name = str_escape(request.args.get('typeName', type=str))
# 查询参数构造 # 查询参数构造
mf = ModelFilter() mf = ModelFilter()
if type_name: if type_name:
@@ -46,10 +46,10 @@ def dict_type_add():
@authorize("admin:dict:add", log=True) @authorize("admin:dict:add", log=True)
def dict_type_save(): def dict_type_save():
req_json = request.json req_json = request.json
description = xss_escape(req_json.get("description")) description = str_escape(req_json.get("description"))
enable = xss_escape(req_json.get("enable")) enable = str_escape(req_json.get("enable"))
type_code = xss_escape(req_json.get("typeCode")) type_code = str_escape(req_json.get("typeCode"))
type_name = xss_escape(req_json.get("typeName")) type_name = str_escape(req_json.get("typeName"))
d = DictType(type_name=type_name, type_code=type_code, enable=enable, description=description) d = DictType(type_name=type_name, type_code=type_code, enable=enable, description=description)
db.session.add(d) db.session.add(d)
db.session.commit() db.session.commit()
@@ -72,11 +72,11 @@ def dict_type_edit():
@authorize("admin:dict:edit", log=True) @authorize("admin:dict:edit", log=True)
def dict_type_update(): def dict_type_update():
req_json = request.json req_json = request.json
id = xss_escape(req_json.get("id")) id = str_escape(req_json.get("id"))
description = xss_escape(req_json.get("description")) description = str_escape(req_json.get("description"))
enable = xss_escape(req_json.get("enable")) enable = str_escape(req_json.get("enable"))
type_code = xss_escape(req_json.get("typeCode")) type_code = str_escape(req_json.get("typeCode"))
type_name = xss_escape(req_json.get("typeName")) type_name = str_escape(req_json.get("typeName"))
DictType.query.filter_by(id=id).update({ DictType.query.filter_by(id=id).update({
"description": description, "description": description,
"enable": enable, "enable": enable,
@@ -126,7 +126,7 @@ def dict_type_delete(_id):
@admin_dict.get('/dictData/data') @admin_dict.get('/dictData/data')
@authorize("admin:dict:main", log=True) @authorize("admin:dict:main", log=True)
def dict_code_data(): def dict_code_data():
type_code = xss_escape(request.args.get('typeCode', type=str)) type_code = str_escape(request.args.get('typeCode', type=str))
dict_data = DictData.query.filter_by(type_code=type_code).layui_paginate() dict_data = DictData.query.filter_by(type_code=type_code).layui_paginate()
count = dict_data.total count = dict_data.total
data = curd.model_to_dicts(schema=DictDataOutSchema, data=dict_data.items) data = curd.model_to_dicts(schema=DictDataOutSchema, data=dict_data.items)
@@ -146,11 +146,11 @@ def dict_data_add():
@authorize("admin:dict:add", log=True) @authorize("admin:dict:add", log=True)
def dict_data_save(): def dict_data_save():
req_json = request.json req_json = request.json
data_label = xss_escape(req_json.get("dataLabel")) data_label = str_escape(req_json.get("dataLabel"))
data_value = xss_escape(req_json.get("dataValue")) data_value = str_escape(req_json.get("dataValue"))
enable = xss_escape(req_json.get("enable")) enable = str_escape(req_json.get("enable"))
remark = xss_escape(req_json.get("remark")) remark = str_escape(req_json.get("remark"))
type_code = xss_escape(req_json.get("typeCode")) type_code = str_escape(req_json.get("typeCode"))
d = DictData(data_label=data_label, data_value=data_value, enable=enable, remark=remark, type_code=type_code) d = DictData(data_label=data_label, data_value=data_value, enable=enable, remark=remark, type_code=type_code)
db.session.add(d) db.session.add(d)
db.session.commit() db.session.commit()
@@ -175,11 +175,11 @@ def dict_data_update():
req_json = request.json req_json = request.json
id = req_json.get("dataId") id = req_json.get("dataId")
DictData.query.filter_by(id=id).update({ DictData.query.filter_by(id=id).update({
"data_label": xss_escape(req_json.get("dataLabel")), "data_label": str_escape(req_json.get("dataLabel")),
"data_value": xss_escape(req_json.get("dataValue")), "data_value": str_escape(req_json.get("dataValue")),
"enable": xss_escape(req_json.get("enable")), "enable": str_escape(req_json.get("enable")),
"remark": xss_escape(req_json.get("remark")), "remark": str_escape(req_json.get("remark")),
"type_code": xss_escape(req_json.get("typeCode")) "type_code": str_escape(req_json.get("typeCode"))
}) })
db.session.commit() db.session.commit()
return success_api(msg="更新成功") return success_api(msg="更新成功")
applications/view/admin/power.py
+17 -17
View File
@@ -3,7 +3,7 @@ from flask import Blueprint, render_template, request, jsonify
from applications.common import curd from applications.common import curd
from applications.common.utils.http import success_api, fail_api from applications.common.utils.http import success_api, fail_api
from applications.common.utils.rights import authorize from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape from applications.common.utils.validate import str_escape
from applications.extensions import db from applications.extensions import db
from applications.models import Power from applications.models import Power
from applications.schemas import PowerOutSchema2 from applications.schemas import PowerOutSchema2
@@ -52,14 +52,14 @@ def select_parent():
@authorize("admin:power:add", log=True) @authorize("admin:power:add", log=True)
def save(): def save():
req = request.json req = request.json
icon = xss_escape(req.get("icon")) icon = str_escape(req.get("icon"))
openType = xss_escape(req.get("openType")) openType = str_escape(req.get("openType"))
parentId = xss_escape(req.get("parentId")) parentId = str_escape(req.get("parentId"))
powerCode = xss_escape(req.get("powerCode")) powerCode = str_escape(req.get("powerCode"))
powerName = xss_escape(req.get("powerName")) powerName = str_escape(req.get("powerName"))
powerType = xss_escape(req.get("powerType")) powerType = str_escape(req.get("powerType"))
powerUrl = xss_escape(req.get("powerUrl")) powerUrl = str_escape(req.get("powerUrl"))
sort = xss_escape(req.get("sort")) sort = str_escape(req.get("sort"))
power = Power( power = Power(
icon=icon, icon=icon,
open_type=openType, open_type=openType,
@@ -96,14 +96,14 @@ def update():
req_json = request.json req_json = request.json
id = request.json.get("powerId") id = request.json.get("powerId")
data = { data = {
"icon": xss_escape(req_json.get("icon")), "icon": str_escape(req_json.get("icon")),
"open_type": xss_escape(req_json.get("openType")), "open_type": str_escape(req_json.get("openType")),
"parent_id": xss_escape(req_json.get("parentId")), "parent_id": str_escape(req_json.get("parentId")),
"code": xss_escape(req_json.get("powerCode")), "code": str_escape(req_json.get("powerCode")),
"name": xss_escape(req_json.get("powerName")), "name": str_escape(req_json.get("powerName")),
"type": xss_escape(req_json.get("powerType")), "type": str_escape(req_json.get("powerType")),
"url": xss_escape(req_json.get("powerUrl")), "url": str_escape(req_json.get("powerUrl")),
"sort": xss_escape(req_json.get("sort")) "sort": str_escape(req_json.get("sort"))
} }
res = Power.query.filter_by(id=id).update(data) res = Power.query.filter_by(id=id).update(data)
db.session.commit() db.session.commit()
applications/view/admin/role.py
+18 -24
View File
@@ -4,7 +4,7 @@ from applications.common.curd import model_to_dicts, enable_status, disable_stat
from applications.common.helper import ModelFilter from applications.common.helper import ModelFilter
from applications.common.utils.http import table_api, success_api, fail_api from applications.common.utils.http import table_api, success_api, fail_api
from applications.common.utils.rights import authorize from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape from applications.common.utils.validate import str_escape
from applications.extensions import db from applications.extensions import db
from applications.models import Role, Power, User from applications.models import Role, Power, User
from applications.schemas import RoleOutSchema, PowerOutSchema2 from applications.schemas import RoleOutSchema, PowerOutSchema2
@@ -23,21 +23,15 @@ def main():
@admin_role.get('/data') @admin_role.get('/data')
@authorize("admin:role:main", log=True) @authorize("admin:role:main", log=True)
def table(): def table():
# 获取请求参数 role_name = str_escape(request.args.get('roleName', type=str))
role_name = xss_escape(request.args.get('roleName', type=str)) role_code = str_escape(request.args.get('roleCode', type=str))
role_code = xss_escape(request.args.get('roleCode', type=str)) filters = []
# 查询参数构造
mf = ModelFilter()
if role_name: if role_name:
mf.vague(field_name="name", value=role_name) filters.append(Role.name.contains(role_name))
if role_code: if role_code:
mf.vague(field_name="code", value=role_code) filters.append(Role.code.contains(role_code))
# orm查询 roles = Role.query.filter(*filters).layui_paginate()
# 使用分页获取data需要.items return table_api(data=model_to_dicts(schema=RoleOutSchema, data=roles.items), count=roles.total)
role = Role.query.filter(mf.get_filter(Role)).layui_paginate()
count = role.total
# 返回api
return table_api(data=model_to_dicts(schema=RoleOutSchema, data=role.items), count=count)
# 角色增加 # 角色增加
@@ -52,11 +46,11 @@ def add():
@authorize("admin:role:add", log=True) @authorize("admin:role:add", log=True)
def save(): def save():
req = request.json req = request.json
details = xss_escape(req.get("details")) details = str_escape(req.get("details"))
enable = xss_escape(req.get("enable")) enable = str_escape(req.get("enable"))
roleCode = xss_escape(req.get("roleCode")) roleCode = str_escape(req.get("roleCode"))
roleName = xss_escape(req.get("roleName")) roleName = str_escape(req.get("roleName"))
sort = xss_escape(req.get("sort")) sort = str_escape(req.get("sort"))
role = Role( role = Role(
details=details, details=details,
enable=enable, enable=enable,
@@ -132,11 +126,11 @@ def update():
req_json = request.json req_json = request.json
id = req_json.get("roleId") id = req_json.get("roleId")
data = { data = {
"code": xss_escape(req_json.get("roleCode")), "code": str_escape(req_json.get("roleCode")),
"name": xss_escape(req_json.get("roleName")), "name": str_escape(req_json.get("roleName")),
"sort": xss_escape(req_json.get("sort")), "sort": str_escape(req_json.get("sort")),
"enable": xss_escape(req_json.get("enable")), "enable": str_escape(req_json.get("enable")),
"details": xss_escape(req_json.get("details")) "details": str_escape(req_json.get("details"))
} }
role = Role.query.filter_by(id=id).update(data) role = Role.query.filter_by(id=id).update(data)
db.session.commit() db.session.commit()
applications/view/admin/user.py
+21 -19
View File
@@ -4,10 +4,9 @@ from sqlalchemy import desc
from applications.common import curd from applications.common import curd
from applications.common.curd import enable_status, disable_status from applications.common.curd import enable_status, disable_status
from applications.common.helper import ModelFilter
from applications.common.utils.http import table_api, fail_api, success_api from applications.common.utils.http import table_api, fail_api, success_api
from applications.common.utils.rights import authorize from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape from applications.common.utils.validate import str_escape
from applications.extensions import db from applications.extensions import db
from applications.models import Role, Dept from applications.models import Role, Dept
from applications.models import User, AdminLog from applications.models import User, AdminLog
@@ -27,17 +26,20 @@ def main():
@authorize("admin:user:main", log=True) @authorize("admin:user:main", log=True)
def data(): def data():
# 获取请求参数 # 获取请求参数
real_name = xss_escape(request.args.get('realName', type=str)) real_name = str_escape(request.args.get('realName', type=str))
username = xss_escape(request.args.get('username', type=str))
username = str_escape(request.args.get('username', type=str))
dept_id = request.args.get('deptId', type=int) dept_id = request.args.get('deptId', type=int)
# 查询参数构造
mf = ModelFilter() filters = []
if real_name: if real_name:
mf.contains(field_name="realname", value=real_name) filters.append(User.realname.contains(real_name))
if username: if username:
mf.contains(field_name="username", value=username) filters.append(User.realname.contains(username))
if dept_id: if dept_id:
mf.exact(field_name="dept_id", value=dept_id) filters.append(User.realname == dept_id)
# print(*filters)
data, count = db.session.query( data, count = db.session.query(
User.id, User.id,
User.username, User.username,
@@ -46,8 +48,8 @@ def data():
User.create_at, User.create_at,
User.update_at, User.update_at,
Dept.dept_name Dept.dept_name
).filter(mf.get_filter(model=User)).filter(User.dept_id==Dept.id).layui_paginate_db_json() ).filter(*filters).filter(User.dept_id == Dept.id).layui_paginate_db_json()
return table_api(data=data,count=count) return table_api(data=data, count=count)
# 用户增加 # 用户增加
@@ -63,9 +65,9 @@ def add():
def save(): def save():
req_json = request.json req_json = request.json
a = req_json.get("roleIds") a = req_json.get("roleIds")
username = xss_escape(req_json.get('username')) username = str_escape(req_json.get('username'))
real_name = xss_escape(req_json.get('realName')) real_name = str_escape(req_json.get('realName'))
password = xss_escape(req_json.get('password')) password = str_escape(req_json.get('password'))
role_ids = a.split(',') role_ids = a.split(',')
if not username or not real_name or not password: if not username or not real_name or not password:
@@ -114,11 +116,11 @@ def edit(id):
@authorize("admin:user:edit", log=True) @authorize("admin:user:edit", log=True)
def update(): def update():
req_json = request.json req_json = request.json
a = xss_escape(req_json.get("roleIds")) a = str_escape(req_json.get("roleIds"))
id = xss_escape(req_json.get("userId")) id = str_escape(req_json.get("userId"))
username = xss_escape(req_json.get('username')) username = str_escape(req_json.get('username'))
real_name = xss_escape(req_json.get('realName')) real_name = str_escape(req_json.get('realName'))
dept_id = xss_escape(req_json.get('deptId')) dept_id = str_escape(req_json.get('deptId'))
role_ids = a.split(',') role_ids = a.split(',')
User.query.filter_by(id=id).update({'username': username, 'realname': real_name, 'dept_id': dept_id}) User.query.filter_by(id=id).update({'username': username, 'realname': real_name, 'dept_id': dept_id})
u = User.query.filter_by(id=id).first() u = User.query.filter_by(id=id).first()
applications/view/department/__init__.py
+15 -15
View File
@@ -4,7 +4,7 @@ from applications.common import curd
from applications.common.utils import validate from applications.common.utils import validate
from applications.common.utils.http import success_api, fail_api from applications.common.utils.http import success_api, fail_api
from applications.common.utils.rights import authorize from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape from applications.common.utils.validate import str_escape
from applications.extensions import db from applications.extensions import db
from applications.models import Dept, User from applications.models import Dept, User
from applications.schemas import DeptOutSchema from applications.schemas import DeptOutSchema
@@ -58,13 +58,13 @@ def save():
req_json = request.json req_json = request.json
dept = Dept( dept = Dept(
parent_id=req_json.get('parentId'), parent_id=req_json.get('parentId'),
dept_name=xss_escape(req_json.get('deptName')), dept_name=str_escape(req_json.get('deptName')),
sort=xss_escape(req_json.get('sort')), sort=str_escape(req_json.get('sort')),
leader=xss_escape(req_json.get('leader')), leader=str_escape(req_json.get('leader')),
phone=xss_escape(req_json.get('phone')), phone=str_escape(req_json.get('phone')),
email=xss_escape(req_json.get('email')), email=str_escape(req_json.get('email')),
status=xss_escape(req_json.get('status')), status=str_escape(req_json.get('status')),
address=xss_escape(req_json.get('address')) address=str_escape(req_json.get('address'))
) )
r = db.session.add(dept) r = db.session.add(dept)
db.session.commit() db.session.commit()
@@ -115,13 +115,13 @@ def update():
json = request.json json = request.json
id = json.get("deptId"), id = json.get("deptId"),
data = { data = {
"dept_name": validate.xss_escape(json.get("deptName")), "dept_name": validate.str_escape(json.get("deptName")),
"sort": validate.xss_escape(json.get("sort")), "sort": validate.str_escape(json.get("sort")),
"leader": validate.xss_escape(json.get("leader")), "leader": validate.str_escape(json.get("leader")),
"phone": validate.xss_escape(json.get("phone")), "phone": validate.str_escape(json.get("phone")),
"email": validate.xss_escape(json.get("email")), "email": validate.str_escape(json.get("email")),
"status": validate.xss_escape(json.get("status")), "status": validate.str_escape(json.get("status")),
"address": validate.xss_escape(json.get("address")) "address": validate.str_escape(json.get("address"))
} }
d = Dept.query.filter_by(id=id).update(data) d = Dept.query.filter_by(id=id).update(data)
if not d: if not d: