From 51ff6dbcc6167d923319c1bdd2d902682d30da21 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 02:06:34 +0000 Subject: [PATCH] =?UTF-8?q?feat(#150):=20=E8=AE=A4=E8=AF=81=20HTTP=20?= =?UTF-8?q?=E7=AB=AF=E7=82=B9=20auth=5Fapi.py=EF=BC=88login/logout/me=20+?= =?UTF-8?q?=20require=5Fauth/can=5Fwrite=20=E5=86=99=E6=93=8D=E4=BD=9C?= =?UTF-8?q?=E5=AE=88=E5=8D=AB=EF=BC=8CPRD=208.2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/auth/auth_api.py | 181 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 181 insertions(+) create mode 100644 core/auth/auth_api.py diff --git a/core/auth/auth_api.py b/core/auth/auth_api.py new file mode 100644 index 0000000..c391325 --- /dev/null +++ b/core/auth/auth_api.py @@ -0,0 +1,181 @@ +# -*- 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()