python-fastapi-mm-zhcs-yj-api/routers/prod_api/auth.py

# -*- coding: utf-8 -*-

from fastapi import APIRouter, Depends, Request, Header, Form, Body
from fastapi.responses import FileResponse, StreamingResponse
from sqlalchemy.orm import Session
from sqlalchemy import text, exists, and_, or_, not_
from fastapi.responses import JSONResponse
from database import get_db
from utils import *
from utils.vcode import *
from utils.redis_util import *
import base64
from common.const import *
from io import BytesIO
from utils.StripTagsHTMLParser import *
from common import security
from datetime import timedelta
from common.security import verify_password
from utils import ase_utils
from common.auth_user import *
from common import YzyApi, TassApi
from models import *
from urllib.parse import quote
from exceptions import *
import traceback
from common.db import db_czrz
from common.enc import mpfun, sys_user_data
import hashlib

router = APIRouter()

@router.get('/tenant/list')
async def tenant_list(
    request: Request, 
    db: Session = Depends(get_db)
):
    return {
        "code": 200,
        "msg": "操作成功",
        "data": {
            "tenantEnabled": False,
            "voList": []
        }
    }

def buildVerificationCodeRedisKey(uuid: str) -> str:
    return VERIFICATION_CODE_REDIS_PREFIX.format(uuid)

@router.get('/code')
async def code(
    request: Request, 
    db: Session = Depends(get_db)
):
    uuid_str = md5(new_guid())
    image, code = getVerifyCode()
    buf = BytesIO()
    image.save(buf, 'png')
    img_data = buf.getvalue()
    image_bytes_io = BytesIO(img_data)
    image_bytes_io.seek(0)
    image_bytes = image_bytes_io.read()
    base64_str = base64.b64encode(image_bytes).decode('utf-8')

    redis_key = "kaptcha_" + buildVerificationCodeRedisKey(uuid_str)
    redis_set(redis_key, code)

    return {
        "code": 200,
        "msg": "操作成功",
        "data": {
            "captchaEnabled": True,
            "uuid": uuid_str,
            "img": base64_str
        }
    }

