# -*- coding: utf-8 -*- """④-1 认证 HTTP 端点 + 写操作守卫 —— issue #150 / PRD 8.2。 最小认证服务（标准库 http.server，零依赖），对外提供： - POST /auth/login {username,password} → 200 {token,user}（Set-Cookie）/ 401 - POST /auth/logout → 204（清 cookie） - GET /auth/me → 200 {user} / 401（凭 token 校验当前用户） 并导出 `require_auth` 装饰器与 `can_write` 守卫：未登录或非写角色 （readonly）对写操作返回 403，落实 PRD 8.2「未登录不可访问写操作」。 可独立运行 `python -m core.auth.auth_api` 冒烟（默认 :8088，绑定 127.0.0.1）。 生产由统一网关接入，本模块只负责认证语义。 """ from __future__ import annotations import json from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Callable, Optional, Tuple from .session import SESSION_COOKIE, issue_token, parse_token from .users import VALID_ROLES, UserStore WRITE_ROLES = ("engineer", "admin") # 写操作允许的角色（rbac.py 语义） class AuthError(Exception): def __init__(self, status: int, message: str) -> None: super().__init__(message) self.status = status self.message = message # --------------------------------------------------------------------------- # 守卫（供其它路由复用） # --------------------------------------------------------------------------- def require_auth(headers, store: UserStore) -> Tuple["object", None]: """从请求头/cookie 取 token 校验，返回 (user, None) 或抛 AuthError(401)。""" token = _extract_token(headers) if not token: raise AuthError(HTTPStatus.UNAUTHORIZED, "未登录") sess = parse_token(token) if sess is None: raise AuthError(HTTPStatus.UNAUTHORIZED, "会话无效或已过期") user = store.get(sess.user_id) if user is None or not user.active: raise AuthError(HTTPStatus.UNAUTHORIZED, "账号不可用") return user, None def can_write(user) -> None: """写操作角色守卫：非 engineer/admin 抛 AuthError(403)。""" if not hasattr(user, "role") or user.role not in WRITE_ROLES: raise AuthError(HTTPStatus.FORBIDDEN, "权限不足：当前角色不可执行写操作") def _extract_token(headers) -> Optional[str]: # 1) Authorization: Bearer auth = headers.get("Authorization", "") if headers else "" if auth.startswith("Bearer "): return auth[7:].strip() # 2) Cookie: iaop_session= cookie = headers.get("Cookie", "") if headers else "" for part in cookie.split(";"): if "=" in part: k, v = part.strip().split("=", 1) if k == SESSION_COOKIE: return v.strip() return None # --------------------------------------------------------------------------- # HTTP 端点 # --------------------------------------------------------------------------- class AuthAPIHandler(BaseHTTPRequestHandler): """认证端点。store 与 server 共享（见 make_server）。""" server_version = "iAOP-AuthAPI/1.0" def log_message(self, fmt, *args): # 安静日志 pass @property def store(self) -> UserStore: return self.server.user_store # type: ignore[attr-defined] def _json(self, status: int, body: dict, *, set_cookie: Optional[str] = None, clear_cookie: bool = False) -> 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))) if set_cookie: self.send_header("Set-Cookie", set_cookie) if clear_cookie: self.send_header("Set-Cookie", "%s=; Max-Age=0; Path=/; HttpOnly; SameSite=Lax" % SESSION_COOKIE) self.end_headers() self.wfile.write(payload) def _read_json(self) -> dict: length = int(self.headers.get("Content-Length", 0) or 0) if length <= 0 or length > 4096: return {} raw = self.rfile.read(length) try: return json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): return {} def do_POST(self) -> None: try: if self.path == "/auth/login": self._handle_login() elif self.path == "/auth/logout": self._json(HTTPStatus.NO_CONTENT, {}, clear_cookie=True) if False else self._send_no_content() else: self._json(HTTPStatus.NOT_FOUND, {"error": "not found"}) except AuthError as e: self._json(e.status, {"error": e.message}) except Exception as e: # noqa: BLE001 self._json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "server error: %s" % e}) def do_GET(self) -> None: try: if self.path == "/auth/me": user, _ = require_auth(self.headers, self.store) self._json(HTTPStatus.OK, {"user": user.to_public()}) elif self.path == "/auth/health": self._json(HTTPStatus.OK, {"status": "ok"}) else: self._json(HTTPStatus.NOT_FOUND, {"error": "not found"}) except AuthError as e: self._json(e.status, {"error": e.message}) except Exception as e: # noqa: BLE001 self._json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "server error: %s" % e}) def _send_no_content(self) -> None: self.send_response(HTTPStatus.NO_CONTENT) self.send_header("Set-Cookie", "%s=; Max-Age=0; Path=/; HttpOnly; SameSite=Lax" % SESSION_COOKIE) self.end_headers() def _handle_login(self) -> None: body = self._read_json() username = str(body.get("username", "")).strip() password = str(body.get("password", "")) if not username or not password: raise AuthError(HTTPStatus.BAD_REQUEST, "用户名和密码不能为空") user = self.store.authenticate(username, password) # 不区分"用户不存在/密码错"，统一 401 防枚举 if user is None: raise AuthError(HTTPStatus.UNAUTHORIZED, "用户名或密码错误") token = issue_token(user.id) cookie = "%s=%s; Path=/; Max-Age=%d; HttpOnly; SameSite=Lax" % ( SESSION_COOKIE, token, 8 * 3600) self._json(HTTPStatus.OK, {"token": token, "user": user.to_public()}, set_cookie=cookie) def make_server(store: UserStore, host: str = "127.0.0.1", port: int = 8088) -> ThreadingHTTPServer: srv = ThreadingHTTPServer((host, port), AuthAPIHandler) srv.user_store = store # type: ignore[attr-defined] return srv def _main() -> None: store = UserStore() store.ensure_bootstrap_admin("admin", "change-me-now") # 生产必须改密 srv = make_server(store) print("iAOP AuthAPI on http://%s:%d (bootstrap admin: admin / change-me-now)" % srv.server_address) srv.serve_forever() if __name__ == "__main__": _main()