no message

common/enc/__init__.py

@@ -0,0 +1,2 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-

common/enc/mpfun.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from common import TassApi
+import base64
+
+# 将原数据按密评要求加密辅助类
+# 启动时检查相关表，如果sign为空，按规则将关键字段进行加密，将记录进行HMAC运算
+
+# 调用密码服务器[敏感信息数据加密]接口
+def enc_data(data: any)->str:
+    return TassApi.TransparentEnc(str(data))
+
+# 调用密码服务器[敏感信息数据解密]接口
+def dec_data(data: any)->str:
+    return TassApi.TransparentDec(str(data))
+
+# 调用密码服务器[计算HMAC]接口
+def sign_data(data: any)->str:
+    return TassApi.Hmac(str(data))
+
+# 对含有中文的数据进行BASE64处理
+def base64_data(val: str):
+    return str(base64.b64encode(val.encode("utf-8")), "utf-8")
+
+# 比较字段合并字符串是否和MAC值匹配上,调用密码服务器[验证HMAC]接口
+def hmac_verify(sign_data: str, sign_hmac: str) -> bool:
+    return TassApi.HmacVerify(sign_data, sign_hmac) 

common/enc/sys_user_data.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from . import mpfun
+from models import *
+from sqlalchemy.orm import Session
+
+# 系统用户表
+
+def sign_row(db: Session, row: SysUser) -> None:
+    if row.sign != '':
+        return
+    
+    user_id = str(row.user_id)
+    user_name = mpfun.enc_data(row.user_name)
+    password = mpfun.enc_data(row.password)
+    nick_name = mpfun.base64_data(row.nick_name)
+    dept_id = str(row.dept_id)
+    dept_name = mpfun.base64_data(row.dept_name)
+    email = mpfun.enc_data(row.email)
+    phonenumber = mpfun.enc_data(row.phonenumber)
+    status = str(row.status)
+    del_flag = row.del_flag
+    yzy_account = mpfun.enc_data(row.yzy_account)
+
+    sign_data = ",".join([user_id, user_name, password, nick_name, dept_id, dept_name, email, phonenumber, status, del_flag, yzy_account])
+    sign_hmac = mpfun.sign_data(sign_data)
+    print('sign_tbl_user sign_data:', sign_data)
+    print('sign_tbl_user sign_hmac:', sign_hmac)
+
+    row.user_name = user_name
+    row.password = password
+    row.email = email
+    row.phonenumber = phonenumber
+    row.yzy_account = yzy_account
+    row.sign = sign_hmac
+    
+    db.commit()
+
+# 比较字段合并字符串是否和MAC值匹配上,调用密码服务器[验证HMAC]接口
+def sign_valid_sign_row(row: SysUser) -> bool:
+    if row.sign == '':
+        return True
+    
+    user_id = str(row.user_id)
+    user_name = row.user_name
+    password = row.password
+    nick_name = mpfun.base64_data(row.nick_name)
+    dept_id = str(row.dept_id)
+    dept_name = mpfun.base64_data(row.dept_name)
+    email = row.email
+    phonenumber = row.phonenumber
+    status = str(row.status)
+    del_flag = row.del_flag
+    yzy_account = row.yzy_account
+
+    # 原HMACSM3数值
+    sign_hmac = row.sign
+    print('sign_hmac:', sign_hmac)
+
+    # 关键字段合并字符串
+    sign_data = ",".join([user_id, user_name, password, nick_name, dept_id, dept_name, email, phonenumber, status, del_flag, yzy_account])
+    print('sys_user resign_data:', sign_data)
+    
+    # 比较字段合并字符串是否和MAC值匹配上,调用密码服务器[验证HMAC]接口
+    return mpfun.hmac_verify(sign_data, sign_hmac)

common/security.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-from fastapi import Header
+from fastapi import Header, Depends
 from datetime import datetime, timedelta
 import jwt
 from passlib.context import CryptContext