@router.post('/login')
async def login(
    request: Request, 
    db: Session = Depends(get_db),
    data: dict = Depends(remove_xss_json)
):
    try:
        # tenantId = data['tenantId']
        username = data['username']
        password = data['password']
        # rememberMe = data['rememberMe']
        uuid_str = data['uuid']
        code = data['code']
        # clientId = data['clientId']

        fromSystem = ''
        if 'fromSystem' in data:
            fromSystem = data['fromSystem']

        logger.info("fromSystem: {}", fromSystem)

        # 仅为了可能的兼容
        clientId = "e5cd7e4891bf95d1d19206ce24a7b32e"
        grantType = "password"

        uuid = buildVerificationCodeRedisKey(uuid_str)
        redis_key = "kaptcha_" + uuid
        redis_code = redis_get(redis_key)
        if code is None or code != redis_code:
            raise AppException(500, "图片验证码不正确")

        redis_login_key = "login_user_" + username
        login_error_times = redis_get(redis_login_key)
        if login_error_times is None:
            login_error_times = 0
        else:
            login_error_times = int(login_error_times)

        if login_error_times >= 5:
            raise AppException(500, "登录错误多，请5分钟后再尝试！")
        
        # 对用户账号进行密码机接口加密处理
        username = mpfun.enc_data(username)

        password = ase_utils.aesDecrypt(uuid_str, password)
        logger.info('userpass: {}', password)

        row = db.query(SysUser).filter(and_(SysUser.user_name == username, SysUser.del_flag == '0')).first()
        if sys_user_data.sign_valid_row(row) == False:
            raise AppException(500, "系统用户表验证异常，已被非法篡改")

        if row is None:
            login_error_times = login_error_times + 1
            redis_set_with_time(redis_login_key, str(login_error_times), 300)

            raise AppException(500, "帐号或者密码错误")

        logger.info('row.plain_user_name: {}', row.plain_user_name)
       
        # 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)

            raise AppException(500, "帐号或者密码错误")
        
        # 校验账号是否停用
        if row.status != "0":
            raise AppException(500, "帐号已停用")
        
        # 校验账号是否过期
        if row.expire_time is not None and row.expire_time < datetime.now():
            raise AppException(500, "帐号已过期")

        # 校验长期（超过1个月）未使用的账号和及开通后未及时（如72小时）修改初始密码的账号做清除
        login_date = row.login_date
        if row.login == 0:
            # 计算初始化的时间和当前时间相差的小时数
            diff_hour = (datetime.now() - login_date).seconds/3600
            if diff_hour > 72:
                raise AppException(500, "你的账号在开通后（72小时）内未登录及修改初始密码，账号已被锁定，请联系管理员处理，否则将被清除。")
            
        else:
            # 计算上次登录到当前时间的相差天数
            diff_day = (datetime.now() - login_date).days
            if diff_day > 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": mpfun.dec_data(row.user_name),  
            "nick_name": row.nick_name,
            "is_yzy_user": "0"
        }

        request.session.update({
            'user_auth': auth,
            'user_auth_sign': data_auth_sign(auth),
            'user_name': username
        })

        action = '登录'
        czrz = '后台管理登录成功'
        if fromSystem == 'yjdp':
            action = '应急一张图'
            czrz = '大屏登录成功'

        db_czrz.log_username(db, row.user_id, auth['user_name'], row.nick_name, action, czrz, 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)
        access_token = security.create_access_token(
            data={"sub": user_id}, expires_delta = access_token_expires
        )

        refresh_token_expires = timedelta(days = 10)
        refresh_token = security.create_access_token(
            data={"sub": user_id}, expires_delta = refresh_token_expires
        )

        return {
            "code": 200,
            "msg": "操作成功",
            "data": {
                "access_token": access_token,
                "refresh_token": refresh_token,
                "expire_in": 7200,
                "refresh_expire_in": 7200,
                "client_id": clientId,
                "scope": "",
                "openid": ""
            }
        }

    except AppException as e:
        return {
            "code": e.code,
            "msg": e.msg
        }
    
    except Exception as e:
        traceback.print_exc()
        return {
            "code": 500,
            "msg": "帐号或者密码错误"
        }

@router.post('/logout')
async def logout(
    request: Request, 
    user: AuthUser = Depends(get_auth_user),
    db: Session = Depends(get_db)
):
    logger.info("logout ok")
    request.session.clear()

    try:
        # db_czrz.log(db, user, "退出", "后台管理退出成功", request.client.host)

        if user.is_yzy_user == 1:
            logout_url = settings.TYRZ_LOGOUT.format(settings.TYRZ_CLIENT_ID) + quote(settings.HOME_URL+"/yjzp/")
            logger.info(logout_url)
        else:
            logout_url = settings.HOME_URL + "/yjzp/"
    except Exception as e:
        traceback.print_exc()
        logout_url = settings.HOME_URL+"/yjzp/"
    return {
        "code": 200,
        "msg": "退出成功",
        "data": logout_url
    }

