跳到正文
DEV Community· Alessio·· 2 小時前AI 評分40

如何審核 AI 爬蟲流量並識別日誌中的偽冒機械人

How to Audit AI Crawler Traffic and Identify Spoofed Bots in Logs

AI 導讀

`aibot_ledger_crawler_log_reporter.py` 可分析伺服器存取日誌、識別 AI 爬蟲，並以正向確認反向 DNS（FCrDNS）核實 IP，協助偵測偽冒爬蟲。它支援 Combined 格式及 JSON 日誌、讀取 `.gz` 壓縮檔，並檢查 robots 排除規則、估算頻寬成本及產生伺服器層級的緩解範本。

正文

Unmanaged web scraping by artificial intelligence agents can consume server bandwidth, increase egress costs, and skew operational analytics. While legitimate crawlers identify themselves via their request headers, malicious scrapers often spoof those same headers to bypass rules or scrape protected assets under the guise of well-known automated systems.

Analyzing access logs manually across multiple servers is inefficient. You need a systematic way to parse server logs, match known crawler signatures, estimate hosting expenses tied to crawler bandwidth, and verify whether the IP addresses sending those headers actually belong to the claimed crawler networks.

The script aibot_ledger_crawler_log_reporter.py addresses this issue directly. It inspects web server access logs, identifies automated AI crawlers, performs Forward-Confirmed Reverse DNS lookups to catch spoofed clients, checks paths against your robots exclusion file, and generates server-level mitigation templates.

Install & Usage

The script relies entirely on standard modules included with Python, such as urllib.robotparser, socket, gzip, and json. You do not need to install any third-party libraries using pip.

To get started, save the source code into a file named aibot_ledger_crawler_log_reporter.py.

You can view the command-line help interface by running:

python aibot_ledger_crawler_log_reporter.py --help

Because the core processing logic is encapsulated in the AIBotLedger class, you can also import and execute the ledger directly within your own administrative scripts:

from aibot_ledger_crawler_log_reporter import AIBotLedger

# Initialize the ledger with custom bandwidth pricing and DNS verification
ledger = AIBotLedger(cost_per_gb=0.09, verify_dns=True)

# Ingest standard or compressed log files
ledger.ingest_file("/var/log/nginx/access.log")
ledger.ingest_file("/var/log/nginx/access.log.1.gz")

# Print the formatted terminal summary
print(ledger.generate_terminal_summary())

How It Works

The tool is organized around log parsing, pattern matching, DNS validation, and rules checking.

1. Dual Log Parsing and Compression Handling

The ingest_file function checks the file extension. If the target file ends in .gz, it transparently reads the stream via Python's built-in gzip module.

Log entries are processed by parse_log_line, which detects both standard Combined log entries (typical for Nginx and Apache) and structured JSON log lines (typical for Caddy):

def parse_log_line(line):
    """Parses standard Combined format or Caddy JSON logs."""
    line = line.strip()
    if not line:
        return None

    # Try Caddy JSON log format
    if line.startswith("{") and line.endswith("}"):
        try:
            data = json.loads(line)
            req = data.get("request", {})
            headers = req.get("headers", {})
            ua_list = headers.get("User-Agent", [""])
            ua = ua_list[0] if isinstance(ua_list, list) and ua_list else str(ua_list)

            ip = req.get("remote_ip") or req.get("client_ip") or ""
            if ":" in ip and not ip.startswith("["):
                # Handle host:port
                ip = ip.rsplit(":", 1)[0]

            return {
                "ip": ip,
                "time": data.get("ts", ""),
                "method": req.get("method", "GET"),
                "path": req.get("uri", "/"),
                "status": int(data.get("status", 200)),
                "bytes": int(data.get("size", 0) or 0),
                "ua": ua,
            }
        except Exception:
            pass

    # Try Combined / Nginx / Apache format
    m = COMBINED_LOG_REGEX.match(line)
    if m:
        d = m.groupdict()
        b = d["bytes"]
        bytes_val = int(b) if b.isdigit() else 0
        return {
            "ip": d["ip"],
            "time": d["time"],
            "method": d["method"],
            "path": d["path"],
            "status": int(d["status"]),
            "bytes": bytes_val,
            "ua": d["ua"],
        }
    return None

2. Signature Detection

The script maintains an internal collection of bot definitions named BOT_PROFILES. It matches user-agent values against regular expressions corresponding to entities such as GPTBot, ClaudeBot, Bytespider, CCBot, Google-Extended, PerplexityBot, Cohere-AI, and Diffbot. Each entry maps the signature to a classification category (such as Live Browsing, Dataset Mining, or Aggressive Scraper) and assigns legitimate root domains used for verification.

3. Spoof Detection with Forward-Confirmed Reverse DNS (FCrDNS)

