# -*- coding: utf-8 -*- """Does the crawler that says it is GPTBot come from an address OpenAI publishes? python verify_agents.py broikos.gr A user agent string is a claim, not evidence. Anybody can send `GPTBot` and nothing stops them, so a paper that counts agent strings and calls the result "what OpenAI took" is counting a claim. This script is the check the plan promised: every operator that publishes its address ranges is checked against them, and every request from an address outside them is reported separately rather than silently credited to that operator. Three verification methods, in the order of how much they prove: published range the operator publishes a JSON list of prefixes. An address inside one is the operator, an address outside is not. reverse DNS the operator documents a hostname suffix instead. The address is resolved to a name, the name is resolved back to an address, and both must agree. A one-way lookup is forgeable and is not accepted here. none published the operator publishes neither. The requests are counted and labelled unverifiable, which is a gap in this study rather than an accusation against the crawler. Output is `verification.csv` and `verification.md`. Neither contains an address: the counts are per operator, which is the same rule `extract.py` follows. """ import argparse import collections import concurrent.futures as cf import csv import glob import gzip import io import ipaddress import json import os import re import socket import sys HERE = os.path.dirname(os.path.abspath(__file__)) RAW = os.path.join(HERE, 'raw') sys.path.insert(0, HERE) from extract import CRAWLER_RE, LINE, is_page # noqa: E402 import sitecfg # noqa: E402 # Where each operator says its crawlers come from. Several are tried per # operator because publishers move these files; whichever answers is recorded in # the output, and an operator whose file cannot be fetched is reported as # unverifiable rather than quietly treated as verified. RANGES = { 'GPTBot': ['https://openai.com/gptbot.json'], 'OAI-SearchBot': ['https://openai.com/searchbot.json'], 'ChatGPT-User': ['https://openai.com/chatgpt-user.json'], # one file for all three Anthropic agents, found through the support article # rather than guessed at: /crawling/bots.json, linked from the sentence about # blocking by address 'ClaudeBot': ['https://claude.com/crawling/bots.json'], 'Claude-User': ['https://claude.com/crawling/bots.json'], 'Claude-SearchBot': ['https://claude.com/crawling/bots.json'], 'PerplexityBot': ['https://www.perplexity.com/perplexitybot.json', 'https://www.perplexity.ai/perplexitybot.json'], 'Perplexity-User': ['https://www.perplexity.com/perplexity-user.json'], # Google splits its crawlers across three files and an address in any of them # is Google. Checking only the first would report Google impersonating itself. 'Googlebot': ['https://developers.google.com/static/search/apis/ipranges/googlebot.json', 'https://developers.google.com/static/search/apis/ipranges/special-crawlers.json', 'https://developers.google.com/static/search/apis/ipranges/user-triggered-fetchers-google.json'], 'Google-Extended': ['https://developers.google.com/static/search/apis/ipranges/googlebot.json', 'https://developers.google.com/static/search/apis/ipranges/special-crawlers.json', 'https://developers.google.com/static/search/apis/ipranges/user-triggered-fetchers-google.json'], 'Bingbot': ['https://www.bing.com/toolbox/bingbot.json'], 'Applebot': ['https://search.developer.apple.com/applebot.json'], 'Applebot-Extended': ['https://search.developer.apple.com/applebot.json'], 'DuckAssistBot': ['https://duckduckgo.com/duckassistbot.json'], } # Operators that document a hostname suffix instead of a prefix list. Both # directions are checked: name -> address must return the address we started # from, because a reverse record alone is set by whoever owns the address. RDNS = { 'Googlebot': ('.googlebot.com', '.google.com'), 'Google-Extended': ('.googlebot.com', '.google.com'), 'Bingbot': ('.search.msn.com',), 'Applebot': ('.applebot.apple.com',), 'Applebot-Extended': ('.applebot.apple.com',), 'Amazonbot': ('.crawl.amazonbot.amazon',), 'meta-externalagent': ('.fbsv.net', '.facebook.com', '.fbcdn.net'), 'Bytespider': ('.bytedance.com', '.byteoversea.com'), } _CACHE = {} def fetch_prefixes(urls): """Every published prefix across the operator's files, and which served them. All of the URLs are unioned rather than the first one that answers, because Google splits its crawlers across three lists and an address in the second is still Google. Each file is fetched once per run. """ try: import requests # noqa: PLC0415 except ImportError: return None, None, 'requests not installed' nets, served, last = [], [], '' for url in urls: if url in _CACHE: got = _CACHE[url] else: got = None for attempt in (1, 2, 3): try: r = requests.get(url, timeout=25, headers={'User-Agent': 'paper-09-verifier'}) if r.status_code == 200: doc = json.loads(r.text) got = [] for p in (doc.get('prefixes') or []): for k in ('ipv4Prefix', 'ipv6Prefix', 'prefix', 'ipv4', 'ipv6'): if p.get(k): try: got.append( ipaddress.ip_network(p[k], strict=False)) except ValueError: pass break last = 'HTTP %d' % r.status_code except Exception as exc: # noqa: BLE001 last = str(exc)[:60] if got is None: # A file that would not load is a hole in the check, and a hole # that is not said out loud reads as a verified result. print(' could not load %s: %s' % (url, last)) _CACHE[url] = got if got: nets.extend(got) served.append(url) if nets: return nets, ' '.join(served), '' return None, None, last or 'no prefixes in the document' # Operators that run their crawler from their own registered network and say so, # but publish neither a prefix file nor a hostname. The registry is then the only # thing left, and it is the weakest of the three: it says who holds the address, # not who sent the request. It is used only as a last resort and it is labelled. RDAP_ORG = { 'meta-externalagent': ('facebook', 'meta platforms'), 'Bytespider': ('bytedance', 'douyin', 'beijing douyin'), 'YouBot': ('you.com', 'you inc'), } _RDAP = {} def rdap_org(addr): """Who the registry says holds this address. Cached by /24 to be polite.""" try: import requests # noqa: PLC0415 except ImportError: return '' key = addr.rsplit('.', 1)[0] if '.' in addr else addr[:16] if key in _RDAP: return _RDAP[key] text = '' try: r = requests.get('https://rdap.org/ip/%s' % addr, timeout=20, headers={'User-Agent': 'paper-09-verifier'}) if r.status_code == 200: doc = r.json() bits = [str(doc.get('name') or ''), str(doc.get('handle') or '')] for e in (doc.get('entities') or []): bits.append(str(e.get('handle') or '')) for v in (e.get('vcardArray') or [None, []])[1]: if isinstance(v, list) and len(v) > 3 and v[0] == 'fn': bits.append(str(v[3])) for rem in (doc.get('remarks') or []): bits.extend(str(x) for x in (rem.get('description') or [])) text = ' '.join(bits).lower() except Exception: # noqa: BLE001 text = '' _RDAP[key] = text return text def ptr(addr): """The reverse record, or empty. Used only to count, never to verify.""" try: return socket.gethostbyaddr(addr)[0].lower().rstrip('.') except Exception: # noqa: BLE001 return '' def rdns_ok(addr, suffixes): """A reverse record that survives a forward lookup back to the same address.""" try: name = socket.gethostbyaddr(addr)[0].lower().rstrip('.') except Exception: # noqa: BLE001 return False if not any(name.endswith(s) for s in suffixes): return False try: _h, _a, addrs = socket.gethostbyname_ex(name) except Exception: # noqa: BLE001 return False return addr in addrs def collect(site): """Requests, pages and distinct addresses per operator, addresses in memory. Pages are carried per address as well as per operator, so that the verified page count is the sum over verified addresses rather than a percentage applied to a total, which would be an estimate wearing a measurement's clothes. """ per = collections.defaultdict(lambda: {'req': 0, 'pages': 0, 'ips': collections.Counter(), 'ip_pages': collections.Counter()}) # (day, operator, address) -> counts, so that the day-level table can be # rebuilt from verified addresses only once the check has run daily = collections.defaultdict(lambda: {'req': 0, 'pages': 0, 'bytes': 0}) for path in sorted(glob.glob(os.path.join(RAW, site, '*.gz'))): with gzip.open(path, 'rt', encoding='utf-8', errors='replace') as f: for line in f: m = LINE.match(line) if not m: continue agent = m.group('agent') or '' for name, rx in CRAWLER_RE: if rx.search(agent): row = per[name] ip = m.group('host') row['req'] += 1 row['ips'][ip] += 1 req = m.group('req') or '' path_ = req.split(' ')[1] if ' ' in req else '-' try: status = int(m.group('status')) except (TypeError, ValueError): status = 0 try: size = int(m.group('bytes')) except (TypeError, ValueError): size = 0 d = daily[((m.group('when') or '')[:11], name, ip)] d['req'] += 1 d['bytes'] += size if status in (200, 206) and is_page(path_): row['pages'] += 1 row['ip_pages'][ip] += 1 d['pages'] += 1 break return per, daily def main(): ap = argparse.ArgumentParser() ap.add_argument('site', nargs='?', default='broikos.gr') a = ap.parse_args() al = sitecfg.alias(a.site) out_dir = sitecfg.outdir(al) per, daily = collect(a.site) if not per: print('no raw logs for %s' % a.site) return 1 good = {} # operator -> the addresses that survived print('%d operator(s), %d distinct address(es) in memory' % (len(per), sum(len(v['ips']) for v in per.values()))) rows = [] for name in sorted(per, key=lambda n: -per[n]['req']): row = per[name] ips = list(row['ips']) nets, src, why = (None, None, 'no range file published') if name in RANGES: nets, src, why = fetch_prefixes(RANGES[name]) # Both methods, not one. An operator can publish a prefix list that is # behind its own fleet, and refusing to try the documented hostname on # the addresses the list missed would report that gap as impersonation. ok_ips, used, source = set(), [], '' if nets: used.append('published range') source = src for ip in ips: try: addr = ipaddress.ip_address(ip) except ValueError: continue if any(addr in n for n in nets): ok_ips.add(ip) rest = [ip for ip in ips if ip not in ok_ips] if rest and name in RDNS: sfx = RDNS[name] with cf.ThreadPoolExecutor(max_workers=24) as ex: for ip, ok in zip(rest, ex.map(lambda i: rdns_ok(i, sfx), rest)): if ok: ok_ips.add(ip) used.append('reverse DNS') source = (source + ' | ' if source else '') + ' '.join(sfx) rest = [ip for ip in ips if ip not in ok_ips] if rest and name in RDAP_ORG: pats = RDAP_ORG[name] with cf.ThreadPoolExecutor(max_workers=8) as ex: for ip, org in zip(rest, ex.map(rdap_org, rest)): if org and any(p in org for p in pats): ok_ips.add(ip) used.append('registry ownership') source = (source + ' | ' if source else '') + ' '.join(pats) method = ' and '.join(used) or 'none published' if not used: source = why good[name] = ok_ips ok_req = sum(n for ip, n in row['ips'].items() if ip in ok_ips) ok_pages = sum(n for ip, n in row['ip_pages'].items() if ip in ok_ips) # An address that resolves to nothing at all is a different thing from # one that resolves to somebody else's domain, and the second is the # only one that looks like impersonation. note = '' left = [ip for ip in ips if ip not in ok_ips] if left: with cf.ThreadPoolExecutor(max_workers=24) as ex: names = list(ex.map(ptr, left)) noname = sum(1 for n2 in names if not n2) note = ('%d unverified, %d of them with no reverse record at all' % (len(left), noname)) rows.append(dict(operator=name, requests=row['req'], pages=row['pages'], distinct_ips=len(ips), verified_ips=len(ok_ips), verified_requests=ok_req, verified_pages=ok_pages, method=method, source=source, note=note)) print(' %-20s %6d req %4d addr %4d verified %-32s %s' % (name, row['req'], len(ips), len(ok_ips), method, note)) with io.open(os.path.join(out_dir, 'verification.csv'), 'w', encoding='utf-8', newline='') as f: w = csv.DictWriter(f, fieldnames=['operator', 'requests', 'pages', 'distinct_ips', 'verified_ips', 'verified_requests', 'verified_pages', 'method', 'source', 'note']) w.writeheader() for r in rows: w.writerow(r) # the day-level table, rebuilt from verified addresses only, so that every # downstream analysis can work from checked identities rather than claims agg = collections.defaultdict(lambda: {'req': 0, 'pages': 0, 'bytes': 0}) for (day, name, ip), d in daily.items(): if ip in good.get(name, ()): row = agg[(day, name)] row['req'] += d['req'] row['pages'] += d['pages'] row['bytes'] += d['bytes'] with io.open(os.path.join(out_dir, 'crawlers_verified.csv'), 'w', encoding='utf-8', newline='') as f: w = csv.writer(f) w.writerow(['day', 'operator', 'requests', 'pages', 'bytes']) for (day, name), row in sorted(agg.items()): w.writerow([day, name, row['req'], row['pages'], row['bytes']]) tot = sum(r['requests'] for r in rows) ver = sum(r['verified_requests'] for r in rows) checkable = sum(r['requests'] for r in rows if r['method'] != 'none published') okable = sum(r['verified_requests'] for r in rows if r['method'] != 'none published') md = ['# Which of these crawlers is who it says it is', '', 'Every operator that publishes address ranges or a documented hostname', 'suffix was checked against it. Addresses are held in memory and never', 'written anywhere, here or in `extract.py`.', '', '| operator | requests | addresses | verified | requests verified | how |', '|---|---|---|---|---|---|'] for r in rows: md.append('| %s | %d | %d | %d | %d (%.0f%%) | %s |' % (r['operator'], r['requests'], r['distinct_ips'], r['verified_ips'], r['verified_requests'], 100.0 * r['verified_requests'] / r['requests'] if r['requests'] else 0, r['method'])) md += ['', '**%d of %d requests (%.1f per cent) came from an address the operator ' 'publishes or a hostname it documents.**' % (ver, tot, 100.0 * ver / tot if tot else 0), '', 'Counting only the operators for which any check is possible at all, ' '%d of %d requests verified, %.1f per cent.' % (okable, checkable, 100.0 * okable / checkable if checkable else 0), '', 'An unverified request is not proof of impersonation. It can equally be ' 'a range published after this window, a proxy the operator uses and does ' 'not list, or a file that moved. It is reported because the alternative ' 'is to count a claim and call it a measurement.', ''] io.open(os.path.join(out_dir, 'verification.md'), 'w', encoding='utf-8').write( '\n'.join(md)) print('\n%d of %d requests verified, %.1f%%' % (ver, tot, 100.0 * ver / tot if tot else 0)) return 0 if __name__ == '__main__': raise SystemExit(main())