asxpi/iac
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1
iac/ansible/roles/crawler_alert/files/crawler-alert.py
Sergei Poljanski 27ccdc4255 ansible: describe what runs on the servers 
The roles had drifted far from production: Forgejo was pinned at 12
while 15.0.11 runs, its config lacked the mailer, packages and the real
database path, and several services existed only on the hosts. Running
a role would have rolled production back. Take the servers as the truth
and make the repository match them.

site.yaml now holds the desired state of both hosts:

 - base: unattended upgrades, fail2ban, Docker daemon config and the
   weekly image prune, as found on the hosts. New: a 500M journal cap,
   and the Docker systemd override is kept absent. That override once
   opened dockerd on tcp/2375 without authentication.
 - traefik: templates as deployed, version in a variable (v3.7.18).
   The Coraza WAF files are removed; no router ever used them.
 - forgejo: 15.0.11, app.ini generated from the live file with the
   secrets taken from the vault.
 - crawler_alert, sillytavern, minecraft, minecraft_fossils: added
   from the hosts.

Forgejo's canonical name becomes git.asxp.io; git.czsk.it stays as an
alias on the same router.

docker_service restarts a stack when its config changes, no longer
forces ownership on existing data directories, and hides template
diffs because they carry secrets. Containers listed in
service_graceful_stop are saved and stopped with a long timeout before
compose recreates them: Docker's 10 s default is too short for a
Minecraft world. Both game servers get stop_grace_period for the same
reason.

Removed: the full-deployment and runner playbooks with their prompts,
the Forgejo backup and restore roles for the old bare-metal host, and
inventory entries for services that were never deployed. SSH hardening
stays a manual playbook, since a mistake there locks everyone out.

New secrets (SMTP, RCON, CurseForge key) went into the vault.

Assisted-by: Claude:fable-5-2
2026-11-12 06:47:08 +04:00

