Python调用Rapid7 API报错排查与代码优化咨询

报错排查 #

你遇到的JSONDecodeError不是因为添加了标签判断逻辑，而是API返回了非JSON格式的内容（比如空响应、5xx服务错误、4xx权限/限流错误）。运行几分钟后才崩溃，大概率是请求次数过多触发了Rapid7的API限流，或者网络波动导致响应中断。

排查步骤：

• 不要直接调用.json()，先检查响应状态码，判断请求是否成功
• 添加异常捕获，记录错误时的响应内容，方便定位问题
• 查看Rapid7 API文档，确认限流规则，添加请求间隔避免触发限制

疑问解答 #

1. 是否可以为resp2分配两个不同变量同时过滤tags和获取cursor？ #

完全可以，而且推荐拆分变量让代码更清晰。resp2是完整的响应数据，你可以把不同字段提取到独立变量中，互不影响：

# 从resp2中拆分数据和元数据
data_list = resp2.get('data', [])
metadata = resp2.get('metadata', {})
# 获取cursor，用get避免KeyError
next_cursor = metadata.get('cursor')

过滤tags是遍历data_list里的内容，获取cursor是读取metadata里的字段，两者完全独立，不需要给resp2分配多个变量，拆分只是让逻辑更直观。

2. 如何封装if tags["name"] == 'OSSWIN'这类逻辑？ #

把标签判断和漏洞统计的重复逻辑封装成函数，减少冗余代码：

def update_vuln_stats(data, tag_name, totals):
    """更新指定标签的漏洞统计"""
    totals[tag_name]['critical'] += data.get('critical_vulnerabilities', 0)
    totals[tag_name]['severe'] += data.get('severe_vulnerabilities', 0)
    totals[tag_name]['moderate'] += data.get('moderate_vulnerabilities', 0)
    print(f"{tag_name} 统计更新: 严重={totals[tag_name]['critical']}, 高危={totals[tag_name]['severe']}, 中等={totals[tag_name]['moderate']}")

调用时只需传入数据、标签名和统计字典：

for tag in data['tags']:
    tag_name = tag.get('name')
    if tag_name in ['OSSWIN', 'DESKTOP_SUPPORT']:
        update_vuln_stats(data, tag_name, totals)

代码优化建议 #

1. 添加异常处理与请求校验 #

避免因API异常直接崩溃，同时记录错误信息：

import requests
import json
import time

def _request():
    third_party_patching_filer = {
        "asset": "asset.agentKey IS NOT NULL", 
        "vulnerability": "vulnerability.categories NOT IN ['microsoft patch']"
    }
    headers = _headers()
    url_base = "https://us.api.insight.rapid7.com/vm/v4/integration/assets"
    
    # 初始化全局统计字典，避免循环内重复创建
    totals = {
        'OSSWIN': {'critical': 0, 'severe': 0, 'moderate': 0},
        'DESKTOP_SUPPORT': {'critical': 0, 'severe': 0, 'moderate': 0}
    }

    # 首次请求
    try:
        resp = requests.post(url_base, headers=headers, json=third_party_patching_filer, verify=False)
        resp.raise_for_status()  # 触发HTTP错误（如403/500）
        json_data = resp.json()
    except requests.exceptions.RequestException as e:
        print(f"首次请求失败: {str(e)}")
        if resp:
            print(f"错误响应内容: {resp.text}")
        return
    except json.JSONDecodeError as e:
        print(f"JSON解析失败: {str(e)}")
        print(f"响应内容: {resp.text}")
        return

    next_key = json_data["metadata"].get("cursor")
    has_next_cursor = next_key is not None

    while has_next_cursor:
        url = f"{url_base}?size=10&cursor={next_key}"
        try:
            resp2 = requests.post(url, headers=headers, json=third_party_patching_filer, verify=False)
            resp2.raise_for_status()
            resp2_data = resp2.json()
        except requests.exceptions.RequestException as e:
            print(f"分页请求失败: {str(e)}")
            if resp2:
                print(f"错误响应内容: {resp2.text}")
            time.sleep(5)  # 出错后等待5秒重试
            continue
        except json.JSONDecodeError as e:
            print(f"分页JSON解析失败: {str(e)}")
            print(f"响应内容: {resp2.text}")
            time.sleep(5)
            continue

        data_list = resp2_data.get('data', [])
        metadata = resp2_data.get('metadata', {})
        next_key = metadata.get('cursor')
        has_next_cursor = next_key is not None

        print(f"下一页cursor: {next_key}")

        # 处理当前页数据
        for data in data_list:
            tags = data.get('tags', [])
            for tag in tags:
                tag_name = tag.get('name')
                if tag_name in totals:
                    update_vuln_stats(data, tag_name, totals)

    # 打印最终统计结果
    print("\n=== 最终统计结果 ===")
    for tag_name, stats in totals.items():
        print(f"{tag_name}:")
        print(f"  严重漏洞: {stats['critical']}")
        print(f"  高危漏洞: {stats['severe']}")
        print(f"  中等漏洞: {stats['moderate']}")

2. 其他优化点 #

• 用.get(key, 默认值)代替直接字典索引，避免KeyError
• 移除循环内重复创建的空列表，直接累加数值统计
• 添加请求间隔（如time.sleep(1)），避免触发API限流
• 去掉无意义的else分支输出，减少冗余信息

内容的提问来源于stack exchange，提问作者FloydLee