feat: 应用部署主机逻辑账号系统（零root+bwrap沙箱）

- deploy_account.py: 逻辑账号（ag_前缀）创建/删除/列表，bwrap沙箱执行，免密目录读写
- 零root：不建真实Linux账号，逻辑隔离目录 + unprivileged user namespace
- init.py: sd_deploy_accounts 表 DDL 自动初始化

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"

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_<username>。"""
+    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),
+    }

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", {})