影刀RPA 网络安全自动化：漏洞扫描与补丁管理

影刀RPA 网络安全自动化：漏洞扫描与补丁管理

什么情况用什么 → 怎么做 → 有什么坑 作者：林焱 | 飞行社出品

什么情况用什么

企业服务器越来越多，漏洞也越来越多。手动登录每台服务器跑yum update或apt upgrade，重复劳动到怀疑人生。

这套方案适合：

• 企业运维团队自动化漏洞扫描
• 自动化补丁管理（Windows/Linux）
• 安全合规审计（等保2.0、网络安全法）

核心工具： 影刀RPA + Nessus/OpenVAS（漏洞扫描） + Ansible（补丁管理） + 合规报告生成

怎么做

第一步：自动化漏洞扫描（OpenVAS）

拼多多店群自动化报活动上架！

import requests
import json
import time

def init_openvas_client(openvas_url, username, password):
"""
初始化OpenVAS客户端
OpenVAS是开源漏洞扫描工具
"""
try:
# 1. 登录获取token
login_url = f"{openvas_url}/login"
login_data = {
"username": username,
"password": password
}

response = requests.post(login_url, json=login_data, verify=False)
result = response.json()

if "token" in result:
print(f"OpenVAS登录成功")
return result["token"]
else:
print(f"⚠️ OpenVAS登录失败: {result}")
return None

except Exception as e:
print(f"⚠️ OpenVAS连接失败: {e}")
return None

def create_scan_task(openvas_url, token, target_ip, scan_config="Full and fast"):
"""
创建扫描任务
target_ip: 目标IP或IP段（如 "192.168.1.0/24"）
scan_config: 扫描配置（"Full and fast", "Full and very deep"等）
"""
headers = {"Authorization": f"Bearer {token}"}

# 1. 创建目标（Target）
target_data = {
"name": f"Auto Target {target_ip}",
"hosts": target_ip,
"port_range": "T:1-65535,U:1-65535", # TCP+UDP全端口
"alive_test": "Consider alive"
}

response = requests.post(f"{openvas_url}/targets", headers=headers, json=target_data, verify=False)
target_id = response.json()["id"]

print(f"目标已创建: {target_id}")

# 2. 创建扫描任务（Task）
task_data = {
"name": f"Auto Scan {target_ip}",
"config_id": get_config_id(openvas_url, headers, scan_config),
"target_id": target_id
}

response = requests.post(f"{openvas_url}/tasks", headers=headers, json=task_data, verify=False)
task_id = response.json()["id"]

print(f"扫描任务已创建: {task_id}")

return task_id

def start_scan_task(openvas_url, token, task_id):
"""启动扫描任务"""
headers = {"Authorization": f"Bearer {token}"}

response = requests.post(f"{openvas_url}/tasks/{task_id}/start", headers=headers, verify=False)

if response.status_code == 202:
print(f"扫描任务已启动: {task_id}")
return True
else:
print(f"⚠️ 启动扫描任务失败: {response.text}")
return False

def get_scan_report(openvas_url, token, task_id):
"""获取扫描报告"""
headers = {"Authorization": f"Bearer {token}"}

# 1. 等待扫描完成（轮询状态）
while True:
response = requests.get(f"{openvas_url}/tasks/{task_id}", headers=headers, verify=False)
task_status = response.json()["status"]

if task_status == "Done":
print(f"扫描已完成")
break
else:
print(f"扫描中… 状态: {task_status}")
time.sleep(30) # 每30秒查询一次

# 2. 获取报告ID
response = requests.get(f"{openvas_url}/tasks/{task_id}", headers=headers, verify=False)
report_id = response.json()["last_report"]["id"]

# 3. 下载报告（PDF格式）
report_url = f"{openvas_url}/reports/{report_id}/download"
response = requests.get(report_url, headers=headers, verify=False)

report_file = f"/tmp/openvas_report_{task_id}.pdf"
with open(report_file, "wb") as f:
f.write(response.content)

print(f"扫描报告已下载: {report_file}")

return report_file

# 完整流程示例
token = init_openvas_client("https://openvas.company.com", "admin", "password")

if token:
task_id = create_scan_task(token, "192.168.1.0/24")

if task_id:
start_scan_task(token, task_id)
report_file = get_scan_report(token, task_id)

第二步：自动化补丁管理（Linux – Ansible）

import subprocess
import json

def run_ansible_playbook(playbook_path, inventory_path, extra_vars=None):
"""
执行Ansible Playbook（自动化补丁管理）
playbook_path: playbook文件路径
inventory_path: inventory文件路径（定义要管理的服务器）
extra_vars: 额外变量（如 {"patch_type": "security"}）
"""
print(f"开始执行Ansible Playbook: {playbook_path}")

