ChatGPT可疑活动监测实战指南：从告警到处理的完整解决方案

1次阅读

没有评论

共计 2262 个字符，预计需要花费 6 分钟才能阅读完成。

随着 ChatGPT 等 AI 应用的普及，API 滥用、敏感信息泄露等安全风险日益突出。本文将分享一套完整的监测与处理方案，帮助开发者构建可靠的 ChatGPT 可疑活动监测系统。

ChatGPT 应用常见的安全风险包括：

API 滥用 ：恶意用户通过高频调用消耗 API 配额
敏感信息泄露 ：用户意外提交包含密码、密钥等敏感内容
内容违规 ：生成违法、不当或侵权内容

现有监测方案往往存在告警信息不明确、处理流程复杂等问题，难以满足实际需求。

通过分析 API 请求日志，可以识别异常行为模式：

请求频率异常 ：短时间内大量请求
内容特征异常 ：包含敏感关键词或特殊字符
用户行为异常 ：非正常时间访问或异常操作序列

设计一个加权评分模型，关键指标包括：

请求频率（权重 30%）
内容敏感度（权重 40%）
用户历史行为（权重 20%）
时间异常度（权重 10%）

根据风险评分实施分级处理：

低风险（0-30 分）：记录日志
中风险（31-70 分）：发送告警
高风险（71-100 分）：阻断请求并通知管理员

import re
from datetime import datetime

def parse_log_entry(log_entry):
    """
    解析单条日志条目
    :param log_entry: 原始日志字符串
    :return: 解析后的字典对象
    """
    try:
        pattern = r'(?P<ip>\d+\.\d+\.\d+\.\d+) - - \[(?P<timestamp>.*?)\]"(?P<method>\w+) (?P<url>.*?) HTTP/\d\.\d"(?P<status>\d+) (?P<size>\d+)"(?P<referrer>.*?)""(?P<user_agent>.*?)"'
        match = re.match(pattern, log_entry)
        if match:
            return match.groupdict()
        return None
    except Exception as e:
        print(f"日志解析错误: {e}")
        return None

def calculate_risk_score(request_data):
    """
    计算请求风险评分
    :param request_data: 请求数据字典
    :return: 风险评分 (0-100)
    """
    try:
        score = 0

        # 频率因素
        request_count = get_request_count(request_data['ip'])
        if request_count > 100:  # 超过 100 次 / 分钟
            score += 30
        elif request_count > 50:
            score += 15

        # 内容因素
        sensitive_keywords = ['password', 'secret', 'key']
        content = request_data.get('content', '').lower()
        for kw in sensitive_keywords:
            if kw in content:
                score += 40
                break

        # 其他因素...

        return min(score, 100)
    except Exception as e:
        print(f"评分计算错误: {e}")
        return 0

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook/alert', methods=['POST'])
def handle_alert():
    """处理告警 webhook"""
    try:
        data = request.json
        risk_level = data.get('risk_level')

        if risk_level == 'high':
            # 高风险处理逻辑
            notify_admin(data)
            block_ip(data['ip'])
            return jsonify({'status': 'blocked'}), 200

        elif risk_level == 'medium':
            # 中风险处理逻辑
            notify_team(data)
            return jsonify({'status': 'alerted'}), 200

        return jsonify({'status': 'ignored'}), 200

    except Exception as e:
        return jsonify({'error': str(e)}), 500