如何用脚本分析代理访问记录？

本文目录导读：

1. 使用说明
2. 注意事项

我可以为您提供一个基础的代理访问记录分析脚本方案,以下是一个Python脚本示例，用于分析常见的代理日志格式：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
代理访问日志分析脚本
支持的日志格式：Squid, Nginx反向代理, Apache等常见格式
"""
import re
import sys
import os
from datetime import datetime
from collections import Counter, defaultdict
import argparse
class ProxyLogAnalyzer:
    """代理日志分析器"""
    def __init__(self, log_file):
        self.log_file = log_file
        self.total_requests = 0
        self.successful_requests = 0
        self.failed_requests = 0
        self.status_codes = Counter()
        self.source_ips = Counter()
        self.requested_urls = Counter()
        self.user_agents = Counter()
        self.hourly_requests = defaultdict(int)
        self.error_logs = []
    def parse_line(self, line):
        """解析单行日志（支持多种格式）"""
        # 尝试匹配标准组合日志格式 (Nginx/Apache)
        pattern = r'(\d+\.\d+\.\d+\.\d+)\s+.*?\s+.*?\s+\[(.*?)\]\s+"(.*?)"\s+(\d+)\s+(\d+)\s+"(.*?)"\s+"(.*?)"'
        match = re.match(pattern, line)
        if not match:
            # 尝试匹配Squid日志格式
            pattern = r'(\d+\.\d+\.\d+\.\d+)\s+.*?\s+.*?\s+(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2})\s+(\w+)\s+(.*?)\s+(.*?)\s+(\d+)\s+(\d+)\s+(\d+)'
            match = re.match(pattern, line)
        if not match:
            return None
        try:
            if len(match.groups()) == 7:  # Nginx格式
                ip = match.group(1)
                timestamp = match.group(2)
                request = match.group(3)
                status = int(match.group(4))
                size = int(match.group(5))
                referer = match.group(6)
                user_agent = match.group(7)
            else:  # Squid格式
                ip = match.group(1)
                timestamp = match.group(2)
                method = match.group(3)
                url = match.group(4)
                user_agent = match.group(5)
                status = int(match.group(6))
                size = int(match.group(7))
                request = f"{method} {url}"
            # 解析时间
            dt = datetime.strptime(timestamp, '%d/%b/%Y:%H:%M:%S %z')
            return {
                'ip': ip,
                'timestamp': dt,
                'request': request,
                'status': status,
                'size': size,
                'user_agent': user_agent[:100] if user_agent else 'unknown'
            }
        except (ValueError, IndexError) as e:
            self.error_logs.append((line, str(e)))
            return None
    def analyze(self):
        """分析日志文件"""
        try:
            with open(self.log_file, 'r', encoding='utf-8', errors='ignore') as f:
                for line_num, line in enumerate(f, 1):
                    line = line.strip()
                    if not line:
                        continue
                    parsed = self.parse_line(line)
                    if not parsed:
                        continue
                    self.total_requests += 1
                    # 统计状态码
                    self.status_codes[parsed['status']] += 1
                    if parsed['status'] < 400:
                        self.successful_requests += 1
                    else:
                        self.failed_requests += 1
                    # 统计IP
                    self.source_ips[parsed['ip']] += 1
                    # 统计URL
                    self.requested_urls[parsed['request']] += 1
                    # 统计User-Agent
                    ua = parsed['user_agent']
                    self.user_agents[ua] += 1
                    # 按小时统计
                    hour = parsed['timestamp'].hour
                    self.hourly_requests[hour] += 1
        except FileNotFoundError:
            print(f"错误：文件 {self.log_file} 不存在")
            sys.exit(1)
        except PermissionError:
            print(f"错误：没有权限读取文件 {self.log_file}")
            sys.exit(1)
        except Exception as e:
            print(f"读取文件时出错: {e}")
            sys.exit(1)
    def generate_report(self):
        """生成分析报告"""
        if self.total_requests == 0:
            print("未找到有效的日志记录")
            return
        print("\n" + "="*60)
        print("代理访问日志分析报告")
        print("="*60)
        # 基本信息
        print(f"\n📊 基本统计")
        print(f"  - 总请求数: {self.total_requests}")
        print(f"  - 成功请求(2xx/3xx): {self.successful_requests} ({self.successful_requests/self.total_requests*100:.1f}%)")
        print(f"  - 失败请求(4xx/5xx): {self.failed_requests} ({self.failed_requests/self.total_requests*100:.1f}%)")
        # 状态码分布
        print(f"\n📈 状态码分布")
        for code, count in self.status_codes.most_common():
            print(f"  {code}: {count}次 ({count/self.total_requests*100:.1f}%)")
        # 来源IP Top 10
        print(f"\n🌐 来源IP Top 10")
        for ip, count in self.source_ips.most_common(10):
            print(f"  {ip}: {count}次")
        # 请求URL Top 10
        print(f"\n🔗 请求URL Top 10")
        for url, count in self.requested_urls.most_common(10):
            print(f"  {url[:80]}: {count}次")
        # 用户代理 Top 10
        print(f"\n🤖 用户代理 Top 10")
        for ua, count in self.user_agents.most_common(10):
            print(f"  {ua[:50]}: {count}次")
        # 小时分布
        print(f"\n⏰ 小时请求分布")
        for hour in range(24):
            count = self.hourly_requests.get(hour, 0)
            bar = '█' * min(count // 10, 50)
            print(f"  {hour:02d}:00 - {hour:02d}:59: {count}次 {bar}")
        # 解析错误
        if self.error_logs:
            print(f"\n⚠️ 解析错误记录 ({len(self.error_logs)}条)")
            for line, error in self.error_logs[:5]:
                print(f"  行内容: {line[:80]}...")
                print(f"  错误: {error}")
        print("\n" + "="*60)
def main():
    parser = argparse.ArgumentParser(
        description='代理访问日志分析工具',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  python log_analyzer.py access.log
  python log_analyzer.py /var/log/squid/access.log
  python log_analyzer.py proxy.log -f csv
        """
    )
    parser.add_argument('logfile', help='代理日志文件路径')
    parser.add_argument('-f', '--format', choices=['text', 'csv'], 
                       default='text', help='输出格式 (默认: text)')
    args = parser.parse_args()
    print(f"开始分析日志文件: {args.logfile}")
    print(f"日志格式: 自动检测")
    analyzer = ProxyLogAnalyzer(args.logfile)
    analyzer.analyze()
    analyzer.generate_report()
