eric/hnefatafl
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

A credibility pass over ee5690e..HEAD: traffic.py parses each request once and counts visits one way, the deploy notes and skills agree with the scripts, and a few edit leftovers in the hall are tidied

Browse Source 
traffic.py gives identical counts before and after over 41 days of the
host's access logs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GwFKMuQnPAEHJ5yA1q4orh
This commit is contained in:
Eric Wagoner
2026-10-11 10:17:01 -04:00
co-authored by Claude Opus 5.5
parent 57ae338cdb
commit 7f66be8016
11 changed files with 76 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/traffic.py
+37 -38
View File
@@ -46,6 +46,28 @@ def unfurled_by(ua):
return None
def parse(r):
"""The parts of one log line the counts read."""
q = r.get("request", {})
h = q.get("headers", {})
uri = q.get("uri", "")
return dict(
ts=r.get("ts", 0), status=r.get("status", 0), method=q.get("method"), uri=uri, path=uri.split("?")[0],
ip=q.get("client_ip") or q.get("remote_ip") or "?",
ua=" ".join(h.get("User-Agent", [""])), referer=" ".join(h.get("Referer", [""])),
)
def visit_starts(times):
"""When each visit began: requests more than VISIT_GAP apart are separate visits."""
starts, last = [], None
for ts in sorted(times):
if last is None or ts - last > VISIT_GAP:
starts.append(ts)
last = ts
return starts
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
@@ -53,8 +75,8 @@ def scan(day, logs=LOGS):
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,
the whole day when it asks for a scanner's paths, or comes back 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."""
@@ -73,55 +95,37 @@ def scan(day, logs=LOGS):
except ValueError:
continue
if t0 <= r.get("ts", 0) < t1:
requests.append(r)
requests.append(parse(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:
if BOT.search(r["ua"]) or not r["ua"]:
continue
path = q.get("uri", "").split("?")[0]
a = seen[ip]
a["times"].append(r.get("ts", 0))
path = r["path"]
a = seen[r["ip"]]
a["times"].append(r["ts"])
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:])]
starts = visit_starts(times)
if len(starts) < 4:
return False
gaps = [b - a for a, b in zip(starts, starts[1:])]
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}
monitors = {ip for ip, a in seen.items() if not (a["scanner"] or a["played"]) and len(visit_starts(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))
if r["ua"] and not BOT.search(r["ua"]) and r["path"] in PAGES and not seen[r["ip"]]["played"]:
by_browser[r["ua"]].append((r["ts"], r["ip"]))
for hits in by_browser.values():
if keeps_time([ts for ts, _ in hits]):
monitors |= {ip for _, ip in hits}
@@ -134,12 +138,7 @@ def scan(day, logs=LOGS):
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]
ip, ua, uri, path = r["ip"], r["ua"], r["uri"], r["path"]
t["requests"] += 1
if ip in posing or BOT.search(ua) or not ua:
t["bot"] += 1
@@ -151,7 +150,7 @@ def scan(day, logs=LOGS):
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:
if r["method"] == "POST" and path == "/api/rooms" and 200 <= r["status"] < 300:
t["opened"] += 1
if path.startswith(("/_app/", "/api/")) or path == "/ws" or path.endswith((".png", ".ico", ".txt")):
continue
@@ -161,7 +160,7 @@ def scan(day, logs=LOGS):
t["pages"][PAGES[path]] += 1
elif path.startswith("/join/"):
t["joins"][path.split("/")[2].upper()] += 1
ref = " ".join(h.get("Referer", [""]))
ref = r["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)