waving-hands/deploy/traffic.py

"""Caddy's access log for one day, people and bots apart. The visitors
digest and the nightly rollup both count through here, so they agree."""

import datetime
import glob
import gzip
import json
import os
import re
from collections import Counter, defaultdict

BOT = re.compile(r"bot|crawl|spider|slurp|facebookexternalhit|slack|discord|twitter|whatsapp|telegram|preview|fetch|curl|python|go-http|wget|headless|scan|monitor|uptime", re.I)
# Link previews: the app that fetched the page to show it in a conversation. The people
# who click such a link arrive with no referrer, so this is the best sign of where it went.
UNFURL = [
    ("Slack", re.compile(r"slack", re.I)),
    ("Discord", re.compile(r"discord", re.I)),
    ("iMessage", re.compile(r"facebookexternalhit.*twitterbot|applebot.*preview", re.I)),
    ("Facebook", re.compile(r"facebookexternalhit|facebot|meta-externalagent", re.I)),
    ("WhatsApp", re.compile(r"whatsapp", re.I)),
    ("Telegram", re.compile(r"telegram", re.I)),
    ("Reddit", re.compile(r"redditbot", re.I)),
    ("LinkedIn", re.compile(r"linkedinbot", re.I)),
    ("Mastodon", re.compile(r"mastodon", re.I)),
    ("Bluesky", re.compile(r"bluesky|cardyb", re.I)),
    ("X", re.compile(r"twitterbot", re.I)),
]
# Paths nobody asks a game for but a vulnerability scanner: its address is a bot's all day.
# Matched against whole path segments, never the site's own hashed assets under /_app/.
SCANNER = re.compile(r"\.php$|(^|/)(\.env|\.git|\.aws|wp-admin|wp-login|wp-content|wp-includes|wordpress|phpmyadmin|cgi-bin|vendor|actuator|boaform|owa|admin)(/|$)", re.I)
PAGES = {"/": "hall", "/play": "play", "/rules": "rules"}
SITE = "hands.kestrelsnest.social"
# Each game on a shared host keeps its own access log.
LOGS = "/var/lib/caddy/waving-hands-access*.log*"
# The site's own addresses, for telling a referrer from a move between its pages: its domain, and its sslip.io name.
OWN = re.compile(re.escape(SITE) + r"|\.sslip\.io\b", re.I)
# Requests from one address more than this far apart are separate visits.
VISIT_GAP = 30 * 60


def unfurled_by(ua):
    """The app a link preview came from, or None."""
    for name, pattern in UNFURL:
        if pattern.search(ua):
            return name
    return None


