feat(deploy): 添加 enable_nginx_site 函数骨架 + 变量安全校验

deploy.py

@@ -22,6 +22,7 @@ Purpose: 统一部署所有服务到生产服务器，替代原有的多个 shel
 
 import io
 import os
+import re
 import sys
 import time
 import subprocess
@@ -35,6 +36,70 @@ if hasattr(sys.stdout, 'buffer'):
 # 当前 Python 可执行文件（Windows 用 python.exe，Linux/Mac 用 python3）
 PYTHON = sys.executable
 
+
+# ============================================================================
+# 环境变量修复（Windows 专用）
+# ============================================================================
+def _build_full_path():
+    """构造包含所有关键目录的完整 PATH 字符串。
+
+    Windows cmd.exe 启动时会用自己的"标准 PATH"（只有 2 段用户 PATH），
+    完全忽略父进程传入的 PATH 变量。解决办法是在每个 cmd.exe 命令前
+    用 `set "PATH=..."` 显式设置。这个函数返回要设置的完整 PATH。
+    """
+    if sys.platform != 'win32':
+        return None
+    parts = [
+        r'C:\Windows\system32',
+        r'C:\Windows',
+        r'C:\Windows\System32\Wbem',
+        r'C:\Windows\System32\OpenSSH',
+    ]
+    # 从已知环境变量自动推断关键目录（不硬编码）
+    if os.environ.get('MAVEN_HOME'):
+        parts.append(os.environ['MAVEN_HOME'] + r'\bin')
+    if os.environ.get('JAVA_HOME'):
+        parts.append(os.environ['JAVA_HOME'] + r'\bin')
+    # Node.js 常见安装位置自动检测
+    for d in [r'C:\Program Files\nodejs', r'F:\Program Files\nodejs',
+              r'C:\Program Files (x86)\nodejs', r'F:\Programs\nvm']:
+        if Path(d).exists():
+            parts.append(d)
+    # 保留用户已有的 PATH（不丢失用户自定义工具）
+    user_path = os.environ.get('PATH', '')
+    for p in user_path.split(';'):
+        if p and p not in parts:
+            parts.append(p)
+    return ';'.join(parts)
+
+
+def _wrap_cmd_for_windows(cmd):
+    """为 Windows cmd.exe 命令包装 set PATH 头部，解决 PATH 截断问题。
+
+    原因：cmd.exe 启动后 PATH 只有 2 段（pnpm/npm），找不到 mvn/node/npm。
+    修复：在每个命令前用 `set "PATH=完整路径" &&` 显式设置 PATH。
+    """
+    if sys.platform != 'win32':
+        return cmd
+    full_path = _build_full_path()
+    return f'set "PATH={full_path}" && {cmd}'
+
+
+def _ensure_env():
+    """修复 Windows 子进程编码问题 + 准备 PATH 工具。
+
+    Windows cmd.exe 启动时强制用 GBK 编码输出（即使父进程是 UTF-8），
+    设置 PYTHONIOENCODING/PYTHONUTF8 强制 Python 子进程使用 UTF-8。
+    实际 PATH 修复在 _wrap_cmd_for_windows() 中按需包装每个命令。
+    """
+    if sys.platform != 'win32':
+        return
+    # 强制 Python 子进程使用 UTF-8 编码
+    os.environ.setdefault('PYTHONIOENCODING', 'utf-8')
+    os.environ.setdefault('PYTHONUTF8', '1')
+    os.environ.setdefault('PYTHONLEGACYWINDOWSSTDIO', '0')
+
+
 # ============================================================================
 # 配置
 # ============================================================================
@@ -84,21 +149,23 @@ def log_section(msg):
 
 
 def run_command(cmd, cwd=None, timeout=120, capture=True):
-    """执行本地命令"""
+    """执行本地命令（自动处理 Windows cmd.exe PATH 截断）"""
     # 设置 PYTHONUNBUFFERED=1 确保 Python 子进程不缓冲 stdout
     env = os.environ.copy()
     env['PYTHONUNBUFFERED'] = '1'
+    # Windows 上包装 set PATH 前缀，解决 cmd.exe PATH 截断问题
+    effective_cmd = _wrap_cmd_for_windows(cmd)
     try:
         if capture:
             result = subprocess.run(
-                cmd, shell=True, cwd=cwd, env=env,
+                effective_cmd, shell=True, cwd=cwd, env=env,
                 capture_output=True, text=True, encoding='utf-8', errors='replace',
                 timeout=timeout
             )
             return result.returncode == 0, result.stdout.strip(), result.stderr.strip()
         else:
             result = subprocess.run(
-                cmd, shell=True, cwd=cwd, env=env, timeout=timeout
+                effective_cmd, shell=True, cwd=cwd, env=env, timeout=timeout
             )
             return result.returncode == 0, "", ""
     except subprocess.TimeoutExpired:
@@ -158,6 +225,57 @@ def check_ssh():
     log_info("SSH 连接正常")
 
 
+# ============================================================================
+# Nginx 站点启用（防御性诊断 + 清理）
+# ============================================================================
+
+# 变量安全校验正则
+_SAFE_DOMAIN = re.compile(r'^[a-zA-Z0-9.-]+$')
+_SAFE_REMOTE_CONF = re.compile(r'^[a-zA-Z0-9./_-]+$')
+
+
+def _validate_nginx_inputs(remote_conf: str, domain: str) -> bool:
+    """校验 domain 和 remote_conf 防止 shell 注入和路径遍历。
+
+    Returns: True 合法；False 非法（已 log_error）
+    """
+    if not domain or not _SAFE_DOMAIN.match(domain):
+        log_error(f"非法 domain 名称: {domain!r}")
+        return False
+    if domain.startswith('-') or domain.endswith('-') \
+            or domain.startswith('.') or domain.endswith('.') \
+            or '..' in domain:
+        log_error(f"非法 domain 结构: {domain!r} (首尾连字符/点 或 连续点非法)")
+        return False
+    if not remote_conf or not _SAFE_REMOTE_CONF.match(remote_conf):
+        log_error(f"非法 remote_conf 路径: {remote_conf!r}")
+        return False
+    if '//' in remote_conf or '..' in remote_conf:
+        log_error(f"remote_conf 包含可疑路径片段: {remote_conf!r}")
+        return False
+    return True
+
+
+def enable_nginx_site(remote_conf: str, domain: str) -> bool:
+    """幂等地启用 nginx 站点（sites-enabled symlink）。
+
+    流程：
+    1. 诊断目标路径当前状态（symlink/dir/file/missing）
+    2. 清理异常状态（目录/文件/断链）；失败短路
+    3. 二次检查 + 创建 symlink
+    4. 移除 default 站点（失败仅 warning）
+    5. 验证 symlink + default 状态
+
+    Returns: True 成功；False 失败（stderr 已打印）
+    """
+    if not _validate_nginx_inputs(remote_conf, domain):
+        return False
+
+    target = f"/etc/nginx/sites-enabled/{domain}.conf"
+    log_info(f"启用 nginx 站点: {target} -> {remote_conf}")
+    return True  # 后续 Task 替换为完整 5 步实现
+
+
 # ============================================================================
 # Nginx 配置
 # ============================================================================
@@ -365,6 +483,9 @@ def print_usage():
 
 
 def main():
+    # 修复 Windows cmd.exe 子进程环境块截断问题
+    _ensure_env()
+
     deploy_type = sys.argv[1] if len(sys.argv) > 1 else "all"
 
     if deploy_type in ("--help", "-h", "help"):