#!/usr/bin/env python3 # -*- coding: utf-8 -*- from . import mpfun from models import * from sqlalchemy.orm import Session from database import get_local_db # 系统部门表 # 加密和HMAC签名 def sign_row(db: Session, row: SysDept) -> None: if row.sign != '': return dept_id = str(row.dept_id) # 部门id parent_id = str(row.parent_id) # 父部门id parent_name = mpfun.base64_data(row.parent_name) # 父部门名称 ancestors = mpfun.base64_data(row.ancestors) # 祖级列表 dept_name = mpfun.base64_data(row.dept_name) # 部门名称 dept_category = mpfun.base64_data(row.dept_category) # 部门类别编码 order_num = str(row.order_num) # 显示顺序 leader_name = mpfun.base64_data(row.leader_name) # 负责人姓名 leader = str(row.leader) # 负责人 phone = mpfun.enc_data(row.phone) # 联系电话 email = mpfun.enc_data(row.email) # 邮箱 status = str(row.status) # 部门状态 del_flag = row.del_flag # 是否已删除 sign_data = ",".join([dept_id, parent_id, parent_name, ancestors, dept_name, dept_category, order_num, leader_name, leader, phone, email, status, del_flag]) sign_hmac = mpfun.sign_data(sign_data) # print('sign_tbl_user sign_data:', sign_data) # print('sign_tbl_user sign_hmac:', sign_hmac) row.phone = phone row.email = email row.sign = sign_hmac db.commit() # 比较字段合并字符串是否和MAC值匹配上,调用密码服务器[验证HMAC]接口 def sign_valid_row(row: SysDept) -> 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: SysDept) -> str: dept_id = str(row.dept_id) # 部门id parent_id = str(row.parent_id) # 父部门id parent_name = mpfun.base64_data(row.parent_name) # 父部门名称 ancestors = mpfun.base64_data(row.ancestors) # 祖级列表 dept_name = mpfun.base64_data(row.dept_name) # 部门名称 dept_category = mpfun.base64_data(row.dept_category) # 部门类别编码 order_num = str(row.order_num) # 显示顺序 leader_name = mpfun.base64_data(row.leader_name) # 负责人姓名 leader = str(row.leader) # 负责人 phone = mpfun.enc_data(row.phone) # 联系电话 email = mpfun.enc_data(row.email) # 邮箱 status = str(row.status) # 部门状态 del_flag = row.del_flag # 是否已删除 # 关键字段合并字符串 sign_data = ",".join([dept_id, parent_id, parent_name, ancestors, dept_name, dept_category, order_num, leader_name, leader, phone, email, status, del_flag]) return sign_data # 生成HAMC签名值 def get_sign_hmac(row: SysDept) -> str: sign_data = get_sign_str(row) return mpfun.sign_data(sign_data) # 对所有数据进行签名 def sign_table(): print('sign_sys_dept table =====>>>') with get_local_db() as db: rows = db.query(SysDept).filter(SysDept.sign == '').all() for row in rows: sign_row(db, row)