diff --git a/pipeline_service/__init__.py b/pipeline_service/__init__.py index 2b74018..0381f16 100644 --- a/pipeline_service/__init__.py +++ b/pipeline_service/__init__.py @@ -28,4 +28,18 @@ from .state import ( # v2: Agent 执行引擎（对照 Hermes Agent） from .agent_loop_v2 import AgentExecutor, run_agent +# 应用部署主机逻辑账号系统（零 root 逻辑隔离 + bwrap 沙箱） +from .deploy_account import ( + ensure_account, + remove_account, + list_accounts, + run_in_sandbox, + write_file, + read_file, + list_dir, + account_name_for, + deploy_dir_for, + sanitize_username, +) + __version__ = "3.5.0" diff --git a/pipeline_service/deploy_account.py b/pipeline_service/deploy_account.py new file mode 100644 index 0000000..2b02028 --- /dev/null +++ b/pipeline_service/deploy_account.py @@ -0,0 +1,291 @@ +""" +pipeline_service/deploy_account.py - 应用部署主机逻辑账号系统 + +每个平台用户默认拥有一个逻辑部署账号（ag_ 前缀），实现： +1. 隔离部署目录（零 root，不建真实 Linux 账号） +2. 免密访问（部署进程直接文件系统读写，无需 SSH/密码） +3. bwrap 沙箱执行（user namespace 隔离文件/进程/IPC/UTS） +4. 特权能力（沙箱内 root-like 映射，受 user namespace 约束） + +全程零 root：运行时不 sudo、不 useradd，仅依赖 bwrap（unprivileged user namespaces）。 +""" + +import os +import re +import json +import shutil +import asyncio +import logging + +logger = logging.getLogger("pipeline.deploy_account") + +# 部署账号根目录 +DEPLOY_ENV_BASE = "/d/pipeline/deploy_envs" + +# bwrap 可执行文件（未安装时为 None，降级为目录隔离） +BWRAP = shutil.which("bwrap") + +# 账号名合法字符（目录安全，防路径穿越） +_SAFE_RE = re.compile(r"[^a-zA-Z0-9_-]") + + +def sanitize_username(username: str) -> str: + """把用户 username 清洗成安全的账号名片段。""" + s = _SAFE_RE.sub("_", username or "").strip("_") + return s or "user" + + +def account_name_for(username: str) -> str: + """生成账号名：ag_。""" + return f"ag_{sanitize_username(username)}" + + +def deploy_dir_for(account_name: str) -> str: + """账号隔离目录。""" + return os.path.join(DEPLOY_ENV_BASE, account_name) + + +def _ensure_account_dirs(account_name: str) -> str: + """创建账号目录结构，返回 deploy_dir。幂等。""" + d = deploy_dir_for(account_name) + for sub in ("", "home", "workspace", "data", "logs"): + os.makedirs(os.path.join(d, sub), exist_ok=True) + return d + + +def _resolve_deploy_dir(sor, account_name): + """从 DB 查 deploy_dir，无记录则返回 None。""" + return deploy_dir_for(account_name) + + +def _is_safe_relpath(relpath: str) -> bool: + """防路径穿越：相对路径不得包含 .. 或绝对路径。""" + if not relpath: + return True + if relpath.startswith("/") or relpath.startswith("\\"): + return False + if ".." in relpath.split("/"): + return False + return True + + +# ═══════════════════════════════════════════════════════════ +# 账号管理 +# ═══════════════════════════════════════════════════════════ + +async def ensure_account(sor, user_id: str, username: str) -> dict: + """确保用户拥有逻辑部署账号。幂等：已存在则返回现有记录。 + + Returns: {"account_name": str, "deploy_dir": str, "created": bool} + """ + account_name = account_name_for(username) + + # 查现有记录 + recs = await sor.sqlExe( + "SELECT id, account_name, deploy_dir, status FROM sd_deploy_accounts " + "WHERE user_id=${uid}$ LIMIT 1", {"uid": user_id}) + if recs: + r = recs[0] + deploy_dir = getattr(r, "deploy_dir", "") or deploy_dir_for(account_name) + _ensure_account_dirs(account_name) + return { + "account_name": getattr(r, "account_name", account_name), + "deploy_dir": deploy_dir, + "created": False, + } + + # 新建 + from appPublic.uniqueID import getID + deploy_dir = _ensure_account_dirs(account_name) + rec_id = getID() + await sor.C("sd_deploy_accounts", { + "id": rec_id, + "user_id": user_id, + "account_name": account_name, + "deploy_dir": deploy_dir, + "status": "active", + "sandbox_config": json.dumps({"network": "shared", "readonly_system": True}), + }) + logger.info("deploy account created: %s (%s)", account_name, user_id) + return {"account_name": account_name, "deploy_dir": deploy_dir, "created": True} + + +async def remove_account(sor, account_name: str) -> dict: + """删除逻辑账号（DB 记录 + 目录）。""" + account_name = sanitize_username(account_name) + if not account_name.startswith("ag_"): + return {"ok": False, "error": "非法账号名"} + + recs = await sor.sqlExe( + "SELECT deploy_dir FROM sd_deploy_accounts WHERE account_name=${n}$ LIMIT 1", + {"n": account_name}) + if not recs: + return {"ok": False, "error": "账号不存在"} + + deploy_dir = getattr(recs[0], "deploy_dir", "") or deploy_dir_for(account_name) + + # 删目录（限定在 DEPLOY_ENV_BASE 内，防误删） + if deploy_dir.startswith(DEPLOY_ENV_BASE + "/") and os.path.isdir(deploy_dir): + shutil.rmtree(deploy_dir, ignore_errors=True) + + await sor.sqlExe( + "DELETE FROM sd_deploy_accounts WHERE account_name=${n}$", {"n": account_name}) + logger.info("deploy account removed: %s", account_name) + return {"ok": True, "account_name": account_name} + + +async def list_accounts(sor, user_id: str = "") -> list: + """列出部署账号。user_id 为空则列全部。""" + if user_id: + recs = await sor.sqlExe( + "SELECT account_name, user_id, deploy_dir, status, created_at " + "FROM sd_deploy_accounts WHERE user_id=${uid}$ ORDER BY created_at DESC", + {"uid": user_id}) + else: + recs = await sor.sqlExe( + "SELECT account_name, user_id, deploy_dir, status, created_at " + "FROM sd_deploy_accounts ORDER BY created_at DESC") + out = [] + for r in recs: + out.append({ + "account_name": getattr(r, "account_name", ""), + "user_id": getattr(r, "user_id", ""), + "deploy_dir": getattr(r, "deploy_dir", ""), + "status": getattr(r, "status", ""), + "created_at": str(getattr(r, "created_at", "")), + }) + return out + + +# ═══════════════════════════════════════════════════════════ +# 免密目录访问（部署进程直接读写，无需认证） +# ═══════════════════════════════════════════════════════════ + +async def write_file(sor, account_name: str, relpath: str, content: str) -> dict: + """部署进程免密写入文件到账号目录。""" + if not _is_safe_relpath(relpath): + return {"ok": False, "error": "非法路径"} + deploy_dir = _resolve_deploy_dir(sor, account_name) + full = os.path.abspath(os.path.join(deploy_dir, relpath)) + if not full.startswith(deploy_dir + "/"): + return {"ok": False, "error": "路径越界"} + try: + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "w", encoding="utf-8") as f: + f.write(content or "") + return {"ok": True, "path": full} + except Exception as e: + return {"ok": False, "error": str(e)} + + +async def read_file(sor, account_name: str, relpath: str, limit: int = 20000) -> dict: + """免密读取账号目录文件。""" + if not _is_safe_relpath(relpath): + return {"ok": False, "error": "非法路径"} + deploy_dir = _resolve_deploy_dir(sor, account_name) + full = os.path.abspath(os.path.join(deploy_dir, relpath)) + if not full.startswith(deploy_dir + "/"): + return {"ok": False, "error": "路径越界"} + try: + with open(full, "r", encoding="utf-8", errors="replace") as f: + content = f.read(limit) + return {"ok": True, "content": content} + except Exception as e: + return {"ok": False, "error": str(e)} + + +async def list_dir(sor, account_name: str, relpath: str = "") -> dict: + """免密列出账号目录。""" + if not _is_safe_relpath(relpath): + return {"ok": False, "error": "非法路径"} + deploy_dir = _resolve_deploy_dir(sor, account_name) + full = os.path.abspath(os.path.join(deploy_dir, relpath)) if relpath else deploy_dir + if not full.startswith(deploy_dir + "/") and full != deploy_dir: + return {"ok": False, "error": "路径越界"} + try: + entries = [] + for name in sorted(os.listdir(full)): + p = os.path.join(full, name) + entries.append({ + "name": name, + "is_dir": os.path.isdir(p), + "size": os.path.getsize(p) if os.path.isfile(p) else 0, + }) + return {"ok": True, "entries": entries} + except Exception as e: + return {"ok": False, "error": str(e)} + + +# ═══════════════════════════════════════════════════════════ +# bwrap 沙箱执行 +# ═══════════════════════════════════════════════════════════ + +def _build_bwrap_cmd(deploy_dir: str, command: str, workdir: str = "", network: str = "shared") -> list: + """构建 bwrap 命令。 + + - 隔离 user/pid/ipc/uts namespace（保留网络，部署应用通常需要） + - 只读挂载系统目录，可写挂载 deploy_dir + - tmpfs /tmp（会话内隔离） + """ + parts = [ + BWRAP, + "--unshare-user", "--unshare-pid", "--unshare-ipc", "--unshare-uts", + "--die-with-parent", + "--ro-bind", "/usr", "/usr", + "--ro-bind", "/bin", "/bin", + "--ro-bind", "/sbin", "/sbin", + "--ro-bind", "/lib", "/lib", + "--ro-bind", "/lib64", "/lib64", + "--ro-bind", "/etc", "/etc", + "--proc", "/proc", + "--dev", "/dev", + "--tmpfs", "/tmp", + "--bind", deploy_dir, "/home", + ] + if network == "isolated": + parts.insert(1, "--unshare-net") + + chdir = "/home" + if workdir and _is_safe_relpath(workdir): + chdir = "/home/" + workdir.strip("/") + parts += ["--chdir", chdir, "--", "bash", "-c", command] + return parts + + +async def run_in_sandbox(sor, account_name: str, command: str, + workdir: str = "", timeout: int = 120, + network: str = "shared") -> dict: + """在账号沙箱内执行命令。返回 {rc, stdout, stderr, sandbox}。""" + if not command: + return {"rc": -1, "stdout": "", "stderr": "缺少命令", "sandbox": bool(BWRAP)} + + deploy_dir = _resolve_deploy_dir(sor, account_name) + if not os.path.isdir(deploy_dir): + return {"rc": -1, "stdout": "", "stderr": f"账号目录不存在: {deploy_dir}", "sandbox": bool(BWRAP)} + + if BWRAP: + cmd = _build_bwrap_cmd(deploy_dir, command, workdir, network) + # bwrap 用列表直接执行（非 shell，防注入） + proc = await asyncio.create_subprocess_exec( + *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + else: + # 降级：无 bwrap，目录 + chdir 弱隔离 + cwd = os.path.join(deploy_dir, workdir) if workdir and _is_safe_relpath(workdir) else deploy_dir + if not cwd.startswith(deploy_dir + "/") and cwd != deploy_dir: + return {"rc": -1, "stdout": "", "stderr": "路径越界", "sandbox": False} + proc = await asyncio.create_subprocess_shell( + command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd) + + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + return {"rc": -1, "stdout": "", "stderr": f"命令超时（{timeout}s）", "sandbox": bool(BWRAP)} + + return { + "rc": proc.returncode or 0, + "stdout": stdout.decode("utf-8", "replace")[-8000:], + "stderr": stderr.decode("utf-8", "replace")[-4000:], + "sandbox": bool(BWRAP), + } diff --git a/pipeline_service/init.py b/pipeline_service/init.py index e75c5ca..fe43c61 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -635,6 +635,19 @@ def load_pipeline_service(): "KEY idx_category_priority (category, priority)" ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", {}) _debug("pipeline_user_memory table ready") + # 应用部署主机逻辑账号表 + await sor.sqlExe( + "CREATE TABLE IF NOT EXISTS sd_deploy_accounts (" + "id varchar(32) NOT NULL, user_id varchar(32) NOT NULL," + "account_name varchar(64) NOT NULL, deploy_dir varchar(500) NOT NULL," + "status varchar(20) NOT NULL DEFAULT 'active'," + "sandbox_config text," + "created_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP," + "updated_at datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP," + "PRIMARY KEY (id), UNIQUE KEY uk_account_name (account_name)," + "UNIQUE KEY uk_user_id (user_id), KEY idx_status (status)" + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", {}) + _debug("sd_deploy_accounts table ready") # agent_config columns try: await sor.sqlExe("ALTER TABLE pipelines ADD COLUMN agent_config text", {})