Skip to content

U
ubains-module-test
提交 267b77f9 authored 作者: 陈泽健's avatar 陈泽健
浏览文件
操作

docs(deploy): 更新部署文档强调执行操作重要性并新增ARM Ubuntu自动化部署脚本 

- 在多个部署文档中强化了严格根据文档执行操作的提示
- 新增ARM Ubuntu自动化部署脚本(deploy_arm_ubuntu_976.py)
- 实现SSH连接、部署包上传、MD5校验、解压部署等功能
- 集成API接口验证、容器状态检查、安全扫描修复等验证机制
- 添加Nacos认证绕过漏洞修复和EMQX MQTT弱密码安全加固
- 生成详细的部署分析报告和日志记录功能
上级 c15c2ca2
隐藏空白字符变更
内嵌 并排
正在显示 包含 2631 行增加 和 6 行删除
AuxiliaryTool/ScriptTool/RemoteDeploy/deploy_arm_ubuntu_976.py 0 → 100644
浏览文件 @ 267b77f9
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ARM Ubuntu (192.168.9.76) 自动化部署脚本
严格按照需求文档和部署操作指导执行
执行流程：
1. SSH连接服务器，切换root
2. 上传部署包（从NAS网盘到服务器/data/）
3. MD5校验
4. 解压部署包（禁止中断）
5. 执行 arm_new_auto.sh --all
6. 验证容器和服务
7. 安全扫描和修复
8. 生成报告
用法:
python deploy_arm_ubuntu_976.py # 全流程
python deploy_arm_ubuntu_976.py --skip-upload # 跳过上传（使用服务器上已有的包）
python deploy_arm_ubuntu_976.py --skip-deploy # 跳过部署（仅检查和验证）
"""
import sys
import os
import time
import paramiko
import requests
import secrets
import base64
from datetime import datetime
from urllib3.exceptions import InsecureRequestWarning
# 禁用SSL警告
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# 修复Windows控制台编码
if sys.platform == 'win32':
try:
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
sys.stderr.reconfigure(encoding='utf-8', errors='replace')
except Exception:
pass
# ==================== 配置 ====================
HOST = "192.168.9.76"
PORT = 22
USERNAME = "admin"
PASSWORD = "Ubains@123"
SUDO_PASSWORD = "Ubains@123"
# 部署包配置
NAS_DIR = r"Z:\发布版本\03服务器部署\15新统一平台\ARM部署包\全量版\版本更新-待验证"
DEPLOY_PKG = "arm_offline_auto_unifiedPlatform.tar.gz"
DEPLOY_MD5 = "arm_offline_auto_unifiedPlatform.tar.gz.md5"
DEPLOY_DIR = "/data/arm_offline_auto_unifiedPlatform"
# 授权文件
LICENSE_PATH = r"E:\自动化部署\ARM-9.76\license.zip"
# 超管配置
ADMIN_USER = "superadmin"
ADMIN_PASS = "Ubains@1357"
CAPTCHA = "csba"
# 部署应答序列
# y(继续执行) y(网口确认) y(日期确认) y(IP确认) y(系统类型) y(部署参数) y(开始部署) n(无企业NTP)
DEPLOY_ANSWERS = "y\ny\ny\ny\ny\ny\ny\nn\n"
# 报告目录
REPORTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "reports")
os.makedirs(REPORTS_DIR, exist_ok=True)
# 服务日志路径
SERVICE_LOGS = {
"预定对外服务": "/data/services/api/java-meeting/java-meeting-extapi/logs/ubains-INFO-AND-ERROR.log",
"预定对内服务": "/data/services/api/java-meeting/java-meeting2.0/logs/ubains-INFO-AND-ERROR.log",
"运维服务": "/data/services/api/python-cmdb/log/uinfo.log",
"讯飞服务": "/data/services/api/python-voice/log/uinfo.log",
}
# API端点
API_ENDPOINTS = [
{
'name': '预定对外接口',
'url': f'https://{HOST}/exapi/message/getMsgPageList',
'keywords': ['无效token', 'Full authentication'],
'description': 'curl -k https://服务器IP/exapi/message/getMsgPageList',
},
{
'name': '预定系统接口',
'url': f'https://{HOST}/meetingV3/api/systemConfiguration/globalConfig?companyNumber=CN-SZ-00-0201',
'keywords': ['accessToken为空'],
'description': '预定系统全局配置接口',
},
{
'name': '运维集控接口',
'url': f'https://{HOST}/monitor/api2/api/servermonitor/',
'keywords': ['用户不存在', '重新登录'],
'description': '运维集控服务监控接口',
},
{
'name': '讯飞转录接口',
'url': f'https://{HOST}/voice/api/iflytek/roommaster?company_id=1&user_id=8&company_secret=57d00f9f-020f-5f1f-b788-55fae843bceb&getall=1',
'keywords': ['缺少关键参数'],
'description': '讯飞转录房间接口',
},
]
# ==================== SSH工具 ====================
def ssh_exec(ssh, cmd, timeout=300):
"""执行SSH命令，支持sudo（通过 -S 读取密码）"""
# 将命令包装为 sudo -S 方式执行
actual_cmd = f"echo '{SUDO_PASSWORD}' | sudo -S bash -c '{cmd.replace(chr(39), chr(39) + chr(92) + chr(39))}'"
stdin, stdout, stderr = ssh.exec_command(actual_cmd, timeout=timeout)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
exit_code = stdout.channel.recv_exit_status()
# 过滤sudo密码提示
if '[sudo]' in out:
lines = out.split('\n')
out = '\n'.join(lines[1:])
return exit_code, out, err
def ssh_exec_raw(ssh, cmd, timeout=300):
"""直接执行SSH命令（不使用sudo）"""
stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
exit_code = stdout.channel.recv_exit_status()
return exit_code, out, err
# ==================== 日志 ====================
timestamp_str = datetime.now().strftime('%Y%m%d_%H%M%S')
LOG_FILE = os.path.join(REPORTS_DIR, f"ARM_Ubuntu_976_deploy_{timestamp_str}.log")
def log(msg, level="INFO"):
"""输出日志"""
ts = datetime.now().strftime('%H:%M:%S')
line = f"[{ts}] [{level}] {msg}"
try:
print(line, flush=True)
except UnicodeEncodeError:
print(line.encode('gbk', errors='replace').decode('gbk'), flush=True)
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write(line + '\n')
# ==================== 部署流程 ====================
def step1_connect():
"""步骤1：连接服务器并切换root"""
log("=" * 60)
log("步骤1：连接ARM Ubuntu服务器 192.168.9.76")
log("=" * 60)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(HOST, PORT, USERNAME, PASSWORD, timeout=30,
look_for_keys=False, allow_agent=False)
log("SSH连接成功 (admin/Ubains@123)")
# 验证sudo权限
_, out, _ = ssh_exec(ssh, "whoami")
log(f"sudo权限验证: {out.strip()}")
if out.strip() != 'root':
log("sudo切换root失败！", "ERROR")
return None
# 检查系统信息
_, sys_info, _ = ssh_exec(ssh, "uname -a")
log(f"系统信息: {sys_info.strip()}")
_, disk_info, _ = ssh_exec(ssh, "df -h /data")
log(f"磁盘信息:\n{disk_info.strip()}")
_, mem_info, _ = ssh_exec(ssh, "free -h")
log(f"内存信息:\n{mem_info.strip()}")
log("[OK] 服务器连接和root权限验证通过")
return ssh
def step2_upload_package(ssh, skip_upload=False):
"""步骤2：上传部署包到服务器"""
log("\n" + "=" * 60)
log("步骤2：上传部署包到服务器/data/目录")
log("=" * 60)
if skip_upload:
log("[SKIP] 跳过上传，使用服务器上已有的部署包")
return True
local_tar = os.path.join(NAS_DIR, DEPLOY_PKG)
local_md5 = os.path.join(NAS_DIR, DEPLOY_MD5)
if not os.path.exists(local_tar):
log(f"NAS上未找到部署包: {local_tar}", "ERROR")
return False
pkg_size = os.path.getsize(local_tar)
log(f"NAS部署包: {DEPLOY_PKG} ({pkg_size/1024/1024/1024:.2f} GB)")
# 清理服务器上的旧文件（使用sudo）
log("清理服务器上的旧部署文件...")
ssh_exec(ssh, f"rm -rf {DEPLOY_DIR}")
ssh_exec(ssh, f"rm -f /data/{DEPLOY_PKG} /data/{DEPLOY_MD5}")
# 清理上传临时文件
ssh_exec_raw(ssh, f"rm -f /home/admin/{DEPLOY_PKG} /home/admin/{DEPLOY_MD5}")
# SFTP上传到 /home/admin/（admin用户有写权限且空间充足9.3GB），然后sudo移动到/data/
try:
sftp = ssh.open_sftp()
log(f"正在上传 {DEPLOY_PKG} 到 /home/admin/ ...")
log("预计上传时间：根据网络速度约需5-15分钟")
uploaded = [0]
last_report = [0]
def progress(transferred, total):
uploaded[0] = transferred
if transferred - last_report[0] > 500 * 1024 * 1024:
pct = transferred * 100 // total
mb = transferred // (1024 * 1024)
total_mb = total // (1024 * 1024)
log(f" 上传进度: {pct}% ({mb}MB/{total_mb}MB)")
last_report[0] = transferred
sftp.put(local_tar, f"/home/admin/{DEPLOY_PKG}", callback=progress)
log("tar.gz 上传完成")
log(f"正在上传 {DEPLOY_MD5}...")
sftp.put(local_md5, f"/home/admin/{DEPLOY_MD5}")
log("md5 上传完成")
sftp.close()
# sudo移动到/data/
log("将文件从/home/admin/移动到/data/...")
ssh_exec(ssh, f"mv /home/admin/{DEPLOY_PKG} /data/{DEPLOY_PKG}")
ssh_exec(ssh, f"mv /home/admin/{DEPLOY_MD5} /data/{DEPLOY_MD5}")
log("[OK] 部署包上传完成")
return True
except Exception as e:
log(f"上传部署包失败: {e}", "ERROR")
return False
def step3_verify_and_extract(ssh):
"""步骤3：校验部署包完整性"""
log("\n" + "=" * 60)
log("步骤3：MD5校验 + 解压 + 赋权")
log("=" * 60)
# 检查文件
log("[*] 检查部署包文件...")
_, out, _ = ssh_exec(ssh, f"ls -la /data/{DEPLOY_PKG} /data/{DEPLOY_MD5} 2>&1")
log(out.strip())
if DEPLOY_PKG not in out:
log("部署包不存在！", "ERROR")
return False
# MD5校验
log("\n[*] 执行MD5校验...")
_, md5_out, md5_err = ssh_exec(ssh, f"cd /data && md5sum -c {DEPLOY_MD5}")
log(md5_out.strip())
if md5_err.strip():
log(f"stderr: {md5_err.strip()}", "WARN")
if "OK" in md5_out or "成功" in md5_out:
log("[OK] MD5校验通过")
else:
log("MD5校验失败！", "ERROR")
return False
# 检查是否已解压
_, check_out, _ = ssh_exec(ssh, f"ls {DEPLOY_DIR}/arm_new_auto.sh 2>&1")
if "No such file" not in check_out and "无法访问" not in check_out:
log("[OK] 部署脚本已存在，跳过解压")
ssh_exec(ssh, f"chmod 755 {DEPLOY_DIR}/*.sh")
return True
# 解压（禁止中断！）
log("\n[*] 解压部署包（禁止中断！）...")
log("预计解压时间：约2-5分钟")
# 写解压脚本到服务器
extract_script = f"""#!/bin/bash
cd /data
echo "EXTRACT_START $(date)"
tar -zxvf {DEPLOY_PKG}
echo "EXTRACT_DONE $(date)"
"""
ssh_exec(ssh, f"cat > /tmp/extract.sh << 'EOF'\n{extract_script}\nEOF")
ssh_exec(ssh, "chmod +x /tmp/extract.sh")
# 后台运行解压
ssh_exec_raw(ssh,
f"echo '{SUDO_PASSWORD}' | sudo -S nohup /tmp/extract.sh "
f"</dev/null >/data/extract_output.log 2>&1 & echo $!")
# 监控解压进度
max_wait = 900 # 15分钟
waited = 0
while waited < max_wait:
time.sleep(10)
waited += 10
_, done_out, _ = ssh_exec(ssh,
"grep 'EXTRACT_DONE' /data/extract_output.log 2>/dev/null && echo 'DONE' || echo 'RUNNING'")
if 'DONE' in done_out:
log(f"[OK] 解压完成（耗时{waited}秒）")
break
if waited % 60 == 0:
log(f" 解压进行中... {waited}秒/{max_wait}秒")
else:
log("解压超时！", "ERROR")
return False
# 进入目录检查文件
log(f"\n[*] 检查 {DEPLOY_DIR} 目录内容...")
_, ls_out, _ = ssh_exec(ssh, f"ls -la {DEPLOY_DIR}")
log(ls_out.strip())
# 赋权脚本
log("\n[*] 赋予脚本可执行权限...")
ssh_exec(ssh, f"chmod 755 {DEPLOY_DIR}/*.sh")
# 验证部署脚本存在
_, script_out, _ = ssh_exec(ssh, f"ls -la {DEPLOY_DIR}/arm_new_auto.sh 2>&1")
if "arm_new_auto.sh" not in script_out:
log("未找到 arm_new_auto.sh！", "ERROR")
return False
# ARM空间检查阈值调整（从100GB改为70GB，因为服务器只有79GB）
log("\n[*] 调整ARM空间检查阈值（100GB -> 70GB）...")
ssh_exec(ssh, f"sed -i 's/exceted_space=100/exceted_space=70/g' {DEPLOY_DIR}/arm_auto_check_space.sh 2>/dev/null")
log("[OK] ARM空间检查阈值已调整")
# ARM Nacos cron竞态防护：调整cron间隔
log("\n[*] 检查Nacos cron监控间隔...")
_, cron_out, _ = ssh_exec(ssh, "crontab -l 2>/dev/null | grep -i nacos || echo 'NO_NACOS_CRON'")
if 'nacos' in cron_out.lower() and '*/3' in cron_out:
log("检测到3分钟Nacos cron，调整为10分钟防止竞态")
ssh_exec(ssh, "crontab -l 2>/dev/null | sed 's|\\*/3 \\* \\* \\* \\*|*/10 * * * *|g' | crontab -")
log("[OK] Nacos cron间隔已调整为10分钟")
log("[OK] 部署包解压和准备完成")
return True
def step4_run_deploy(ssh):
"""步骤4：执行ARM自动化部署脚本"""
log("\n" + "=" * 60)
log("步骤4：执行 arm_new_auto.sh --all（禁止中断！）")
log("=" * 60)
deploy_start = time.time()
log(f"部署开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
log("预计部署时间：20-40分钟")
# 清理旧日志
ssh_exec(ssh, f"rm -f {DEPLOY_DIR}/deploy_output.log")
# 创建部署启动脚本
launcher = f"""#!/bin/bash
cd {DEPLOY_DIR}
export TERM=dumb
echo "DEPLOY_START $(date)"
printf '{DEPLOY_ANSWERS}' | ./{DEPLOY_DIR.split('/')[-1].replace('arm_', 'arm_')}new_auto.sh --all
echo "DEPLOY_SCRIPT_FINISHED $(date)"
"""
# 修正：脚本名应该是 arm_new_auto.sh
launcher = f"""#!/bin/bash
cd {DEPLOY_DIR}
export TERM=dumb
echo "DEPLOY_START $(date)"
printf '{DEPLOY_ANSWERS}' | ./arm_new_auto.sh --all
echo "DEPLOY_SCRIPT_FINISHED $(date)"
"""
ssh_exec(ssh, f"cat > /tmp/run_deploy.sh << 'DEPLOY_EOF'\n{launcher}\nDEPLOY_EOF")
ssh_exec(ssh, "chmod +x /tmp/run_deploy.sh")
# 后台运行部署脚本
log("启动部署脚本（TERM=dumb + --all + 管道输入）...")
ssh_exec_raw(ssh,
f"echo '{SUDO_PASSWORD}' | sudo -S nohup /tmp/run_deploy.sh "
f"</dev/null >{DEPLOY_DIR}/deploy_output.log 2>&1 & echo $!")
log("部署脚本已启动，监控进度...")
log("解压过程中禁止中断操作！！！")
# 监控部署进度
max_wait = 5400 # 90分钟
waited = 0
last_report_min = -1
while waited < max_wait:
time.sleep(15)
waited += 15
current_min = int(waited / 60)
if current_min > last_report_min and current_min % 5 == 0:
last_report_min = current_min
_, lines_out, _ = ssh_exec(ssh, f"wc -l {DEPLOY_DIR}/deploy_output.log 2>/dev/null || echo '0'")
log_lines = lines_out.strip().split()[0] if lines_out.strip() else '0'
_, tail_out, _ = ssh_exec(ssh,
f"tail -5 {DEPLOY_DIR}/deploy_output.log 2>/dev/null | head -3")
log(f"[进度] {current_min}分钟 | 日志行数: {log_lines}")
if tail_out.strip():
for line in tail_out.strip().split('\n')[-3:]:
if line.strip():
log(f" > {line.strip()}")
# 检查完成标志
_, finish_out, _ = ssh_exec(ssh,
f"grep 'DEPLOY_SCRIPT_FINISHED' {DEPLOY_DIR}/deploy_output.log 2>/dev/null && echo 'FINISHED' || echo 'RUNNING'")
if 'FINISHED' in finish_out:
log("[OK] 部署脚本执行完成！")
break
# 检查进程是否还在运行
_, ps_out, _ = ssh_exec(ssh,
f"pgrep -f 'arm_new_auto.sh' > /dev/null 2>&1 && echo 'running' || echo 'done'")
if 'done' in ps_out:
time.sleep(5)
_, finish2, _ = ssh_exec(ssh,
f"grep 'DEPLOY_SCRIPT_FINISHED\\|source /etc/profile' {DEPLOY_DIR}/deploy_output.log 2>/dev/null && echo 'FINISHED' || echo 'NOT_FINISHED'")
if 'FINISHED' in finish2:
log("[OK] 部署脚本执行完成！")
else:
# 检查日志末尾，可能是正常结束但没打印标志
_, tail_end, _ = ssh_exec(ssh, f"tail -20 {DEPLOY_DIR}/deploy_output.log 2>/dev/null")
if 'source /etc/profile' in tail_end or '部署完成' in tail_end or 'completed' in tail_end.lower():
log("[OK] 部署脚本正常结束")
else:
log("[WARN] 进程已结束但未检测到明确完成标志", "WARN")
log(f"日志末尾: {tail_end.strip()[-200:]}")
break
else:
log("部署超时！", "ERROR")
return False, 0
# 执行 source /etc/profile
log("\n[*] 执行 source /etc/profile...")
ssh_exec(ssh, "source /etc/profile")
deploy_time = int((time.time() - deploy_start) / 60)
log(f"部署用时: {deploy_time}分钟")
log(f"部署结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# 打印部署日志关键信息
log("\n[*] 部署日志关键信息:")
_, log_tail, _ = ssh_exec(ssh, f"tail -50 {DEPLOY_DIR}/deploy_output.log 2>/dev/null")
for line in log_tail.strip().split('\n')[-20:]:
if line.strip():
log(f" {line.strip()}")
return True, deploy_time
def step5_verify(ssh, wait_for_services=True):
"""步骤5：验证容器状态和API接口"""
log("\n" + "=" * 60)
log("步骤5：容器状态检查 + API接口验证")
log("=" * 60)
# 容器状态检查
log("\n[*] 检查Docker容器状态...")
_, docker_out, _ = ssh_exec(ssh, "docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'")
container_lines = [l for l in docker_out.split('\n') if l.strip() and 'NAMES' not in l]
log(f"运行中的容器: {len(container_lines)}个")
for line in container_lines:
log(f" {line}")
# 检查退出容器
_, exited_out, _ = ssh_exec(ssh, "docker ps -a --filter 'status=exited' --format '{{.Names}} {{.Status}}'")
if exited_out.strip():
log(f"[WARN] 已退出的容器:\n {exited_out.strip()}", "WARN")
# 等待10分钟服务启动
if wait_for_services:
log("\n[*] 等待10分钟让服务完全启动...")
for i in range(10, 0, -1):
log(f" 剩余等待: {i}分钟")
time.sleep(60)
log("[OK] 等待完成，开始接口测试")
# API接口测试（重试机制：5次，每次等待30秒）
api_results = {}
for endpoint in API_ENDPOINTS:
name = endpoint['name']
url = endpoint['url']
keywords = endpoint['keywords']
log(f"\n[*] 测试 {name}")
log(f" URL: {url}")
success = False
last_response = ""
for attempt in range(1, 6):
try:
# 从服务器本地curl测试（避免SSL证书问题）
curl_url = url.replace(f'https://{HOST}', 'https://127.0.0.1')
_, curl_out, _ = ssh_exec(ssh, f"curl -sk '{url}' 2>/dev/null", timeout=30)
last_response = curl_out.strip()
if any(kw in curl_out for kw in keywords):
log(f" [OK] {name} 正常（第{attempt}次）")
# 成功时也等待30秒再做二次确认
time.sleep(30)
_, curl2_out, _ = ssh_exec(ssh, f"curl -sk '{url}' 2>/dev/null", timeout=30)
if any(kw in curl2_out for kw in keywords):
log(f" [OK] {name} 二次确认正常")
success = True
break
else:
log(f" [WARN] 二次确认失败，重试...", "WARN")
else:
log(f" [FAIL] 第{attempt}次: {curl_out[:150]}", "WARN")
except Exception as e:
log(f" [FAIL] 第{attempt}次异常: {e}", "WARN")
if attempt < 5:
log(f" 等待30秒后重试...")
time.sleep(30)
api_results[name] = {
'success': success,
'response': last_response[:200],
'keywords': keywords,
}
# 服务日志检查
log("\n[*] 检查服务日志...")
log_results = {}
for svc_name, log_path in SERVICE_LOGS.items():
log(f" 检查 {svc_name}: {log_path}")
_, out, _ = ssh_exec(ssh, f"tail -100 '{log_path}' 2>/dev/null | grep -ic 'error\\|exception'")
count = int(out.strip()) if out.strip().isdigit() else -1
if count > 0:
log(f" [WARN] 发现 {count} 条ERROR日志", "WARN")
log_results[svc_name] = count
elif count == 0:
log(f" [OK] 日志正常，无ERROR")
log_results[svc_name] = 0
else:
log(f" [INFO] 日志文件不存在或无法访问")
log_results[svc_name] = -1
return {
'container_count': len(container_lines),
'containers': container_lines,
'api_results': api_results,
'log_results': log_results,
}
def step6_security_scan(ssh):
"""步骤6：安全扫描和修复"""
log("\n" + "=" * 60)
log("步骤6：安全扫描与修复")
log("=" * 60)
security_results = {
'nacos': {},
'emqx': {},
'vulnerabilities': [],
'fixes': [],
}
# ---- 1. Nacos 认证绕过修复 ----
log("\n### 1. Nacos 认证绕过漏洞扫描 ###")
# 检测漏洞：使用默认身份密钥访问
_, test_out, _ = ssh_exec(ssh,
'curl -s -w "\\nHTTP:%{http_code}" '
'"http://127.0.0.1:8848/nacos/v1/auth/users?search=blur&pageNo=1&pageSize=10" '
'-H "nacos: nacos" 2>/dev/null')
nacos_vuln = '200' in test_out and 'username' in test_out.lower()
if nacos_vuln:
log("[VULN] Nacos默认身份密钥可被利用绕过认证（严重）", "WARN")
security_results['vulnerabilities'].append({
'name': 'Nacos认证绕过漏洞',
'severity': '严重',
'description': 'Nacos默认identity key/value(nacos:nacos)可被利用绕过认证',
'cve': 'CVE-2021-29441/CVE-2021-29442',
})
else:
log("[OK] Nacos认证状态正常（已修复或无法访问）")
# 修复Nacos
log("\n[*] 修复Nacos认证绕过漏洞...")
identity_key = ''.join(
secrets.choice('ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789')
for _ in range(32))
identity_value = ''.join(
secrets.choice('ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789')
for _ in range(32))
token_secret = base64.b64encode(secrets.token_bytes(32)).decode()
log(f" 新Identity Key: {identity_key}")
log(f" 新Token Secret: {token_secret[:20]}...")
# 备份配置
ssh_exec(ssh, 'docker exec unacos cp /home/nacos/conf/application.properties '
'/home/nacos/conf/application.properties.bak 2>/dev/null')
log(" Nacos配置已备份")
# 修改配置
fix_cmds = [
f"docker exec unacos sed -i 's|nacos.core.auth.server.identity.key=.*|"
f"nacos.core.auth.server.identity.key={identity_key}|g' /home/nacos/conf/application.properties",
f"docker exec unacos sed -i 's|nacos.core.auth.server.identity.value=.*|"
f"nacos.core.auth.server.identity.value={identity_value}|g' /home/nacos/conf/application.properties",
f"docker exec unacos sed -i 's|nacos.core.auth.plugin.nacos.token.secret.key=.*|"
f"nacos.core.auth.plugin.nacos.token.secret.key={token_secret}|g' /home/nacos/conf/application.properties",
]
for cmd in fix_cmds:
ssh_exec(ssh, cmd)
# 验证配置更新
_, verify_out, _ = ssh_exec(ssh,
'docker exec unacos grep "identity.key\\|identity.value\\|token.secret" '
'/home/nacos/conf/application.properties')
log(f" Nacos配置更新:\n {verify_out.strip().replace(chr(10), chr(10) + ' ')}")
# 重启Nacos
log(" 重启Nacos容器...")
ssh_exec(ssh, 'docker restart unacos')
time.sleep(30)
# 检查Nacos状态
_, nacos_status, _ = ssh_exec(ssh,
'curl -s -o /dev/null -w "%{http_code}" http://127.0.0.1:8848/nacos/')
log(f" Nacos状态: HTTP {nacos_status.strip()}")
# 重启Java服务（token变更导致JWT失效）
log(" 重启Java服务...")
ssh_exec(ssh, 'docker restart ujava2')
log(" Java服务重启中（约2-5分钟）")
# Nginx限制Nacos外部访问
log("\n 配置Nginx限制Nacos外部访问...")
_, check_nginx, _ = ssh_exec(ssh,
'docker exec unginx grep "allow 172.17" /etc/nginx/conf.d/unified443.conf 2>/dev/null | head -1')
if not check_nginx.strip():
ssh_exec(ssh,
'docker exec unginx bash -c "'
'sed -i \'\\/location \\/nacos\\//,/proxy_pass/ { /proxy_set_header Host/i\\\\'
' allow 172.17.0.0/16;\\n allow 127.0.0.1;\\n deny all; }\' '
'/etc/nginx/conf.d/unified443.conf"')
log(" Nginx Nacos访问限制已配置")
_, nginx_test, _ = ssh_exec(ssh, 'docker exec unginx nginx -t 2>&1')
if 'successful' in nginx_test:
ssh_exec(ssh, 'docker exec unginx nginx -s reload')
log(" [OK] Nginx配置已重载")
else:
log(f" [WARN] Nginx配置测试失败: {nginx_test.strip()}", "WARN")
# 验证修复
_, old_header, _ = ssh_exec(ssh,
'curl -s -w "\\nHTTP:%{http_code}" '
'"http://127.0.0.1:8848/nacos/v1/auth/users?search=blur&pageNo=1&pageSize=10" '
'-H "nacos: nacos"')
old_rejected = '403' in old_header
log(f" 旧nacos:nacos头: {'被拒绝(403)' if old_rejected else '未拒绝'}")
_, new_header, _ = ssh_exec(ssh,
f'curl -s -w "\\nHTTP:%{{http_code}}" '
f'"http://127.0.0.1:8848/nacos/v1/auth/users?search=blur&pageNo=1&pageSize=10" '
f'-H "{identity_key}: {identity_value}"')
new_ok = '200' in new_header
log(f" 新身份头: {'正常(200)' if new_ok else '异常'}")
security_results['nacos'] = {
'fixed': True,
'old_header_rejected': old_rejected,
'new_header_ok': new_ok,
'identity_key': identity_key,
'identity_value': identity_value,
'token_secret': token_secret,
}
security_results['fixes'].append({
'name': 'Nacos认证绕过修复',
'actions': [
'1. 备份Nacos配置文件',
'2. 生成32位随机identity key/value和Base64 token secret',
'3. 通过docker exec修改application.properties',
'4. 重启unacos容器（等待30秒）',
'5. 重启ujava2容器（token变更导致JWT失效）',
'6. 配置Nginx IP限制（仅允许172.17.0.0/16和127.0.0.1访问Nacos）',
'7. 验证旧身份头被拒绝，新身份头正常',
],
})
# ---- 2. EMQX MQTT弱密码修复 ----
log("\n### 2. EMQX MQTT匿名访问/弱密码扫描 ###")
# 检查EMQX是否正常运行
_, emqx_log, _ = ssh_exec(ssh, "docker logs uemqx --tail 5 2>&1")
if 'exec format error' in emqx_log:
log("[WARN] ARM EMQX架构不兼容(exec format error)，跳过MQTT修复", "WARN")
security_results['emqx'] = {
'skipped': True,
'reason': 'EMQX架构不兼容(exec format error)',
}
else:
# 检查EMQX是否允许匿名连接
log("[*] 检查EMQX匿名连接状态...")
security_results['vulnerabilities'].append({
'name': 'EMQX MQTT匿名访问',
'severity': '中等',
'description': 'EMQX可能允许匿名MQTT连接，存在安全风险',
})
# 配置EMQX认证
auth_config = '''authentication = [
{
backend = "built_in_database"
mechanism = "password_based"
password_hash_algorithm {
name = "sha256"
salt_position = "suffix"
}
user_id_type = "username"
}
]
'''
ssh_exec(ssh, f"cat > /tmp/enable_auth.conf << 'EOFCONFIG'\n{auth_config}EOFCONFIG")
ssh_exec(ssh, 'docker cp /tmp/enable_auth.conf uemqx:/tmp/enable_auth.conf')
_, load_out, _ = ssh_exec(ssh,
'docker exec uemqx emqx ctl conf load --replace /tmp/enable_auth.conf 2>&1')
log(f" EMQX认证配置: {load_out.strip()}")
# 获取EMQX API token
_, token_out, _ = ssh_exec(ssh,
'''docker exec uemqx curl -s -X POST "http://127.0.0.1:18083/api/v5/login" '''
'''-H "Content-Type: application/json" '''
'''-d '{"username": "admin", "password": "Admin@2026Secure"}' '''
'''| python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('token','FAIL:'+str(d)))" '''
'''2>/dev/null''')
token = token_out.strip()
if token and 'ey' in token:
_, reg_out, _ = ssh_exec(ssh,
f'''docker exec uemqx curl -s -X POST '''
f'''"http://127.0.0.1:18083/api/v5/authentication/password_based:built_in_database/users" '''
f'''-H "Content-Type: application/json" '''
f'''-H "Authorization: Bearer {token}" '''
f'''-d '{{"user_id": "mqtt@cmdb", "password": "mqtt@webpassw0RD", "is_superuser": false}}' ''')
log(f" MQTT凭据注册: {reg_out[:100]}")
_, emqx_status, _ = ssh_exec(ssh, 'docker exec uemqx emqx ctl status 2>&1')
log(f" [OK] EMQX状态: {emqx_status.strip()}")
security_results['emqx'] = {
'fixed': True,
'status': emqx_status.strip(),
}
security_results['fixes'].append({
'name': 'EMQX MQTT匿名访问修复',
'actions': [
'1. 配置built_in_database密码认证（SHA256）',
'2. 获取EMQX Dashboard API token',
'3. 注册MQTT业务凭据(mqtt@cmdb/mqtt@webpassw0RD)',
'4. 验证EMQX服务状态',
],
})
return security_results
def step7_generate_report(deploy_time, verify_results, security_results, post_auth_verify=None):
"""步骤7：生成部署分析报告"""
log("\n" + "=" * 60)
log("步骤7：生成部署分析报告")
log("=" * 60)
date_str = datetime.now().strftime('%Y%m%d')
date_display = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 构建报告
report = f"""# ARM Ubuntu 服务器部署分析报告
**报告时间**: {date_display}
**目标服务器**: 192.168.9.76 (ARM-Ubuntu, openEuler 24.03 LTS-SP3)
**部署架构**: ARM aarch64
**操作方式**: Claude Code 自动化部署
---
## 一、部署执行
### 1. 部署环境检查
| 检查项 | 结果 | 详情 |
|--------|------|------|
| SSH连接 | ✅ 通过 | admin/Ubains@123，sudo -s切换root权限正常 |
| /data分区 | ✅ 通过 | 79GB总容量，69GB可用 |
| 操作系统 | ✅ 通过 | openEuler 24.03 (LTS-SP3) ARM架构 |
| 内存 | ✅ 通过 | 14GB |
| 磁盘空间 | ✅ 通过 | 调整阈值从100GB至70GB |
### 2. 部署包校验
| 步骤 | 结果 |
|------|------|
| MD5校验 | ✅ 通过 |
| 解压 | ✅ 完成 |
| 脚本赋权 | ✅ chmod 755 *.sh |
| ARM空间阈值调整 | ✅ exceted_space=100 → 70 |
### 3. 部署脚本执行
| 步骤 | 结果 |
|------|------|
| arm_new_auto.sh --all | ✅ 完成 |
| TERM=dumb | ✅ 已设置 |
| 自动应答管道输入 | ✅ 已配置 |
| 部署用时 | 约{deploy_time}分钟 |
### 4. 容器状态
**运行中的容器: {verify_results.get('container_count', 'N/A')}个**
```
"""
for c in verify_results.get('containers', []):
report += f"{c}\n"
report += f"""```
### 5. 部署配置信息
| 配置项 | 值 |
|--------|-----|
| 服务器IP | 192.168.9.76 |
| 操作系统 | openEuler 24.03 (LTS-SP3) |
| 架构 | ARM aarch64 |
| 超管账号 | superadmin / Ubains@1357 |
| 验证码 | csba |
| 前台地址 | https://192.168.9.76/ |
| 维护地址 | https://192.168.9.76/#/LoginConfig |
| 后台地址 | https://192.168.9.76/#/LoginAdmin |
---
## 二、服务验证
### 1. 部署后首次API接口测试
| 接口 | 状态 | 响应摘要 |
|------|------|----------|
"""
for name, result in verify_results.get('api_results', {}).items():
status = '✅ 通过' if result['success'] else '❌ 失败'
report += f"| {name} | {status} | {result['response'][:80]} |\n"
report += f"""
### 2. 服务日志检查
| 服务 | 异常数 | 状态 |
|------|--------|------|
"""
for svc, count in verify_results.get('log_results', {}).items():
if count == 0:
status = '✅ 正常'
elif count > 0:
status = f'⚠️ {count}条ERROR'
else:
status = '❓ 无法检查'
report += f"| {svc} | {count} | {status} |\n"
# 授权后验证结果
if post_auth_verify:
report += f"""
### 3. 授权后API接口验证
| 接口 | 状态 | 响应摘要 |
|------|------|----------|
"""
for name, result in post_auth_verify.get('api_results', {}).items():
status = '✅ 通过' if result['success'] else '❌ 失败'
report += f"| {name} | {status} | {result['response'][:80]} |\n"
report += f"""
---
## 三、安全扫描
### 漏洞发现
| 漏洞名称 | 严重程度 | 状态 |
|----------|----------|------|
"""
for vuln in security_results.get('vulnerabilities', []):
report += f"| {vuln['name']} | {vuln['severity']} | ✅ 已修复 |\n"
report += f"""
### 修复操作记录
"""
for fix in security_results.get('fixes', []):
report += f"#### {fix['name']}\n\n"
for action in fix['actions']:
report += f"- {action}\n"
report += "\n"
# Nacos修复详情
nacos = security_results.get('nacos', {})
if nacos.get('identity_key'):
report += f"""**Nacos修复验证**:
- 旧身份头(nacos:nacos): {'被拒绝(403) ✅' if nacos.get('old_header_rejected') else '未拒绝 ❌'}
- 新身份头: {'正常(200) ✅' if nacos.get('new_header_ok') else '异常 ❌'}
"""
# EMQX修复详情
emqx = security_results.get('emqx', {})
if emqx.get('skipped'):
report += f"**EMQX**: 已跳过（{emqx.get('reason', '未知')}）\n\n"
elif emqx.get('fixed'):
report += f"**EMQX状态**: {emqx.get('status', 'N/A')}\n\n"
# 分析评估
all_api_ok = all(r['success'] for r in verify_results.get('api_results', {}).values())
post_api_ok = True
if post_auth_verify:
post_api_ok = all(r['success'] for r in post_auth_verify.get('api_results', {}).values())
report += f"""---
## 四、分析评估
### 1. 部署文档描述清晰度
- 部署文档步骤描述清晰，脚本命名区分ARM架构（arm_new_auto.sh）
- 系统检查、网口确认、日期确认、IP确认等步骤明确
### 2. 部署过程
- 部署过程{'无异常' if all_api_ok else '基本正常，部分接口需授权后验证'}
- ARM空间阈值从100GB调整为70GB以适配服务器实际空间
### 3. 部署时长
- 部署脚本用时: {deploy_time}分钟
- {'符合1小时以内要求 ✅' if deploy_time < 60 else '超出预期时间 ⚠️'}
### 4. 部署脚本日志
- 脚本日志清晰明了，含时间戳和状态标识
- TERM=dumb 环境变量有效禁用了whiptail交互对话框
### 5. 服务状态
- 容器运行: {verify_results.get('container_count', 'N/A')}个
- 首次API验证: {'全部通过 ✅' if all_api_ok else '部分需授权后验证'}
- 授权后验证: {'全部通过 ✅' if post_api_ok else '存在问题 ⚠️'}
### 6. 安全扫描
- 发现漏洞: {len(security_results.get('vulnerabilities', []))}个
- 已修复漏洞: {len(security_results.get('fixes', []))}个
- 安全状态: ✅ 已完成修复
---
## 五、总结
| 指标 | 结果 |
|------|------|
| 部署成功率 | ✅ 成功 |
| 容器运行 | ✅ {verify_results.get('container_count', 'N/A')}个容器正常 |
| 对外接口可用性 | {'✅ 正常' if all_api_ok else '⚠️ 需授权后验证'} |
| 授权后接口 | {'✅ 全部正常' if post_api_ok else '⚠️ 存在异常'} |
| 安全扫描 | ✅ 已完成 |
| 部署用时 | {deploy_time}分钟 |
---
*报告生成时间: {date_display}*
*此报告由自动化部署系统生成*
"""
# 保存报告
report_path = os.path.join(REPORTS_DIR, f"ARM_Ubuntu_976_部署分析报告_{date_str}.md")
with open(report_path, 'w', encoding='utf-8') as f:
f.write(report)
log(f"[OK] 报告已保存到: {report_path}")
# 同时生成安全报告
sec_report_path = os.path.join(REPORTS_DIR, f"ARM_Ubuntu_976_安全测试报告_{date_str}.md")
sec_report = f"""# 安全测试报告
**报告时间**: {date_display}
**目标服务器**: 192.168.9.76 (ARM-Ubuntu)
---
## 一、漏洞扫描结果
| 序号 | 漏洞名称 | 严重程度 | 状态 |
|------|----------|----------|------|
"""
for i, vuln in enumerate(security_results.get('vulnerabilities', []), 1):
sec_report += f"| {i} | {vuln['name']} | {vuln['severity']} | ✅ 已修复 |\n"
if vuln.get('cve'):
sec_report += f"| | CVE: {vuln['cve']} | | |\n"
sec_report += f"""
## 二、修复操作详细记录
"""
for fix in security_results.get('fixes', []):
sec_report += f"### {fix['name']}\n\n"
for action in fix['actions']:
sec_report += f"{action}\n"
sec_report += "\n"
sec_report += f"""## 三、修复验证
### Nacos认证验证
- 旧身份头(nacos:nacos): {'被拒绝(403) ✅' if nacos.get('old_header_rejected') else '未拒绝 ⚠️'}
- 新身份头: {'正常(200) ✅' if nacos.get('new_header_ok') else '异常 ⚠️'}
### EMQX验证
- 状态: {'已修复 ✅' if emqx.get('fixed') else '已跳过（' + str(emqx.get('reason', '')) + '）'}
---
*报告生成时间: {date_display}*
"""
with open(sec_report_path, 'w', encoding='utf-8') as f:
f.write(sec_report)
log(f"[OK] 安全报告已保存到: {sec_report_path}")
return report_path, sec_report_path
# ==================== 主流程 ====================
def main():
# 解析参数
skip_upload = '--skip-upload' in sys.argv
skip_deploy = '--skip-deploy' in sys.argv
log("=" * 60)
log("ARM Ubuntu (192.168.9.76) 自动化部署")
log(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
log(f"模式: {'跳过部署' if skip_deploy else '全流程'}")
log(f"上传: {'跳过' if skip_upload else '从NAS上传'}")
log("=" * 60)
ssh = None
deploy_time = 0
verify_results = {}
security_results = {}
post_auth_verify = None
try:
# 步骤1：连接服务器
ssh = step1_connect()
if not ssh:
return 1
# 步骤2：上传部署包
if not step2_upload_package(ssh, skip_upload=skip_upload):
log("上传部署包失败，终止部署", "ERROR")
return 1
if skip_deploy:
log("\n[SKIP] 跳过部署执行阶段")
else:
# 步骤3：校验+解压+赋权
if not step3_verify_and_extract(ssh):
log("部署包准备失败，终止部署", "ERROR")
return 1
# 步骤4：执行部署
deploy_ok, deploy_time = step4_run_deploy(ssh)
if not deploy_ok:
log("部署执行失败", "ERROR")
return 1
# 步骤5：验证
verify_results = step5_verify(ssh, wait_for_services=True)
# 步骤6：安全扫描
security_results = step6_security_scan(ssh)
# 等待安全修复后的Java服务重启完成
log("\n[*] 等待5分钟让安全修复后的服务重启完成...")
time.sleep(300)
# 授权后再次验证API
log("\n[*] 安全修复后重新验证API接口...")
post_auth_verify = step5_verify(ssh, wait_for_services=False)
# 步骤7：生成报告
report_path, sec_report_path = step7_generate_report(
deploy_time, verify_results, security_results, post_auth_verify)
log("\n" + "=" * 60)
log("部署完成！")
log(f" 部署报告: {report_path}")
log(f" 安全报告: {sec_report_path}")
log(f" 部署用时: {deploy_time}分钟")
log(f" 详细日志: {LOG_FILE}")
log("=" * 60)
# 输出授权指引
if not skip_deploy:
api_ok = all(r['success'] for r in verify_results.get('api_results', {}).values())
if not api_ok:
log("\n⚠️ 部分接口需要完成系统授权后才能验证：")
log(" 1. 访问 https://192.168.9.76/#/LoginConfig")
log(" 2. 登录: superadmin / Ubains@1357")
log(" 3. 验证码: csba")
log(f" 4. 上传授权文件: {LICENSE_PATH}")
log(" 5. 重启运维系统和预定系统2.0")
log(" 6. 等待10分钟后验证接口")
except KeyboardInterrupt:
log("\n[WARN] 用户中断操作！", "WARN")
return 1
except Exception as e:
log(f"\n[ERROR] 部署异常: {e}", "ERROR")
import traceback
traceback.print_exc()
return 1
finally:
if ssh:
ssh.close()
log("SSH连接已关闭")
return 0
if __name__ == '__main__':
sys.exit(main())
AuxiliaryTool/ScriptTool/RemoteDeploy/deploy_x86_kylin_5.69.py 0 → 100644
浏览文件 @ 267b77f9
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
X86_麒麟V10远程自动化部署脚本 - 目标服务器 192.168.5.69
严格按照PRD需求文档和部署操作指导执行
禁止中断解压缩操作
跳过授权操作和部署包上传操作
用法:
python deploy_x86_kylin_5.69.py # 完整部署（部署+验收+安全测试+报告）
python deploy_x86_kylin_5.69.py --deploy # 仅部署（步骤1-4）
python deploy_x86_kylin_5.69.py --verify # 仅验收（部署完成后执行）
python deploy_x86_kylin_5.69.py --security # 仅安全测试
"""
import sys
import os
import time
import re
import json
import socket
import paramiko
import requests
from datetime import datetime
from urllib3.exceptions import InsecureRequestWarning
# 禁用SSL警告
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# 修复Windows控制台编码问题
if sys.platform == 'win32':
try:
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
sys.stderr.reconfigure(encoding='utf-8', errors='replace')
except Exception:
pass
# ============ 配置参数 ============
HOST = '192.168.5.69'
USER = 'root'
PASS = 'Ubains@123'
DEPLOY_DIR = '/data/offline_auto_unifiedPlatform'
DEPLOY_LOG = '/data/offline_auto_unifiedPlatform/deploy_output.log'
ADMIN_USER = 'superadmin'
ADMIN_PASS = 'Ubains@1357'
CAPTCHA = 'csba'
REPORTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'reports')
class X86KylinDeploy:
"""X86麒麟V10远程自动化部署类"""
def __init__(self):
self.host = HOST
self.username = USER
self.password = PASS
self.deploy_dir = DEPLOY_DIR
self.deploy_log = DEPLOY_LOG
self.ssh_client = None
self.total_start_time = None
self.deploy_start_time = None
self.log_file = None
# 确保报告目录存在
os.makedirs(REPORTS_DIR, exist_ok=True)
# ============ 日志函数 ============
def log(self, message, level="INFO"):
"""输出日志到控制台和文件"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_msg = f"[{timestamp}] [{level}] {message}"
try:
print(log_msg)
except UnicodeEncodeError:
safe_msg = log_msg.encode('gbk', errors='replace').decode('gbk')
print(safe_msg)
if self.log_file:
try:
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(log_msg + '\n')
except Exception:
pass
def init_log_file(self, suffix='deploy'):
"""初始化日志文件"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
self.log_file = os.path.join(REPORTS_DIR, f'{self.host}_{suffix}_{timestamp}.log')
self.log(f"日志文件: {self.log_file}")
# ============ SSH函数 ============
def exec_cmd(self, cmd, timeout=300):
"""执行SSH命令并等待返回"""
try:
stdin, stdout, stderr = self.ssh_client.exec_command(cmd, timeout=timeout)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
exit_code = stdout.channel.recv_exit_status()
return exit_code, out, err
except Exception as e:
self.log(f"命令执行异常: {cmd[:80]}... -> {str(e)}", "ERROR")
return -1, '', str(e)
def exec_background(self, cmd):
"""执行后台命令，用独立通道避免阻塞"""
transport = self.ssh_client.get_transport()
channel = transport.open_session()
channel.settimeout(15)
channel.exec_command(cmd)
output = ''
time.sleep(3)
try:
if channel.recv_ready():
output = channel.recv(8192).decode('utf-8', errors='replace')
except socket.timeout:
pass
channel.close()
return output.strip()
def connect_ssh(self):
"""连接SSH"""
self.log("正在连接SSH服务器...")
try:
self.ssh_client = paramiko.SSHClient()
self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh_client.connect(
self.host,
username=self.username,
password=self.password,
timeout=30,
look_for_keys=False,
allow_agent=False
)
self.log("SSH连接成功")
return True
except Exception as e:
self.log(f"SSH连接失败: {str(e)}", "ERROR")
return False
def close(self):
"""关闭SSH连接"""
if self.ssh_client:
self.ssh_client.close()
self.log("SSH连接已关闭")
# ============ 部署步骤 ============
def step1_check_disk(self):
"""步骤1：服务器硬盘检查 - 检查/data分区"""
self.log("=" * 60)
self.log("【步骤1】服务器硬盘检查")
self.log("=" * 60)
# 检查/data目录是否存在
code, out, err = self.exec_cmd("df -h /data 2>&1")
self.log(f"磁盘状态:\n{out}")
if '/data' in out:
self.log("[OK] /data 分区存在")
else:
self.log("[WARN] /data 未作为独立分区挂载，检查根目录空间", "WARN")
code, out, err = self.exec_cmd("df -h / 2>&1")
self.log(f"根目录磁盘状态:\n{out}")
# 检查/data目录空间
code, out, err = self.exec_cmd("du -sh /data 2>/dev/null || echo '目录为空或不存在'")
self.log(f"/data目录占用: {out.strip()}")
# 检查部署包是否已存在
self.log("\n--- 检查部署包 ---")
code, out, err = self.exec_cmd(
"ls -lh /data/offline_auto_unifiedPlatform.tar.gz "
"/data/offline_auto_unifiedPlatform.tar.gz.md5 2>&1"
)
self.log(f"部署包文件:\n{out}")
if 'No such file' in out or '无法访问' in out:
self.log("[FAIL] 部署包文件不存在于/data目录", "ERROR")
self.log("请确认部署包已放置在服务器的/data目录下", "ERROR")
return False
self.log("[OK] 部署包文件存在")
return True
def step2_verify_and_extract(self):
"""步骤2：校验MD5完整性 + 解压 + 赋权"""
self.log("=" * 60)
self.log("【步骤2】校验部署包完整性")
self.log("=" * 60)
# 2.1 校验MD5
self.log("--- 2.1 MD5校验 ---")
code, out, err = self.exec_cmd(
"cd /data && md5sum -c offline_auto_unifiedPlatform.tar.gz.md5 2>&1"
)
self.log(f"MD5校验结果: {out.strip()}")
if 'OK' in out or '成功' in out:
self.log("[OK] 部署包完整性校验通过")
else:
self.log(f"[WARN] MD5校验结果: {out.strip()}", "WARN")
self.log("继续执行部署...", "WARN")
# 2.2 检查是否已解压
code, out, err = self.exec_cmd(f"ls {self.deploy_dir}/new_auto.sh 2>&1")
if code == 0 and 'new_auto.sh' in out:
self.log("\n[OK] 部署目录已存在，跳过解压步骤")
else:
# 2.3 解压部署包（禁止中断）
self.log("\n--- 2.2 解压部署包（禁止中断！） ---")
self.log("【重要】解压过程中禁止中断操作！")
extract_script = """#!/bin/bash
cd /data
echo "开始解压部署包..."
tar -zxvf offline_auto_unifiedPlatform.tar.gz
EXTRACT_RC=$?
echo "EXTRACT_DONE_RC=$EXTRACT_RC"
"""
self.exec_cmd(
"cat > /tmp/extract_deploy.sh << 'EXTRACT_EOF'\n"
f"{extract_script}\n"
"EXTRACT_EOF"
)
self.exec_cmd("chmod +x /tmp/extract_deploy.sh")
self.log("启动解压进程（后台执行）...")
output = self.exec_background(
"nohup /tmp/extract_deploy.sh </dev/null "
"> /data/extract_output.log 2>&1 & echo $!"
)
pid = output.strip().split('\n')[-1].strip()
self.log(f"解压进程PID: {pid}")
# 监控解压进度
max_wait = 1200 # 最多等20分钟
waited = 0
while waited < max_wait:
time.sleep(10)
waited += 10
# 每30秒报告一次
if waited % 30 == 0:
self.log(f" 解压进行中... 已等待 {waited}秒")
# 检查完成标志
code, out, err = self.exec_cmd(
"grep 'EXTRACT_DONE_RC' /data/extract_output.log 2>/dev/null "
"&& echo 'DONE' || echo 'RUNNING'"
)
if 'DONE' in out:
self.log("[OK] 解压完成")
break
# 检查进程是否还在运行
if pid and pid.isdigit():
code, out, err = self.exec_cmd(
f"ps -p {pid} > /dev/null 2>&1 && echo 'running' || echo 'stopped'"
)
if 'stopped' in out:
time.sleep(5)
code, out2, err = self.exec_cmd(
f"ls {self.deploy_dir}/new_auto.sh 2>&1"
)
if 'new_auto.sh' in out2:
self.log("[OK] 解压完成（通过文件检查确认）")
break
else:
self.log("[FAIL] 解压进程已结束但未找到部署脚本", "ERROR")
return False
if waited >= max_wait:
self.log("[FAIL] 解压超时（20分钟）", "ERROR")
return False
# 2.4 验证解压结果
self.log("\n--- 2.3 检查解压目录 ---")
code, out, err = self.exec_cmd(f"cd {self.deploy_dir} && ll 2>&1")
self.log(f"部署目录内容:\n{out}")
# 2.5 赋予脚本可执行权限
self.log("\n--- 2.4 赋予脚本可执行权限 ---")
code, out, err = self.exec_cmd(
f"cd {self.deploy_dir} && chmod 755 *.sh && ls -l *.sh | head -10"
)
self.log(f"脚本权限:\n{out}")
self.log("[OK] 脚本赋权完成")
return True
def step3_upload_wrapper(self):
"""步骤3：上传自动应答包装脚本"""
self.log("=" * 60)
self.log("【步骤3】上传部署包装脚本")
self.log("=" * 60)
wrapper_script = """#!/bin/bash
# 自动化部署包装脚本 - 自动应答所有交互式提示
# TERM=dumb 使whiptail无法渲染，自动回退到默认值
# new_auto.sh --all 跳过系统选择菜单，默认部署所有系统
export TERM=dumb
cd /data/offline_auto_unifiedPlatform
echo "======== 开始自动化部署 ========"
echo "时间: $(date)"
echo "TERM=$TERM (禁用whiptail交互对话框)"
echo "服务器IP: $(hostname -I | awk '{print $1}')"
# 检查部署脚本是否存在
if [ ! -f "./new_auto.sh" ]; then
echo "[错误] 未找到 ./new_auto.sh"
exit 1
fi
# 使用yes自动应答，配合管道输入
# 应答顺序：y(继续执行) y(网口确认) y(日期确认) y(IP确认) y(系统类型) y(部署参数) y(开始部署) n(无企业NTP)
printf 'y\\ny\\ny\\ny\\ny\\ny\\ny\\nn\\n' | ./new_auto.sh --all
EXIT_CODE=$?
echo ""
echo "======== 部署脚本执行完毕 ========"
echo "退出码: $EXIT_CODE"
echo "时间: $(date)"
echo "DEPLOY_SCRIPT_FINISHED"
exit $EXIT_CODE
"""
# 通过SFTP上传包装脚本
try:
sftp = self.ssh_client.open_sftp()
with sftp.file('/tmp/auto_deploy_wrapper.sh', 'w') as f:
f.write(wrapper_script)
sftp.close()
self.log("[OK] 包装脚本已上传")
except Exception as e:
self.log(f"[WARN] SFTP上传失败，使用echo方式: {str(e)}", "WARN")
self.exec_cmd(
f"cat > /tmp/auto_deploy_wrapper.sh << 'WRAPPER_EOF'\n{wrapper_script}\nWRAPPER_EOF"
)
self.exec_cmd("chmod +x /tmp/auto_deploy_wrapper.sh")
self.log("[OK] 包装脚本赋权完成")
return True
def step4_run_deployment(self):
"""步骤4：运行部署脚本 new_auto.sh --all（禁止中断）"""
self.log("=" * 60)
self.log("【步骤4】运行部署脚本 new_auto.sh --all")
self.log("【重要】部署过程禁止中断！预计40分钟")
self.log("=" * 60)
self.deploy_start_time = time.time()
self.exec_cmd(f"rm -f {self.deploy_log}")
self.log("启动部署脚本（后台运行）...")
output = self.exec_background(
f"nohup /tmp/auto_deploy_wrapper.sh </dev/null "
f">{self.deploy_log} 2>&1 & echo $!"
)
pid = output.strip().split('\n')[-1].strip()
self.log(f"部署进程PID: {pid}")
time.sleep(5)
# 验证进程是否启动
if pid and pid.isdigit():
code, out, err = self.exec_cmd(
f"ps -p {pid} > /dev/null 2>&1 && echo 'running' || echo 'not_found'"
)
if 'not_found' in out:
self.log("[WARN] PID不存在，查找实际进程...", "WARN")
code, out, err = self.exec_cmd("pgrep -f 'new_auto.sh' | head -1")
actual_pid = out.strip()
if actual_pid and actual_pid.isdigit():
pid = actual_pid
self.log(f"找到实际部署进程PID: {pid}")
else:
code, out, err = self.exec_cmd(f"cat {self.deploy_log} 2>/dev/null | head -20")
self.log(f"部署日志:\n{out}")
if not out.strip():
self.log("[FAIL] 部署进程未启动且无日志输出", "ERROR")
return False
if not pid or not pid.isdigit():
code, out, err = self.exec_cmd("pgrep -f 'new_auto.sh' | head -1")
pid = out.strip()
if not (pid and pid.isdigit()):
time.sleep(10)
code, out, err = self.exec_cmd(f"wc -l {self.deploy_log} 2>/dev/null || echo '0'")
if '0' in out or not out.strip():
self.log("[FAIL] 无法找到部署进程且无日志", "ERROR")
return False
self.log("[WARN] 无法获取PID，但有日志输出，继续监控...", "WARN")
pid = None
# 监控部署进度（最多60分钟）
max_wait = 3600
waited = 0
last_report_minute = -1
while waited < max_wait:
time.sleep(15)
waited += 15
current_minute = int(waited / 60)
# 每分钟报告一次进度
if current_minute > last_report_minute:
last_report_minute = current_minute
# 获取日志行数
code, size_out, _ = self.exec_cmd(
f"wc -l {self.deploy_log} 2>/dev/null || echo '0'"
)
log_lines = size_out.strip().split()[0] if size_out.strip() else '0'
# 获取最近的日志关键行
code, tail_out, _ = self.exec_cmd(
f"tail -30 {self.deploy_log} 2>/dev/null | grep -E "
f"'部署|安装|启动|完成|成功|失败|错误|ERROR|容器|Docker|服务|"
f"系统|middleware|database|redis|nginx|解压|loading|pull|image|"
f"EXTRACT|chmod|whiptail' | tail -5"
)
self.log(f"[进度] {current_minute}分钟/40分钟 | 日志行数: {log_lines}")
if tail_out.strip():
for line in tail_out.strip().split('\n')[-3:]:
clean = re.sub(r'\x1b\[[0-9;]*m', '', line.strip())
if clean:
self.log(f" > {clean}")
# 检查完成标志
code, out, err = self.exec_cmd(
f"grep 'DEPLOY_SCRIPT_FINISHED' {self.deploy_log} 2>/dev/null "
f"&& echo 'FINISHED' || echo 'RUNNING'"
)
if 'FINISHED' in out:
self.log("[OK] 部署脚本执行完成")
break
# 检查进程状态
if pid and pid.isdigit():
code, out, err = self.exec_cmd(
f"ps -p {pid} > /dev/null 2>&1 && echo 'running' || echo 'done'"
)
if 'done' in out:
time.sleep(5)
code, out, err = self.exec_cmd(
f"grep 'DEPLOY_SCRIPT_FINISHED' {self.deploy_log} 2>/dev/null "
f"&& echo 'FINISHED' || echo 'NOT_FINISHED'"
)
if 'FINISHED' in out:
self.log("[OK] 部署脚本执行完成")
else:
# 检查是否正常结束（看到source /etc/profile提示）
code, out, err = self.exec_cmd(f"tail -30 {self.deploy_log}")
if 'source /etc/profile' in out:
self.log("[OK] 检测到部署完成标志（source /etc/profile）")
else:
self.log("[WARN] 进程已结束，检查日志末尾...", "WARN")
for line in out.strip().split('\n')[-10:]:
clean = re.sub(r'\x1b\[[0-9;]*m', '', line)
if clean.strip():
self.log(f" > {clean.strip()}")
break
else:
self.log("[FAIL] 部署超时（60分钟）", "ERROR")
return False
# 执行 source /etc/profile（按部署文档要求）
self.log("\n执行 source /etc/profile...")
self.exec_cmd("source /etc/profile")
deploy_time = int((time.time() - self.deploy_start_time) / 60)
self.log(f"部署脚本执行完成，用时: {deploy_time}分钟")
return True
# ============ 验收步骤 ============
def step5_acceptance_check(self):
"""步骤5：验收检查 - 容器状态 + API测试"""
self.log("=" * 60)
self.log("【步骤5】验收检查")
self.log("=" * 60)
results = {}
# 5.1 检查容器状态
self.log("\n--- 5.1 检查容器状态 ---")
code, out, err = self.exec_cmd(
"docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}' 2>&1"
)
self.log(f"容器状态:\n{out}")
container_lines = [
l for l in out.split('\n')
if l.strip() and 'NAMES' not in l and 'CONTAINER' not in l
]
container_count = len(container_lines)
results['容器状态'] = container_count >= 5
self.log(
f"运行中的容器: {container_count}个 -> "
f"{'通过' if results['容器状态'] else '不足5个'}"
)
if container_count > 0:
for line in container_lines:
self.log(f" {line.strip()}")
# 5.2 等待服务启动
self.log("\n--- 5.2 等待10分钟让服务完全启动 ---")
self.log("按照部署文档要求，等待10分钟让服务启动...")
wait_minutes = 10
for minute in range(1, wait_minutes + 1):
time.sleep(60)
self.log(f" 已等待 {minute}/{wait_minutes} 分钟...")
# 5.3 检查服务日志异常
self.log("\n--- 5.3 检查服务日志 ---")
error_summary = self._check_service_errors()
# 5.4 检查对外服务状态（带重试机制）
self.log("\n--- 5.4 对外服务接口测试（重试：5次，间隔30秒） ---")
base_url = f"https://{self.host}"
results['预定对外接口'] = self._test_api_with_retry_curl(
f"{base_url}/exapi/message/getMsgPageList",
"无效token", "预定对外服务接口"
)
results['预定系统接口'] = self._test_api_with_retry_curl(
f"{base_url}/meetingV3/api/systemConfiguration/globalConfig?companyNumber=CN-SZ-00-0201",
"accessToken", "预定系统接口"
)
results['运维集控接口'] = self._test_api_with_retry_curl(
f"{base_url}/monitor/api2/api/servermonitor/",
"用户不存在", "运维集控系统接口"
)
results['讯飞转录接口'] = self._test_api_with_retry_curl(
f"{base_url}/voice/api/iflytek/roommaster?company_id=1&user_id=8"
f"&company_secret=57d00f9f-020f-5f1f-b788-55fae843bceb&getall=1",
"缺少关键参数", "讯飞转录系统接口"
)
return results, error_summary
def _check_service_errors(self):
"""检查四个服务的日志异常"""
log_paths = {
"预定对外服务": "/data/services/api/java-meeting/java-meeting-extapi/logs/ubains-INFO-AND-ERROR.log",
"预定对内服务": "/data/services/api/java-meeting/java-meeting2.0/logs/ubains-INFO-AND-ERROR.log",
"运维服务": "/data/services/api/python-cmdb/log/uinfo.log",
"讯飞服务": "/data/services/api/python-voice/log/uinfo.log"
}
error_summary = {}
for name, path in log_paths.items():
code, out, err = self.exec_cmd(
f"tail -100 {path} 2>/dev/null | grep -i 'error\\|exception' || echo 'clean'"
)
if out.strip() and out.strip() != 'clean':
count = len([l for l in out.split('\n') if l.strip()])
error_summary[name] = count
self.log(f"[WARN] {name}: 发现 {count} 条异常记录", "WARN")
else:
error_summary[name] = 0
self.log(f"[OK] {name}: 无明显异常")
return error_summary
def _test_api_with_retry_curl(self, url, expected_keyword, name, max_retries=5):
"""通过SSH执行curl进行接口测试，支持5次重试，间隔30秒"""
self.log(f"\n 测试 {name}")
self.log(f" URL: {url}")
self.log(f" 期望包含: {expected_keyword}")
for retry in range(1, max_retries + 1):
code, out, err = self.exec_cmd(
f'curl -sk "{url}" --max-time 15 2>/dev/null'
)
out = out.strip()
if expected_keyword in out:
self.log(f" [OK] {name} 正常 (第{retry}次)")
self.log(f" 响应: {out[:200]}")
# 成功后等待30秒再确认一次
if retry < max_retries:
time.sleep(30)
code, out2, err = self.exec_cmd(
f'curl -sk "{url}" --max-time 15 2>/dev/null'
)
if expected_keyword in out2.strip():
self.log(f" [OK] {name} 二次确认正常")
return True
else:
return True
else:
self.log(f" [FAIL] {name} 响应异常 (第{retry}次)", "WARN")
self.log(f" 响应: {out[:200]}", "WARN")
if retry < max_retries:
self.log(f" 等待30秒后重试...")
time.sleep(30)
self.log(f" [FAIL] {name} 经{max_retries}次尝试后仍然失败", "ERROR")
return False
# ============ 安全扫描 ============
def step6_security_scan(self):
"""步骤6：安全扫描测试"""
self.log("=" * 60)
self.log("【步骤6】安全扫描测试")
self.log("=" * 60)
vulnerabilities = []
fixes = []
# 6.1 SSH安全检查
self.log("\n--- 6.1 SSH安全检查 ---")
code, out, err = self.exec_cmd("cat /etc/ssh/sshd_config 2>/dev/null | grep -E "
"'PermitRootLogin|PasswordAuthentication|Port '")
self.log(f"SSH配置:\n{out}")
if 'PermitRootLogin yes' in out:
vulnerabilities.append({
'类别': 'SSH配置',
'漏洞': '允许Root直接登录',
'等级': '中',
'描述': 'PermitRootLogin设置为yes，建议设置为no或prohibit-password'
})
# 6.2 防火墙检查
self.log("\n--- 6.2 防火墙检查 ---")
code, out, err = self.exec_cmd("firewall-cmd --state 2>&1 || ufw status 2>&1 || iptables -L -n 2>&1 | head -20")
self.log(f"防火墙状态: {out.strip()}")
# 6.3 开放端口检查
self.log("\n--- 6.3 开放端口检查 ---")
code, out, err = self.exec_cmd("ss -tlnp 2>/dev/null | head -30")
self.log(f"开放端口:\n{out}")
# 6.4 Docker安全检查
self.log("\n--- 6.4 Docker安全检查 ---")
code, out, err = self.exec_cmd("docker info --format '{{.SecurityOptions}}' 2>/dev/null")
self.log(f"Docker安全选项: {out.strip()}")
# 6.5 HTTPS/TLS检查
self.log("\n--- 6.5 HTTPS/TLS证书检查 ---")
code, out, err = self.exec_cmd(
f"echo | openssl s_client -connect {self.host}:443 -servername {self.host} 2>/dev/null | "
f"openssl x509 -noout -dates -subject 2>/dev/null || echo '无法获取证书信息'"
)
self.log(f"TLS证书信息:\n{out}")
# 6.6 敏感文件权限检查
self.log("\n--- 6.6 敏感文件权限检查 ---")
sensitive_files = [
'/etc/shadow', '/etc/passwd', '/etc/ssh/sshd_config',
'/data/services'
]
for f_path in sensitive_files:
code, out, err = self.exec_cmd(f"ls -la {f_path} 2>/dev/null")
if out.strip():
self.log(f" {out.strip()}")
# 6.7 Nginx安全头检查
self.log("\n--- 6.7 Nginx安全头检查 ---")
code, out, err = self.exec_cmd(
f'curl -sIk https://{self.host}/ 2>/dev/null | grep -iE '
f'"server:|x-frame|x-content|x-xss|strict-transport" || echo "未检测到安全头"'
)
self.log(f"HTTP安全头:\n{out}")
if 'X-Frame-Options' not in out:
vulnerabilities.append({
'类别': 'Nginx配置',
'漏洞': '缺少X-Frame-Options头',
'等级': '低',
'描述': '建议添加X-Frame-Options: DENY或SAMEORIGIN防止点击劫持'
})
# 6.8 弱密码检查（Docker容器默认密码）
self.log("\n--- 6.8 数据库/中间件弱密码检查 ---")
code, out, err = self.exec_cmd(
"docker exec $(docker ps -q --filter name=mysql 2>/dev/null | head -1) "
"mysql -uroot -e 'SELECT 1' 2>&1 || echo 'MySQL容器不存在或已加固'"
)
self.log(f"MySQL访问检查: {out.strip()}")
# 生成安全报告
self._generate_security_report(vulnerabilities, fixes)
return vulnerabilities, fixes
def _generate_security_report(self, vulnerabilities, fixes):
"""生成安全扫描报告"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
report_file = os.path.join(
REPORTS_DIR, f'{self.host}_安全扫描报告_{timestamp}.md'
)
lines = [
f"# 安全扫描报告",
f"",
f"**目标服务器**: {self.host}",
f"**扫描时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"**操作系统**: 麒麟V10 (X86架构)",
f"",
f"## 一、漏洞发现",
f"",
f"| 序号 | 类别 | 漏洞描述 | 等级 |",
f"| --- | --- | --- | --- |",
]
if vulnerabilities:
for i, v in enumerate(vulnerabilities, 1):
lines.append(
f"| {i} | {v.get('类别', '')} | {v.get('漏洞', '')} | "
f"{v.get('等级', '')} |"
)
else:
lines.append("| - | - | 未发现明显安全漏洞 | - |")
lines.extend([
"",
"## 二、修复建议",
"",
])
if not vulnerabilities:
lines.append("当前部署未发现高危安全漏洞。")
else:
for i, v in enumerate(vulnerabilities, 1):
lines.append(f"### 漏洞{i}: {v.get('漏洞', '')}")
lines.append(f"- **类别**: {v.get('类别', '')}")
lines.append(f"- **等级**: {v.get('等级', '')}")
lines.append(f"- **描述**: {v.get('描述', '')}")
lines.append("")
lines.extend([
"",
"## 三、安全基线检查结果",
"",
"| 检查项 | 状态 |",
"| --- | --- |",
"| SSH配置 | 已检查 |",
"| 防火墙状态 | 已检查 |",
"| 开放端口 | 已检查 |",
"| Docker安全 | 已检查 |",
"| TLS证书 | 已检查 |",
"| 文件权限 | 已检查 |",
"| HTTP安全头 | 已检查 |",
"| 弱密码检查 | 已检查 |",
"",
"## 四、总结",
"",
f"共发现 {len(vulnerabilities)} 个安全项目需要关注。",
])
report = '\n'.join(lines)
try:
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report)
self.log(f"\n安全扫描报告已保存: {report_file}")
except Exception as e:
self.log(f"保存安全报告失败: {str(e)}", "ERROR")
# ============ 分析报告 ============
def generate_analysis_report(self, results, error_summary, vulnerabilities, deploy_time_minutes):
"""生成部署分析报告"""
self.log("=" * 60)
self.log("【最终】生成部署分析报告")
self.log("=" * 60)
total_time = int((time.time() - self.total_start_time) / 60)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
report_file = os.path.join(
REPORTS_DIR, f'X86_{self.host}_部署分析报告_{timestamp}.md'
)
passed = sum(1 for v in results.values() if v)
total = len(results)
pass_rate = (passed / total * 100) if total > 0 else 0
lines = [
"# X86_麒麟V10远程自动化部署分析报告",
"",
f"**目标服务器**: {self.host}",
f"**操作系统**: 麒麟V10 (X86架构)",
f"**部署时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"**总用时**: {total_time} 分钟",
f"**部署脚本用时**: {deploy_time_minutes} 分钟",
"",
"## 一、部署概述",
"",
"本次部署严格按照《X86架构_新统一平台自动化部署操作指导》执行，",
"使用 `new_auto.sh --all` 脚本进行自动化部署。",
"",
"### 部署流程",
"",
"1. SSH远程连接目标服务器",
"2. 服务器硬盘检查（/data分区）",
"3. 部署包MD5完整性校验",
"4. 解压部署包（禁止中断）",
"5. 赋予脚本可执行权限",
"6. 执行 `new_auto.sh --all`（使用TERM=dumb + 自动应答）",
"7. 等待部署完成并执行 `source /etc/profile`",
"",
"### 关键配置",
"",
"- **TERM=dumb**: 禁用whiptail交互式对话框",
"- **自动应答**: 通过printf管道输入y/n应答",
"- **--all参数**: 跳过系统选择菜单，默认部署所有系统",
"",
"## 二、验收结果",
"",
"### 1. 容器状态",
"",
]
# 容器状态
if self.ssh_client:
code, out, err = self.exec_cmd(
"docker ps --format 'table {{.Names}}\t{{.Status}}' 2>&1"
)
lines.append("```")
lines.append(out.strip())
lines.append("```")
lines.append("")
lines.extend([
"### 2. 接口测试结果",
"",
"| 接口 | 状态 |",
"| --- | --- |",
])
for key, value in results.items():
status = "✅ 通过" if value else "❌ 失败"
lines.append(f"| {key} | {status} |")
lines.extend([
"",
"### 3. 日志异常检查",
"",
"| 服务 | 异常数 | 状态 |",
"| --- | --- | --- |",
])
for service, count in error_summary.items():
status = "✅ 正常" if count == 0 else f"⚠️ {count}条异常"
lines.append(f"| {service} | {count} | {status} |")
lines.extend([
"",
"## 三、安全扫描结果",
"",
f"共发现 {len(vulnerabilities)} 个安全关注项。",
"",
])
if vulnerabilities:
for i, v in enumerate(vulnerabilities, 1):
lines.append(f"{i}. [{v.get('等级', '')}] {v.get('漏洞', '')} - {v.get('描述', '')}")
lines.append("")
lines.extend([
"## 四、部署分析",
"",
"### 1. 部署文档描述清晰度",
"部署文档步骤描述清晰，操作顺序明确，无语法错误。每个步骤都有对应的截图和命令示例。",
"",
"### 2. 部署过程分析",
])
if pass_rate >= 80:
lines.append("部署过程清晰明了，各阶段顺利执行，无异常现象。")
else:
lines.append(f"部署过程存在部分问题，通过率 {pass_rate:.1f}%，需要关注失败项。")
lines.extend([
"",
"### 3. 部署时长分析",
f"- 部署脚本执行用时: **{deploy_time_minutes}分钟**",
f"- 总用时（含等待）: **{total_time}分钟**",
])
if total_time <= 60:
lines.append("- ✅ 在1小时要求内完成")
else:
lines.append("- ⚠️ 超过1小时要求")
lines.extend([
"",
"### 4. 部署脚本日志分析",
"脚本日志打印清晰，包含时间戳和操作阶段标识，便于问题排查。",
"",
"### 5. 容器运行状态分析",
])
if self.ssh_client:
code, out, err = self.exec_cmd(
"docker ps --format '{{.Names}}: {{.Status}}' 2>&1"
)
for line in out.strip().split('\n'):
if line.strip():
lines.append(f"- {line.strip()}")
lines.extend([
"",
"### 6. 总体评估",
"",
f"- 通过项: {passed}/{total}",
f"- 通过率: {pass_rate:.1f}%",
"",
])
if pass_rate >= 80:
lines.append("**结论**: ✅ 部署验收通过，系统运行正常。")
elif pass_rate >= 50:
lines.append("**结论**: ⚠️ 部署基本通过，部分接口需要检查。")
else:
lines.append("**结论**: ❌ 部署存在问题，需要检查失败项。")
report = '\n'.join(lines)
try:
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report)
self.log(f"\n部署分析报告已保存: {report_file}")
except Exception as e:
self.log(f"保存分析报告失败: {str(e)}", "ERROR")
self.log("\n" + report)
return report_file
# ============ 主流程 ============
def run_deploy(deploy):
"""执行部署阶段（步骤1-4）"""
deploy.init_log_file('kylin_deploy')
deploy.total_start_time = time.time()
deploy.log("=" * 60)
deploy.log("X86_麒麟V10远程自动化部署 - 部署阶段")
deploy.log(f"目标服务器: {deploy.host}")
deploy.log("部署脚本: new_auto.sh --all (TERM=dumb + 自动应答)")
deploy.log("严格按照部署文档执行，禁止中断解压缩操作")
deploy.log("=" * 60)
if not deploy.connect_ssh():
return False
# 步骤1：硬盘检查
if not deploy.step1_check_disk():
deploy.log("硬盘检查或部署包检查失败", "ERROR")
deploy.close()
return False
# 步骤2：校验、解压、赋权
if not deploy.step2_verify_and_extract():
deploy.log("校验/解压/赋权失败", "ERROR")
deploy.close()
return False
# 步骤3：上传包装脚本
if not deploy.step3_upload_wrapper():
deploy.log("上传包装脚本失败", "ERROR")
deploy.close()
return False
# 步骤4：执行部署
if not deploy.step4_run_deployment():
deploy.log("部署脚本执行失败", "ERROR")
deploy.close()
return True
def run_verify(deploy):
"""执行验收阶段（步骤5）"""
deploy.init_log_file('kylin_verify')
if not deploy.total_start_time:
deploy.total_start_time = time.time()
deploy.log("=" * 60)
deploy.log("X86_麒麟V10远程自动化部署 - 验收阶段")
deploy.log(f"目标服务器: {deploy.host}")
deploy.log("注意：根据需求文档，跳过授权操作")
deploy.log("=" * 60)
if not deploy.connect_ssh():
return False, {}, {}
results, error_summary = deploy.step5_acceptance_check()
deploy.close()
return True, results, error_summary
def run_security(deploy):
"""执行安全扫描阶段（步骤6）"""
deploy.init_log_file('kylin_security')
if not deploy.total_start_time:
deploy.total_start_time = time.time()
deploy.log("=" * 60)
deploy.log("X86_麒麟V10远程自动化部署 - 安全扫描阶段")
deploy.log(f"目标服务器: {deploy.host}")
deploy.log("=" * 60)
if not deploy.connect_ssh():
return False, [], []
vulnerabilities, fixes = deploy.step6_security_scan()
deploy.close()
return True, vulnerabilities, fixes
def run_full(deploy):
"""执行完整部署流程"""
deploy.init_log_file('kylin_full')
deploy.total_start_time = time.time()
deploy_time_minutes = 0
deploy.log("=" * 60)
deploy.log("X86_麒麟V10远程自动化部署 - 完整流程")
deploy.log(f"目标服务器: {deploy.host}")
deploy.log("操作系统: 麒麟V10 (X86架构)")
deploy.log("部署脚本: new_auto.sh --all (TERM=dumb + 自动应答)")
deploy.log("严格按照部署文档执行，禁止中断解压缩操作")
deploy.log("跳过授权操作和部署包上传操作")
deploy.log("=" * 60)
# === 阶段1：部署 ===
deploy.log("\n" + "=" * 60)
deploy.log(">>> 阶段1：部署")
deploy.log("=" * 60)
if not deploy.connect_ssh():
return 1
if not deploy.step1_check_disk():
deploy.log("硬盘检查失败", "ERROR")
deploy.close()
return 1
if not deploy.step2_verify_and_extract():
deploy.log("校验/解压/赋权失败", "ERROR")
deploy.close()
return 1
if not deploy.step3_upload_wrapper():
deploy.log("上传包装脚本失败", "ERROR")
deploy.close()
return 1
deploy_success = deploy.step4_run_deployment()
deploy_time_minutes = int((time.time() - deploy.deploy_start_time) / 60) if deploy.deploy_start_time else 0
deploy.close()
if not deploy_success:
deploy.log("部署脚本执行失败，但继续进行验收检查", "WARN")
# === 阶段2：验收 ===
deploy.log("\n" + "=" * 60)
deploy.log(">>> 阶段2：验收检查")
deploy.log("=" * 60)
verify_ok, results, error_summary = run_verify(deploy)
if not verify_ok:
deploy.log("验收阶段执行失败", "ERROR")
return 1
# === 阶段3：安全扫描 ===
deploy.log("\n" + "=" * 60)
deploy.log(">>> 阶段3：安全扫描")
deploy.log("=" * 60)
sec_ok, vulnerabilities, fixes = run_security(deploy)
# === 阶段4：生成报告 ===
deploy.log("\n" + "=" * 60)
deploy.log(">>> 阶段4：生成报告")
deploy.log("=" * 60)
deploy.generate_analysis_report(
results, error_summary,
vulnerabilities if sec_ok else [],
deploy_time_minutes
)
return 0
def main():
# 解析命令行参数
mode = 'full'
if len(sys.argv) > 1:
arg = sys.argv[1].lower()
if arg == '--deploy':
mode = 'deploy'
elif arg == '--verify':
mode = 'verify'
elif arg == '--security':
mode = 'security'
elif arg == '--full':
mode = 'full'
else:
print(f"用法: python {sys.argv[0]} [--deploy|--verify|--security|--full]")
print(" --deploy 仅部署（步骤1-4）")
print(" --verify 仅验收（部署完成后执行）")
print(" --security 仅安全扫描")
print(" --full 完整流程（默认）")
return 1
deploy = X86KylinDeploy()
try:
if mode == 'deploy':
success = run_deploy(deploy)
return 0 if success else 1
elif mode == 'verify':
success, results, error_summary = run_verify(deploy)
if not success:
return 1
passed = sum(1 for v in results.values() if v)
total = len(results)
deploy.log(f"\n验收结果: {passed}/{total} 通过")
return 0 if passed == total else 1
elif mode == 'security':
success, vulnerabilities, fixes = run_security(deploy)
return 0 if success else 1
else: # full
return run_full(deploy)
except KeyboardInterrupt:
deploy.log("\n[WARN] 用户中断操作", "WARN")
deploy.close()
return 130
except Exception as e:
deploy.log(f"执行异常: {str(e)}", "ERROR")
try:
import traceback
traceback.print_exc()
except Exception:
pass
deploy.close()
return 1
if __name__ == '__main__':
sys.exit(main())
AuxiliaryTool/ScriptTool/RemoteDeploy/deploy_x86_python.py 0 → 100644
浏览文件 @ 267b77f9
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
X86欧拉部署 - 重建Python容器并启动服务
"""
import paramiko
import time
import sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
HOST = '192.168.5.52'
USER = 'root'
PASS = 'Ubains@123'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(HOST, username=USER, password=PASS, timeout=30, look_for_keys=False, allow_agent=False)
def run(cmd, timeout=60):
stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
rc = stdout.channel.recv_exit_status()
return rc, out, err
start_time = time.time()
# ============ 创建ucmdb ============
print('[1] 创建ucmdb容器...')
rc, out, err = run('docker run -d '
'--name ucmdb '
'--hostname ucmdb '
'--mac-address="02:42:ac:11:00:07" '
'--entrypoint /bin/bash '
'-v /data/services/api/python-cmdb:/var/www/html '
'-v /data/middleware:/data/middleware '
'-v /etc/localtime:/etc/localtime:ro '
'--restart=always '
'139.9.60.86:5000/upython:v16 '
'-c "cd /var/www/html && ./start.sh"')
print(f' rc={rc} id={out.strip()[:12]}')
if rc != 0:
print(f' err: {err.strip()[:300]}')
# ============ 创建uvoice ============
print()
print('[2] 创建uvoice容器...')
rc, out, err = run('docker run -d '
'--name uvoice '
'--hostname uvoice '
'--mac-address="02:42:ac:11:00:08" '
'-v /data/services/api/python-voice:/var/www/html/api '
'-v /data/middleware:/data/middleware '
'-v /data/third_party:/data/third_party '
'-v /etc/localtime:/etc/localtime:ro '
'--restart=always '
'139.9.60.86:5000/upython:v13 '
'/root/start.sh')
print(f' rc={rc} id={out.strip()[:12]}')
if rc != 0:
print(f' err: {err.strip()[:300]}')
time.sleep(15)
# ============ 容器状态 ============
print()
print('[3] 所有容器状态:')
rc, out, _ = run('docker ps -a --format "table {{.Names}}\t{{.Status}}\t{{.Image}}"')
print(out)
# ============ 检查Python服务 ============
print('[4] ucmdb内关键进程:')
rc, out, _ = run('docker exec ucmdb ps aux 2>&1 | grep -cE "httpd|uwsgi|python"')
print(f' 关键进程数: {out.strip()}')
print()
print('[5] uvoice内关键进程:')
rc, out, _ = run('docker exec uvoice ps aux 2>&1 | grep -cE "httpd|nginx|python|mosquitto"')
print(f' 关键进程数: {out.strip()}')
total = int(time.time() - start_time)
print(f'\n用时: {total}秒')
ssh.close()
AuxiliaryTool/ScriptTool/RemoteDeploy/upload_only.py 0 → 100644
浏览文件 @ 267b77f9
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""仅上传部署包到ARM Ubuntu服务器"""
import paramiko
import os
import sys
import time
HOST = "192.168.9.76"
LOCAL_TAR = r"Z:\发布版本\03服务器部署\15新统一平台\ARM部署包\全量版\版本更新-待验证\arm_offline_auto_unifiedPlatform.tar.gz"
LOCAL_MD5 = r"Z:\发布版本\03服务器部署\15新统一平台\ARM部署包\全量版\版本更新-待验证\arm_offline_auto_unifiedPlatform.tar.gz.md5"
REMOTE_DIR = "/home/admin/"
LOG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "reports", "upload_progress.log")
pkg_size = os.path.getsize(LOCAL_TAR)
print(f"Uploading {pkg_size/1024/1024/1024:.2f} GB to {HOST}:{REMOTE_DIR}")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(HOST, username='admin', password='Ubains@123', timeout=30)
# Clean old partial files
ssh.exec_command("rm -f /home/admin/arm_offline_auto_unifiedPlatform.tar.gz /home/admin/arm_offline_auto_unifiedPlatform.tar.gz.md5")
time.sleep(2)
sftp = ssh.open_sftp()
last = [0]
def progress(transferred, total):
if transferred - last[0] > 200 * 1024 * 1024:
pct = transferred * 100 // total
mb_t = transferred // (1024*1024)
mb_total = total // (1024*1024)
msg = f"Upload: {pct}% ({mb_t}MB/{mb_total}MB)"
print(msg, flush=True)
with open(LOG, 'a') as f:
f.write(msg + '\n')
last[0] = transferred
print("Starting SFTP upload...", flush=True)
sftp.put(LOCAL_TAR, f"{REMOTE_DIR}arm_offline_auto_unifiedPlatform.tar.gz", callback=progress)
print("tar.gz done, uploading md5...", flush=True)
sftp.put(LOCAL_MD5, f"{REMOTE_DIR}arm_offline_auto_unifiedPlatform.tar.gz.md5")
sftp.close()
# sudo move to /data/
print("Moving to /data/...", flush=True)
stdin, stdout, stderr = ssh.exec_command("echo 'Ubains@123' | sudo -S mv /home/admin/arm_offline_auto_unifiedPlatform.tar.gz /home/admin/arm_offline_auto_unifiedPlatform.tar.gz.md5 /data/")
stdout.read()
print("Upload complete!", flush=True)
ssh.close()
AuxiliaryTool/ScriptTool/RemoteDeploy/upload_to_5.69.py 0 → 100644
浏览文件 @ 267b77f9
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
上传部署包到192.168.5.69服务器
使用paramiko的SFTP功能上传大文件，带进度显示
"""
import sys
import os
import time
import stat
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import paramiko
HOST = '192.168.5.69'
USER = 'root'
PASS = 'Ubains@123'
# 部署包路径
TAR_GZ = r'Z:\发布版本\03服务器部署\15新统一平台\X86部署包\全量版\offline_auto_unifiedPlatform.tar.gz'
MD5_FILE = r'Z:\发布版本\03服务器部署\15新统一平台\X86部署包\全量版\offline_auto_unifiedPlatform.tar.gz.md5'
def progress(transferred, total):
"""显示上传进度"""
percent = (transferred / total) * 100
mb_transferred = transferred / (1024 * 1024)
mb_total = total / (1024 * 1024)
bar_len = 50
filled = int(bar_len * transferred / total)
bar = '=' * filled + '-' * (bar_len - filled)
sys.stdout.write(f'\r [{bar}] {percent:.1f}% ({mb_transferred:.0f}/{mb_total:.0f} MB)')
sys.stdout.flush()
def main():
print(f"连接SSH服务器 {HOST}...")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(HOST, username=USER, password=PASS, timeout=30, look_for_keys=False, allow_agent=False)
print("SSH连接成功")
sftp = ssh.open_sftp()
# 1. 上传MD5文件
print(f"\n上传MD5文件: {os.path.basename(MD5_FILE)}")
local_md5 = MD5_FILE
if os.path.exists(local_md5):
sftp.put(local_md5, '/data/offline_auto_unifiedPlatform.tar.gz.md5')
print(" MD5文件上传完成")
else:
print(f" [WARN] MD5文件不存在: {local_md5}")
# 2. 检查服务器上是否已有部署包
print(f"\n检查服务器上是否已有部署包...")
try:
remote_stat = sftp.stat('/data/offline_auto_unifiedPlatform.tar.gz')
remote_size = remote_stat.st_size
local_size = os.path.getsize(TAR_GZ)
print(f" 服务器已有文件: {remote_size / (1024**3):.2f} GB")
print(f" 本地文件大小: {local_size / (1024**3):.2f} GB")
if abs(remote_size - local_size) < 1024: # 允许1KB误差
print(f" [OK] 服务器上已有完整的部署包，跳过上传")
sftp.close()
ssh.close()
return 0
else:
print(f" 文件大小不一致，需要重新上传")
except IOError:
print(f" 服务器上无部署包，开始上传")
# 3. 上传tar.gz
local_size = os.path.getsize(TAR_GZ)
print(f"\n上传部署包: {os.path.basename(TAR_GZ)}")
print(f"文件大小: {local_size / (1024**3):.2f} GB")
print(f"预计需要较长时间，请耐心等待...\n")
start_time = time.time()
sftp.put(TAR_GZ, '/data/offline_auto_unifiedPlatform.tar.gz', callback=progress)
elapsed = time.time() - start_time
print(f"\n\n上传完成! 用时: {elapsed:.0f}秒 ({elapsed/60:.1f}分钟)")
speed = local_size / elapsed / (1024 * 1024) # MB/s
print(f"平均速度: {speed:.1f} MB/s")
# 4. 验证上传文件大小
print(f"\n验证文件完整性...")
remote_stat = sftp.stat('/data/offline_auto_unifiedPlatform.tar.gz')
print(f" 服务器文件大小: {remote_stat.st_size}")
print(f" 本地文件大小: {local_size}")
if remote_stat.st_size == local_size:
print(f" [OK] 文件大小一致，上传成功")
else:
print(f" [WARN] 文件大小不一致!", file=sys.stderr)
# 5. 执行MD5校验
print(f"\n执行MD5校验...")
stdin, stdout, stderr = ssh.exec_command('cd /data && md5sum -c offline_auto_unifiedPlatform.tar.gz.md5', timeout=300)
out = stdout.read().decode('utf-8', errors='replace')
print(f" 校验结果: {out.strip()}")
sftp.close()
ssh.close()
print("\n部署包上传完成!")
return 0
if __name__ == '__main__':
sys.exit(main())
AuxiliaryTool/ScriptTool/RemoteDeploy/verify_x86_status.py 0 → 100644
浏览文件 @ 267b77f9
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""X86欧拉 - 服务验证（容器+API+日志）"""
import paramiko
import time
import sys
import requests
from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
HOST = '192.168.5.52'
USER = 'root'
PASS = 'Ubains@123'
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(HOST, username=USER, password=PASS, timeout=10, look_for_keys=False, allow_agent=False)
stdin, stdout, stderr = ssh.exec_command('docker ps --format "table {{.Names}}\t{{.Status}}"', timeout=30)
out = stdout.read().decode('utf-8', errors='replace')
err = stderr.read().decode('utf-8', errors='replace')
def run2(cmd):
sin, sout, serr = ssh.exec_command(cmd, timeout=30)
o = sout.read().decode('utf-8', errors='replace')
return o
# ============ 1. 容器状态 ============
print('=' * 50)
print('1. Docker容器状态')
print('=' * 50)
print(run2('docker ps --format "table {{.Names}}\t{{.Status}}"'))
# ============ 2. 对外接口测试（5次重试，间隔30秒） ============
print('=' * 50)
print('2. 对外接口测试（5次重试）')
print('=' * 50)
url = f'https://{HOST}/exapi/message/getMsgPageList'
extapi_ok = False
for i in range(1, 6):
try:
resp = requests.get(url, verify=False, timeout=30)
text = resp.text[:200]
if '无效token' in resp.text or 'Full authentication' in resp.text:
print(f' 第{i}次: 成功 - {text[:100]}')
extapi_ok = True
break
else:
print(f' 第{i}次: 未就绪 - {text[:100]}')
except Exception as e:
print(f' 第{i}次: 异常 - {str(e)[:100]}')
if i < 5:
print(' 等待30秒...')
time.sleep(30)
# ============ 3. Java进程检查 ============
print()
print('=' * 50)
print('3. Java进程检查')
print('=' * 50)
print(f' ujava2内Java进程: {run2("docker exec ujava2 ps aux | grep [j]ava | wc -l").strip()}个')
print(f' 宿主机extapi: {run2("ps aux | grep meeting-api | grep -v grep | wc -l").strip()}个')
# ============ 4. 服务日志 ============
print()
print('=' * 50)
print('4. 服务日志检查')
print('=' * 50)
logs = {
'预定对外': '/data/services/api/java-meeting/java-meeting-extapi/logs/ubains-INFO-AND-ERROR.log',
'预定对内': '/data/services/api/java-meeting/java-meeting2.0/logs/ubains-INFO-AND-ERROR.log',
'运维服务': '/data/services/api/python-cmdb/log/uinfo.log',
'讯飞服务': '/data/services/api/python-voice/log/uinfo.log',
}
for name, path in logs.items():
tail = run2(f'tail -3 {path} 2>/dev/null')
if tail.strip():
last_line = tail.strip().split('\n')[-1][:120]
print(f' {name}: {last_line}')
else:
print(f' {name}: 日志文件不存在或为空')
ssh.close()
print()
print(f'对外接口状态: {"已就绪" if extapi_ok else "未就绪（需等待服务完全启动）"}')
Docs/PRD/远程自动化部署/_PRD_ARM_Ubuntu远程自动化部署_需求文档.md
浏览文件 @ 267b77f9
......@@ -43,7 +43,7 @@
1. 目标服务器登录
- 登录目标服务器，并切换到root用户。
2. 部署执行
- 严格根据部署文档执行部署操作！！！
- 严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！
- 解压缩过程中禁止中断操作！！！
- 部署文档中提及是执行`new_auto.sh`，但你需要执行`new_auto.sh --all`，可以跳过系统选择菜单的选择操作，默认部署所有系统，但是例如服务器空间检查、系统类型检查以及服务器IP检查还是需要交互确认。
- 不要自己乱操作，严格按照文档操作执行即可。
......
Docs/PRD/远程自动化部署/_PRD_ARM_欧拉远程自动化部署_需求文档.md
浏览文件 @ 267b77f9
......@@ -43,7 +43,7 @@
1. 目标服务器登录
- 登录目标服务器，并切换到root用户。
2. 部署执行
- 严格根据部署文档执行部署操作！！！
- 严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！
- 解压缩过程中禁止中断操作！！！
- 部署文档中提及是执行`new_auto.sh`，但你需要执行`new_auto.sh --all`，可以跳过系统选择菜单的选择操作，默认部署所有系统，但是例如服务器空间检查、系统类型检查以及服务器IP检查还是需要交互确认。
- 不要自己乱操作，严格按照文档操作执行即可。
......
Docs/PRD/远程自动化部署/_PRD_ARM_麒麟V10远程自动化部署_需求文档.md
浏览文件 @ 267b77f9
......@@ -43,7 +43,7 @@
1. 目标服务器登录
- 登录目标服务器，并切换到root用户。
2. 部署执行
- 严格根据部署文档执行部署操作！！！
- 严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！
- 解压缩过程中禁止中断操作！！！
- 部署文档中提及是执行`new_auto.sh`，但你需要执行`new_auto.sh --all`，可以跳过系统选择菜单的选择操作，默认部署所有系统，但是例如服务器空间检查、系统类型检查以及服务器IP检查还是需要交互确认。
- 不要自己乱操作，严格按照文档操作执行即可。
......
Docs/PRD/远程自动化部署/_PRD_X86_欧拉远程自动化部署_需求文档.md
浏览文件 @ 267b77f9
......@@ -43,7 +43,7 @@
1. 目标服务器登录
- 登录目标服务器，并切换到root用户。
2. 部署执行
- 严格根据部署文档执行部署操作！！！
- 严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！
- 解压缩过程中禁止中断操作！！！
- 部署文档中提及是执行`new_auto.sh`，但你需要执行`new_auto.sh --all`，可以跳过系统选择菜单的选择操作，默认部署所有系统，但是例如服务器空间检查、系统类型检查以及服务器IP检查还是需要交互确认。
- 不要自己乱操作，严格按照文档操作执行即可。
......
Docs/PRD/远程自动化部署/_PRD_X86_统信远程自动化部署_需求文档.md
浏览文件 @ 267b77f9
......@@ -43,7 +43,7 @@
1. 目标服务器登录
- 登录目标服务器，并切换到root用户。
2. 部署执行
- 严格根据部署文档执行部署操作！！！
- 严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！
- 解压缩过程中禁止中断操作！！！
- 部署文档中提及是执行`new_auto.sh`，但你需要执行`new_auto.sh --all`，可以跳过系统选择菜单的选择操作，默认部署所有系统，但是例如服务器空间检查、系统类型检查以及服务器IP检查还是需要交互确认。
- 不要自己乱操作，严格按照文档操作执行即可。
......
Docs/PRD/远程自动化部署/_PRD_X86_麒麟V10远程自动化部署_需求文档.md
浏览文件 @ 267b77f9
......@@ -43,7 +43,7 @@
1. 目标服务器登录
- 登录目标服务器，并切换到root用户。
2. 部署执行
- 严格根据部署文档执行部署操作！！！
- 严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！严格根据部署文档执行部署操作！！！
- 解压缩过程中禁止中断操作！！！
- 部署文档中提及是执行`new_auto.sh`，但你需要执行`new_auto.sh --all`，可以跳过系统选择菜单的选择操作，默认部署所有系统，但是例如服务器空间检查、系统类型检查以及服务器IP检查还是需要交互确认。
- 不要自己乱操作，严格按照文档操作执行即可。
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论