def scan(day, logs=LOGS):
    """Counts for one UTC day (YYYY-MM-DD): every request, page requests by
    people, the addresses of people and of bots, requests by page, room links
    opened, rooms opened from the site itself, where links were shared,
    referrers, and campaign tags (each counted once per address).

    A request is a bot's when its browser says so. An address is a bot's for
    the whole day when it asks for a scanner's paths, or comes back for pages
    three visits or more without ever opening or joining a game: a monitor,
    not a person. A monitor that takes a new address every visit gives itself
    away by keeping time: one browser, page loads at a steady interval, four
    visits or more, never a game."""
    d0 = datetime.datetime.fromisoformat(day).replace(tzinfo=datetime.timezone.utc)
    t0, t1 = d0.timestamp(), (d0 + datetime.timedelta(days=1)).timestamp()
    requests = []
    for f in sorted(glob.glob(logs)):
        if os.path.getmtime(f) < t0:
            continue
        opener = gzip.open if f.endswith(".gz") else open
        try:
            with opener(f, "rt", errors="ignore") as fh:
                for line in fh:
                    try:
                        r = json.loads(line)
                    except ValueError:
                        continue
                    if t0 <= r.get("ts", 0) < t1:
                        requests.append(r)
        except OSError:
            continue

    # First pass: what each address did all day under a browser's name, to find scanners and monitors.
    seen = defaultdict(lambda: dict(times=[], played=False, scanner=False))
    for r in requests:
        q = r.get("request", {})
        ip = q.get("client_ip") or q.get("remote_ip") or "?"
        ua = " ".join(q.get("headers", {}).get("User-Agent", [""]))
        if BOT.search(ua) or not ua:
            continue
        path = q.get("uri", "").split("?")[0]
        a = seen[ip]
        a["times"].append(r.get("ts", 0))
        if not path.startswith("/_app/") and SCANNER.search(path):
            a["scanner"] = True
        if path.startswith(("/join/", "/api/rooms")) or path == "/ws":
            a["played"] = True

    def visits(times):
        n, last = 0, None
        for ts in sorted(times):
            if last is None or ts - last > VISIT_GAP:
                n += 1
            last = ts
        return n

    def keeps_time(times):
        starts, last = [], None
        for ts in sorted(times):
            if last is None or ts - last > VISIT_GAP:
                starts.append(ts)
            last = ts
        gaps = [b - a for a, b in zip(starts, starts[1:])]
        if len(starts) < 4:
            return False
        typical = sorted(gaps)[len(gaps) // 2]
        return all(0.75 * typical <= g <= 1.25 * typical for g in gaps)

    scanners = {ip for ip, a in seen.items() if a["scanner"]}
    monitors = {ip for ip, a in seen.items() if not (a["scanner"] or a["played"]) and visits(a["times"]) >= 3}
    by_browser = defaultdict(list)
    for r in requests:
        q = r.get("request", {})
        ip = q.get("client_ip") or q.get("remote_ip") or "?"
        ua = " ".join(q.get("headers", {}).get("User-Agent", [""]))
        if ua and not BOT.search(ua) and q.get("uri", "").split("?")[0] in PAGES and not seen[ip]["played"]:
            by_browser[ua].append((r.get("ts", 0), ip))
    for hits in by_browser.values():
        if keeps_time([ts for ts, _ in hits]):
            monitors |= {ip for _, ip in hits}
    posing = scanners | monitors

    t = dict(
        requests=0, human=0, bot=0, addresses=set(), botAddresses=set(), pages=Counter(), joins=Counter(),
        opened=0, shared=Counter(), scanners=len(scanners), monitors=len(monitors), referrers=Counter(), campaigns=Counter(),
    )
    seen_campaign = set()
    seen_share = set()
    for r in requests:
        q = r.get("request", {})
        h = q.get("headers", {})
        ua = " ".join(h.get("User-Agent", [""]))
        ip = q.get("client_ip") or q.get("remote_ip") or "?"
        uri = q.get("uri", "")
        path = uri.split("?")[0]
        t["requests"] += 1
        if ip in posing or BOT.search(ua) or not ua:
            t["bot"] += 1
            t["botAddresses"].add(ip)
            service = unfurled_by(ua)
            # One share per app and page: a preview is often fetched twice.
            if service and (service, path) not in seen_share:
                seen_share.add((service, path))
                t["shared"][service] += 1
            continue
        # A room opened from the site itself never passes through /join: count the API's answer.
        if q.get("method") == "POST" and path == "/api/rooms" and 200 <= r.get("status", 0) < 300:
            t["opened"] += 1
        if path.startswith(("/_app/", "/api/")) or path == "/ws" or path.endswith((".png", ".ico", ".txt")):
            continue
        t["human"] += 1
        t["addresses"].add(ip)
        if path in PAGES:
            t["pages"][PAGES[path]] += 1
        elif path.startswith("/join/"):
            t["joins"][path.split("/")[2].upper()] += 1
        ref = " ".join(h.get("Referer", [""]))
        if ref and not OWN.search(ref):
            t["referrers"][re.sub(r"^https?://", "", ref).split("/")[0]] += 1
        m = re.search(r"[?&](ref|utm_source|fbclid)=([^&]*)", uri)
        if m and (ip, m.group(1)) not in seen_campaign:
            seen_campaign.add((ip, m.group(1)))
            t["campaigns"][m.group(1) + ("=" + m.group(2)[:24] if m.group(1) != "fbclid" else "")] += 1
    return t