# 1. 构造ansible-playbook命令
cmd = [
"ansible-playbook",
"-i", inventory_path, # inventory文件
playbook_path
]

if extra_vars:
cmd.extend(["–extra-vars", json.dumps(extra_vars)])

try:
# 2. 执行playbook
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=3600 # 最多执行1小时
)

if result.returncode == 0:
print(f"✅ Playbook执行成功")
print(result.stdout)
return True
else:
print(f"⚠️ Playbook执行失败")
print(result.stderr)
return False

except Exception as e:
print(f"⚠️ 执行Playbook失败: {e}")
return False

# Ansible Playbook示例（update_packages.yml）
"""
—
– name: Update packages on Linux servers
hosts: all
become: yes # 使用sudo权限
tasks:
– name: Update yum cache (CentOS/RHEL)
yum:
update_cache: yes
when: ansible_os_family == "RedHat"

– name: Update apt cache (Ubuntu/Debian)
apt:
update_cache: yes
when: ansible_os_family == "Debian"

– name: Install security updates (CentOS/RHEL)
yum:
name: '*'
security: yes
state: latest
when: ansible_os_family == "RedHat"

– name: Install security updates (Ubuntu/Debian)
apt:
upgrade: 'security'
when: ansible_os_family == "Debian"

– name: Reboot if kernel updated
reboot:
reboot_timeout: 300
when: kernel_updated | default(false)
"""

# 使用示例
success = run_ansible_playbook(
playbook_path="/opt/ansible/update_packages.yml",
inventory_path="/opt/ansible/inventory/hosts",
extra_vars={"patch_type": "security"}
)

第三步：自动化补丁管理（Windows – PowerShell）

import subprocess

def run_windows_update(remote_host, username, password):
"""
远程执行Windows Update（使用PowerShell）
Windows补丁管理需要使用PSWindowsUpdate模块
"""
print(f"开始执行Windows Update: {remote_host}")

# 1. 构造PowerShell脚本（远程执行）
ps_script = """
# 安装PSWindowsUpdate模块（如果没装）
Install-Module -Name PSWindowsUpdate -Force -AllowClobber

# 检查可用更新
Get-WindowsUpdate

# 安装所有可用更新（自动重启）
Install-WindowsUpdate -AcceptAll -AutoReboot
"""

# 2. 将脚本保存到临时文件
import tempfile

with tempfile.NamedTemporaryFile(mode="w", suffix=".ps1", delete=False) as f:
f.write(ps_script)
ps_script_path = f.name

# 3. 使用WinRM远程执行（需要开启WinRM服务）
cmd = [
"python",
"-m", "pywinrm",
"–host", remote_host,
"–username", username,
"–password", password,
"–script", ps_script_path
]

try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600)

if result.returncode == 0:
print(f"✅ Windows Update执行成功")
print(result.stdout)
return True
else:
print(f"⚠️ Windows Update执行失败")
print(result.stderr)
return False

except Exception as e:
print(f"⚠️ 执行Windows Update失败: {e}")
return False

finally:
# 删除临时脚本文件
os.remove(ps_script_path)

# 或者使用Ansible管理Windows补丁（推荐）
"""
Ansible Playbook for Windows Update（win_update.yml）：

—
– name: Update Windows servers
hosts: windows
gather_facts: no
tasks:
– name: Install all updates
win_updates:
category_names:
– SecurityUpdates
– CriticalUpdates
reboot: yes
reboot_timeout: 1800
"""

第四步：生成安全合规报告（等保2.0）

def generate_security_compliance_report(scan_results, output_file):
"""
生成安全合规报告（等保2.0要求）
scan_results: 漏洞扫描结果（OpenVAS报告）
"""
import openpyxl
from openpyxl.styles import Font, Alignment

print(f"生成安全合规报告: {output_file}")

# 1. 创建Excel工作簿
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "安全合规报告"

# 2. 写入报告头
ws["A1"] = "安全合规报告（等保2.0）"
ws["A1"].font = Font(bold=True, size=14)
ws["A2"] = f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

# 3. 漏洞汇总
ws["A4"] = "漏洞汇总"
ws["A4"].font = Font(bold=True)

ws["A5"] = "高危漏洞数量"
ws["B5"] = scan_results.get("high_vulns", 0)

ws["A6"] = "中危漏洞数量"
ws["B6"] = scan_results.get("medium_vulns", 0)

ws["A7"] = "低危漏洞数量"
ws["B7"] = scan_results.get("low_vulns", 0)

# 4. 补丁状态
ws["A9"] = "补丁状态"
ws["A9"].font = Font(bold=True)

ws["A10"] = "已安装补丁数"
ws["B10"] = scan_results.get("patches_installed", 0)

