487 lines
20 KiB
Python
487 lines
20 KiB
Python
"""认证服务：JWT 签发、企微 OAuth、密码校验。"""
|
||
import jwt
|
||
import bcrypt
|
||
import time
|
||
import hashlib
|
||
import re
|
||
import secrets
|
||
import requests
|
||
from datetime import datetime, timedelta
|
||
from flask import current_app
|
||
from insurance.config import get_config as _get_wecom_config
|
||
from insurance.db.compat import db
|
||
from insurance.models.email_verification import EmailVerificationCode
|
||
from insurance.models.wecom_user import WeComUserMapping
|
||
from insurance.utils.email import EmailSendError, send_email
|
||
|
||
|
||
class AuthService:
|
||
"""认证业务逻辑。"""
|
||
|
||
CODE_EXPIRE_MINUTES = 10
|
||
CODE_RESEND_SECONDS = 60
|
||
CODE_MAX_ATTEMPTS = 5
|
||
|
||
def _normalize_email(self, email: str) -> str:
|
||
return (email or "").strip().lower()
|
||
|
||
def _validate_email(self, email: str) -> bool:
|
||
return bool(re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", email or ""))
|
||
|
||
def _validate_password(self, password: str) -> str:
|
||
if not password or len(password) < 8 or len(password) > 128:
|
||
return "密码长度需在 8-128 之间"
|
||
if not any(c.isalpha() for c in password) or not any(c.isdigit() for c in password):
|
||
return "密码需包含字母和数字"
|
||
return ""
|
||
|
||
def _hash_code(self, email: str, purpose: str, code: str) -> str:
|
||
secret = current_app.config.get("JWT_SECRET", "change-this")
|
||
raw = f"{email}:{purpose}:{code}:{secret}"
|
||
return hashlib.sha256(raw.encode()).hexdigest()
|
||
|
||
def _user_payload(self, mapping: WeComUserMapping) -> dict:
|
||
return {
|
||
"id": str(mapping.id),
|
||
"username": mapping.username,
|
||
"email": mapping.email,
|
||
"real_name": mapping.real_name,
|
||
"wecom_userid": mapping.wecom_userid,
|
||
"role": mapping.role,
|
||
"department": mapping.department or "",
|
||
}
|
||
|
||
def send_email_code(self, email: str, purpose: str, ip: str = "") -> dict:
|
||
"""发送注册或找回密码验证码。"""
|
||
email = self._normalize_email(email)
|
||
if purpose not in ("register", "reset_password"):
|
||
return {"code": 1001, "message": "验证码用途无效", "data": None}
|
||
if not self._validate_email(email):
|
||
return {"code": 1001, "message": "邮箱格式不正确", "data": None}
|
||
|
||
if purpose == "register":
|
||
existing = db.session.query(WeComUserMapping).filter_by(email=email).first()
|
||
if existing:
|
||
return {"code": 1001, "message": "邮箱已被注册", "data": None}
|
||
elif not db.session.query(WeComUserMapping).filter_by(email=email).first():
|
||
return {"code": 0, "message": "验证码已发送", "data": None}
|
||
|
||
now = datetime.utcnow()
|
||
latest = db.session.query(EmailVerificationCode).filter_by(
|
||
email=email,
|
||
purpose=purpose,
|
||
).order_by(EmailVerificationCode.created_at.desc()).first()
|
||
if latest and latest.created_at and latest.created_at > now - timedelta(seconds=self.CODE_RESEND_SECONDS):
|
||
return {"code": 1001, "message": "验证码发送过于频繁，请稍后再试", "data": None}
|
||
|
||
code = f"{secrets.randbelow(1000000):06d}"
|
||
record = EmailVerificationCode(
|
||
email=email,
|
||
purpose=purpose,
|
||
code_hash=self._hash_code(email, purpose, code),
|
||
expires_at=now + timedelta(minutes=self.CODE_EXPIRE_MINUTES),
|
||
ip=ip,
|
||
)
|
||
|
||
subject = "保险智能客服系统验证码"
|
||
action = "注册账号" if purpose == "register" else "重置密码"
|
||
content = (
|
||
f"您正在{action}，验证码为：{code}\n\n"
|
||
f"验证码 {self.CODE_EXPIRE_MINUTES} 分钟内有效。"
|
||
)
|
||
db.session.add(record)
|
||
try:
|
||
db.session.flush()
|
||
if not send_email(email, subject, content):
|
||
db.session.rollback()
|
||
return {"code": 5001, "message": "邮件服务未配置，请联系管理员", "data": None}
|
||
except EmailSendError as e:
|
||
db.session.rollback()
|
||
current_app.logger.warning("发送邮箱验证码失败: %s", e)
|
||
return {"code": 5001, "message": e.user_message, "data": None}
|
||
except Exception:
|
||
db.session.rollback()
|
||
current_app.logger.exception("发送邮箱验证码失败")
|
||
return {"code": 5001, "message": "验证码发送失败，请稍后重试", "data": None}
|
||
|
||
try:
|
||
db.session.commit()
|
||
except Exception:
|
||
db.session.rollback()
|
||
raise
|
||
|
||
return {"code": 0, "message": "验证码已发送", "data": None}
|
||
|
||
def _verify_email_code(self, email: str, purpose: str, code: str, commit_success: bool = True) -> tuple[bool, str]:
|
||
email = self._normalize_email(email)
|
||
code = (code or "").strip()
|
||
if not code:
|
||
return False, "请输入邮箱验证码"
|
||
|
||
record = db.session.query(EmailVerificationCode).filter_by(
|
||
email=email,
|
||
purpose=purpose,
|
||
consumed_at=None,
|
||
).order_by(EmailVerificationCode.created_at.desc()).first()
|
||
now = datetime.utcnow()
|
||
if not record or record.expires_at < now:
|
||
return False, "验证码已过期，请重新获取"
|
||
if (record.attempts or 0) >= self.CODE_MAX_ATTEMPTS:
|
||
return False, "验证码错误次数过多，请重新获取"
|
||
|
||
record.attempts = (record.attempts or 0) + 1
|
||
if record.code_hash != self._hash_code(email, purpose, code):
|
||
db.session.commit()
|
||
return False, "验证码错误"
|
||
|
||
record.consumed_at = now
|
||
if commit_success:
|
||
db.session.commit()
|
||
return True, ""
|
||
|
||
def wework_login(self, code: str, state: str) -> dict:
|
||
"""企微 OAuth 登录：code 换 token -> 获取用户信息 -> 签发 JWT。"""
|
||
corp_id = _get_wecom_config("WECOM_CORP_ID")
|
||
secret = _get_wecom_config("WECOM_SECRET")
|
||
|
||
# 1. 用 code 换取企微 access_token
|
||
token_url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
|
||
resp = requests.get(token_url, params={"corpid": corp_id, "corpsecret": secret})
|
||
token_data = resp.json()
|
||
if token_data.get("errcode") != 0:
|
||
return {"code": 3001, "message": "企微 API 调用失败", "data": None}
|
||
|
||
access_token = token_data["access_token"]
|
||
|
||
# 2. 用 code 换取 userid（标准企微 OAuth 流程）
|
||
getuserinfo_url = "https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo"
|
||
resp = requests.get(getuserinfo_url, params={"access_token": access_token, "code": code})
|
||
user_info = resp.json()
|
||
if user_info.get("errcode") != 0:
|
||
return {"code": 3001, "message": "获取企微用户信息失败", "data": None}
|
||
|
||
wecom_userid = user_info.get("userid", "")
|
||
|
||
# 3. 用 userid 获取详细用户信息
|
||
user_detail_url = "https://qyapi.weixin.qq.com/cgi-bin/user/get"
|
||
resp = requests.get(user_detail_url, params={"access_token": access_token, "userid": wecom_userid})
|
||
user_data = resp.json()
|
||
if user_data.get("errcode") != 0:
|
||
return {"code": 3001, "message": "获取企微用户详情失败", "data": None}
|
||
|
||
username = user_data.get("name", "")
|
||
department = user_data.get("department", [""])[0] if user_data.get("department") else ""
|
||
|
||
# 4. 查找或创建用户映射
|
||
mapping = db.session.query(WeComUserMapping).filter_by(wecom_userid=wecom_userid).first()
|
||
if not mapping:
|
||
mapping = WeComUserMapping(
|
||
wecom_userid=wecom_userid,
|
||
internal_user_id=f"wecom_{wecom_userid}",
|
||
username=username,
|
||
department=str(department),
|
||
role="sales",
|
||
)
|
||
db.session.add(mapping)
|
||
try:
|
||
db.session.commit()
|
||
except Exception:
|
||
db.session.rollback()
|
||
raise
|
||
|
||
# 4. 签发 JWT
|
||
token = self._generate_token(mapping.id, mapping.role, mapping.department or "")
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"token": token,
|
||
"expires_in": current_app.config.get("JWT_EXPIRE_SECONDS", 7200),
|
||
"user": self._user_payload(mapping),
|
||
},
|
||
}
|
||
|
||
def register(self, username: str, password: str, email: str, real_name: str, verification_code: str) -> dict:
|
||
"""用户注册。"""
|
||
email = self._normalize_email(email)
|
||
real_name = (real_name or "").strip()
|
||
|
||
# 用户名校验
|
||
if not username or len(username) < 3 or len(username) > 64:
|
||
return {"code": 1001, "message": "用户名长度需在 3-64 之间", "data": None}
|
||
if not self._validate_email(email):
|
||
return {"code": 1001, "message": "邮箱格式不正确", "data": None}
|
||
if not real_name or len(real_name) > 128:
|
||
return {"code": 1001, "message": "请输入真实姓名", "data": None}
|
||
|
||
# 密码校验
|
||
password_error = self._validate_password(password)
|
||
if password_error:
|
||
return {"code": 1001, "message": password_error, "data": None}
|
||
|
||
# 检查用户名是否已存在
|
||
existing = db.session.query(WeComUserMapping).filter_by(username=username).first()
|
||
if existing:
|
||
return {"code": 1001, "message": "用户名已存在", "data": None}
|
||
existing_email = db.session.query(WeComUserMapping).filter_by(email=email).first()
|
||
if existing_email:
|
||
return {"code": 1001, "message": "邮箱已被注册", "data": None}
|
||
|
||
ok, message = self._verify_email_code(email, "register", verification_code, commit_success=False)
|
||
if not ok:
|
||
return {"code": 1001, "message": message, "data": None}
|
||
|
||
# 生成密码哈希
|
||
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
||
|
||
# 创建用户
|
||
import uuid
|
||
mapping = WeComUserMapping(
|
||
wecom_userid=f"local_{uuid.uuid4().hex[:8]}",
|
||
internal_user_id=f"user_{uuid.uuid4().hex[:8]}",
|
||
username=username,
|
||
email=email,
|
||
real_name=real_name,
|
||
email_verified="true",
|
||
password_hash=password_hash,
|
||
role="client",
|
||
status="active",
|
||
)
|
||
db.session.add(mapping)
|
||
try:
|
||
db.session.commit()
|
||
except Exception:
|
||
db.session.rollback()
|
||
raise
|
||
|
||
# 同步到 Dify accounts 表
|
||
self._sync_to_dify(username, password, mapping.id, email)
|
||
|
||
# 签发 JWT
|
||
token = self._generate_token(mapping.id, mapping.role)
|
||
|
||
from insurance.utils.audit import log_operation
|
||
log_operation(str(mapping.id), "register", "user", str(mapping.id))
|
||
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"token": token,
|
||
"expires_in": current_app.config.get("JWT_EXPIRE_SECONDS", 7200),
|
||
"user": self._user_payload(mapping),
|
||
},
|
||
}
|
||
|
||
def _sync_to_dify(self, username: str, password: str, local_user_id: int, email: str = "") -> bool:
|
||
"""同步用户到 Dify accounts 表。"""
|
||
try:
|
||
from insurance.db.compat import db
|
||
import uuid as uuid_mod
|
||
from datetime import datetime, timezone
|
||
|
||
# 检查 Dify 中是否已存在同名用户
|
||
result = db.session.execute(
|
||
db.text("SELECT id FROM accounts WHERE name = :name"),
|
||
{"name": username}
|
||
)
|
||
existing = result.fetchone()
|
||
|
||
if existing:
|
||
# 已存在，记录映射关系
|
||
return True
|
||
|
||
# 创建 Dify 账号
|
||
account_id = str(uuid_mod.uuid4())
|
||
account_email = email or f"{username}@baodan.local"
|
||
now = datetime.now(timezone.utc)
|
||
|
||
# Dify 期望密码已经是 bcrypt 哈希后的值
|
||
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
||
|
||
# 插入 accounts 表
|
||
db.session.execute(
|
||
db.text("""
|
||
INSERT INTO accounts (id, name, email, password, avatar, interface_language,
|
||
interface_theme, timezone, last_login_at, last_active_at,
|
||
status, created_at, updated_at)
|
||
VALUES (:id, :name, :email, :password, '', 'zh-Hans', 'light', 'Asia/Shanghai',
|
||
:now, :now, 'active', :now, :now)
|
||
"""),
|
||
{
|
||
"id": account_id,
|
||
"name": username,
|
||
"email": account_email,
|
||
"password": password_hash,
|
||
"now": now,
|
||
}
|
||
)
|
||
|
||
# 获取默认租户
|
||
tenant_result = db.session.execute(
|
||
db.text("SELECT id FROM tenants LIMIT 1")
|
||
)
|
||
tenant = tenant_result.fetchone()
|
||
|
||
if tenant:
|
||
# 关联到租户
|
||
db.session.execute(
|
||
db.text("""
|
||
INSERT INTO tenant_account_joins (id, tenant_id, account_id, role,
|
||
invited_by, created_at, updated_at)
|
||
VALUES (:id, :tenant_id, :account_id, 'normal', NULL, :now, :now)
|
||
"""),
|
||
{
|
||
"id": str(uuid_mod.uuid4()),
|
||
"tenant_id": str(tenant[0]),
|
||
"account_id": account_id,
|
||
"now": now,
|
||
}
|
||
)
|
||
|
||
db.session.commit()
|
||
return True
|
||
|
||
except Exception as e:
|
||
# 同步失败不影响注册
|
||
print(f"同步到 Dify 失败: {e}")
|
||
db.session.rollback()
|
||
return False
|
||
|
||
def password_login(self, username: str, password: str) -> dict:
|
||
"""账密登录：bcrypt 校验，返回 JWT。"""
|
||
mapping = db.session.query(WeComUserMapping).filter_by(username=username).first()
|
||
if not mapping or not mapping.password_hash:
|
||
return {"code": 1002, "message": "用户名或密码错误", "data": None}
|
||
|
||
if not bcrypt.checkpw(password.encode(), mapping.password_hash.encode()):
|
||
return {"code": 1002, "message": "用户名或密码错误", "data": None}
|
||
|
||
if mapping.status != "active":
|
||
return {"code": 1002, "message": "账号已停用", "data": None}
|
||
|
||
token = self._generate_token(mapping.id, mapping.role, mapping.department or "")
|
||
|
||
from insurance.utils.audit import log_operation
|
||
log_operation(str(mapping.id), "login", "user", str(mapping.id))
|
||
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"token": token,
|
||
"expires_in": current_app.config.get("JWT_EXPIRE_SECONDS", 7200),
|
||
"user": self._user_payload(mapping),
|
||
},
|
||
}
|
||
|
||
def refresh_token(self, user_id: str) -> dict:
|
||
"""刷新 Token：签发新的 JWT。"""
|
||
mapping = db.session.query(WeComUserMapping).filter_by(id=int(user_id)).first()
|
||
if not mapping:
|
||
return {"code": 1002, "message": "用户不存在", "data": None}
|
||
|
||
token = self._generate_token(mapping.id, mapping.role, mapping.department or "")
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"token": token,
|
||
"expires_in": current_app.config.get("JWT_EXPIRE_SECONDS", 7200),
|
||
},
|
||
}
|
||
|
||
def change_password(self, user_id: str, old_password: str, new_password: str) -> dict:
|
||
"""修改密码。"""
|
||
mapping = db.session.query(WeComUserMapping).filter_by(id=int(user_id)).first()
|
||
if not mapping:
|
||
return {"code": 1005, "message": "用户不存在", "data": None}
|
||
|
||
if not mapping.password_hash:
|
||
return {"code": 1002, "message": "该账号无密码，请使用企微登录", "data": None}
|
||
|
||
if not bcrypt.checkpw(old_password.encode(), mapping.password_hash.encode()):
|
||
return {"code": 1002, "message": "旧密码错误", "data": None}
|
||
|
||
mapping.password_hash = bcrypt.hashpw(new_password.encode(), bcrypt.gensalt()).decode()
|
||
try:
|
||
db.session.commit()
|
||
except Exception:
|
||
db.session.rollback()
|
||
raise
|
||
|
||
self._sync_password_to_dify(mapping.username, new_password)
|
||
|
||
from insurance.utils.audit import log_operation
|
||
log_operation(user_id, "change_password", "user", user_id)
|
||
|
||
return {"code": 0, "message": "密码修改成功", "data": None}
|
||
|
||
def reset_password(self, email: str, verification_code: str, new_password: str) -> dict:
|
||
"""通过邮箱验证码重置密码。"""
|
||
email = self._normalize_email(email)
|
||
password_error = self._validate_password(new_password)
|
||
if password_error:
|
||
return {"code": 1001, "message": password_error, "data": None}
|
||
|
||
ok, message = self._verify_email_code(email, "reset_password", verification_code, commit_success=False)
|
||
if not ok:
|
||
return {"code": 1001, "message": message, "data": None}
|
||
|
||
mapping = db.session.query(WeComUserMapping).filter_by(email=email).first()
|
||
if not mapping:
|
||
return {"code": 1005, "message": "用户不存在", "data": None}
|
||
|
||
mapping.password_hash = bcrypt.hashpw(new_password.encode(), bcrypt.gensalt()).decode()
|
||
mapping.email_verified = "true"
|
||
try:
|
||
db.session.commit()
|
||
except Exception:
|
||
db.session.rollback()
|
||
raise
|
||
|
||
self._sync_password_to_dify(mapping.username, new_password)
|
||
|
||
from insurance.utils.audit import log_operation
|
||
log_operation(str(mapping.id), "reset_password", "user", str(mapping.id))
|
||
|
||
return {"code": 0, "message": "密码重置成功", "data": None}
|
||
|
||
def _sync_password_to_dify(self, username: str, password: str) -> bool:
|
||
"""同步本地密码到 Dify accounts 表。"""
|
||
try:
|
||
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
||
db.session.execute(
|
||
db.text("UPDATE accounts SET password = :password, updated_at = :updated_at WHERE name = :name"),
|
||
{"password": password_hash, "updated_at": datetime.utcnow(), "name": username}
|
||
)
|
||
db.session.commit()
|
||
return True
|
||
except Exception as e:
|
||
print(f"同步密码到 Dify 失败: {e}")
|
||
db.session.rollback()
|
||
return False
|
||
|
||
def logout(self, token: str, user_id: str) -> dict:
|
||
"""吊销 Token：加入 Redis 黑名单。"""
|
||
try:
|
||
from insurance.db.compat import redis_client
|
||
expire = current_app.config.get("JWT_EXPIRE_SECONDS", 7200)
|
||
redis_client.setex(f"token:blacklist:{token}", expire, "1")
|
||
except Exception:
|
||
pass # Redis 不可用时仍允许 logout
|
||
|
||
from insurance.utils.audit import log_operation
|
||
log_operation(user_id, "logout", "user", user_id)
|
||
|
||
return {"code": 0, "message": "success", "data": None}
|
||
|
||
def _generate_token(self, user_id: int, role: str, department: str = "") -> str:
|
||
"""生成 JWT Token。"""
|
||
secret = current_app.config.get("JWT_SECRET", "change-this")
|
||
expire = int(current_app.config.get("JWT_EXPIRE_SECONDS", 7200))
|
||
payload = {
|
||
"user_id": str(user_id),
|
||
"role": role,
|
||
"department": department,
|
||
"exp": int(time.time()) + expire,
|
||
"iat": int(time.time()),
|
||
}
|
||
return jwt.encode(payload, secret, algorithm="HS256")
|