707 lines
32 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Crawler watch for a Traefik-fronted Forgejo (or any Traefik site).
Separates real AI crawlers from traffic that only claims to be, by
checking the source IP against each vendor's published range list.
Modes:
(default) daily threshold alert: mail only if yesterday exceeded
CRAWLER_THRESHOLD requests
--weekly summary: always mail stats for the past 7 days
--dry-run print instead of mailing (combinable with --weekly)
Configuration is by environment variable; see README.md. The only
value with no sensible default is CRAWLER_RECIPIENT.
"""
import configparser, glob, gzip, ipaddress, json, os, re, smtplib, socket, subprocess, sys, urllib.request
from collections import Counter
from datetime import date, timedelta
from email.mime.text import MIMEText
# --- configuration ------------------------------------------------------
THRESHOLD = int(os.environ.get("CRAWLER_THRESHOLD", "100000"))
RECIPIENT = os.environ.get("CRAWLER_RECIPIENT", "")
SITE = os.environ.get("CRAWLER_SITE", "this site")
LOG_GLOB = os.environ.get("CRAWLER_LOG", "/var/log/traefik/access.log*")
# Traefik router name to filter on, e.g. "forgejo@docker". Empty = all
# traffic reaching Traefik, which is what you want for a single-site host.
ROUTER = os.environ.get("CRAWLER_ROUTER", "")
# Container to read SMTP settings from (Forgejo/Gitea app.ini [mailer]).
# Set CRAWLER_SMTP_* instead to configure SMTP directly.
MAIL_CONTAINER = os.environ.get("CRAWLER_MAIL_CONTAINER", "forgejo")
MAIL_INI = os.environ.get("CRAWLER_MAIL_INI", "/data/gitea/conf/app.ini")
WEEKLY = "--weekly" in sys.argv
DRY = "--dry-run" in sys.argv
if not RECIPIENT and not DRY:
sys.exit("CRAWLER_RECIPIENT is not set (no --dry-run either); refusing "
"to run. See README.md.")
# --- verified-crawler support -------------------------------------------
# Vendors publishing machine-readable IP ranges. A UA token alone proves
# nothing: on the instance this was written for, ~83% of AI-bot-labelled
# traffic came from hosts outside these ranges. Only IP-verified hits count.
RANGE_SOURCES = {
"openai": ["https://openai.com/chatgpt-user.json",
"https://openai.com/gptbot.json",
"https://openai.com/searchbot.json"],
"anthropic": ["https://claude.com/crawling/bots.json"],
"perplexity":["https://www.perplexity.ai/perplexitybot.json"],
"apple": ["https://search.developer.apple.com/applebot.json"],
"google": ["https://developers.google.com/static/crawling/ipranges/special-crawlers.json",
"https://developers.google.com/static/crawling/ipranges/common-crawlers.json",
"https://developers.google.com/static/crawling/ipranges/user-triggered-fetchers.json",
"https://developers.google.com/static/crawling/ipranges/user-triggered-fetchers-google.json"],
}
# UA token -> vendor whose ranges must contain the source IP
BOT_VENDOR = {
"ChatGPT-User": "openai", "GPTBot": "openai", "OAI-SearchBot": "openai",
"ClaudeBot": "anthropic", "Claude-User": "anthropic", "Claude-SearchBot": "anthropic",
"PerplexityBot": "perplexity", "Perplexity-User": "perplexity",
"Applebot-Extended": "apple", "Applebot": "apple",
"Googlebot": "google", "Google-Extended": "google", "GoogleOther": "google",
}
# --- reverse-DNS verification -------------------------------------------
# For vendors that publish no IP ranges. The check is forward-confirmed
# (FCrDNS): PTR of the address must end in one of the vendor's domains,
# AND that hostname must resolve back to the same address. The forward
# step is what makes it meaningful - a PTR record alone is set by
# whoever controls the address block, so it proves nothing on its own.
BOT_RDNS = {
"Amazonbot": (".crawl.amazonbot.amazon",),
"meta-externalagent": (".facebook.com", ".fbsv.net"),
"meta-externalfetcher": (".facebook.com", ".fbsv.net"),
"FacebookBot": (".facebook.com", ".fbsv.net"),
"YouBot": (".search.you.com",),
"Bytespider": (".bytedance.com", ".byteoversea.com"),
"PetalBot": (".petalsearch.com", ".aspiegel.com"),
"DuckAssistBot": (".duckduckgo.com",),
# Apple and Google publish ranges, but also answer rDNS; listed so a
# range miss can still be resolved when the published file is stale.
"Applebot": (".applebot.apple.com",),
"Googlebot": (".googlebot.com", ".google.com"),
}
# rDNS is slow (two lookups per address), so only unique IPs are probed
# and results are memoised for the run.
RDNS_MAX = int(os.environ.get("CRAWLER_RDNS_MAX", "400"))
RDNS_ENABLE = os.environ.get("CRAWLER_RDNS", "1") != "0"
# --- ASN verification, for single-tenant networks only -------------------
# An ASN match usually proves nothing: genuine YouBot and genuine
# Amazonbot both live in AS14618, which is also every EC2 instance a
# spoofer could rent. It is only evidence when the ASN belongs to the
# vendor exclusively and sells no compute in it. Meta qualifies -
# AS32934 is Meta's own network and cannot be rented - and Meta
# publishes no IP range file (their docs say to email webmasters@meta.com),
# so without this there is nothing to check them against at all.
BOT_ASN = {
"meta-externalagent": {"32934"},
"meta-externalfetcher": {"32934"},
"FacebookBot": {"32934"},
"facebookexternalhit": {"32934"},
}
ASN_ENABLE = os.environ.get("CRAWLER_ASN", "1") != "0"
# Team Cymru's DNS interface: <reversed-addr>.origin[6].asn.cymru.com TXT
CYMRU4 = "origin.asn.cymru.com"
CYMRU6 = "origin6.asn.cymru.com"
# robots.txt to audit verified crawlers against. Fetched once per run;
# empty or unreachable simply skips the compliance check.
ROBOTS_URL = os.environ.get("CRAWLER_ROBOTS", "")
CACHE = os.environ.get("CRAWLER_CACHE", "/var/cache/crawler-alert/ranges.json")
CF_URL = "https://www.cloudflare.com/ips-v4"
# claude.com and cloudflare.com sit behind Cloudflare, which 403s the
# default Python-urllib UA. Send a real one.
UA = "Mozilla/5.0 (compatible; crawler-alert/1.0; +https://codeberg.org/)"
def fetch(url, timeout=20):
req = urllib.request.Request(url, headers={"User-Agent": UA})
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.read()
def load_ranges():
"""Fetch vendor prefixes; fall back to cache so a vendor outage
degrades the report instead of breaking the mail."""
out, stale = {}, []
for vendor, urls in RANGE_SOURCES.items():
nets = []
for u in urls:
try:
for p in json.loads(fetch(u)).get("prefixes", []):
for k, v in p.items():
if k.startswith("ipv"):
nets.append(v)
except Exception:
stale.append(vendor)
if nets:
out[vendor] = nets
cf = []
try:
cf = [l.strip() for l in fetch(CF_URL).decode().splitlines() if l.strip()]
except Exception:
stale.append("cloudflare")
try:
cached = json.load(open(CACHE))
except Exception:
cached = {}
for vendor in RANGE_SOURCES:
if vendor not in out and vendor in cached.get("vendors", {}):
out[vendor] = cached["vendors"][vendor]
if not cf:
cf = cached.get("cloudflare", [])
if out:
try:
os.makedirs(os.path.dirname(CACHE), exist_ok=True)
json.dump({"vendors": out, "cloudflare": cf}, open(CACHE, "w"))
except Exception:
pass
# bucket by first octet so lookups stay cheap over ~1425 prefixes
idx = {}
for vendor, nets in out.items():
b = {}
for s in nets:
n = ipaddress.ip_network(s)
b.setdefault(int(n.network_address) >> 24 if n.version == 4 else -1, []).append(n)
idx[vendor] = b
cfnets = [ipaddress.ip_network(s) for s in cf]
return idx, cfnets, sorted(set(stale))
def in_index(idx_v, addr):
if idx_v is None:
return False
key = int(addr) >> 24 if addr.version == 4 else -1
return any(addr in n for n in idx_v.get(key, []))
def load_robots():
"""Disallow patterns from robots.txt, as (prefix, contains) pairs.
Only the `User-agent: *` group is read: the point is to audit what
well-behaved crawlers were asked not to fetch, not to model every
per-agent rule. Returns [] when unset or unreachable, which disables
the compliance check rather than reporting false violations.
"""
if not ROBOTS_URL:
return []
try:
txt = fetch(ROBOTS_URL, timeout=15).decode("utf-8", "replace")
except Exception:
return []
rules, in_star = [], False
for line in txt.splitlines():
line = line.split("#")[0].strip()
if not line:
continue
k, _, v = line.partition(":")
k, v = k.strip().lower(), v.strip()
if k == "user-agent":
in_star = (v == "*")
elif k == "disallow" and in_star and v:
# Forgejo-style /*/*/thing/ means "contains /thing/".
if v.startswith("/*"):
rules.append(("", v.lstrip("/*")))
else:
rules.append((v.rstrip("*"), ""))
return rules
def robots_blocked(path, rules):
for prefix, contains in rules:
if prefix and path.startswith(prefix):
return True
if contains and contains in path:
return True
return False
_asn_cache = {}
def asn_of(ip):
"""Origin ASN for an address via Team Cymru's DNS interface.
Returns the ASN as a string, or "" if it cannot be determined.
Needs `dig`; absence degrades to no ASN evidence, never an error.
"""
if ip in _asn_cache:
return _asn_cache[ip]
asn = ""
try:
a = ipaddress.ip_address(ip)
if a.version == 4:
q = ".".join(reversed(ip.split("."))) + "." + CYMRU4
else:
nibbles = a.exploded.replace(":", "")
q = ".".join(reversed(nibbles)) + "." + CYMRU6
out = subprocess.run(["dig", "+short", "+time=3", "+tries=1", q, "TXT"],
capture_output=True, text=True, timeout=12).stdout
# "32934 | 2a03:2880:f814::/48 | IE | ripencc | 2011-08-22"
first = out.strip().strip('"').split("|")[0].strip()
if first.split(" ")[0].isdigit():
asn = first.split(" ")[0]
except (OSError, ValueError, subprocess.SubprocessError):
pass
_asn_cache[ip] = asn
return asn
_rdns_cache = {}
def fcrdns(ip, suffixes):
"""Forward-confirmed reverse DNS.
PTR must end in one of `suffixes`, and the name it gives must resolve
back to `ip`. Returns (ok, hostname). Without the forward step this
would only prove the address owner can write their own PTR record.
"""
key = (ip, suffixes)
if key in _rdns_cache:
return _rdns_cache[key]
result = (False, "", "mismatch")
try:
host = socket.gethostbyaddr(ip)[0].rstrip(".").lower()
if any(host.endswith(s) or host == s.lstrip(".") for s in suffixes):
_, _, addrs = socket.gethostbyname_ex(host)
try:
v6 = socket.getaddrinfo(host, None, socket.AF_INET6)
addrs = addrs + [a[4][0] for a in v6]
except OSError:
pass
if ip in addrs:
result = (True, host, "ok")
else:
result = (False, host, "mismatch")
else:
result = (False, host, "mismatch")
except OSError:
# No PTR at all. This is absence of evidence, not evidence of
# forgery: Meta's IPv6 crawler space, for one, publishes no PTR
# records. Kept distinct so a real mismatch stays visible.
result = (False, "", "no-ptr")
_rdns_cache[key] = result
return result
days = [date.today() - timedelta(days=i) for i in range(7 if WEEKLY else 1, 0, -1)]
stamps = [d.strftime("%d/%b/%Y") for d in days]
RANGES, CFNETS, STALE = (load_ranges() if WEEKLY else ({}, [], []))
per_day = {s: [0, set()] for s in stamps}
repos = Counter()
uas = Counter()
verified = Counter() # bot -> IP-verified hits
spoofed = Counter() # bot -> UA claimed, IP outside vendor ranges
proxied = Counter() # bot -> arrived via Cloudflare, unverifiable
ver_paths = Counter() # paths fetched by verified bots
ver_ips = {} # bot -> set of verified source IPs
# What the verified crawlers actually did, once confirmed: status mix,
# bytes served, and any path robots.txt asked them not to fetch.
ver_status = Counter()
ver_bytes = [0]
ver_violations = Counter() # path -> hits, robots.txt disallowed
ver_bot_bytes = Counter() # bot -> bytes served
# Repository *content* fetched by verified crawlers, as opposed to
# policy files and chrome. This is the question a permissive robots.txt
# is asking: now that they may read the code, do they? Counting top
# paths cannot answer it - robots.txt and sitemap.xml drown everything.
ver_content = Counter() # bot -> content hits
ver_content_paths = Counter()
CONTENT_RE = re.compile(r"/(src|raw|commit|commits|media|blame|archive|compare)/")
# Which repositories each vendor is actually interested in.
ver_repos = {} # bot -> Counter(owner/repo)
ver_pkgs = Counter() # "type/name" -> hits, registry artifacts
ver_pkg_index = Counter() # owner -> hits on the bare package index
# Forge paths are /owner/repo/...; these first segments are not owners,
# and a two-segment path like /pix/photo.jpg on another vhost is a file,
# not a repository - hence the extension check.
NOT_OWNER = {"assets", "fonts", "explore", "user", "api", "avatars",
"repo-avatars", "swagger.v1.json", "attachments", "notifications",
"issues", "pulls", "milestones", "admin", "org", "repo", "login",
"-", "favicon.ico", "robots.txt", "sitemap.xml", "pix", "static"}
# Container/package registry: /{owner}/-/packages/{type}/{name}[/{version}]
# Worth counting separately from repositories - it says which published
# artifacts crawlers are indexing, not which source they are reading.
PKG_RE = re.compile(r"^/([^/]+)/-/packages(?:/([^/?]+)/([^/?]+))?")
def package_of(path):
"""(owner, 'type/name') for a registry path, else None. Bare
/owner/-/packages (the index) returns name None."""
m = PKG_RE.match(path.split("?")[0])
if not m:
return None
owner, ptype, name = m.group(1), m.group(2), m.group(3)
return (owner, f"{ptype}/{name}" if ptype and name else None)
def repo_of(path):
"""owner/repo for a forge path, else None."""
seg = path.lstrip("/").split("?")[0].split("/")
if len(seg) < 2 or not seg[0] or not seg[1]:
return None
if seg[0].lower() in NOT_OWNER or seg[1].lower() in NOT_OWNER:
return None
if "." in seg[1] and len(seg) == 2: # /pix/photo.jpg -> a file
return None
return f"{seg[0]}/{seg[1]}"
# rDNS-verified vendors: {bot: {ip: hits}}, resolved after the log pass
rdns_hits = {b: Counter() for b in BOT_RDNS}
rdns_proxied = Counter()
# {bot: {ip: Counter(path)}} - kept so that once an address is confirmed
# by DNS/ASN its paths can join the verified-path breakdown, which
# otherwise only ever showed IP-range-verified vendors.
rdns_paths = {b: {} for b in BOT_RDNS}
ROBOTS_RULES = load_robots() if WEEKLY else []
def record_verified(bot, path, status, nbytes):
"""Log what a confirmed crawler did. Called only after verification,
so these counters describe legitimate traffic and nothing else."""
ver_paths[path] += 1
ver_status[status] += 1
ver_bytes[0] += nbytes
ver_bot_bytes[bot] += nbytes
if CONTENT_RE.search(path):
ver_content[bot] += 1
ver_content_paths[path] += 1
pkg = package_of(path)
if pkg:
owner, name = pkg
if name:
ver_pkgs[f"{owner}/{name}"] += 1
else:
ver_pkg_index[owner] += 1
r = repo_of(path)
if r:
ver_repos.setdefault(bot, Counter())[r] += 1
if ROBOTS_RULES and robots_blocked(path, ROBOTS_RULES):
ver_violations[path] += 1
for path in sorted(glob.glob(LOG_GLOB)):
opener = gzip.open if path.endswith(".gz") else open
with opener(path, "rt", errors="replace") as f:
for line in f:
if ROUTER and ROUTER not in line:
continue
for s in stamps:
if s in line:
per_day[s][0] += 1
per_day[s][1].add(line.split(" ", 1)[0])
parts = line.split()
if len(parts) > 6 and parts[6].count("/") >= 2:
repos["/".join(parts[6].split("/")[1:3])] += 1
# Traefik CLF: ... "<referer>" "<user-agent>" <n> "<router>" ...
# Anchor on the request counter so this works whatever
# the router is called.
ua = re.search(r"\"[^\"]*\" \"([^\"]*)\" \d+ \"", line)
if ua:
u = ua.group(1)
m = re.search(r"(bot|crawler|spider|scrapy|externalagent|gpt|claude|perplexity)[\w./-]*", u, re.I)
uas[m.group(0) if m else ("(none)" if u == "-" else "(browser-like)")] += 1
if WEEKLY:
# longest token first so Applebot-Extended wins over Applebot
bot = next((b for b in sorted(BOT_VENDOR, key=len, reverse=True) if b in u), None)
if bot:
try:
addr = ipaddress.ip_address(parts[0])
except ValueError:
addr = None
if addr is None:
pass
elif any(addr in n for n in CFNETS):
proxied[bot] += 1
elif in_index(RANGES.get(BOT_VENDOR[bot]), addr):
verified[bot] += 1
ver_ips.setdefault(bot, set()).add(parts[0])
if len(parts) > 9:
record_verified(
bot, parts[6][:60], parts[8],
int(parts[9]) if parts[9].isdigit() else 0)
else:
spoofed[bot] += 1
if WEEKLY and RDNS_ENABLE:
rb = next((b for b in sorted(BOT_RDNS, key=len, reverse=True)
if b in u), None)
# skip if already counted by IP-range check
if rb and not (rb in BOT_VENDOR and rb == bot):
try:
a2 = ipaddress.ip_address(parts[0])
except ValueError:
a2 = None
if a2 is None:
pass
elif any(a2 in n for n in CFNETS):
rdns_proxied[rb] += 1
else:
rdns_hits[rb][parts[0]] += 1
if len(parts) > 9 and not a2.is_private:
rdns_paths[rb].setdefault(parts[0], []).append(
(parts[6][:60], parts[8],
int(parts[9]) if parts[9].isdigit() else 0))
break
total = sum(v[0] for v in per_day.values())
all_ips = set().union(*(v[1] for v in per_day.values()))
# --- resolve rDNS candidates ---------------------------------------------
# One pass over unique addresses, busiest first, capped by RDNS_MAX so a
# flood of distinct forgeries cannot stall the report on DNS timeouts.
rdns_ok = Counter() # bot -> forward-confirmed by PTR
rdns_mismatch = Counter() # bot -> PTR exists but points elsewhere: forgery
rdns_noptr = Counter() # bot -> no PTR at all: cannot be judged by DNS
rdns_asn = Counter() # bot -> no PTR, but origin ASN is the vendor's
rdns_skipped = Counter() # bot -> hits left unresolved by the cap
rdns_private = Counter() # bot -> hits from RFC1918/loopback (see below)
rdns_ok_ips = {}
if WEEKLY and RDNS_ENABLE:
socket.setdefaulttimeout(3)
for bot, counter in rdns_hits.items():
budget = RDNS_MAX
for ip, hits in counter.most_common():
# A private source address means the real client IP never
# reached the log: something in front (Docker's bridge
# gateway, a local proxy) is rewriting it. Neither verified
# nor spoofed - the evidence simply is not there.
if ipaddress.ip_address(ip).is_private:
rdns_private[bot] += hits
continue
if budget <= 0:
rdns_skipped[bot] += hits
continue
budget -= 1
ok, _host, why = fcrdns(ip, BOT_RDNS[bot])
confirmed = False
if ok:
rdns_ok[bot] += hits
rdns_ok_ips.setdefault(bot, set()).add(ip)
confirmed = True
elif why == "no-ptr":
# Fall back to ASN, but only for vendors whose network is
# single-tenant (see BOT_ASN). Otherwise leave it unknown.
if ASN_ENABLE and bot in BOT_ASN and asn_of(ip) in BOT_ASN[bot]:
rdns_asn[bot] += hits
rdns_ok_ips.setdefault(bot, set()).add(ip)
confirmed = True
else:
rdns_noptr[bot] += hits
else:
rdns_mismatch[bot] += hits
if confirmed:
for p, st, nb in rdns_paths.get(bot, {}).get(ip, ()):
record_verified(bot, p, st, nb)
socket.setdefaulttimeout(None)
if not WEEKLY:
s0 = stamps[0]
count, ips = per_day[s0][0], len(per_day[s0][1])
print(f"{s0}: {count} requests, {ips} unique IPs (threshold {THRESHOLD})")
if DRY or count <= THRESHOLD:
sys.exit(0)
subject = f"{SITE}: {count} requests on {s0} (over {THRESHOLD}/day threshold)"
body = (f"Forgejo served {count} requests from {ips} unique IPs on {s0} - "
f"above the {THRESHOLD}/day watch threshold.\n\n"
"Time to consider Anubis (PoW challenge) in front of Forgejo.\n")
else:
lines = [f"{SITE} weekly crawler report ({stamps[0]} - {stamps[-1]})", ""]
lines.append(f"Total: {total} requests, {len(all_ips)} unique IPs")
lines.append(f"Daily threshold alert fires above {THRESHOLD} req/day (none = quiet week)")
lines.append("")
lines.append("Per day:")
for s in stamps:
lines.append(f" {s}: {per_day[s][0]:>7} requests, {len(per_day[s][1]):>6} unique IPs")
lines.append("")
lines.append("Top repos:")
for r, c in repos.most_common(5):
lines.append(f" {c:>7} {r}")
lines.append("")
lines.append("User-agent classes (browser-like = mostly the spoofing swarm):")
for u, c in uas.most_common(8):
lines.append(f" {c:>7} {u}")
lines.append("")
# --- legitimate (IP-verified) crawlers -------------------------------
lines.append("=" * 62)
lines.append("LEGITIMATE CRAWLERS (source IP inside vendor's published ranges)")
lines.append("=" * 62)
if STALE:
lines.append(f"NOTE: range fetch failed for {', '.join(STALE)} - used cache")
tv, ts, tp = sum(verified.values()), sum(spoofed.values()), sum(proxied.values())
claimed = tv + ts + tp
if not claimed:
lines.append(" No AI-bot user-agents seen this week.")
else:
lines.append(f" {'bot':<20}{'verified':>9}{'spoofed':>9}{'via CF':>8} {'IPs':>4}")
for bot in sorted(set(verified) | set(spoofed) | set(proxied),
key=lambda b: -(verified[b] + spoofed[b] + proxied[b])):
lines.append(f" {bot:<20}{verified[bot]:>9}{spoofed[bot]:>9}"
f"{proxied[bot]:>8} {len(ver_ips.get(bot, ())):>4}")
pct = tv / claimed * 100
lines.append("")
lines.append(f" {claimed} requests claimed an AI-bot identity; {tv} verified ({pct:.1f}%).")
lines.append(f" {ts} failed IP verification (spoofed). {tp} arrived via Cloudflare")
lines.append(" and cannot be verified by IP - not counted either way.")
# --- vendors verified by reverse DNS ---------------------------------
if RDNS_ENABLE and (rdns_ok or rdns_mismatch or rdns_noptr or rdns_asn
or rdns_proxied or rdns_private):
lines.append("")
lines.append("-" * 62)
lines.append("VERIFIED BY DNS / ASN (vendors publishing no IP ranges)")
lines.append("-" * 62)
lines.append(f" {'bot':<20}{'PTR ok':>7}{'ASN ok':>7}{'FORGED':>8}"
f"{'no-PTR':>7}{'CF':>6}{'priv':>7} {'IPs':>4}")
allbots = (set(rdns_ok) | set(rdns_mismatch) | set(rdns_noptr)
| set(rdns_asn) | set(rdns_proxied) | set(rdns_private))
for bot in sorted(allbots, key=lambda b: -(rdns_ok[b] + rdns_mismatch[b]
+ rdns_noptr[b] + rdns_asn[b]
+ rdns_proxied[b] + rdns_private[b])):
lines.append(f" {bot:<20}{rdns_ok[bot]:>7}{rdns_asn[bot]:>7}"
f"{rdns_mismatch[bot]:>8}{rdns_noptr[bot]:>7}"
f"{rdns_proxied[bot]:>6}{rdns_private[bot]:>7}"
f" {len(rdns_ok_ips.get(bot, ())):>4}")
if rdns_skipped:
sk = ", ".join(f"{b} {c}" for b, c in rdns_skipped.most_common(4))
lines.append(f" (unresolved, over CRAWLER_RDNS_MAX={RDNS_MAX}: {sk})")
lines.append("")
lines.append(" PTR ok forward-confirmed: PTR ends in a vendor domain AND")
lines.append(" that hostname resolves back to the same address.")
lines.append(" ASN ok no PTR, but the address sits in an ASN belonging")
lines.append(" solely to the vendor (Meta's AS32934). Only used")
lines.append(" where the ASN sells no compute - never for clouds.")
lines.append(" FORGED a PTR exists and points somewhere else. This is")
lines.append(" the only column that is evidence of forgery.")
lines.append(" no-PTR no PTR record and no usable ASN: unknown, not bad.")
if sum(rdns_private.values()):
lines.append(" priv logged source was RFC1918: a proxy or Docker")
lines.append(" bridge masked the client. Unjudgeable. See README.")
# --- what the verified crawlers actually fetched ---------------------
# Last, because it spans both tables above: an address only counts here
# once it has been confirmed, whether by IP range, PTR or ASN.
if ver_paths:
lines.append("")
lines.append("=" * 62)
lines.append("WHAT THE VERIFIED CRAWLERS DID (all methods)")
lines.append("=" * 62)
tot_v = sum(ver_paths.values())
mb = ver_bytes[0] / 1048576
lines.append(f" {tot_v} requests, {mb:.1f} MB served.")
if ver_status:
mix = " ".join(f"{s}:{c}" for s, c in ver_status.most_common(6))
lines.append(f" Status: {mix}")
if ver_bot_bytes:
top = " ".join(f"{b} {v/1048576:.1f}MB"
for b, v in ver_bot_bytes.most_common(4))
lines.append(f" Bandwidth: {top}")
lines.append("")
if not ROBOTS_RULES:
lines.append(" robots.txt compliance: not checked "
"(set CRAWLER_ROBOTS to enable)")
elif ver_violations:
n = sum(ver_violations.values())
lines.append(f" ROBOTS.TXT VIOLATIONS: {n} requests to disallowed paths")
for p, c in ver_violations.most_common(6):
lines.append(f" {c:>6} {p}")
lines.append(" A verified crawler ignoring robots.txt is a policy")
lines.append(" problem, not a security one - the identity is genuine.")
else:
lines.append(f" robots.txt compliance: clean, 0 violations in "
f"{tot_v} requests")
lines.append("")
# Repository content vs everything else. If robots.txt permits
# /src/ and /raw/, this is what says whether anyone took it up.
nc = sum(ver_content.values())
pct_c = nc / tot_v * 100 if tot_v else 0
lines.append(f" Repository content fetched: {nc} of {tot_v} requests "
f"({pct_c:.1f}%)")
if nc:
by_bot = " ".join(f"{b} {c}" for b, c in ver_content.most_common(5))
lines.append(f" by: {by_bot}")
for p, c in ver_content_paths.most_common(8):
lines.append(f" {c:>6} {p}")
else:
lines.append(" none - crawlers indexed the site but read no code")
# Which repositories each vendor cares about. Grouped by vendor
# rather than by UA token: OpenAI alone sends three tokens, and
# "what is OpenAI reading" is the useful question.
if ver_repos:
groups = {}
for b, c in ver_repos.items():
if b in BOT_VENDOR:
v = BOT_VENDOR[b] # openai, anthropic, ...
elif b in BOT_ASN:
v = "meta"
elif b in BOT_RDNS:
v = b.lower() # amazonbot, youbot, ...
else:
v = b.lower()
groups.setdefault(v, Counter()).update(c)
lines.append("")
lines.append(" Repositories by vendor:")
for v in sorted(groups, key=lambda x: -sum(groups[x].values())):
tot = sum(groups[v].values())
top = ", ".join(f"{r} ({n})" for r, n in groups[v].most_common(3))
lines.append(f" {v:<12} {tot:>5} hits {top}")
if ver_pkgs or ver_pkg_index:
np = sum(ver_pkgs.values()) + sum(ver_pkg_index.values())
lines.append("")
lines.append(f" Package registry: {np} requests")
for p, c in ver_pkgs.most_common(6):
lines.append(f" {c:>6} {p}")
if ver_pkg_index:
idx = ", ".join(f"{o} ({n})" for o, n in ver_pkg_index.most_common(4))
lines.append(f" index pages: {idx}")
lines.append("")
lines.append(" Top paths overall:")
for p, c in ver_paths.most_common(12):
lines.append(f" {c:>6} {p}")
body = "\n".join(lines) + "\n"
subject = f"{SITE} weekly crawler report: {total} requests, {len(all_ips)} IPs"
if DRY:
print(subject); print(); print(body)
sys.exit(0)
# SMTP settings: use CRAWLER_SMTP_* if given, otherwise borrow them from
# a Forgejo/Gitea app.ini [mailer] section so there is no second copy of
# the credentials to keep in sync.
if os.environ.get("CRAWLER_SMTP_ADDR"):
addr = os.environ["CRAWLER_SMTP_ADDR"]
port = int(os.environ.get("CRAWLER_SMTP_PORT", "587"))
user = os.environ.get("CRAWLER_SMTP_USER", "")
passwd = os.environ.get("CRAWLER_SMTP_PASSWORD", "")
sender = os.environ.get("CRAWLER_SMTP_FROM", user or RECIPIENT)
proto = os.environ.get("CRAWLER_SMTP_PROTOCOL", "smtp+starttls")
else:
ini = subprocess.run(["docker", "exec", MAIL_CONTAINER, "cat", MAIL_INI],
capture_output=True, text=True).stdout
if not ini.strip():
sys.exit(f"could not read {MAIL_INI} from container "
f"'{MAIL_CONTAINER}' and CRAWLER_SMTP_ADDR is unset")
cp = configparser.ConfigParser(interpolation=None, strict=False)
cp.read_string("[DEFAULT]\n" + ini)
m = cp["mailer"]
proto = m.get("PROTOCOL", "smtps").strip()
addr, port = m.get("SMTP_ADDR").strip(), int(m.get("SMTP_PORT", "465").strip())
user, passwd = m.get("USER", "").strip(), m.get("PASSWD", "").strip().strip("`\"")
sender = m.get("FROM", user).strip()
msg = MIMEText(body)
msg["Subject"], msg["From"], msg["To"] = subject, sender, RECIPIENT
if proto == "smtps":
s = smtplib.SMTP_SSL(addr, port, timeout=30)
else:
s = smtplib.SMTP(addr, port, timeout=30)
if proto != "smtp": # plain smtp = no TLS (local relay)
s.starttls()
if user: # unauthenticated local relays exist
s.login(user, passwd)
s.sendmail(sender, [RECIPIENT], msg.as_string())
s.quit()
print(f"mailed to {RECIPIENT}")
Reference in a new issue View git blame Copy permalink