ws["A11"] = "待安装补丁数"
ws["B11"] = scan_results.get("patches_pending", 0)

# 5. 合规检查项（等保2.0要求）
ws["A13"] = "等保2.0合规检查"
ws["A13"].font = Font(bold=True)

compliance_checks = [
("身份鉴别", "通过" if scan_results.get("auth_check") else "不通过"),
("访问控制", "通过" if scan_results.get("access_control") else "不通过"),
("安全审计", "通过" if scan_results.get("audit_check") else "不通过"),
("入侵防范", "通过" if scan_results.get("intrusion_prevention") else "不通过"),
]

for i, (check_item, status) in enumerate(compliance_checks):
ws[f"A{14+i}"] = check_item
ws[f"B{14+i}"] = status

# 不通过的项标红
if status == "不通过":
ws[f"B{14+i}"].font = Font(color="FF0000")

# 6. 保存报告
wb.save(output_file)

print(f"✅ 报告已生成: {output_file}")

return output_file

第五步：影刀RPA完整流程编排

【定时触发】每周一凌晨1:00
↓
【Python节点】init_openvas_client() → 登录OpenVAS
↓
【Python节点】create_scan_task() → 创建扫描任务
↓
【Python节点】start_scan_task() → 启动扫描
↓
【等待】扫描完成（OpenVAS回调或轮询状态）
↓
【Python节点】get_scan_report() → 获取扫描报告
↓
【条件判断】是否有高危漏洞？
├─ 是 → 【企微告警】发送高危漏洞告警
└─ 否 → 继续
↓
【Python节点】run_ansible_playbook() → 执行补丁管理
↓
【条件判断】补丁安装是否成功？
├─ 否 → 【企微告警】发送补丁安装失败通知
└─ 是 → 继续
↓
【Python节点】generate_security_compliance_report() → 生成合规报告
↓
【发送邮件】将报告发送给安全团队
↓
【生成报告】"安全运维周报.xlsx"
→ 包含：漏洞数量、补丁状态、合规检查结果

有什么坑

坑1：漏洞扫描误报（False Positive）

OpenVAS等工具可能把正常服务误判为漏洞，导致运维人员疲于应对。

解决方案： 人工复核 + 建立白名单：

# 建立漏洞白名单（已知误报）
VULN_WHITELIST = [
"TCP Timestamp", # 通常误报
"ICMP Timestamp", # 通常误报
# …
]

def filter_false_positives(vuln_list):
"""过滤误报漏洞"""
filtered = []

for vuln in vuln_list:
if vuln["name"] not in VULN_WHITELIST:
filtered.append(vuln)

return filtered

坑2：补丁安装后需要重启，业务中断

Windows补丁安装后经常需要重启，如果在业务高峰期重启，会导致服务中断。

解决方案： 使用维护窗口 + 负载均衡：

# 1. 只在维护窗口执行补丁安装（如凌晨2:00-4:00）
# 在Ansible Playbook中加条件判断
"""
—
– name: Install Windows updates
win_updates:
category_names:
– SecurityUpdates
reboot: no # 先不重启
when: ansible_date_time.hour >= 2 and ansible_date_time.hour <= 4

– name: Reboot if needed
win_reboot:
reboot_timeout: 300
when: ansible_date_time.hour >= 2 and ansible_date_time.hour <= 4
"""

# 2. 如果使用负载均衡，先摘除节点 → 补丁安装+重启 → 重新加入集群

坑3：等保2.0合规检查项很多，手动检查耗时

等保2.0有上百个检查项，手动检查需要几天时间。

TEMU店群矩阵自动化运营核价报活动

解决方案： 使用自动化合规检查工具（如Lynis、OpenSCAP）：

# 使用Lynis进行自动化安全审计（Linux）
wget https://downloads.cisofy.com/lynis/lynis.tar.gz
tar xvf lynis.tar.gz
cd lynis
./lynis audit system

# 使用OpenSCAP（支持等保2.0基线检查）
oscap xccdf eval –profile xccdf_org.ssgproject.content_profile_standard \\
/usr/share/xml/scap/ssg/content/ssg-rhel7-ds.xml

坑4：漏洞扫描对生产环境有影响

某些漏洞扫描会发送恶意payload，可能触发WAF/IPS告警，甚至导致服务崩溃。

解决方案： 使用被动扫描 + 在测试环境先扫描：

# 1. 先在测试环境扫描（镜像生产环境）
create_scan_task(openvas_url, token, "192.168.2.0/24") # 测试环境

# 2. 生产环境使用被动扫描（不发送payload，只分析流量）

# 使用Nessus的"Passive Vulnerability Scanner"（PVS）

总结

实际落地建议：

网络安全自动化能为运维/安全团队节省70%以上的日常漏洞管理时间，同时提高系统整体安全性。