Anyone can configure an HTTP client to send User-Agent: GPTBot. To prove authenticity, the script uses Forward-Confirmed Reverse DNS via verify_fcrdns:

def verify_fcrdns(ip, expected_domains, timeout=2.0):
    """Forward-Confirmed Reverse DNS (FCrDNS) check.

    Returns: 'VERIFIED', 'SPOOFED', 'UNVERIFIED', or 'SKIPPED'.
    """
    if not expected_domains:
        return "SKIPPED"
    if ip in _DNS_CACHE:
        return _DNS_CACHE[ip]

    # Don't resolve private or special addresses
    if ip.startswith(("127.", "10.", "192.168.", "172.16.", "172.17.", "172.18.", "172.19.", "172.20.", "172.21.", "172.22.", "172.23.", "172.24.", "172.25.", "172.26.", "172.27.", "172.28.", "172.29.", "172.30.", "172.31.")):
        result = "SKIPPED"
        _DNS_CACHE[ip] = result
        return result

    orig_timeout = socket.getdefaulttimeout()
    try:
        socket.setdefaulttimeout(timeout)
        try:
            hostname, _, _ = socket.gethostbyaddr(ip)
        except (socket.herror, socket.gaierror, socket.timeout, OSError):
            result = "SPOOFED"
            _DNS_CACHE[ip] = result
            return result

        hostname_lower = hostname.lower()
        domain_matched = any(hostname_lower.endswith(domain.lower()) for domain in expected_domains)
        if not domain_matched:
            result = "SPOOFED"
            _DNS_CACHE[ip] = result
            return result

        try:
            forward_ips = socket.gethostbyname_ex(hostname)[2]
            if ip in forward_ips:
                result = "VERIFIED"
            else:
                result = "SPOOFED"
        except (socket.gaierror, socket.timeout, OSError):
            result = "SPOOFED"

    finally:
        socket.setdefaulttimeout(orig_timeout)

    _DNS_CACHE[ip] = result
    return result

This verification process consists of three distinct checks:

  1. It queries the reverse DNS record for the IP address to obtain a hostname.
  2. It verifies that the returned hostname ends with one of the legitimate domain suffixes configured for that bot.
  3. It performs a forward lookup on that hostname to ensure that the originating IP address is present in the resolved address list.

If any check fails, the record is flagged as SPOOFED and recorded for auditing.

4. Cost Calculations and Defense Export

By accumulating byte counts associated with verified and unverified crawlers, AIBotLedger multiplies total gigabytes transferred by the cost_per_gb parameter to provide an estimate of hosting bandwidth charges.

It also includes standalone helper functions—such as generate_nginx_snippets and generate_robots_snippet—to output firewall maps, rate-limiting directives, and exclusion lines targeted at the specific crawler names encountered in your logs.

Limitations

While practical for audits, the current code has several built-in constraints:

  • Synchronous DNS Resolution: The verify_fcrdns function runs synchronously using Python's standard socket module. Although lookups are saved to an in-memory dictionary cache (_DNS_CACHE), processing very large uncompressed logs containing thousands of unique IP addresses will experience delays due to network timeouts.
  • Rigid Log Formats: The parser handles standard Apache/Nginx Combined log patterns and Caddy JSON logs. If your server outputs custom access log layouts missing status codes, request sizes, or user-agent fields, lines will fail to match and will be silently ignored.
  • Hardcoded Profile List: The tool identifies only the twelve specific agents defined inside BOT_PROFILES. Unlisted crawlers, custom research bots, or newer AI agents will not be cataloged unless you manually add new entries to the profile list.
  • Robots Rule Interpretation: Compliance tracking relies on urllib.robotparser. It does not interpret non-standard directives like Crawl-delay or vendor-specific path expansions that are outside standard robot exclusions.

Frequently Asked Questions

Does running this script require elevated root privileges?

No. The script only reads log files from disk and performs standard outbound DNS resolution queries. As long as your operating system user has read permissions for the target log files, root or administrative access is not required.

Can the script process logs stored in .gz archives alongside plain text logs?

Yes. The ingest_file method inspects the file name. If it detects a .gz suffix, it uncompresses and reads the file stream via Python's gzip module without requiring prior extraction on disk.

Why does an IP show up as spoofed even if the user agent looks valid?

A request is marked as spoofed when the client supplies a known AI crawler user agent, but its IP address either has no reverse DNS entry, resolves to a domain outside the provider's authorized domain list, or fails forward verification when matching the hostname back to the originating IP address.

Full version and the rest of the toolkit: AIBot Ledger crawler log reporter on PyForge. Product page: https://py-forge.com/products/aibot-ledger-crawler-log-reporter

來源：DEV Community · dev.to

#教學/實踐#部署/工程