From 609fda05052857efc8bd12d664934b99f62811c2 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Thu, 6 Aug 2026 15:30:57 +0800 Subject: [PATCH] =?UTF-8?q?feat(#182):=20=E6=A8=A1=E5=9E=8B=E6=B3=A8?= =?UTF-8?q?=E5=86=8C=E8=A1=A8=E5=90=8E=E7=AB=AF=E5=8C=96=20TemplateRegistr?= =?UTF-8?q?y=20=E6=9C=8D=E5=8A=A1=E5=8C=96=20HTTP=20API=EF=BC=88E1?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 core/model-framework/registry_api.py（标准库 http.server，对齐 auth_api 风格，零依赖），把已交付的 TemplateRegistry 暴露为 HTTP 服务： - GET /api/v1/registry/models 列表（可按 stage 过滤） - POST /api/v1/registry/models 注册新版本（服务端校验） - POST /api/v1/registry/models/{name}/{version}/promote dev→staging→prod - POST /api/v1/registry/models/{name}/rollback?version=&stage= 回滚指定 stage 服务端持久化：PersistedRegistry 每次写操作原子落盘到 deploy/data/registry/registry.json（可挂卷），重启不丢；首次启动自动播种 4 条演示模型（对齐 web/admin/admin.js seedModels）。 鉴权双轨：读接 FBA JWT（core/auth/fba_jwt.py），写要求权限码 iaop:admin 或 iaop:studio；未配置 FBA_TOKEN_SECRET_KEY 降级旧 core/auth 会话； 两者均无时为本地/CI 演示模式（读免鉴权、写需 token 占位）。 测试：tests/test_registry_api.py（16 例，含四接口回环联调、401/403 路径、 promote/rollback/持久化/种子）+ _fba_stub.py（FBA 桩）。全 284 例通过。 部署：deploy/fba/README.md 补 nginx 反代 /api/v1/registry/ → :8090 约定。 --- core/model-framework/registry_api.py | 513 ++++++++++++++++++ core/model-framework/tests/_fba_stub.py | 52 ++ .../tests/test_registry_api.py | 267 +++++++++ deploy/fba/README.md | 38 ++ 4 files changed, 870 insertions(+) create mode 100644 core/model-framework/registry_api.py create mode 100644 core/model-framework/tests/_fba_stub.py create mode 100644 core/model-framework/tests/test_registry_api.py diff --git a/core/model-framework/registry_api.py b/core/model-framework/registry_api.py new file mode 100644 index 0000000..c19992f --- /dev/null +++ b/core/model-framework/registry_api.py @@ -0,0 +1,513 @@ +# -*- coding: utf-8 -*- +"""模型注册表 HTTP API（issue #182 / PRD 5.3 ③）。 + +把已交付的 ``TemplateRegistry``（``core/model-framework/template_registry.py``） +通过标准库 ``http.server`` 暴露为 HTTP 服务，对外提供四类接口： + +- ``GET /api/v1/registry/models`` 列表（可按 stage 过滤） +- ``POST /api/v1/registry/models`` 注册新版本（服务端校验） +- ``POST /api/v1/registry/models/{name}/{version}/promote`` dev→staging→prod +- ``POST /api/v1/registry/models/{name}/rollback?version=&stage=`` 回滚到指定 stage 版本 + +设计要点 +-------- + +- **零依赖**：与 ``core/auth/auth_api.py`` 同款风格（``http.server.ThreadingHTTPServer``）， + 纯标准库，CI 可加载、可单测。 +- **服务端持久化**：注册表底层 ``TemplateRegistry.save/load`` JSON 落盘到 + ``deploy/data/registry/registry.json``（生产可挂卷），重启不丢数据。 +- **鉴权双轨**：读接口接 FBA JWT（``core/auth/fba_jwt.py`` 的 ``FbaAuth``），写操作 + 要求权限码 ``iaop:admin`` 或 ``iaop:studio``；未配置 ``FBA_TOKEN_SECRET_KEY`` 时 + 降级旧 ``core/auth`` 会话（``auth_api.require_auth``）。两者均未配置时， + 仅本地/CI 演示用——读接口免鉴权、写接口要求带任意有效 FBA/会话 token， + 否则写接口返回 401（见 ``_require_write``）。 +- **种子**：首次启动数据文件不存在时，用旧版 4 条演示模型初始化 + （与 ``web/admin/admin.js`` 的 ``seedModels()`` 对齐，便于前端无缝迁移）。 + +可独立运行冒烟：``python -m core.model_framework.registry_api``（默认 :8090）。 +nginx 反代约定见 ``deploy/fba/README.md``。 +""" +from __future__ import annotations + +import json +import os +import re +import threading +import time +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import parse_qs, urlsplit + +# 兼容两种导入位置：包内 ``from .template_registry`` 与 +# 单测 / 独立运行时 ``from model_framework.template_registry``。 +try: # pragma: no cover - 取决于运行入口 + from .template_registry import ( # type: ignore + ALLOWED_BACKBONES, + ModelTemplate, + Stage, + TemplateRegistry, + TemplateRegistryError, + is_valid_version, + ) +except ImportError: # pragma: no cover + from model_framework.template_registry import ( # type: ignore + ALLOWED_BACKBONES, + ModelTemplate, + Stage, + TemplateRegistry, + TemplateRegistryError, + is_valid_version, + ) + +# 写操作要求的权限码（与 deploy/fba/seed_iaop_menus.py 播种一致） +WRITE_PERMS = ("iaop:admin", "iaop:studio") + +# 默认服务端数据目录（生产可挂卷）。相对仓库根。 +DEFAULT_DATA_DIR = os.path.normpath( + os.path.join(os.path.dirname(os.path.abspath(__file__)), + "..", "..", "deploy", "data", "registry")) +DEFAULT_DATA_FILE = "registry.json" + +#: 旧版演示模型种子（与 web/admin/admin.js seedModels 对齐，便于前端无缝迁移） +SEED_MODELS: List[Dict[str, Any]] = [ + {"name": "quality_forecast", "version": "1.2.0", "stage": "prod", + "backbone": "quality_forecast", "author": "bot_dev1", + "registered_at": "2026-08-01T09:00:00Z"}, + {"name": "anomaly_detection", "version": "1.0.3", "stage": "staging", + "backbone": "anomaly_detection", "author": "bot_dev1", + "registered_at": "2026-08-02T09:00:00Z"}, + {"name": "process_optimizer", "version": "0.9.1", "stage": "dev", + "backbone": "recipe_opt", "author": "engineer", + "registered_at": "2026-08-03T09:00:00Z"}, + {"name": "cross_process_optimizer", "version": "0.5.0", "stage": "dev", + "backbone": "cross_process_opt", "author": "engineer", + "registered_at": "2026-08-04T09:00:00Z"}, +] + + +class RegistryApiError(Exception): + """注册表 HTTP API 统一异常（携带 HTTP 状态码）。""" + + def __init__(self, status: int, message: str) -> None: + super().__init__(message) + self.status = status + self.message = message + + +# --------------------------------------------------------------------------- +# 持久化注册表：写操作自动落盘 +# --------------------------------------------------------------------------- + +class PersistedRegistry: + """包一层 ``TemplateRegistry``，每次写操作后自动 JSON 落盘。 + + 线程安全：所有公共方法持锁（``ThreadingHTTPServer`` 多线程服务）。 + """ + + def __init__(self, data_path: str, *, autoload: bool = True) -> None: + self._path = data_path + self._reg = TemplateRegistry() + self._lock = threading.RLock() + os.makedirs(os.path.dirname(data_path), exist_ok=True) + if autoload and os.path.exists(data_path): + try: + self._reg = TemplateRegistry.load(data_path) + except (ValueError, OSError, TemplateRegistryError): + # 损坏的 JSON 不致命：退回空表，重新种子即可 + self._reg = TemplateRegistry() + if autoload and len(self._reg) == 0: + self._seed(SEED_MODELS) + + # -- 查询（只读）------------------------------------------------------- + + def list_models(self, stage: Optional[str] = None) -> List[Dict[str, Any]]: + with self._lock: + names = self._reg.list_names() + out: List[Dict[str, Any]] = [] + for n in names: + for v in self._reg.list_versions(n): + tpl = self._reg.get(n, v) + if stage and tpl.stage.value != stage: + continue + out.append(self._tpl_view(tpl)) + return out + + def get(self, name: str, version: Optional[str] = None) -> Dict[str, Any]: + with self._lock: + try: + tpl = self._reg.get(name, version) + except TemplateRegistryError as exc: + raise RegistryApiError(HTTPStatus.NOT_FOUND, str(exc)) + return self._tpl_view(tpl) + + # -- 写操作（落盘）----------------------------------------------------- + + def register(self, payload: Dict[str, Any]) -> Dict[str, Any]: + with self._lock: + tpl = self._build_template(payload) + try: + self._reg.register(tpl) + except TemplateRegistryError as exc: + raise RegistryApiError(HTTPStatus.CONFLICT, str(exc)) + self._persist() + return self._tpl_view(tpl) + + def promote(self, name: str, version: str) -> Dict[str, Any]: + with self._lock: + try: + tpl = self._reg.promote(name, version) + except TemplateRegistryError as exc: + # 版本不存在 → 404；已在 prod → 409 + status = (HTTPStatus.NOT_FOUND if "无版本" in str(exc) or "未注册" in str(exc) + else HTTPStatus.CONFLICT) + raise RegistryApiError(status, str(exc)) + self._persist() + return self._tpl_view(tpl) + + def rollback(self, name: str, version: str, stage: str) -> Dict[str, Any]: + with self._lock: + try: + st = Stage.from_str(stage) + except TemplateRegistryError as exc: + raise RegistryApiError(HTTPStatus.BAD_REQUEST, str(exc)) + try: + tpl = self._reg.rollback(name, st, version) + except TemplateRegistryError as exc: + status = (HTTPStatus.NOT_FOUND if "无版本" in str(exc) or "未注册" in str(exc) + else HTTPStatus.BAD_REQUEST) + raise RegistryApiError(status, str(exc)) + self._persist() + return self._tpl_view(tpl) + + def history(self, name: Optional[str] = None) -> List[Dict[str, Any]]: + with self._lock: + return self._reg.history(name) + + # -- 内部 -------------------------------------------------------------- + + def _persist(self) -> None: + tmp = self._path + ".tmp" + self._reg.save(tmp) + os.replace(tmp, self._path) # 原子替换，避免半写 + + def _seed(self, seeds: List[Dict[str, Any]]) -> None: + for s in seeds: + try: + ts = s.get("registered_at") + # template_registry 用 epoch；字符串时间转 epoch，失败回退当前 + ra: float + if isinstance(ts, (int, float)): + ra = float(ts) + elif isinstance(ts, str): + try: + from datetime import datetime + ra = datetime.fromisoformat( + ts.replace("Z", "+00:00")).timestamp() + except ValueError: + ra = time.time() + else: + ra = time.time() + tpl = ModelTemplate( + name=s["name"], version=s["version"], + backbone=s.get("backbone", "generic"), + stage=Stage.from_str(s.get("stage", "dev")), + registered_at=ra, + description=s.get("description", ""), + extra={"author": s.get("author", "")}, + ) + self._reg.register(tpl, force=True) + except TemplateRegistryError: + continue # 脏种子跳过，不阻断启动 + if len(self._reg) > 0: + self._persist() + + def _build_template(self, payload: Dict[str, Any]) -> ModelTemplate: + name = str(payload.get("name", "")).strip() + version = str(payload.get("version", "")).strip() + if not name: + raise RegistryApiError(HTTPStatus.BAD_REQUEST, "name 不能为空") + if not is_valid_version(version): + raise RegistryApiError( + HTTPStatus.BAD_REQUEST, + f"非法版本号 {version!r}（例：v1 / 1.0.0 / v1.2-rc1）") + backbone = str(payload.get("backbone", "generic")).strip() or "generic" + if backbone not in ALLOWED_BACKBONES: + raise RegistryApiError( + HTTPStatus.BAD_REQUEST, + f"非法主干 {backbone!r}，允许：{list(ALLOWED_BACKBONES)}") + try: + stage = Stage.from_str(str(payload.get("stage", "dev"))) + except TemplateRegistryError as exc: + raise RegistryApiError(HTTPStatus.BAD_REQUEST, str(exc)) + metrics = payload.get("metrics") or {} + if not isinstance(metrics, dict): + raise RegistryApiError(HTTPStatus.BAD_REQUEST, "metrics 必须是对象") + try: + return ModelTemplate( + name=name, version=version, backbone=backbone, stage=stage, + hyperparams=dict(payload.get("hyperparams") or {}), + feature_columns=tuple(payload.get("feature_columns") or []), + target_column=str(payload.get("target_column", "") or ""), + metrics={k: float(v) for k, v in metrics.items()}, + description=str(payload.get("description", "") or ""), + extra=dict(payload.get("extra") or {}), + ) + except TemplateRegistryError as exc: + raise RegistryApiError(HTTPStatus.BAD_REQUEST, str(exc)) + + @staticmethod + def _tpl_view(tpl: ModelTemplate) -> Dict[str, Any]: + d = tpl.to_dict() + # 补 author（种子放进 extra），便于前端展示与迁移 + d["author"] = (tpl.extra.get("author") or "") + return d + + +# --------------------------------------------------------------------------- +# 鉴权（双轨：FBA JWT / 旧 core-auth 会话；均未配置时仅读免鉴权） +# --------------------------------------------------------------------------- + +class AuthService: + """双轨鉴权封装。读接口要求有效 token；写接口额外要求权限码。 + + - 配置了 ``FBA_TOKEN_SECRET_KEY``：走 ``FbaAuth.require_perm``； + - 否则若提供了 ``UserStore``：走 ``auth_api.require_auth``（旧会话， + 写权限由 ``can_write`` 角色判定，对齐 iaop:admin/studio 语义）； + - 两者都无：本地/CI 演示模式——读接口免鉴权；写接口必须有 token + （``_require_write`` 在无任何鉴权后端时直接 401，避免裸奔写）。 + """ + + def __init__(self, fba=None, user_store=None) -> None: + self._fba = fba + self._store = user_store + + def require_read(self, headers) -> None: + # 读接口：有鉴权后端则校验；无后端（演示）放行 + if self._fba is not None: + self._verify_fba(headers) + return + if self._store is not None: + self._require_session(headers) + return + # 无后端：读放行 + + def require_write(self, headers) -> None: + # 写接口：必须鉴权 + 权限码 + if self._fba is not None: + for perm in WRITE_PERMS: + # iaop:admin 或 iaop:studio 任一即可 + try: + self._fba.require_perm(self._authorization(headers), perm) + return + except Exception as exc: # noqa: BLE001 + status = getattr(exc, "status", None) + if status == HTTPStatus.FORBIDDEN: + continue # 试下一个权限码 + raise # 401 / 503 直接抛 + raise RegistryApiError( + HTTPStatus.FORBIDDEN, + f"权限不足：需要 {list(WRITE_PERMS)} 之一") + if self._store is not None: + user = self._require_session(headers) + role = getattr(user, "role", "") + if role not in ("admin", "engineer"): + raise RegistryApiError( + HTTPStatus.FORBIDDEN, + "权限不足：当前角色不可执行写操作") + return + # 无任何后端：写操作必须带 token 才放行（演示模式仍需 token 占位） + if not self._authorization(headers): + raise RegistryApiError(HTTPStatus.UNAUTHORIZED, + "写操作需要认证（配置 FBA_TOKEN_SECRET_KEY 或旧会话）") + + # -- 内部 -------------------------------------------------------------- + + def _verify_fba(self, headers) -> None: + try: + self._fba.verify(self._authorization(headers)) + except Exception as exc: # noqa: BLE001 + raise RegistryApiError(getattr(exc, "status", HTTPStatus.UNAUTHORIZED), + str(exc)) + + def _require_session(self, headers): + try: + from core.auth.auth_api import require_auth # type: ignore + except ImportError: # pragma: no cover + from .auth_api import require_auth # type: ignore + try: + user, _ = require_auth(headers, self._store) + except Exception as exc: # noqa: BLE001 + raise RegistryApiError(getattr(exc, "status", HTTPStatus.UNAUTHORIZED), + str(exc)) + return user + + @staticmethod + def _authorization(headers) -> str: + if not headers: + return "" + return headers.get("Authorization", "") or "" + + +# --------------------------------------------------------------------------- +# HTTP 端点 +# --------------------------------------------------------------------------- + +#: 路由：GET /api/v1/registry/models +_ROUTES_GET = ( + re.compile(r"^/api/v1/registry/models/?$"), + re.compile(r"^/api/v1/registry/models/history/?$"), +) +#: POST .../models/{name}/{version}/promote +_PROMOTE_RE = re.compile( + r"^/api/v1/registry/models/([^/]+)/([^/]+)/promote/?$") +#: POST .../models/{name}/rollback +_ROLLBACK_RE = re.compile( + r"^/api/v1/registry/models/([^/]+)/rollback/?$") + + +class RegistryAPIHandler(BaseHTTPRequestHandler): + """注册表 HTTP 端点。``registry`` / ``auth`` 与 server 共享（见 make_server）。""" + + server_version = "iAOP-RegistryAPI/1.0" + # HTTP/1.1 + Content-Length 让客户端能稳定读取错误响应体（401/403 等） + protocol_version = "HTTP/1.1" + + def log_message(self, fmt, *args): # 安静日志 + pass + + def handle_one_request(self): # type: ignore[override] + # 每个请求后关闭连接：避免 keep-alive 下 ThreadingHTTPServer 对 + # 缺 Content-Length 的客户端请求误判，从而中断错误响应。 + super().handle_one_request() + self.close_connection = True + + @property + def registry(self) -> PersistedRegistry: + return self.server.registry_store # type: ignore[attr-defined] + + @property + def auth(self) -> AuthService: + return self.server.auth_service # type: ignore[attr-defined] + + # -- 通用响应 ---------------------------------------------------------- + + def _json(self, status: int, body: Any) -> None: + payload = json.dumps(body, ensure_ascii=False).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def _read_json(self) -> Dict[str, Any]: + length = int(self.headers.get("Content-Length", 0) or 0) + if length <= 0 or length > 65536: + return {} + raw = self.rfile.read(length) + try: + data = json.loads(raw.decode("utf-8")) + return data if isinstance(data, dict) else {} + except (ValueError, UnicodeDecodeError): + raise RegistryApiError(HTTPStatus.BAD_REQUEST, "请求体不是合法 JSON") + + def _handle_error(self, exc: Exception) -> None: + if isinstance(exc, RegistryApiError): + self._json(exc.status, {"error": exc.message}) + else: + self._json(HTTPStatus.INTERNAL_SERVER_ERROR, + {"error": "server error: %s" % exc}) + + # -- GET --------------------------------------------------------------- + + def do_GET(self) -> None: # noqa: N802 - http.server 约定 + try: + split = urlsplit(self.path) + path = split.path + qs = parse_qs(split.query) + if _ROUTES_GET[0].match(path): + self.auth.require_read(self.headers) + stage = (qs.get("stage", [None])[0]) + models = self.registry.list_models(stage) + self._json(HTTPStatus.OK, {"models": models, "total": len(models)}) + elif _ROUTES_GET[1].match(path): + self.auth.require_read(self.headers) + name = qs.get("name", [None])[0] + self._json(HTTPStatus.OK, + {"history": self.registry.history(name)}) + elif path in ("/api/v1/registry/health", "/health"): + self._json(HTTPStatus.OK, {"status": "ok", + "models": len(self.registry.list_models())}) + else: + self._json(HTTPStatus.NOT_FOUND, {"error": "not found"}) + except Exception as exc: # noqa: BLE001 + self._handle_error(exc) + + # -- POST -------------------------------------------------------------- + + def do_POST(self) -> None: # noqa: N802 - http.server 约定 + try: + path = urlsplit(self.path).path + qs = parse_qs(urlsplit(self.path).query) + if _ROUTES_GET[0].match(path): # POST .../models → 注册 + self.auth.require_write(self.headers) + body = self._read_json() + view = self.registry.register(body) + self._json(HTTPStatus.CREATED, view) + elif _PROMOTE_RE.match(path): + self.auth.require_write(self.headers) + m = _PROMOTE_RE.match(path) + view = self.registry.promote(m.group(1), m.group(2)) + self._json(HTTPStatus.OK, view) + elif _ROLLBACK_RE.match(path): + self.auth.require_write(self.headers) + m = _ROLLBACK_RE.match(path) + name = m.group(1) + version = (qs.get("version", [None])[0] or "").strip() + stage = (qs.get("stage", [None])[0] or "prod").strip() + if not version: + raise RegistryApiError(HTTPStatus.BAD_REQUEST, + "缺少 query 参数 version") + view = self.registry.rollback(name, version, stage) + self._json(HTTPStatus.OK, view) + else: + self._json(HTTPStatus.NOT_FOUND, {"error": "not found"}) + except Exception as exc: # noqa: BLE001 + self._handle_error(exc) + + +def make_server(data_path: Optional[str] = None, host: str = "127.0.0.1", + port: int = 8090, *, fba=None, user_store=None, + autoload: bool = True) -> ThreadingHTTPServer: + """构造注册表 HTTP 服务。 + + - ``data_path``：注册表 JSON 落盘路径，默认 ``deploy/data/registry/registry.json``。 + - ``fba``：``FbaAuth`` 实例（读 ``FBA_TOKEN_SECRET_KEY``），None 则不启用 FBA 轨。 + - ``user_store``：旧 ``core/auth`` ``UserStore``，None 则不启用旧会话轨。 + """ + if data_path is None: + data_path = os.path.join(DEFAULT_DATA_DIR, DEFAULT_DATA_FILE) + srv = ThreadingHTTPServer((host, port), RegistryAPIHandler) + srv.registry_store = PersistedRegistry(data_path, autoload=autoload) # type: ignore[attr-defined] + srv.auth_service = AuthService(fba=fba, user_store=user_store) # type: ignore[attr-defined] + return srv + + +def _main() -> None: + # FBA 轨：读环境变量 + fba = None + try: + from core.auth.fba_jwt import FbaAuth + fba = FbaAuth.from_env() + except Exception: # noqa: BLE001 - 无 FBA 时降级 + fba = None + srv = make_server(fba=fba) + print("iAOP RegistryAPI on http://%s:%d (data=%s, fba=%s)" % ( + srv.server_address[0], srv.server_address[1], + srv.registry_store._path, # noqa: SLF001 + "on" if fba else "off")) # pragma: no cover + srv.serve_forever() + + +if __name__ == "__main__": # pragma: no cover + _main() diff --git a/core/model-framework/tests/_fba_stub.py b/core/model-framework/tests/_fba_stub.py new file mode 100644 index 0000000..14ca4b6 --- /dev/null +++ b/core/model-framework/tests/_fba_stub.py @@ -0,0 +1,52 @@ +# -*- coding: utf-8 -*- +"""测试用 FBA 鉴权桩（issue #182 注册表 API 测试）。 + +避免真实回源 FBA /auth/codes，本地固定权限码映射： + +- token == "good" → 拥有 iaop:admin（写权限） +- token == "readonly" → 仅读，无任何写权限码 +- 其它 / 缺失 → 401 + +模拟 ``core/auth/fba_jwt.FbaAuth`` 的 ``verify`` / ``require_perm`` 接口形状。 +""" +from __future__ import annotations + +from http import HTTPStatus + + +class _FbaError(Exception): + def __init__(self, status: int, message: str) -> None: + super().__init__(message) + self.status = status + self.message = message + + +class StubFba: + """FbaAuth 桩：固定权限码，不回源网络。""" + + _PERMS = { + "good": {"iaop:admin", "iaop:studio"}, + "readonly": set(), + } + + def verify(self, authorization: str) -> dict: + token = self._extract(authorization) + if token not in self._PERMS: + raise _FbaError(HTTPStatus.UNAUTHORIZED, "Token 无效") + return {"sub": token} + + def require_perm(self, authorization: str, perm: str) -> dict: + claims = self.verify(authorization) + token = self._extract(authorization) + if perm not in self._PERMS.get(token, set()): + raise _FbaError(HTTPStatus.FORBIDDEN, f"缺少权限：{perm}") + return claims + + @staticmethod + def _extract(authorization: str) -> str: + if not authorization: + raise _FbaError(HTTPStatus.UNAUTHORIZED, "未携带认证信息") + parts = authorization.split(None, 1) + if len(parts) != 2 or parts[0].lower() != "bearer": + raise _FbaError(HTTPStatus.UNAUTHORIZED, "认证头格式应为 Bearer ") + return parts[1].strip() diff --git a/core/model-framework/tests/test_registry_api.py b/core/model-framework/tests/test_registry_api.py new file mode 100644 index 0000000..6655e88 --- /dev/null +++ b/core/model-framework/tests/test_registry_api.py @@ -0,0 +1,267 @@ +# -*- coding: utf-8 -*- +"""模型注册表 HTTP API 单元测试（issue #182）。 + +覆盖验收要求： +- 四接口 curl 联调（通过 make_server + urllib 真实起服务回环联调）； +- 写操作 403 / 401 路径（无 token / 权限码缺失）； +- promote / rollback 正确流转 + 落盘持久化（重启不丢）； +- 种子首次启动初始化 4 条演示模型。 + +运行（仓库根）：PYTHONPATH=. python -m unittest core.model_framework.tests.test_registry_api -v +""" +from __future__ import annotations + +import json +import os +import sys +import tempfile +import time +import unittest +import urllib.error +import urllib.request + +HERE = os.path.dirname(os.path.abspath(__file__)) +if HERE not in sys.path: + sys.path.insert(0, HERE) + +import _bootstrap # noqa: E402 加载 model_framework 包 + +from model_framework.registry_api import ( # noqa: E402 + AuthService, + PersistedRegistry, + RegistryApiError, + make_server, +) + + +def _free_port() -> int: + import socket + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +class _ServerCtx: + """起一个回环 RegistryAPI 服务，返回 (base_url, stop)。""" + + def __init__(self, data_path: str, *, fba=None, user_store=None, + autoload: bool = True): + self.port = _free_port() + self.srv = make_server(data_path, "127.0.0.1", self.port, + fba=fba, user_store=user_store, autoload=autoload) + import threading + self._t = threading.Thread(target=self.srv.serve_forever, daemon=True) + self._t.start() + self.base = f"http://127.0.0.1:{self.port}" + + def stop(self): + self.srv.shutdown() + self.srv.server_close() + + +def _req(method: str, url: str, body=None, headers=None): + data = None + h = dict(headers or {}) + if body is not None: + data = json.dumps(body).encode("utf-8") + h.setdefault("Content-Type", "application/json") + req = urllib.request.Request(url, data=data, method=method, headers=h) + try: + with urllib.request.urlopen(req, timeout=5) as r: + return r.status, json.loads(r.read().decode("utf-8")) + except urllib.error.HTTPError as e: + payload = e.read().decode("utf-8") + try: + return e.code, json.loads(payload) + except ValueError: + return e.code, {"raw": payload} + + +class PersistedRegistryTest(unittest.TestCase): + """底层 PersistedRegistry：注册 / promote / rollback / 持久化 / 种子。""" + + def test_seed_on_first_start(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "registry.json") + reg = PersistedRegistry(path) + models = reg.list_models() + self.assertEqual(len(models), 4) + names = {m["name"] for m in models} + self.assertIn("quality_forecast", names) + # 文件已落盘 + self.assertTrue(os.path.exists(path)) + + def test_persist_across_restart(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "registry.json") + reg = PersistedRegistry(path) + reg.register({"name": "quality_forecast", "version": "9.9.9", + "stage": "dev"}) + # 重新加载同一文件 + reg2 = PersistedRegistry(path) + versions = reg2._reg.list_versions("quality_forecast") # noqa: SLF001 + self.assertIn("9.9.9", versions) + + def test_promote_and_rollback(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "registry.json") + reg = PersistedRegistry(path) + # 种子 quality_forecast 1.2.0 在 prod；用新 dev 版本走完整流转 + reg.register({"name": "quality_forecast", "version": "2.0.0", + "stage": "dev"}) + promoted = reg.promote("quality_forecast", "2.0.0") + self.assertEqual(promoted["stage"], "staging") + # rollback staging 指针回到 1.2.0 + rb = reg.rollback("quality_forecast", "1.2.0", "staging") + self.assertEqual(rb["version"], "1.2.0") + self.assertEqual( + reg._reg.stage_pointer("quality_forecast", # noqa: SLF001 + __import__( + "model_framework.template_registry", + fromlist=["Stage"]).Stage.STAGING), + "1.2.0") + + def test_register_validates_bad_input(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "registry.json") + reg = PersistedRegistry(path) + with self.assertRaises(RegistryApiError): + reg.register({"name": "", "version": "1.0.0"}) + with self.assertRaises(RegistryApiError): + reg.register({"name": "x", "version": "not-a-version"}) + with self.assertRaises(RegistryApiError): + reg.register({"name": "x", "version": "1.0.0", + "backbone": "unknown_backbone"}) + + def test_duplicate_register_conflict(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "registry.json") + reg = PersistedRegistry(path) + with self.assertRaises(RegistryApiError) as ctx: + reg.register({"name": "quality_forecast", "version": "1.2.0", + "stage": "dev"}) + self.assertEqual(ctx.exception.status, 409) + + +class RegistryHttpApiTest(unittest.TestCase): + """四接口 HTTP 回环联调 + 鉴权 401/403 路径。""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.path = os.path.join(self._tmp.name, "registry.json") + self.ctx = _ServerCtx(self.path) + + def tearDown(self): + self.ctx.stop() + self._tmp.cleanup() + + def test_list_and_seed(self): + status, body = _req("GET", self.ctx.base + "/api/v1/registry/models") + self.assertEqual(status, 200) + self.assertEqual(body["total"], 4) + # stage 过滤 + status, body = _req("GET", + self.ctx.base + "/api/v1/registry/models?stage=prod") + self.assertEqual(status, 200) + self.assertTrue(all(m["stage"] == "prod" for m in body["models"])) + self.assertEqual(body["total"], 1) + + def test_health(self): + status, body = _req("GET", self.ctx.base + "/api/v1/registry/health") + self.assertEqual(status, 200) + self.assertEqual(body["status"], "ok") + + def test_register_then_promote_then_rollback(self): + # 演示模式（无鉴权后端）：写接口需要 token 占位 + st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models", + body={"name": "demo_model", "version": "0.1.0", + "stage": "dev", "backbone": "generic"}, + headers={"Authorization": "Bearer demo-token"}) + self.assertEqual(st, 201) + self.assertEqual(bd["name"], "demo_model") + # promote dev→staging + st, bd = _req("POST", + self.ctx.base + "/api/v1/registry/models/demo_model/0.1.0/promote", + headers={"Authorization": "Bearer demo-token"}) + self.assertEqual(st, 200) + self.assertEqual(bd["stage"], "staging") + # rollback staging → 重新指向自身（验证回滚路径无错） + st, bd = _req("POST", + self.ctx.base + "/api/v1/registry/models/demo_model/rollback?version=0.1.0&stage=staging", + headers={"Authorization": "Bearer demo-token"}) + self.assertEqual(st, 200) + self.assertEqual(bd["version"], "0.1.0") + + def test_write_requires_token_in_demo_mode(self): + # 无 token → 401 + st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models", + body={"name": "x", "version": "1.0.0"}) + self.assertEqual(st, 401) + + def test_promote_not_found(self): + st, bd = _req("POST", + self.ctx.base + "/api/v1/registry/models/no_such/9.9.9/promote", + headers={"Authorization": "Bearer demo-token"}) + self.assertEqual(st, 404) + + def test_rollback_missing_version_param(self): + st, bd = _req("POST", + self.ctx.base + "/api/v1/registry/models/demo_model/rollback", + headers={"Authorization": "Bearer demo-token"}) + self.assertEqual(st, 400) + + def test_persist_after_register(self): + _req("POST", self.ctx.base + "/api/v1/registry/models", + body={"name": "persist_model", "version": "1.0.0", "stage": "dev"}, + headers={"Authorization": "Bearer demo-token"}) + # 重启服务读同一文件 + ctx2 = _ServerCtx(self.path) + try: + st, bd = _req("GET", ctx2.base + "/api/v1/registry/models") + names = {m["name"] for m in bd["models"]} + self.assertIn("persist_model", names) + finally: + ctx2.stop() + + +class AuthWithFbaTest(unittest.TestCase): + """配置 FBA 轨时：读需有效 token；写需权限码，401/403 路径。""" + + def setUp(self): + from model_framework.tests._fba_stub import StubFba + self._tmp = tempfile.TemporaryDirectory() + self.path = os.path.join(self._tmp.name, "registry.json") + # StubFba：token == "good" 有 iaop:admin；"readonly" 无写权限；无 token 401 + self.ctx = _ServerCtx(self.path, fba=StubFba()) + + def tearDown(self): + self.ctx.stop() + self._tmp.cleanup() + + def test_read_without_token_unauthorized(self): + st, _ = _req("GET", self.ctx.base + "/api/v1/registry/models") + self.assertEqual(st, 401) + + def test_read_with_token_ok(self): + st, bd = _req("GET", self.ctx.base + "/api/v1/registry/models", + headers={"Authorization": "Bearer good"}) + self.assertEqual(st, 200) + self.assertEqual(bd["total"], 4) + + def test_write_without_perm_forbidden(self): + st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models", + body={"name": "x", "version": "1.0.0"}, + headers={"Authorization": "Bearer readonly"}) + self.assertEqual(st, 403) + + def test_write_with_perm_ok(self): + st, bd = _req("POST", self.ctx.base + "/api/v1/registry/models", + body={"name": "y", "version": "1.0.0", "stage": "dev"}, + headers={"Authorization": "Bearer good"}) + self.assertEqual(st, 201) + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/fba/README.md b/deploy/fba/README.md index d17cabe..6767347 100644 --- a/deploy/fba/README.md +++ b/deploy/fba/README.md @@ -150,3 +150,41 @@ docker exec -it fba_postgres psql -U postgres -d fba # 进数据库 ### 旧架构存档 - `web/` 静态演示外壳保留（FBA 停掉时兜底；/index.html、/auth/login.html 已加 C1 自动收口跳转） - 旧登录页三轨会话已精简为 FBA 单轨（session.js，FBA 接入版） + +--- + +## 模型注册表 API 反代（issue #182，PRD 5.3 ③） + +iAOP 后端的模型模板注册表（`core/model-framework/registry_api.py`）以独立 +`http.server` 服务运行（默认 `127.0.0.1:8090`，零依赖标准库实现）。前端模型管理 / +版本发布（E2，issue #183）通过 nginx 把 `/api/v1/registry/` 反代到该端口： + +``` +location /api/v1/registry/ { + proxy_pass http://127.0.0.1:8090/api/v1/registry/; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_http_version 1.1; + proxy_set_header Connection ""; +} +``` + +服务启动： + +```bash +cd /opt/apps/iAOP +# 配置 FBA 密钥后（与 fba.env 的 TOKEN_SECRET_KEY 一致）写操作接 FBA JWT 鉴权 +export FBA_TOKEN_SECRET_KEY=<同 fba.env> +python -m core.model_framework.registry_api +# → iAOP RegistryAPI on http://127.0.0.1:8090 (data=deploy/data/registry/registry.json, fba=on) +``` + +数据落盘到 `deploy/data/registry/registry.json`（生产可挂卷），重启不丢。 +首次启动自动播种 4 条演示模型（与 `web/admin/admin.js` 的 `seedModels()` 对齐）。 + +鉴权双轨：读接口接 FBA JWT（`FbaAuth.verify`）；写接口要求权限码 `iaop:admin` 或 +`iaop:studio`。未配置 `FBA_TOKEN_SECRET_KEY` 时降级旧 `core/auth` 会话（写权限由 +角色 admin/engineer 判定）；两者均未配置时为本地/CI 演示模式——读免鉴权、写需 +带任意 token 占位（避免裸奔写）。 -- 2.54.0