yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: remote_host 清理复制粘贴带入的零宽/BOM 不可见字符

Browse Source
This commit is contained in:
ymq 2026-08-14 14:49:01 +08:00
parent 70185771d0
commit e479b347bd
1 changed files with 14 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
pipeline_service/work_env.py
View File

@ -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])?$") _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._-]+$") _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: def _validate_remote_dir(remote_dir: str) -> str:
"""校验 remote_dir（相对路径），返回错误消息（None 表示通过）。""" """校验 remote_dir（相对路径），返回错误消息（None 表示通过）。"""
d = (remote_dir or "").strip() d = _clean_field(remote_dir)
if not d: if not d:
return "缺少 remote_dir" return "缺少 remote_dir"
if d.startswith("/"): if d.startswith("/"):
@ -166,7 +175,7 @@ def _validate_remote(remote_config: dict) -> str:
防注入（host/user/port）与防破坏（remote_dir 不能是系统目录、 防注入（host/user/port）与防破坏（remote_dir 不能是系统目录、
remote_key_path 不能读任意私钥文件）。 remote_key_path 不能读任意私钥文件）。
""" """
host = (remote_config.get("remote_host") or "").strip() host = _clean_field(remote_config.get("remote_host"))
if not host: if not host:
return "缺少 remote_host" return "缺少 remote_host"
# 允许主机名/IPv4；IPv6 用 [] 包裹后取内部 # 允许主机名/IPv4；IPv6 用 [] 包裹后取内部
@ -174,7 +183,7 @@ def _validate_remote(remote_config: dict) -> str:
if not _HOST_RE.match(_host): if not _HOST_RE.match(_host):
return "remote_host 格式非法" return "remote_host 格式非法"
user = (remote_config.get("remote_user") or "").strip() user = _clean_field(remote_config.get("remote_user"))
if not user: if not user:
return "缺少 remote_user" return "缺少 remote_user"
if not _USER_RE.match(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: async def ensure_remote_bwrap(env: dict) -> dict:
"""确保远程主机已安装 bwrap（零 root：apt download + dpkg -x）。""" """确保远程主机已安装 bwrap（零 root：apt download + dpkg -x）。"""
# 校验（host/user/port/key，无需 remote_dir） # 校验（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 _host = host[1:-1] if host.startswith("[") and host.endswith("]") else host
if not host or not _HOST_RE.match(_host): if not host or not _HOST_RE.match(_host):
return {"ok": False, "error": "remote_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): if not user or not _USER_RE.match(user):
return {"ok": False, "error": "remote_user 格式非法"} return {"ok": False, "error": "remote_user 格式非法"}
key = (env.get("remote_key_path") or "").strip() key = (env.get("remote_key_path") or "").strip()