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

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

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


什么情况用什么

在这里插入图片描述

企业服务器越来越多,漏洞也越来越多。手动登录每台服务器跑yum updateapt 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,只分析流量)
![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/4f0f4437fa6a42cda9bffd2ae9dc9c36.png#pic_center)
# 使用Nessus的"Passive Vulnerability Scanner"(PVS)

总结

功能 节省时间 附加价值
自动化漏洞扫描 每周省2小时 及时发现漏洞,降低风险
自动化补丁管理 每周省3小时 避免人为遗漏,提高安全性
合规报告生成 每次省1小时 满足等保2.0要求,避免合规风险
自动化加固 提高系统整体安全性

实际落地建议:

  1. 先在测试环境验证:避免扫描/补丁导致生产故障
  2. 使用维护窗口:在低峰期执行补丁安装
  3. 做好回滚准备:补丁安装失败时能快速回滚
  4. 定期审计:每月做一次完整的安全审计

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

在这里插入图片描述


在这里插入图片描述

© 版权声明

相关文章