@@ -22,26 +22,26 @@ def valid_access_token(Authorization: str = Header(..., alias="Authorization"))
 
         token_exception = TokenException()
         payload = jwt.decode(access_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
-        print(payload,payload.get("sub"))
+        # print(payload,payload.get("sub"))
         user_id: str = payload.get("sub")
-        logger.info('sub user_id: {}', user_id)
+        # logger.info('sub user_id: {}', user_id)
     except Exception:
         raise token_exception
     
     return int(user_id)
 
 
-def valid_access_token_role(Authorization: str = Header(..., alias="Authorization")) -> int:
+def valid_access_token_role(Authorization: str = Header(..., alias="Authorization"), db: Session = Depends(get_db)) -> int:
     try:
         access_token = Authorization.removeprefix("Bearer ")
 
         token_exception = TokenException()
         payload = jwt.decode(access_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
-        print(payload, payload.get("sub"))
+        # print(payload, payload.get("sub"))
         user_id: str = payload.get("sub")
-        logger.info('sub user_id: {}', user_id)
+        # logger.info('sub user_id: {}', user_id)
         role_list = ["superadmin","super_ld","super_worker"]
-        db= get_db_local()
+        # db= get_db_local()
         role_id_list = [info.role_id for info in db.query(SysRole).filter(SysRole.role_key.in_(role_list)).all()]
         if db.query(SysUserRole).filter(SysUserRole.role_id.in_(role_id_list),SysUserRole.user_id==user_id).first() is None:
             raise RoleException(errcode=4003, errmsg="权限不够")

database.py

@@ -15,14 +15,32 @@ mysql_dwd_config = {
     'database': settings.MYSQL_DB_NAME
 }
 
+mysql_dwd_config_sharedb = {
+    'drivername': 'mysql+pymysql',
+    'username': settings.MYSQL_USER,
+    'password': settings.MYSQL_PASSWORD,
+    'host': settings.MYSQL_SERVER,
+    'port':settings.MYSQL_PORT,
+    'database': 'sharedb'
+}
+
 if sqlalchemy.__version__ >= '1.4':
     mysql_engine_url = sqlalchemy.engine.URL.create(**mysql_dwd_config)
     mysql_engine_url = mysql_engine_url.update_query_dict({'charset': 'utf8mb4'})
+
+    mysql_engine_url_sharedb = sqlalchemy.engine.URL.create(**mysql_dwd_config_sharedb)
+    mysql_engine_url_sharedb = mysql_engine_url_sharedb.update_query_dict({'charset': 'utf8mb4'})
 else:
     mysql_engine_url = '{drivername}://{username}:{password}@{host}:{port}/{database}?charset=utf8mb4'.format(**mysql_dwd_config)
 
-engine = create_engine(mysql_engine_url, echo=False, pool_size=100, pool_recycle=3600, pool_pre_ping=True)
+    mysql_engine_url_sharedb = '{drivername}://{username}:{password}@{host}:{port}/{database}?charset=utf8mb4'.format(**mysql_dwd_config_sharedb)
+
+engine = create_engine(mysql_engine_url, echo=False, pool_size=10, pool_recycle=360, pool_pre_ping=True)
 SessionLocal = sessionmaker(bind=engine)
+
+engine_sharedb = create_engine(mysql_engine_url_sharedb, echo=False, pool_size=3, pool_recycle=360, pool_pre_ping=True)
+SessionLocalShareDb = sessionmaker(bind=engine_sharedb)
+
 Base = declarative_base()
 
 # Dependency
@@ -42,9 +60,20 @@ def get_local_db():
     finally:
         db.close()
 
+@contextmanager
+def get_share_db():
+    try:
+        db = SessionLocalShareDb()
+        yield db
+    finally:
+        db.close()
+
+
 def get_db_local():
     return SessionLocal()
 
+
+
 # from database import engine
 # from models.geojson_base import *
 # from shapely.geometry import shape

jobs/__init__.py

@@ -13,13 +13,9 @@ from .avcon_job import proc as avcon_proc
 from .duty_job import proc as duty_proc
 from .vehicle_job import proc as vehicle_proc
 from .hkvideo_job import proc as hkvideo_proc
-from common.security import encrypt_password
-from common.TassApi import *
+from .sign_data_job import sign_data_proc
 
 def register_jobs(scheduler: BaseScheduler):
-    encrptData = TransparentEnc("将原数据按密评要求加密辅助类")
-    print("TransparentDec:", TransparentDec(encrptData))
-
     # scheduler.add_job(yzy_proc, next_run_time=(datetime.now() + timedelta(seconds=3)))
     # scheduler.add_job(yzy_proc, CronTrigger.from_crontab('0 */5 * * *'))
     # scheduler.add_job(yzy_msg_queue_proc, CronTrigger.from_crontab('* * * * *'))
@@ -44,9 +40,13 @@ def register_jobs(scheduler: BaseScheduler):
     # scheduler.add_job(vehicle_proc, CronTrigger.from_crontab('*/5 * * * *'))
 
     # 视频状态更新
-    scheduler.add_job(hkvideo_proc, next_run_time=(datetime.now() + timedelta(seconds=9)))
+    scheduler.add_job(hkvideo_proc, next_run_time=(datetime.now() + timedelta(seconds=19)))
     scheduler.add_job(hkvideo_proc, CronTrigger.from_crontab('0 1 * * *'))
 
+    # 数据库关键数据加密
+    scheduler.add_job(sign_data_proc, next_run_time=(datetime.now() + timedelta(seconds=3)))
+    scheduler.add_job(sign_data_proc, CronTrigger.from_crontab('0 2 * * *'))
+
 
 def tick():
     print(datetime.now())

jobs/hkvideo_job.py

@@ -8,7 +8,7 @@ from utils.video_util import unitIndexCode_get_video_region_info
 from utils.redis_util import *
 from models import *
 from exceptions import *
-from database import get_db_local
+from database import get_local_db
 from extensions import logger
 import traceback
 import time
@@ -19,16 +19,13 @@ def proc():
     if redis_lock(lock_key):
         logger.info(datetime.now())
 
-        db = get_db_local()
-
-        try:
-            refresh_hkvideo(db)
-            refresh_hkvideo_region_list(db)
-            refresh_hkvideo_list(db)
-        except Exception as e:
-            traceback.print_exc()
-        finally:
-            db.close()
+        with get_local_db() as db:
+            try:
+                refresh_hkvideo(db)
+                refresh_hkvideo_region_list(db)
+                refresh_hkvideo_list(db)
+            except Exception as e:
+                traceback.print_exc()
 
         redis_unlock(lock_key)
 
@@ -102,7 +99,7 @@ def refresh_hkvideo_list(db: Session):
             infos = []
 
             for page in range(total_pages):
-                print(f'视频条数{total_pages},正在获取视频信息第{page}页')
+                # print(f'视频条数{total_pages},正在获取视频信息第{page}页')
                 data = api.get_cameras(page + 1, page_size)
                 if data['code'] == '0':
                     camera_list = data['data']['list']
@@ -167,7 +164,7 @@ def refresh_hkvideo_region_list(db: Session):
             infos = []
 
             for page in range(total_pages):
-                print(f'视频条数{total_pages},正在获取视频信息第{page}页')
+                # print(f'视频条数{total_pages},正在获取视频信息第{page}页')
                 data = api.get_regions(page + 1, page_size)
                 if data['code'] == '0':
                     region_list = data['data']['list']

jobs/sign_data_job.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from models.base import *
+from config import settings
+from utils import *
+from database import get_local_db, get_share_db
+from sqlalchemy import text
+from sqlalchemy.sql import func
+from sqlalchemy.orm import Session
+from extensions import logger
+import traceback
+import base64
+from datetime import datetime
+from common.enc import sys_user_data
+from utils.redis_util import *
+
+# 对数据进行加密处理
+def sign_data_proc():
+    lock_key = "sign_data_proc"
+    if redis_lock(lock_key):
+        logger.info(datetime.now())
+
+        # 系统用户表
+        sign_tbl_user()
+        # sign_tbl_role()
+        # sign_tbl_menu()
+        # sign_tbl_role_menu()
+        # sign_tbl_user_role()
+
+        with get_share_db() as db:
+            logger.info('sharedb ok!!!!!!!!!1')
+
+            c1 = db.query(ChemicalCompany).count()
+            print('c1: ', c1)
+
+        redis_unlock(lock_key)
+
+def sign_tbl_user():
+    print('sign_tbl_user =====>>>')
+    with get_local_db() as db:
+        rows = db.query(SysUser).filter(SysUser.sign == '').all()
+        for row in rows:
+            sys_user_data.sign_row(db, row)

main.py

@@ -18,7 +18,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
 from common.websocketManager import manager
 
-app = FastAPI()
+app = FastAPI(docs_url=None, redoc_url=None)
 # app.include_router(home.router)
 app.include_router(api.router, prefix="/api")
 app.include_router(prod_api.router, prefix="")
@@ -43,6 +43,12 @@ def app_startup():
     logger.info('---------------------------------------')
     logger.info("server started.")
 
+    try:
+        from common.enc import mpfun
+        print(mpfun.enc_data('admin123'))
+    except Exception as e:
+        print(str(e))
+
     sys = platform.system()
     if sys == "Windows":
         from warnings import filterwarnings
@@ -53,7 +59,13 @@ def app_startup():
         'default': ThreadPoolExecutor(20),
         'processpool': ProcessPoolExecutor(5)
     }
-    scheduler = AsyncIOScheduler(timezone='Asia/Shanghai', executors=executors)
+
+    job_defaults = {
+        'coalesce': True,
+        'max_instance': 1
+    }
+    
+    scheduler = AsyncIOScheduler(timezone='Asia/Shanghai', executors=executors, job_defaults=job_defaults)
     register_jobs(scheduler)
     scheduler.start()
 

models/__init__.py

@@ -24,4 +24,7 @@ from .datafilling_base import *
 from .company_management_base import *
 from .resource_provision_base import *
 from .three_proofing_responsible_base import *
-from .city_base import *
+from .city_base import *
+
+# sharedb 库
+from .sharedb import *

models/ry_sys_base.py

@@ -141,6 +141,7 @@ class SysUser(Base):
     password = Column(String(100), default='', comment='密码')
     status = Column(String(1), default='0', comment='帐号状态（0正常 1停用）')
     del_flag = Column(String(1), default='0', comment='删除标志（0代表存在 2代表删除）')
+    login = Column(Integer, default=0, comment='登录次数')
     login_ip = Column(String(128), default='', comment='最后登录IP')
     login_date = Column(DateTime, comment='最后登录时间')
     create_dept = Column(BigInteger, default=None, comment='创建部门')
@@ -150,6 +151,7 @@ class SysUser(Base):
     update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment='更新时间')
     remark = Column(String(500), default=None, comment='备注')
     yzy_account = Column(String(50), default=None, comment='粤政易账号')