# 小屏专用
@router.post('/yzy/callback')
async def yzy(
    request: Request, 
    db: Session = Depends(get_db),
    data: dict = Depends(remove_xss_json)
):
    code = data['code']
    state = data['state']
    logger.info("code:{}, {}", code, state)

    '''
    resp = YzyApi.get_user_info(code)
    if resp['errcode'] != 0:
        return {
            "code": 500,
            "msg": "Code异常"
        }

    user_id = resp['UserId']
    '''
    # 管理中心通过授权码获取用户信息接口
    # 获取用户基本信息
    result = YzyApi.getuserbycode(code)
    errcode = int(result['errcode'])
    if errcode == 0:
        data = result['data']
        user_id = data['userid']
        username = data['username']

        phone = ''    
        try:
            # 敏感数据加密算法(DES 对称加密)
            phone = YzyApi.desDecryptValue(settings.YZY_CORPSECRET, data['mobile'])
        except:
            traceback.print_exc()



    '''
    row = db.query(YzyOrgUserEntity).filter(YzyOrgUserEntity.userid == user_id).first()
    if row is None:
        return {
            "code": 500,
            "msg": "user_id异常"
        }
    
    yzy_account = row.account
    '''
    
    # where = and_(SysUser.del_flag == '0', SysUser.yzy_account == user_id)
    where = and_(SysUser.del_flag == '0', SysUser.user_name == mpfun.enc_data('mmyj_admin'))
    row = db.query(SysUser).filter(where).first()
    if row is None:
        return {
            "code": 500,
            "msg": "用户不是本系统用户"
        }
    
    user_id = str(row.user_id)
    access_token_expires = timedelta(seconds = 7200)
    access_token = security.create_access_token(
        data={"sub": user_id}, expires_delta = access_token_expires
    )

    refresh_token_expires = timedelta(seconds = 7200)
    refresh_token = security.create_access_token(
        data={"sub": user_id}, expires_delta = refresh_token_expires
    )

    return {
        "code": 200,
        "msg": "小屏粤政易登录成功",
        "data": {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "expire_in": 7200,
            "refresh_expire_in": 7200,
            "scope": ""
        }
    }

# 统一身份认证（粤政易扫码登录）token登录，中屏、大屏使用
@router.post('/yzylogin')
async def login(
    request: Request,
    data: dict = Depends(remove_xss_json),
    db: Session = Depends(get_db)
):
    code = data['code']
    redis_key = "yzy_" + code
    user_id = redis_get(redis_key)
    if user_id is None:
        return {
            "code": 500,
            "msg": "用户不是本系统用户"
        }

    row = db.query(SysUser).filter(SysUser.user_id == int(user_id)).first()
    if row is None:
        return {
            "code": 500,
            "msg": "用户不是本系统用户"
        }
    
    user_id = str(row.user_id)

    auth = {
        "user_id": user_id,
        "user_name": row.user_name,  
        "nick_name": row.nick_name,
        "is_yzy_user": "1"
    }

    logger.info('auth {}', auth)

    request.session['user_auth'] = auth
    request.session['user_auth_sign'] = data_auth_sign(auth)
    request.session['user_name'] = row.user_name

    db_czrz.log_username(db, row.user_id, auth['user_name'], row.nick_name, "登录", "粤政易登录登录成功", 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)
    access_token = security.create_access_token(
        data={"sub": user_id}, expires_delta = access_token_expires
    )

    refresh_token_expires = timedelta(days = 10)
    refresh_token = security.create_access_token(
        data={"sub": user_id}, expires_delta = refresh_token_expires
    )

    return {
        "code": 200,
        "msg": "操作成功",
        "data": {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "expire_in": 7200,
            "refresh_expire_in": 7200,
            "client_id": 0,
            "scope": "",
            "openid": ""
        }
    }


