增加xss过滤

applications/service/admin/dept_curd.py

@@ -2,6 +2,7 @@ from applications.models import db
 from applications.models.admin_dept import Dept, DeptSchema
 from applications.models.admin_user import User
 from applications.service.common.curd import model_to_dicts
+from applications.service.common.validate import xss_escape
 
 
 def get_dept_dict():
@@ -11,14 +12,15 @@ def get_dept_dict():
 
 
 def save_dept(req):
-    address = req.get("address")
-    deptName = req.get("deptName")
-    email = req.get("email")
-    leader = req.get("leader")
-    parentId = req.get("parentId")
-    phone = req.get("phone")
-    sort = req.get("sort")
-    status = req.get("status")
+
+    address = xss_escape(req.get("address"))
+    deptName = xss_escape(req.get("deptName"))
+    email = xss_escape(req.get("email"))
+    leader = xss_escape(req.get("leader"))
+    parentId = xss_escape(req.get("parentId"))
+    phone = xss_escape(req.get("phone"))
+    sort = xss_escape(req.get("sort"))
+    status = xss_escape(req.get("status"))
     dept = Dept(
         parent_id=parentId,
         dept_name=deptName,
@@ -61,15 +63,14 @@ def disable_status(id):
 
 def update_dept(json):
     id = json.get("deptId"),
-    print(json.get("leader"))
     data = {
-        "dept_name": json.get("deptName"),
-        "sort": json.get("sort"),
-        "leader": json.get("leader"),
-        "phone": json.get("phone"),
-        "email": json.get("email"),
-        "status": json.get("status"),
-        "address": json.get("address")
+        "dept_name": xss_escape(json.get("deptName")),
+        "sort": xss_escape(json.get("sort")),
+        "leader": xss_escape(json.get("leader")),
+        "phone": xss_escape(json.get("phone")),
+        "email": xss_escape(json.get("email")),
+        "status": xss_escape(json.get("status")),
+        "address": xss_escape(json.get("address"))
     }
     d = Dept.query.filter_by(id=id).update(data)
     if not d:

applications/service/admin/dict_curd.py

@@ -5,7 +5,7 @@
 from applications.models import db
 from applications.models.admin_dict import DictType, DictData, DictTypeSchema, DictDataSchema
 from applications.service.common.curd import model_to_dicts
-
+from applications.service.common.validate import xss_escape
 
 # 通过type_code获取字典dict
 # 例：get_dict('user_sex')
@@ -47,10 +47,10 @@ def get_dict_data(page, limit, type_code):
 
 # 增加dicttype
 def save_dict_type(req_json):
-    description = req_json.get("description")
-    enable = req_json.get("enable")
-    type_code = req_json.get("typeCode")
-    type_name = req_json.get("typeName")
+    description = xss_escape(req_json.get("description"))
+    enable = xss_escape(req_json.get("enable"))
+    type_code = xss_escape(req_json.get("typeCode"))
+    type_name = xss_escape(req_json.get("typeName"))
     d = DictType(type_name=type_name, type_code=type_code, enable=enable, description=description)
     db.session.add(d)
     db.session.commit()
@@ -59,11 +59,11 @@ def save_dict_type(req_json):
 
 # 编辑字典类型
 def update_dict_type(req_json):
-    id = req_json.get("id")
-    description = req_json.get("description")
-    enable = req_json.get("enable")
-    type_code = req_json.get("typeCode")
-    type_name = req_json.get("typeName")
+    id = xss_escape(req_json.get("id"))
+    description = xss_escape(req_json.get("description"))
+    enable = xss_escape(req_json.get("enable"))
+    type_code = xss_escape(req_json.get("typeCode"))
+    type_name = xss_escape(req_json.get("typeName"))
     DictType.query.filter_by(id=id).update({
         "description": description,
         "enable": enable,
@@ -103,11 +103,11 @@ def delete_type_by_id(id):
 
 # 增加dictdata
 def save_dict_data(req_json):
-    data_label = req_json.get("dataLabel")
-    data_value = req_json.get("dataValue")
-    enable = req_json.get("enable")
-    remark = req_json.get("remark")
-    type_code = req_json.get("typeCode")
+    data_label = xss_escape(req_json.get("dataLabel"))
+    data_value = xss_escape(req_json.get("dataValue"))
+    enable = xss_escape(req_json.get("enable"))
+    remark = xss_escape(req_json.get("remark"))
+    type_code = xss_escape(req_json.get("typeCode"))
     d = DictData(data_label=data_label, data_value=data_value, enable=enable, remark=remark, type_code=type_code)
     db.session.add(d)
     db.session.commit()
@@ -118,11 +118,11 @@ def save_dict_data(req_json):
 def update_dict_data(req_json):
     id = req_json.get("dataId")
     DictData.query.filter_by(id=id).update({
-        "data_label": req_json.get("dataLabel"),
-        "data_value": req_json.get("dataValue"),
-        "enable": req_json.get("enable"),
-        "remark": req_json.get("remark"),
-        "type_code": req_json.get("typeCode")
+        "data_label": xss_escape(req_json.get("dataLabel")),
+        "data_value": xss_escape(req_json.get("dataValue")),
+        "enable": xss_escape(req_json.get("enable")),
+        "remark": xss_escape(req_json.get("remark")),
+        "type_code": xss_escape(req_json.get("typeCode"))
     })
     db.session.commit()
     return

applications/service/admin/power_curd.py

@@ -2,6 +2,7 @@ from applications.models import db
 from applications.models.admin_power import Power, PowerSchema2
 from applications.models.admin_role import Role
 from applications.service.common.curd import model_to_dicts
+from applications.service.common.validate import xss_escape
 
 
 def get_power_dict():
@@ -19,14 +20,14 @@ def select_parent():
 
 # 增加权限
 def save_power(req):
-    icon = req.get("icon")
-    openType = req.get("openType")
-    parentId = req.get("parentId")
-    powerCode = req.get("powerCode")
-    powerName = req.get("powerName")
-    powerType = req.get("powerType")
-    powerUrl = req.get("powerUrl")
-    sort = req.get("sort")
+    icon = xss_escape(req.get("icon"))
+    openType = xss_escape(req.get("openType"))
+    parentId = xss_escape(req.get("parentId"))
+    powerCode = xss_escape(req.get("powerCode"))
+    powerName = xss_escape(req.get("powerName"))
+    powerType = xss_escape(req.get("powerType"))
+    powerUrl = xss_escape(req.get("powerUrl"))
+    sort = xss_escape(req.get("sort"))
     power = Power(
         icon=icon,
         open_type=openType,
@@ -53,19 +54,19 @@ def get_power_by_id(id):
 def update_power(req_json):
     id = req_json.get("powerId")
     data = {
-        "icon": req_json.get("icon"),
-        "open_type": req_json.get("openType"),
-        "parent_id": req_json.get("parentId"),
-        "code": req_json.get("powerCode"),
-        "name": req_json.get("powerName"),
-        "type": req_json.get("powerType"),
-        "url": req_json.get("powerUrl"),
-        "sort": req_json.get("sort")
+        "icon": xss_escape(req_json.get("icon")),
+        "open_type": xss_escape(req_json.get("openType")),
+        "parent_id": xss_escape(req_json.get("parentId")),
+        "code": xss_escape(req_json.get("powerCode")),
+        "name": xss_escape(req_json.get("powerName")),
+        "type": xss_escape(req_json.get("powerType")),
+        "url": xss_escape(req_json.get("powerUrl")),
+        "sort": xss_escape(req_json.get("sort"))
     }
-    print(data)
+    # print(data)
     power = Power.query.filter_by(id=id).update(data)
     db.session.commit()
-    print(power)
+    # print(power)
     return power
 
 

applications/service/admin/role_curd.py

@@ -6,6 +6,7 @@ from applications.models.admin_user import User
 
 # 获取角色对象
 from applications.service.common.curd import model_to_dicts
+from applications.service.common.validate import xss_escape
 
 
 def get_role_data(page, limit, filters):
@@ -25,11 +26,11 @@ def get_role_data_dict(page, limit, filters):
 
 # 增加角色
 def add_role(req):
-    details = req.get("details")
-    enable = req.get("enable")
-    roleCode = req.get("roleCode")
-    roleName = req.get("roleName")
-    sort = req.get("sort")
+    details = xss_escape(req.get("details"))
+    enable = xss_escape(req.get("enable"))
+    roleCode = xss_escape(req.get("roleCode"))
+    roleName = xss_escape(req.get("roleName"))
+    sort = xss_escape(req.get("sort"))
     role = Role(
         details=details,
         enable=enable,
@@ -51,13 +52,12 @@ def get_role_by_id(id):
 def update_role(req_json):
     id = req_json.get("roleId")
     data = {
-        "code": req_json.get("roleCode"),
-        "name": req_json.get("roleName"),
-        "sort": req_json.get("sort"),
-        "enable": req_json.get("enable"),
-        "details": req_json.get("details")
+        "code": xss_escape(req_json.get("roleCode")),
+        "name": xss_escape(req_json.get("roleName")),
+        "sort": xss_escape(req_json.get("sort")),
+        "enable": xss_escape(req_json.get("enable")),
+        "details": xss_escape(req_json.get("details"))
     }
-    print(data)
     role = Role.query.filter_by(id=id).update(data)
     db.session.commit()
     return role
@@ -87,8 +87,8 @@ def update_role_power(id, power_list):
     power_id_list = []
     for p in role.power:
         power_id_list.append(p.id)
-        print(p.id)
-    print(power_id_list)
+        # print(p.id)
+    # print(power_id_list)
     powers = Power.query.filter(Power.id.in_(power_id_list)).all()
     for p in powers:
         role.power.remove(p)

applications/service/common/validate.py

@@ -0,0 +1,6 @@
+# xss过滤
+def xss_escape(s: str):
+    if s is None:
+        return None
+    else:
+        return s.replace("&", "&amp;").replace(">", "&gt;").replace("<", "&lt;").replace("'", "&#39;").replace('"', "&#34;")

applications/views/admin/dept.py

@@ -41,7 +41,7 @@ def tree():
 
 
 @admin_dept.route('/save', methods=['POST'])
-@authorize("admin:dept:edit", log=True)
+@authorize("admin:dept:add", log=True)
 def save():
     req = request.json
     dept_curd.save_dept(req)

applications/views/admin/dict.py

@@ -2,6 +2,7 @@ from flask import Blueprint, render_template, request, jsonify
 from applications.models.admin_dict import DictType, DictData
 from applications.service.admin import dict_curd
 from applications.service.common.response import table_api, success_api, fail_api
+from applications.service.common.validate import xss_escape
 from applications.service.route_auth import authorize
 
 admin_dict = Blueprint('adminDict', __name__, url_prefix='/admin/dict')
@@ -19,7 +20,7 @@ def main():
 def dictType_data():
     page = request.args.get('page', type=int)
     limit = request.args.get('limit', type=int)
-    type_name = request.args.get('typeName', type=str)
+    type_name = xss_escape(request.args.get('typeName', type=str))
     data, count = dict_curd.get_dict_type(page=page, limit=limit, type_name=type_name)
     return table_api(data=data,count=count)
 
@@ -44,7 +45,7 @@ def dictType_save():
 @admin_dict.route('/dictType/edit', methods=['GET', 'POST'])
 @authorize("admin:dict:edit", log=True)
 def dictType_edit():
-    id = request.args.get('dictTypeId', type=str)
+    id = request.args.get('dictTypeId', type=int)
     dict_type = DictType.query.filter_by(id=id).first()
     return render_template('admin/dict/edit.html', dict_type=dict_type)
 
@@ -99,7 +100,7 @@ def dictType_delete(id):
 def dictCode_data():
     page = request.args.get('page', type=int)
     limit = request.args.get('limit', type=int)
-    type_code = request.args.get('typeCode', type=str)
+    type_code = xss_escape(request.args.get('typeCode', type=str))
     data, count = dict_curd.get_dict_data(page=page, limit=limit, type_code=type_code)
     return table_api(data=data,count=count)
 
@@ -117,7 +118,7 @@ def dictData_add():
 @authorize("admin:dict:add", log=True)
 def dictData_save():
     req_json = request.json
-    res = save_dict_data(req_json=req_json)
+    res = dict_curd.save_dict_data(req_json=req_json)
     if res == None:
         return jsonify(success=False, msg="增加失败")
     return jsonify(success=True, msg="增加成功")
@@ -137,7 +138,7 @@ def dictData_edit():
 @authorize("admin:dict:edit", log=True)
 def dictData_update():
     req_json = request.json
-    update_dict_data(req_json)
+    dict_curd.update_dict_data(req_json)
     return success_api(msg="更新成功")
 
 
@@ -147,7 +148,7 @@ def dictData_update():
 def dictData_enable():
     id = request.json.get('dataId')
     if id:
-        res = enable_dict_data_status(id)
+        res = dict_curd.enable_dict_data_status(id)
         if not res:
             return fail_api(msg="出错啦")
         return success_api(msg="启动成功")
@@ -160,7 +161,7 @@ def dictData_enable():
 def dictData_disenable():
     id = request.json.get('dataId')
     if id:
-        res = disable_dict_data_status(id)
+        res = dict_curd.isable_dict_data_status(id)
         if not res:
             return fail_api(msg="出错啦")
         return success_api(msg="禁用成功")
@@ -171,7 +172,7 @@ def dictData_disenable():
 @admin_dict.route('dictData/remove/<int:id>', methods=['DELETE'])
 @authorize("admin:dict:remove", log=True)
 def dictData_delete(id):
-    res = delete_data_by_id(id)
+    res = dict_curd.delete_data_by_id(id)
     if not res:
         return fail_api(msg="删除失败")
     return success_api(msg="删除成功")

applications/views/admin/role.py

@@ -2,6 +2,7 @@ from flask import Blueprint, render_template, request, jsonify
 from flask_login import login_required
 from applications.service.admin import role_curd
 from applications.service.common.response import table_api, success_api, fail_api
+from applications.service.common.validate import xss_escape
 from applications.service.route_auth import authorize
 
 admin_role = Blueprint('adminRole', __name__, url_prefix='/admin/role')
@@ -20,8 +21,8 @@ def main():
 def table():
     page = request.args.get('page', type=int)
     limit = request.args.get('limit', type=int)
-    roleName = request.args.get('roleName', type=str)
-    roleCode = request.args.get('roleCode', type=str)
+    roleName = xss_escape(request.args.get('roleName', type=str))
+    roleCode = xss_escape(request.args.get('roleCode', type=str))
     filters = {}
     if roleName:
         filters["name"] = ('%' + roleName + '%')
@@ -102,7 +103,7 @@ def update():
 @authorize("admin:role:edit", log=True)
 def enable():
     id = request.json.get('roleId')
-    print(id)
+    # print(id)
     if id:
         res = role_curd.enable_status(id)
         if not res:

applications/views/admin/user.py

@@ -4,6 +4,7 @@ from applications.models.admin_user import User
 from applications.models.admin_role import Role
 from applications.service.admin import user_curd
 from applications.service.common.response import table_api, fail_api, success_api
+from applications.service.common.validate import xss_escape
 from applications.service.route_auth import authorize
 
 admin_user = Blueprint('adminUser', __name__, url_prefix='/admin/user')
@@ -22,8 +23,8 @@ def main():
 def data():
     page = request.args.get('page', type=int)
     limit = request.args.get('limit', type=int)
-    realName = request.args.get('realName', type=str)
-    username = request.args.get('username', type=str)
+    realName = xss_escape(request.args.get('realName', type=str))
+    username = xss_escape(request.args.get('username', type=str))
     deptId = request.args.get('deptId', type=int)
     filters = {}
     if realName:
@@ -47,9 +48,9 @@ def add():
 def save():
     req_json = request.json
     a = req_json.get("roleIds")
-    username = req_json.get('username')
-    realName = req_json.get('realName')
-    password = req_json.get('password')
+    username = xss_escape(req_json.get('username'))
+    realName = xss_escape(req_json.get('realName'))
+    password = xss_escape(req_json.get('password'))
     role_ids = a.split(',')
 
     if not username or not realName or not password:
@@ -91,11 +92,11 @@ def edit(id):
 @authorize("admin:user:edit", log=True)
 def update():
     req_json = request.json
-    a = req_json.get("roleIds")
-    id = req_json.get("userId")
-    username = req_json.get('username')
-    realName = req_json.get('realName')
-    deptId = req_json.get('deptId')
+    a = xss_escape(req_json.get("roleIds"))
+    id = xss_escape(req_json.get("userId"))
+    username = xss_escape(req_json.get('username'))
+    realName = xss_escape(req_json.get('realName'))
+    deptId = xss_escape(req_json.get('deptId'))
     role_ids = a.split(',')
     user_curd.update_user(id, username, realName,deptId)
     user_curd.update_user_role(id, role_ids)