diff --git a/pipeline_service/work_env.py b/pipeline_service/work_env.py index 70fe9f4..a696739 100644 --- a/pipeline_service/work_env.py +++ b/pipeline_service/work_env.py @@ -129,6 +129,15 @@ def get_org_pubkey(org_id: str) -> dict: # ── 远程配置安全校验 ─────────────────────────────────────── +# 复制粘贴易带入的不可见字符:零宽空格/连字符/BOM 等,str.strip() 去不掉(非 isspace) +_INVISIBLE_RE = re.compile(r"[\s\u200b\u200c\u200d\u2060\ufeff]+") + + +def _clean_field(s) -> str: + """清理字段:去所有空白(含 Unicode)与零宽/BOM 不可见字符。""" + return _INVISIBLE_RE.sub("", s or "") + + _HOST_RE = re.compile(r"^[a-zA-Z0-9]([a-zA-Z0-9\-\.]*[a-zA-Z0-9])?$") _USER_RE = re.compile(r"^[a-zA-Z0-9._-]+$") @@ -145,7 +154,7 @@ def _rel_remote_dir(remote_dir: str) -> str: def _validate_remote_dir(remote_dir: str) -> str: """校验 remote_dir(相对路径),返回错误消息(None 表示通过)。""" - d = (remote_dir or "").strip() + d = _clean_field(remote_dir) if not d: return "缺少 remote_dir" if d.startswith("/"): @@ -166,7 +175,7 @@ def _validate_remote(remote_config: dict) -> str: 防注入(host/user/port)与防破坏(remote_dir 不能是系统目录、 remote_key_path 不能读任意私钥文件)。 """ - host = (remote_config.get("remote_host") or "").strip() + host = _clean_field(remote_config.get("remote_host")) if not host: return "缺少 remote_host" # 允许主机名/IPv4;IPv6 用 [] 包裹后取内部 @@ -174,7 +183,7 @@ def _validate_remote(remote_config: dict) -> str: if not _HOST_RE.match(_host): return "remote_host 格式非法" - user = (remote_config.get("remote_user") or "").strip() + user = _clean_field(remote_config.get("remote_user")) if not user: return "缺少 remote_user" if not _USER_RE.match(user): @@ -445,11 +454,11 @@ async def run_in_work_env(sor, account_name: str, user_id: str, org_id: str, async def ensure_remote_bwrap(env: dict) -> dict: """确保远程主机已安装 bwrap(零 root:apt download + dpkg -x)。""" # 校验(host/user/port/key,无需 remote_dir) - host = (env.get("remote_host") or "").strip() + host = _clean_field(env.get("remote_host")) _host = host[1:-1] if host.startswith("[") and host.endswith("]") else host if not host or not _HOST_RE.match(_host): return {"ok": False, "error": "remote_host 格式非法"} - user = (env.get("remote_user") or "").strip() + user = _clean_field(env.get("remote_user")) if not user or not _USER_RE.match(user): return {"ok": False, "error": "remote_user 格式非法"} key = (env.get("remote_key_path") or "").strip()