if __name__ == "__main__":
    main()

使用说明

基本使用

# 分析代理日志文件
python proxy_log_analyzer.py /var/log/squid/access.log
# 指定日志格式（目前自动检测）
python proxy_log_analyzer.py access.log -f text

支持的日志格式

• Nginx/Apache 标准组合日志格式
• Squid 代理日志格式
• 可扩展支持其他格式

分析功能

扩展用法

实时监控

# 使用tail -f 监控实时日志
import subprocess
log_analyzer = ProxyLogAnalyzer("access.log")
process = subprocess.Popen(['tail', '-f', 'access.log'], 
                          stdout=subprocess.PIPE)
while True:
    line = process.stdout.readline().decode()
    if line:
        log_analyzer.parse_line(line)

异常检测

# 检测异常IP行为
from collections import defaultdict
class AnomalyDetector:
    def detect_ddos(self, analyzer, threshold=100):
        """检测潜在的DDoS攻击"""
        for ip, count in analyzer.source_ips.items():
            if count > threshold:
                print(f"警告: IP {ip} 请求次数异常 ({count}次)")
    def detect_scanning(self, analyzer, threshold=10):
        """检测端口扫描"""
        # 分析404错误分布
        error_ips = defaultdict(list)
        for log in analyzer.error_logs:
            error_ips[log[0]].append(log[1])
        # 进一步分析...

注意事项

1. 日志格式：不同代理服务器日志格式不同，可能需要调整解析规则
2. 文件大小：处理大文件时考虑使用流式处理
3. 性能优化：对于GB级日志，建议使用分批处理
4. 隐私保护：注意IP等敏感信息的处理合规性

此脚本提供了一个基础框架,您可以根据实际需求添加：

• 地理IP查询
• 用户行为分析
• 安全威胁检测
• 实时告警功能
• 数据库存储分析结果