#!/usr/bin/env python3 # -*- coding: utf-8 -*- from . import mpfun from models import * from utils import * from sqlalchemy.orm import Session from database import get_local_db # 操作日志表 # 加密和HMAC签名 def sign_row(db: Session, row: CzrzEntity) -> None: if row.sign != '': return user_id = str(row.user_id) # 用户ID user_name = mpfun.enc_data(row.user_name) # 用户账号 nick_name = mpfun.base64_data(row.nick_name) # 用户昵称 czrz = mpfun.base64_data(row.czrz) # 操作日志 gxsj = get_datetime_str(row.gxsj) # 更新时间 ip = str(row.ip) # IP地址 action = mpfun.base64_data(row.action) # 动作 sign_data = ",".join([user_id, user_name, nick_name, czrz, gxsj, ip, action]) sign_hmac = mpfun.sign_data(sign_data) # print('sign_tbl_czrz sign_data:', sign_data) # print('sign_tbl_czrz sign_hmac:', sign_hmac) row.user_name = user_name row.sign = sign_hmac db.commit() # 比较字段合并字符串是否和MAC值匹配上,调用密码服务器[验证HMAC]接口 def sign_valid_row(row: CzrzEntity) -> bool: if row.sign == '': return True # 关键字段合并字符串 sign_data = get_sign_str(row) # print('sys_user sign_data:', sign_data) # 原HMACSM3数值 sign_hmac = row.sign # print('sign_hmac:', sign_hmac) return mpfun.hmac_verify(sign_data, sign_hmac) # 生成待签名的字符串 def get_sign_str(row: CzrzEntity) -> str: user_id = str(row.user_id) # 用户ID user_name = row.user_name # 用户账号 nick_name = mpfun.base64_data(row.nick_name) # 用户昵称 czrz = mpfun.base64_data(row.czrz) # 操作日志 gxsj = get_datetime_str(row.gxsj) # 更新时间 ip = str(row.ip) # IP地址 action = mpfun.base64_data(row.action) # 动作 # 关键字段合并字符串 sign_data = ",".join([user_id, user_name, nick_name, czrz, gxsj, ip, action]) return sign_data # 生成HAMC签名值 def get_sign_hmac(row: CzrzEntity) -> str: sign_data = get_sign_str(row) return mpfun.sign_data(sign_data) # 对所有数据进行签名 def sign_table(): print('sign_tp_czrz table =====>>>') with get_local_db() as db: rows = db.query(CzrzEntity).filter(CzrzEntity.sign == '').all() for row in rows: sign_row(db, row)