"""门户身份适配器：验签 + 拉取用户信息（返回统一中间结构 PortalIdentity）。 ════════════════════════════════════════════════════════════════ 对端门户系统确定后，本文件是唯一可能需要修改的后端代码： 1. 标准REST形态 → .env 配 DMS_PORTAL_PROVIDER=http + 相关键，无需改代码； 2. 特殊形态（本地公钥验JWT / HMAC共享密钥 / 怪异接口）→ .env 配 DMS_PORTAL_PROVIDER=custom，并替换本文件底部 _custom_provider 函数体（按其内的示例模板手写）。 ════════════════════════════════════════════════════════════════ """ from __future__ import annotations from dataclasses import dataclass from urllib.parse import parse_qs import requests as http_client from dms.services.portal_config import PortalConfigError, load_portal_config @dataclass(frozen=True) class PortalIdentity: """对端门户用户身份的统一中间结构（与对端具体格式解耦）。""" external_id: str username: str real_name: str org_name: str | None = None class PortalTokenError(Exception): """对端验签未通过（token非法/过期），端点层转为401。""" class PortalIdentityError(Exception): """对端返回的用户信息缺少必要字段。""" def extract_token(query_string: str) -> str | None: """从跳转URL的query串中提取token（参数名来自配置）。""" cfg = load_portal_config() values = parse_qs(query_string.lstrip("?")).get(cfg.token_param, []) value = values[0].strip() if values else "" return value or None def resolve_portal_identity(token: str) -> PortalIdentity: """按配置的provider类型分发验签与身份拉取。""" cfg = load_portal_config() if cfg.provider == "mock": return _mock_provider(token) if cfg.provider == "http": return _http_provider(token) if cfg.provider == "custom": return _custom_provider(token) raise PortalConfigError("DMS_PORTAL_PROVIDER未配置") # ---------------------------------------------------------------- mock def _mock_provider(token: str) -> PortalIdentity: """开发/测试用。token自描述格式：``mock:::[:orgName]``。 示例（浏览器直接访问验证）： http://localhost:9345/?token=mock:1001:portal_user:门户联调用户:机关 """ parts = token.split(":") if len(parts) != 4 and len(parts) != 5: raise PortalTokenError( "mock token格式应为 mock:::[:orgName]" ) org_name = parts[4] if len(parts) == 5 and parts[4] else None return PortalIdentity( external_id=parts[1], username=parts[2], real_name=parts[3], org_name=org_name, ) # ---------------------------------------------------------------- http def _dig(data: object, path: str) -> object: """按 ``a.b`` 点路径从嵌套dict取值（仅支持dict层级，不支持数组）。""" current = data for key in path.split("."): if not isinstance(current, dict): return None current = current.get(key) return current def _mapped_str(user_obj: dict, local_field: str, *, required: bool, fallback: str = "") -> str | None: cfg = load_portal_config() remote = cfg.field_map.get(local_field, "") value = user_obj.get(remote) if remote else None if isinstance(value, str) and value.strip(): return value.strip() if isinstance(value, int): return str(value) if required and not fallback: raise PortalIdentityError(f"对端用户信息缺少{local_field}（远端字段名：{remote}）") return fallback or None def _http_provider(token: str) -> PortalIdentity: """声明式HTTP对接：按.env配置调用对端验签（及可选的用户信息接口）。""" cfg = load_portal_config() if not cfg.verify_url: raise PortalConfigError("DMS_PORTAL_PROVIDER=http时必须配置DMS_PORTAL_VERIFY_URL") headers: dict[str, str] = {} params: dict[str, str] = {} body: dict[str, str] = {} if cfg.verify_token_in == "header": headers[cfg.verify_header] = cfg.verify_header_template.format(token=token) elif cfg.verify_token_in == "query": params[cfg.token_param] = token else: body[cfg.token_param] = token try: resp = http_client.request( cfg.verify_method, cfg.verify_url, headers=headers, params=params, json=body if body else None, timeout=cfg.http_timeout_seconds, ) data = resp.json() except (http_client.RequestException, ValueError) as exc: raise PortalTokenError(f"无法访问对端验签接口：{exc}") from exc if str(_dig(data, cfg.verify_success_field)) != cfg.verify_success_value: raise PortalTokenError("对端验签未通过（token非法或已过期）") user_obj = _dig(data, cfg.verify_user_field) if not isinstance(user_obj, dict): raise PortalIdentityError( f"对端验签响应中未找到用户信息（字段路径：{cfg.verify_user_field}）" ) if cfg.userinfo_url: # 验签响应未含完整用户信息时，再调一次用户信息接口（GET + uid参数 + Bearer token） uid = _mapped_str(user_obj, "externalId", required=True) or "" try: info_resp = http_client.get( cfg.userinfo_url, params={cfg.userinfo_uid_param: uid}, headers={"Authorization": f"Bearer {token}"}, timeout=cfg.http_timeout_seconds, ) user_obj = info_resp.json() except (http_client.RequestException, ValueError) as exc: raise PortalTokenError(f"无法访问对端用户信息接口：{exc}") from exc external_id = _mapped_str(user_obj, "externalId", required=True) username = _mapped_str(user_obj, "username", required=False, fallback=external_id or "") real_name = _mapped_str(user_obj, "realName", required=False, fallback=username or "") if not external_id or not username: raise PortalIdentityError("对端用户信息缺少必要的身份字段") return PortalIdentity( external_id=external_id, username=username, real_name=real_name or username, org_name=_mapped_str(user_obj, "orgName", required=False), ) # ---------------------------------------------------------------- custom def _custom_provider(token: str) -> PortalIdentity: """手写适配模板：DMS_PORTAL_PROVIDER=custom 时本函数被调用。 ═══════════════ 手写示例（直接替换本函数体） ═══════════════ 场景：对端是「GET http://portal.internal/api/sso/verify?ticket=xxx」， 返回 ``{"ok": true, "data": {"uid": "1001", "name": "张三", "dept": "作战部"}}``。 import requests as http_client try: resp = http_client.get( "http://portal.internal/api/sso/verify", params={"ticket": token}, timeout=10, ) data = resp.json() except Exception: raise PortalTokenError("无法访问对端验签接口") if not data.get("ok"): raise PortalTokenError("对端验签未通过") info = data.get("data") or {} uid = str(info.get("uid") or "").strip() name = str(info.get("name") or "").strip() dept = str(info.get("dept") or "").strip() or None if not uid or not name: raise PortalIdentityError("对端用户信息缺少uid或name") return PortalIdentity(external_id=uid, username=uid, real_name=name, org_name=dept) ═══════════════ 说明 ═══════════════ - 只需返回 PortalIdentity，其余（落库/匹配/设密码/登录）由本系统自动完成； - username 缺乏明确登录名时可用 external_id 代替（如上例）； - 验签不通过抛 PortalTokenError，字段缺失抛 PortalIdentityError。 """ raise PortalConfigError( "DMS_PORTAL_PROVIDER=custom 需要在 dms/services/portal_provider.py 的 " "_custom_provider 中按模板实现对端验签逻辑" )