Skip to content

A
Automat
Commit 678e6045 by Aeolus
Options

update

parent 78d26505
Showing with 4 additions and 9 deletions
myapps/automat/api/rent_portal.py
......@@ -83,7 +83,6 @@ def create_rent():
Hatch.left_count >= 1).order_by(func.random()).first()
if not open_hatch:
print("....")
return jsonify(HATCH_NOT_EXIST_ERROR)
break
......@@ -212,7 +211,6 @@ def wx_pay_callback():
logger.info(xml_data)
logger.info(callback_data)
attach = re.sub('\'', '\"', callback_data["attach"])
print(attach)
rent_data = json.loads(attach)
platform = rent_data["platform"]
machine_no = rent_data["machine_no"]
......
myapps/pc_management/api/file_protal.py
......@@ -33,7 +33,6 @@ def run_upload_img():
if filetype and filetype in ALLOWED_EXTENSIONS: # 后缀格式必须是bmp结尾
uid = uuid.uuid4() # 生成随机名称
save_file_name = str(uid) + "." + filetype # 拼接名称
print(os.path.join(img_file_path, filename))
file.save(os.path.join(img_file_path, save_file_name)) # 保存文件
return BaseResponse(data={"filename": save_file_name})
......
myapps/pc_management/api/hatch_portal.py
......@@ -66,7 +66,6 @@ def run_hatch_list():
return BaseResponse(data={"list": [], "page": page, "pageSize": page_size, "total_count": 0})
else:
total_count = count_result.total_count
print(select_sql + from_sql + where_sql + order_sql + limit_sql)
result = db.session.execute(select_sql + from_sql + where_sql + order_sql + limit_sql).fetchall()
return_data = []
......
myapps/pc_management/api/machine_portal.py
......@@ -62,7 +62,6 @@ def run_machine_list():
return BaseResponse(data={"list": [], "page": page, "pageSize": page_size, "total_count": 0})
else:
total_count = count_result.total_count
print(select_sql + from_sql + where_sql + order_sql + limit_sql)
result = db.session.execute(select_sql + from_sql + where_sql + order_sql + limit_sql).fetchall()
return_data = []
......
utils/middlewares.py
#!usr/bin/.env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @time: 2021/03/26 @file: middlewares.py @function: @modify: """ import logging from flask import g, request, url_for, current_app, make_response, jsonify from config.wechat_config import platform_config_list from models.models import WxUser, TallymanAccount, AdminAccount from utils.error_code import TOKEN_NOT_VALID_ERROR from utils.my_response import BaseResponse from utils.jwt_util import verify_jwt logger = logging.getLogger(__name__) def log_enter_interface(): """ 日志打印进入接口 :return: """ logger.info("######################### 进入 {} 接口 ################################ ".format(request.path)) def log_out_interface(environ): """ 日志打印退出接口 :return: """ logger.info("######################### 退出 {} 接口 ################################\n".format(request.path)) return environ def close_db_session(environ): from models.base_model import db db.session.close() return environ """用户认证机制==>每次请求前获取并校验token""" "@myapps.before_request 不使@调用装饰器 在 init文件直接装饰" def jwt_authentication(): """ 1.获取请求头Authorization中的token 2.判断是否以 Bearer开头 3.使用jwt模块进行校验 4.判断校验结果,成功就提取token中的载荷信息,赋值给g对象保存 """ path_list = request.path.split("/") if current_app.name == "automat": NO_AUTH_CHECK_URL = [url_for('wx_auth.my_test'), url_for('wx_auth.mini_login'), url_for('rent.wx_pay_callback'), url_for('hatch.get_production_list'), url_for('tallyman.run_tallyman_login'), url_for('machine.run_get_machine_no'), url_for('machine.run_create_machine_no'), url_for('machine.run_bind_serial_num'), url_for('nfc_card.run_nfc_card_wx_pay_callback'), url_for('nfc_card.run_nfc_card_user_pay_record'), url_for('nfc_card.run_nfc_card_load_succeed'), url_for('nfc_card.run_nfc_card_user_load_record'), ] if request.path not in NO_AUTH_CHECK_URL: token = request.headers.get('Authorization') # "校验token" payload = verify_jwt(token) # "判断token的校验结果" if payload: # "获取载荷中的信息赋值给g对象" if request.path.split("/")[2] == "tallyman": user_no = payload.get('user_no') if not user_no: return BaseResponse(**TOKEN_NOT_VALID_ERROR) try: g.user = TallymanAccount.query.filter_by(user_no=user_no).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: print(e) if request.path.split("/")[2] == "machine": user_no = payload.get('user_no', None) user_id = payload.get('user_id', None) if user_no: try: g.user = TallymanAccount.query.filter_by(user_no=user_no).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: print(e) return BaseResponse(**TOKEN_NOT_VALID_ERROR) if user_id: try: g.user = WxUser.query.filter_by(id=user_id).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: print(e) return BaseResponse(**TOKEN_NOT_VALID_ERROR) return BaseResponse(**TOKEN_NOT_VALID_ERROR) user_id = payload.get('user_id') if not user_id: return BaseResponse(**TOKEN_NOT_VALID_ERROR) try: g.user = WxUser.query.filter_by(id=user_id).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: print(e) return BaseResponse(**TOKEN_NOT_VALID_ERROR) else: return BaseResponse(**TOKEN_NOT_VALID_ERROR) elif current_app.name == "pc_management": NO_AUTH_CHECK_URL = [url_for("admin.user_login") ] if request.path.split("/")[2] == "file" and request.path.split("/")[3] == "img": # 图片接口不验证token return if request.path not in NO_AUTH_CHECK_URL: token = request.headers.get('Authorization') # "校验token" payload = verify_jwt(token) # "判断token的校验结果" if payload: # "获取载荷中的信息赋值给g对象" user_id = payload.get('user_id', None) if user_id: g.user = AdminAccount.query.filter_by(id=user_id).first() if g.user: return return BaseResponse(**TOKEN_NOT_VALID_ERROR) else: NO_AUTH_CHECK_URL = [] return def get_platform(): """ :return: """ g.platform = request.headers.get('platform', "sukang24h") def all_options_pass(): """ :return: """ if request.method == "OPTIONS": headers = {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'POST', 'Access-Control-Allow-Headers': 'Content-Type, Content-Length, Authorization, Accept, X-Requested-With , platform', } return make_response((jsonify({'error_code': 0}), 200, headers))
\ No newline at end of file
#!usr/bin/.env python # -*- coding:utf-8 _*- """ @version: author:Aeolus @time: 2021/03/26 @file: middlewares.py @function: @modify: """ import logging from flask import g, request, url_for, current_app, make_response, jsonify from config.wechat_config import platform_config_list from models.models import WxUser, TallymanAccount, AdminAccount from utils.error_code import TOKEN_NOT_VALID_ERROR from utils.my_response import BaseResponse from utils.jwt_util import verify_jwt logger = logging.getLogger(__name__) def log_enter_interface(): """ 日志打印进入接口 :return: """ logger.info("######################### 进入 {} 接口 ################################ ".format(request.path)) def log_out_interface(environ): """ 日志打印退出接口 :return: """ logger.info("######################### 退出 {} 接口 ################################\n".format(request.path)) return environ def close_db_session(environ): from models.base_model import db db.session.close() return environ """用户认证机制==>每次请求前获取并校验token""" "@myapps.before_request 不使@调用装饰器 在 init文件直接装饰" def jwt_authentication(): """ 1.获取请求头Authorization中的token 2.判断是否以 Bearer开头 3.使用jwt模块进行校验 4.判断校验结果,成功就提取token中的载荷信息,赋值给g对象保存 """ path_list = request.path.split("/") if current_app.name == "automat": NO_AUTH_CHECK_URL = [url_for('wx_auth.my_test'), url_for('wx_auth.mini_login'), url_for('rent.wx_pay_callback'), url_for('hatch.get_production_list'), url_for('tallyman.run_tallyman_login'), url_for('machine.run_get_machine_no'), url_for('machine.run_create_machine_no'), url_for('machine.run_bind_serial_num'), url_for('nfc_card.run_nfc_card_wx_pay_callback'), url_for('nfc_card.run_nfc_card_user_pay_record'), url_for('nfc_card.run_nfc_card_load_succeed'), url_for('nfc_card.run_nfc_card_user_load_record'), ] if request.path not in NO_AUTH_CHECK_URL: token = request.headers.get('Authorization') # "校验token" payload = verify_jwt(token) # "判断token的校验结果" if payload: # "获取载荷中的信息赋值给g对象" if request.path.split("/")[2] == "tallyman": user_no = payload.get('user_no') if not user_no: return BaseResponse(**TOKEN_NOT_VALID_ERROR) try: g.user = TallymanAccount.query.filter_by(user_no=user_no).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: logger.info(e) if request.path.split("/")[2] == "machine": user_no = payload.get('user_no', None) user_id = payload.get('user_id', None) if user_no: try: g.user = TallymanAccount.query.filter_by(user_no=user_no).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: logger.info(e) return BaseResponse(**TOKEN_NOT_VALID_ERROR) if user_id: try: g.user = WxUser.query.filter_by(id=user_id).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: logger.info(e) return BaseResponse(**TOKEN_NOT_VALID_ERROR) return BaseResponse(**TOKEN_NOT_VALID_ERROR) user_id = payload.get('user_id') if not user_id: return BaseResponse(**TOKEN_NOT_VALID_ERROR) try: g.user = WxUser.query.filter_by(id=user_id).first() if not g.user: return BaseResponse(**TOKEN_NOT_VALID_ERROR) return except Exception as e: logger.info(e) return BaseResponse(**TOKEN_NOT_VALID_ERROR) else: return BaseResponse(**TOKEN_NOT_VALID_ERROR) elif current_app.name == "pc_management": NO_AUTH_CHECK_URL = [url_for("admin.user_login") ] if request.path.split("/")[2] == "file" and request.path.split("/")[3] == "img": # 图片接口不验证token return if request.path not in NO_AUTH_CHECK_URL: token = request.headers.get('Authorization') # "校验token" payload = verify_jwt(token) # "判断token的校验结果" if payload: # "获取载荷中的信息赋值给g对象" user_id = payload.get('user_id', None) if user_id: g.user = AdminAccount.query.filter_by(id=user_id).first() if g.user: return return BaseResponse(**TOKEN_NOT_VALID_ERROR) else: NO_AUTH_CHECK_URL = [] return def get_platform(): """ :return: """ g.platform = request.headers.get('platform', "sukang24h") def all_options_pass(): """ :return: """ if request.method == "OPTIONS": headers = {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'POST', 'Access-Control-Allow-Headers': 'Content-Type, Content-Length, Authorization, Accept, X-Requested-With , platform', } return make_response((jsonify({'error_code': 0}), 200, headers))
\ No newline at end of file
......
utils/wechat/WXBizDataCrypt.py
import base64import jsonfrom Crypto.Cipher import AES class WXBizDataCrypt: def __init__(self, appId, sessionKey): self.appId = appId self.sessionKey = sessionKey def decrypt(self, encryptedData, iv): sessionKey = base64.decodebytes(bytes(self.sessionKey, encoding='utf8')) encryptedData = base64.decodebytes(bytes(encryptedData, encoding='utf8')) iv = base64.decodebytes(bytes(iv, encoding='utf8')) cipher = AES.new(sessionKey, AES.MODE_CBC, iv) des_str = cipher.decrypt(encryptedData) print("==================================") print(des_str) des_str = self._unpad(des_str) print(des_str) des_str = str(des_str,encoding='utf-8') decrypted = json.loads(des_str) if decrypted['watermark']['appid'] != self.appId: raise Exception('Invalid Buffer') return decrypted def _unpad(self, s): return s[:-ord(s[len(s) - 1:])] if __name__ == '__main__': appId = 'wx3185fb4a3633beb0' sessionKey='S7CMDfC6jXJKSaWKanG8oQ==' encryptedData='E7LZhvK7mOcaYsv9xcAfsBN9eSbzFh9FyMtFJ0zsFB0M62zRJ0cosZWksUujUR5WYUmNoIfIJnTIF8gRskxxbFU3fm5X7z4ChZecMSaFM65aEK1suRUD1U0ubB7mOwBBlY4ftdPT5kRwWgXKVkM4VAkYGN8A4fjWE93yGtjzxXs9dypQkCLSNWs6Kw5USEzjhtDZnptVy+lHF5fTXRuzoCstW2Cto4YI3G9hmnS64QuWjRteSqIgh8GN1zEPN0dROJjaWBjqraBCt/BfMsk4HBeL4PA75K8WdqVgKGfQ7/rnmPFOsNXWfajx9jl7XcrfoPaaPL1DmIJ1BlQne2GuLFtzZ3O4/8cdVQ9Lb0N/3kFAzjgzNFNLSYj2VNctmWyLdWi8hH90yslvrODIhMzIsuux2GIAfp0rQd/iVIVvtd7PXBOCe5iZ7aaqD0b0mLF4CmsuBpl8Eh20ZHkYw2SqO0x9uFrS/gy1vwtkmsTpcDw=' iv = 'DQcmcXyQkU+VKqb2mKmasQ==' pc = WXBizDataCrypt(appId, sessionKey) pc.decrypt(encryptedData, iv)#
\ No newline at end of file
import base64import jsonfrom Crypto.Cipher import AES class WXBizDataCrypt: def __init__(self, appId, sessionKey): self.appId = appId self.sessionKey = sessionKey def decrypt(self, encryptedData, iv): sessionKey = base64.decodebytes(bytes(self.sessionKey, encoding='utf8')) encryptedData = base64.decodebytes(bytes(encryptedData, encoding='utf8')) iv = base64.decodebytes(bytes(iv, encoding='utf8')) cipher = AES.new(sessionKey, AES.MODE_CBC, iv) des_str = cipher.decrypt(encryptedData) des_str = self._unpad(des_str) des_str = str(des_str, encoding='utf-8') decrypted = json.loads(des_str) if decrypted['watermark']['appid'] != self.appId: raise Exception('Invalid Buffer') return decrypted def _unpad(self, s): return s[:-ord(s[len(s) - 1:])] if __name__ == '__main__': appId = 'wx3185fb4a3633beb0' sessionKey = 'S7CMDfC6jXJKSaWKanG8oQ==' encryptedData = 'E7LZhvK7mOcaYsv9xcAfsBN9eSbzFh9FyMtFJ0zsFB0M62zRJ0cosZWksUujUR5WYUmNoIfIJnTIF8gRskxxbFU3fm5X7z4ChZecMSaFM65aEK1suRUD1U0ubB7mOwBBlY4ftdPT5kRwWgXKVkM4VAkYGN8A4fjWE93yGtjzxXs9dypQkCLSNWs6Kw5USEzjhtDZnptVy+lHF5fTXRuzoCstW2Cto4YI3G9hmnS64QuWjRteSqIgh8GN1zEPN0dROJjaWBjqraBCt/BfMsk4HBeL4PA75K8WdqVgKGfQ7/rnmPFOsNXWfajx9jl7XcrfoPaaPL1DmIJ1BlQne2GuLFtzZ3O4/8cdVQ9Lb0N/3kFAzjgzNFNLSYj2VNctmWyLdWi8hH90yslvrODIhMzIsuux2GIAfp0rQd/iVIVvtd7PXBOCe5iZ7aaqD0b0mLF4CmsuBpl8Eh20ZHkYw2SqO0x9uFrS/gy1vwtkmsTpcDw=' iv = 'DQcmcXyQkU+VKqb2mKmasQ==' pc = WXBizDataCrypt(appId, sessionKey) pc.decrypt(encryptedData, iv)#
\ No newline at end of file
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment