sys_post_data.py 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
  1. #!/usr/bin/env python3
  2. # -*- coding: utf-8 -*-
  3. from . import mpfun
  4. from models import *
  5. from utils import *
  6. from sqlalchemy.orm import Session
  7. from database import get_local_db
  8. # 岗位表
  9. # 加密和HMAC签名
  10. def sign_row(db: Session, row: SysPost) -> None:
  11. if row.sign != '':
  12. return
  13. post_id = str(row.post_id) # 岗位ID
  14. dept_id = str(row.dept_id) # 部门ID
  15. post_code = row.post_code # 岗位代码
  16. post_name = mpfun.base64_data(row.post_name) # 岗位名称
  17. post_sort = str(row.post_sort) # 排序
  18. status = str(row.status) # 状态
  19. sign_data = ",".join([post_id, dept_id, post_code, post_name, post_sort, status])
  20. sign_hmac = mpfun.sign_data(sign_data)
  21. # print('sign_tbl_post sign_data:', sign_data)
  22. # print('sign_tbl_post sign_hmac:', sign_hmac)
  23. row.sign = sign_hmac
  24. db.commit()
  25. # 比较字段合并字符串是否和MAC值匹配上,调用密码服务器[验证HMAC]接口
  26. def sign_valid_row(row: SysPost) -> bool:
  27. return True
  28. if row.sign == '':
  29. return True
  30. # 关键字段合并字符串
  31. sign_data = get_sign_str(row)
  32. # print('sys_post sign_data:', sign_data)
  33. # 原HMACSM3数值
  34. sign_hmac = row.sign
  35. # print('sign_hmac:', sign_hmac)
  36. return mpfun.hmac_verify(sign_data, sign_hmac)
  37. # 生成待签名的字符串
  38. def get_sign_str(row: SysPost) -> str:
  39. post_id = str(row.post_id) # 岗位ID
  40. dept_id = str(row.dept_id) # 部门ID
  41. post_code = row.post_code # 岗位代码
  42. post_name = mpfun.base64_data(row.post_name) # 岗位名称
  43. post_sort = str(row.post_sort) # 排序
  44. status = str(row.status) # 状态
  45. # 关键字段合并字符串
  46. sign_data = ",".join([post_id, dept_id, post_code, post_name, post_sort, status])
  47. return sign_data
  48. # 生成HAMC签名值
  49. def get_sign_hmac(row: SysPost) -> str:
  50. sign_data = get_sign_str(row)
  51. return mpfun.sign_data(sign_data)
  52. # 对所有数据进行签名
  53. def sign_table():
  54. print('sign_sys_post table =====>>>')
  55. with get_local_db() as db:
  56. rows = db.query(SysPost).filter(SysPost.sign == '').all()
  57. for row in rows:
  58. sign_row(db, row)