+    sign = Column(String, server_default='', default='', comment='HMAC值')
 
     class Config:
         orm_mode = True

models/sharedb.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-
+from sqlalchemy import String, Column, Integer, DateTime,Text,BigInteger,Float
+from sqlalchemy.dialects.mysql import TINYINT
+from sqlalchemy.sql import func
+from database import Base
+from datetime import datetime
+
+class ChemicalCompany(Base):
+    __tablename__ = 'chemical_company'
+
+    id = Column(BigInteger, primary_key=True,autoincrement=True, comment='主键')
+    province = Column(String, default='', comment='省')
+    city = Column(String, default='', comment='地市')
+    area = Column(String, default='', comment='区县')
+    company_name = Column(String, default='', comment='企业名称')
+    company_code = Column(String, default='', comment='企业编码')
+    company_type = Column(String, default='', comment='企业类型')
+    is_import = Column(String, default='', comment='是否涉及进口')
+    is_manage = Column(String, default='', comment='是否涉及经营')
+    registration_date = Column(String, default='', comment='工商注册时间')
+    registration_address = Column(String, default='', comment='工商注册地址')
+    credit_code = Column(String, default='', comment='统一社会信用代码')
+    industry_classification = Column(String, default='', comment='行业分类')
+    production_scale = Column(Text, default='', comment='主要产品及生产规模')
+    chemical_park_location = Column(String, default='', comment='所在化工园区')
+    emergency_contact_number = Column(String, default='', comment='安全值班电话')
+    emergency_response_hotline = Column(String, default='', comment='应急咨询服务电话')
+    qyagjgfzr_phone = Column(String, default='', comment='企业安管机构负责人手机号')
+    qyfgaqfzr_phone = Column(String, default='', comment='企业分管安全负责人手机号')
+    sfqdwxhxpaqscxkz = Column(String, default='', comment='是否取得危险化学品安全生产许可证')
+    xkzbh = Column(String, default='', comment='许可证编号')
+    sfqdwxhxpjyxkz = Column(String, default='', comment='是否取得危险化学品经营许可证')
+    sfqdwxhxpsyxkz = Column(String, default='', comment='是否取得危险化学品使用许可证')
+    factory_address = Column(String, default='', comment='厂区地址')
+    hazard_source = Column(Text, default='', comment='重大危险源')
+    hazardous_chemical_process = Column(Text, default='', comment='危险化工工艺')
+    issuance_date = Column(String, default='', comment='发证日期')
+    certificate_expiry_date = Column(String, default='', comment='证书有效期')
+    company_registration_number = Column(String, default='', comment='企业登记证编码')
+    number_of_employees = Column(Integer, default='0', comment='职工人数')
+    deregistration_date = Column(String, default='', comment='注销时间')
+    scale_of_enterprise = Column(String, default='', comment='企业规模')
+    chemical_industry_classification = Column(String, default='', comment='化工行业分类')
+    safe_level = Column(String, default='', comment='安全标准化等级')
+    priority = Column(String, default='', comment='重点行业')
+    longitude = Column(Float, comment='经度')
+    latitude = Column(Float, comment='纬度')
+    formatted_address = Column(String, default='', comment='地址')
+    s_last_updatetime = Column(DateTime, comment='更新时间')
+
+    class Config:
+        orm_mode = True