# USBKEY登录
@router.post("/login_with_usbkey")
def login_with_usbkey(
    request: Request,
    username: str = Body(...),
    keyID: str = Body(...),
    p7SignData: str = Body(...),
    p7SignValue: str = Body(...),
    db: Session = Depends(get_db)    
):
    result = TassApi.verifyP7Sign(p7SignData, p7SignValue)
    if result is None:
        return {
            "code": 500, 
            "msg": "证书验签失败",
        }
    logger.info('keyID: {}', keyID)    
    logger.info('verifyP7Sign: {}', result)
    
    try:
        # 对用户账号进行密码机接口加密处理
        username = mpfun.enc_data(username)
        
        redis_login_key = "login_user_" + username
        login_error_times = redis_get(redis_login_key)
        if login_error_times is None:
            login_error_times = 0
        else:
            login_error_times = int(login_error_times)

        if login_error_times >= 5:
            raise AppException(500, "登录错误多，请5分钟后再尝试！")

        row = db.query(SysUser).filter(SysUser.user_name == username).first()
        if row is None:
            login_error_times = login_error_times + 1
            redis_set_with_time(redis_login_key, str(login_error_times), 300)

            raise AppException(500, "帐号或者密码错误")
        
        if sys_user_data.sign_valid_row(row) == False:
            raise AppException(500, "系统用户表验证异常，已被非法篡改")

        # 校验账号是否停用
        if row.status != "0":
            raise AppException(500, "帐号已停用")
        
        # 校验长期（超过1个月）未使用的账号和及开通后未及时（如72小时）修改初始密码的账号做清除
        login_date = row.login_date
        if row.login == 0:
            # 计算初始化的时间和当前时间相差的小时数
            diff_hour = (datetime.now() - login_date).seconds/3600
            if diff_hour > 72:
                raise AppException(500, "你的账号在开通后（72小时）内未登录及修改初始密码，账号已被锁定，请联系管理员处理，否则将被清除。")
            
        else:
            # 计算上次登录到当前时间的相差天数
            diff_day = (datetime.now() - login_date).days
            if diff_day > 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": mpfun.dec_data(row.user_name), 
            "nick_name": row.nick_name,
            "is_yzy_user": "0"
        }

        logger.info('auth {}', auth)

        request.session['user_auth'] = auth
        request.session['user_auth_sign'] = data_auth_sign(auth)
        request.session['username'] = username

        db_czrz.log_username(db, row.user_id, auth['user_name'], row.nick_name, "登录", "后台USBKEY登录成功", 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)
        access_token = security.create_access_token(
            data={"sub": user_id}, expires_delta = access_token_expires
        )

        refresh_token_expires = timedelta(days = 10)
        refresh_token = security.create_access_token(
            data={"sub": user_id}, expires_delta = refresh_token_expires
        )
    
        return {
            "code": 200,
            "msg": "操作成功",
            "data": {
                "access_token": access_token,
                "refresh_token": refresh_token,
                "expire_in": 7200,
                "refresh_expire_in": 7200,
                "client_id": "e5cd7e4891bf95d1d19206ce24a7b32e",
                "scope": "",
                "openid": ""
            }
        }
    
    except AppException as e:
        return {
            "code": e.code,
            "msg": e.msg
        }
    
    except Exception as e:
        traceback.print_exc()
        return {
            "code": 500,
            "msg": "帐号或者密码错误"
        }
    
@router.post("/yzy/wxconfig")
def wxconfig(
    request: Request,
    page: str = None
):
    timestamp = int(time.time())
    noncestr = YzyApi.ranstr(20)

    if settings.IS_PROD == False:
        return {
            "code": 200,
            "msg": "测试环境模拟",
            "data": {
                "appId": settings.YZY_CORPID,
                "timestamp": timestamp,
                "nonceStr": noncestr,
                "signature": "",
                "jsApiList": ["getLocation", "hideOptionMenu"]
            }
        }

    jsapi_ticket = YzyApi.get_cache_jsapi_ticket()

    params = {
        'timestamp': timestamp,
        'noncestr': noncestr,
        'jsapi_ticket': jsapi_ticket,
        'url': page
    }

    # 1. 按 key 的 ASCII 码（字典序）升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])

    # 2. 拼接成 key=value 形式
    string1_parts = []
    for key, value in sorted_params:
        string1_parts.append(f"{key}={value}")

    # 3. 用 & 拼接所有键值对
    string1 = "&".join(string1_parts)
    signature = hashlib.sha1(string1.encode('utf-8')).hexdigest()

    return {
        "code": 200,
        "msg": "操作成功",
        "data": {
            "appId": settings.YZY_CORPID,
            "timestamp": params['timestamp'],
            "nonceStr": params['noncestr'],
            "signature": signature,
            "jsApiList": ["getLocation", "hideOptionMenu"]
        }
    }