feat: 账套模式后端 — token绑定entity_id + user_entities授权表 + 解析链倒置

- auth_middleware: create_token存JSON{user_id,entity_id},旧int格式token强制下线
- models: 新增UserEntity授权表(user_id↔entity_id多对多,唯一约束)
- database: init_db自动建表+存量用户×active企业默认授权(平滑迁移)
- deps: get_entity_id解析链倒置 token优先 → Bot白名单(query/header校验entity active) → 默认1
- auth: login加entity_id+授权校验; 新增switch-entity/my-entities/登录页entities接口; register自动授权
This commit is contained in:
Hermes CI Fix
2026-08-11 11:24:52 +08:00
parent 9d0f070c14
commit b31f9b80c4
5 changed files with 336 additions and 20 deletions
+70 -8
View File
@@ -60,7 +60,7 @@ except Exception:
logger.warning("Redis不可用,token存储降级到内存(不支持多worker)")
# 内存 fallback
_token_store: dict[str, int] = {}
_token_store: dict[str, dict] = {}
TOKEN_PREFIX = "cma:token:"
TOKEN_TTL = 86400 # 24小时
@@ -95,22 +95,84 @@ def _load_permissions(db: Session = None):
return DEFAULT_ROUTE_PERMISSIONS, DEFAULT_ACTION_PERMISSIONS
def create_token(user_id: int) -> str:
def create_token(user_id: int, entity_id: int = None) -> str:
"""签发tokenRedis存储 JSON {user_id, entity_id}(账套模式)
兼容旧调用 create_token(user_id) → entity_id=None(切换器会重新签发)
"""
token = secrets.token_hex(32)
payload = json.dumps({"user_id": user_id, "entity_id": entity_id}, ensure_ascii=False)
if _redis_available:
_redis.setex(f"{TOKEN_PREFIX}{token}", TOKEN_TTL, user_id)
_redis.setex(f"{TOKEN_PREFIX}{token}", TOKEN_TTL, payload)
else:
_token_store[token] = user_id
_token_store[token] = payload
return token
def _resolve_user_id(token: str) -> int | None:
def _resolve_token_data(token: str) -> dict | None:
"""解析token → {user_id, entity_id}
- 新格式 JSON → 返回 dict
- 旧格式 int(改造前)→ 返回 None(强制下线,账套模式需重新登录)
- 不存在 → None
"""
if _redis_available:
val = _redis.get(f"{TOKEN_PREFIX}{token}")
if val is not None:
return int(val)
if val is None:
return None
try:
data = json.loads(val)
if isinstance(data, dict) and "user_id" in data:
return data
except (json.JSONDecodeError, ValueError, TypeError):
pass
# 旧格式纯 int → 强制下线
return None
return _token_store.get(token)
val = _token_store.get(token)
if val is None:
return None
if isinstance(val, dict):
return val
try:
data = json.loads(val)
if isinstance(data, dict) and "user_id" in data:
return data
except (json.JSONDecodeError, ValueError, TypeError):
pass
return None
def _resolve_user_id(token: str) -> int | None:
data = _resolve_token_data(token)
return int(data["user_id"]) if data else None
def get_token_entity_id(token: str) -> int | None:
"""从token解析绑定的entity_idtoken不存在/旧格式 → None"""
data = _resolve_token_data(token)
if not data:
return None
eid = data.get("entity_id")
return int(eid) if eid else None
def user_has_entity(db: Session, user_id: int, entity_id: int) -> bool:
"""校验用户是否被授权访问指定企业(账套授权表 user_entities"""
from app.models import UserEntity, Entity
ent = db.query(Entity).filter(Entity.id == entity_id).first()
if not ent or ent.status != "active":
return False
rel = db.query(UserEntity).filter(
UserEntity.user_id == user_id,
UserEntity.entity_id == entity_id,
).first()
return rel is not None
def extract_bearer_token(request) -> str | None:
"""从Request提取Bearer token(无/非Bearer格式 → None"""
auth = request.headers.get("Authorization", "")
if auth.startswith("Bearer "):
return auth[7:].strip()
return None
def require_auth(