routers/prod_api/auth.py

@@ -20,8 +20,9 @@ from common.auth_user import *
 from common import YzyApi, TassApi
 from models import *
 from urllib.parse import quote
-import requests
+from exceptions import *
 import traceback
+from common.enc import mpfun
 
 router = APIRouter()
 
@@ -94,10 +95,7 @@ async def login(
         redis_key = "kaptcha_" + uuid
         redis_code = redis_get(redis_key)
         if code is None or code != redis_code:
-            return {
-                "code": 500, 
-                "msg": "图片验证码不正确",
-            }
+            raise AppException(500, "图片验证码不正确")
 
         redis_login_key = "login_user_" + username
         login_error_times = redis_get(redis_login_key)
@@ -106,11 +104,11 @@ async def login(
         else:
             login_error_times = int(login_error_times)
 
-        if login_error_times >= 5:
-            return {
-                "code": 500, 
-                "msg": "登录错误多，请5分钟后再尝试！",
-            }
+        if login_error_times >= 50:
+            raise AppException(500, "登录错误多，请5分钟后再尝试！")
+        
+        # 对用户账号进行密码机接口加密处理
+        username = mpfun.enc_data(username)
 
         password = ase_utils.aesDecrypt(uuid_str, password)
         logger.info('userpass: {}', password)
@@ -120,62 +118,42 @@ async def login(
         if row is None:
             login_error_times = login_error_times + 1
             redis_set_with_time(redis_login_key, str(login_error_times), 300)
-            return JSONResponse(status_code=404, content={"code": 404, "msg": "帐号或者密码错误"})
-            # return {
-            #     "error": 1,
-            #     "errmsg": "帐号或者密码错误",
-            # }
-
-        logger.info('row.password: {}', row.password)
-        if verify_password(password, row.password) == False:
-            login_error_times = login_error_times + 1
-            redis_set_with_time(redis_login_key, str(login_error_times), 300)
-
-            return JSONResponse(status_code=404, content={"code":404,"msg":"帐号或者密码错误"})
-
-        '''
-        m = hashlib.md5()
-        m.update(userpass.encode('utf-8'))
-        password_md5 = m.hexdigest()
 
-        password_db = row.password
+            raise AppException(500, "帐号或者密码错误")
 
-        if password_md5 != password_db:
+        logger.info('row.password: {}', row.password)
+       
+        # bcrypt 加密校验
+        # if verify_password(password, row.password) == False:
+        
+        # 密码机加密校验
+        if mpfun.enc_data(password) != row.password:
             login_error_times = login_error_times + 1
             redis_set_with_time(redis_login_key, str(login_error_times), 300)
 
-            return {
-                "error": 1, 
-                "errmsg": "帐号或者密码错误",
-            }
+            raise AppException(500, "帐号或者密码错误")
 
         # 校验长期（超过1个月）未使用的账号和及开通后未及时（如72小时）修改初始密码的账号做清除
-        last_login_time = datetime.fromtimestamp(row.last_login_time)
+        login_date = row.login_date
         if row.login == 0:
             # 计算初始化的时间和当前时间相差的小时数
-            diff_hour = (datetime.now() - last_login_time).seconds/3600
+            diff_hour = (datetime.now() - login_date).seconds/3600
             if diff_hour > 72:
-                return {
-                    "error": 1, 
-                    "errmsg": "你的账号在开通后（72小时）内未登录及修改初始密码，账号已被锁定，请联系管理员处理，否则将被清除。",
-                }
+                raise AppException(500, "你的账号在开通后（72小时）内未登录及修改初始密码，账号已被锁定，请联系管理员处理，否则将被清除。")
+            
         else:
             # 计算上次登录到当前时间的相差天数
-            diff_day = (datetime.now() - last_login_time).days
+            diff_day = (datetime.now() - login_date).days
             if diff_day > 30:
-                return {
-                    "error": 1, 
-                    "errmsg": "你的账号在超过30天未登录使用，账号已被锁定，请联系管理员处理，否则将被清除。",
-                }
+                raise AppException(500, "你的账号在超过30天未登录使用，账号已被锁定，请联系管理员处理，否则将被清除。")
 
         redis_set_with_time(redis_login_key, str(0), 1)
         
-        '''
         user_id = str(row.user_id)
 
         auth = {
             "user_id": user_id,
-            "user_name": row.user_name,  
+            "user_name": mpfun.dec_data(row.user_name),  
             "nick_name": row.nick_name,
             "is_yzy_user": "0"
         }
@@ -187,7 +165,6 @@ async def login(
         # db_czrz_serv.log_username(db, row.uid, row.username, "登录", "后台管理账号+密码登录成功", request.client.host)
         row.login_date = datetime.now()
         row.login_ip = request.client.host
-        # row.login = row.login + 1
         db.commit()
 
         access_token_expires = timedelta(days = 5)
@@ -195,7 +172,7 @@ async def login(
             data={"sub": user_id}, expires_delta = access_token_expires
         )
 
-        refresh_token_expires = timedelta(days = 5)
+        refresh_token_expires = timedelta(days = 10)
         refresh_token = security.create_access_token(
             data={"sub": user_id}, expires_delta = refresh_token_expires
         )
@@ -213,6 +190,12 @@ async def login(
                 "openid": ""
             }
         }
+
+    except AppException as e:
+        return {
+            "code": e.code,
+            "msg": e.msg
+        }
     
     except Exception as e:
         traceback.print_exc()