SPB Git forge
spboucher.ai ↗

spb/company-atlas

Public
28commits 1branches 0releases
7.7 MBsize
maindefault branch
10 days agolast push
Python 66.3% TypeScript 22.7% JavaScript 8.6% HTML 1.4% CSS 0.7%

fix: Brotli codec + charset sniffing, binary-body guard, corruption guard in pipeline/events, purge script for the 2026-09-12 incident

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed 11 days ago (Sep 13, 2026) parent 90d4862

6 changed files +298 −12

modified pyproject.toml +1 −0
@@ -27,6 +27,7 @@ dependencies = [
27 27 "zstandard>=0.23",
28 28 "tldextract>=5.1",
29 29 "sse-starlette>=2.1",
30 + "brotli>=1.1",
30 31 ]
31 32
32 33 [project.optional-dependencies]
added scripts/purge_corrupt.py +93 −0
@@ -0,0 +1,93 @@
1 +"""One-off repair: remove snapshots/entities/events produced from mis-decoded bodies (2026-09-12 Brotli incident).
2 +
3 +Before the fix, the fetcher advertised `Accept-Encoding: br` without the `brotli` codec installed, so Brotli-compressed responses
4 +were stored and "parsed" as text: garbage titles (U+FFFD runs), 100 %-changed pages, meaningless events. This script:
5 +
6 + 1. scans every snapshot's stored text (and title) and flags the corrupt ones (text quality < 0.99 or U+FFFD in the title),
7 + 2. retracts events derived from them (or carrying U+FFFD) — status 'retracted' + reason, never deleted,
8 + 3. deletes the corrupt snapshots, their change rows and the corrupt extracted entities (labels with U+FFFD),
9 + 4. resets the affected sensors (hashes, last_snapshot_id, snapshot_count) and schedules them to run now, so the next observation
10 + becomes a clean baseline (entities flagged baseline, no bogus "change").
11 +
12 +Raw observation objects are kept (content-addressed archive). Usage: `.venv/bin/python scripts/purge_corrupt.py [--apply]`.
13 +"""
14 +from __future__ import annotations
15 +
16 +import asyncio
17 +import sys
18 +from datetime import UTC, datetime
19 +
20 +from companyatlas import archive
21 +from companyatlas.db import dispose, execute, fetch_all, fetch_val, transaction
22 +from companyatlas.fetch import text_quality
23 +
24 +APPLY = "--apply" in sys.argv
25 +REASON = "corrupt extraction: response body was not decoded (Brotli without codec); repaired 2026-09-13"
26 +
27 +
28 +async def main() -> None:
29 + async with transaction() as conn:
30 + rows = await fetch_all(conn, "select id, sensor_id, title, text_key from snapshots order by sensor_id, fetched_at")
31 + corrupt: list[str] = []
32 + sensors: set[str] = set()
33 + for r in rows:
34 + bad = bool(r["title"] and "�" in r["title"])
35 + if not bad and r["text_key"]:
36 + try:
37 + bad = text_quality(archive.get_text(r["text_key"])) < 0.99
38 + except FileNotFoundError:
39 + bad = False
40 + if bad:
41 + corrupt.append(r["id"])
42 + sensors.add(r["sensor_id"])
43 + print(f"snapshots scanned={len(rows)} corrupt={len(corrupt)} sensors_affected={len(sensors)}")
44 + async with transaction() as conn:
45 + n_ev_text = await fetch_val(conn, "select count(*) from events where status = 'active' and (title ~ E'\\uFFFD' or coalesce(summary, '') ~ E'\\uFFFD')")
46 + n_ev_snap = await fetch_val(conn, "select count(*) from events where status = 'active' and (snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[])))",
47 + s=corrupt or ["-"])
48 + counts = {}
49 + for table, col in (("news_items", "title"), ("jobs", "title"), ("people", "name"), ("products", "name"), ("pricing_plans", "plan_name"), ("locations", "name")):
50 + counts[table] = await fetch_val(conn, f"select count(*) from {table} where {col} ~ E'\\uFFFD'")
51 + n_changes = await fetch_val(conn, "select count(*) from changes where snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))", s=corrupt or ["-"])
52 + print(f"events to retract: text={n_ev_text} snapshot-linked={n_ev_snap} · changes to delete={n_changes} · corrupt entities={counts}")
53 + if not APPLY:
54 + print("dry-run (pass --apply to repair)")
55 + return
56 + now = datetime.now(UTC)
57 + async with transaction() as conn:
58 + await execute(conn, """update events set status = 'retracted', retracted_reason = :r
59 + where status = 'active' and (title ~ E'\\uFFFD' or coalesce(summary, '') ~ E'\\uFFFD'
60 + or snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))
61 + or change_id in (select id from changes where snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))))""",
62 + r=REASON, s=corrupt or ["-"])
63 + await execute(conn, "delete from changes where snapshot_before = any(cast(:s as text[])) or snapshot_after = any(cast(:s as text[]))", s=corrupt or ["-"])
64 + for table, col in (("news_items", "title"), ("jobs", "title"), ("people", "name"), ("products", "name"), ("pricing_plans", "plan_name"), ("locations", "name")):
65 + await execute(conn, f"delete from {table} where {col} ~ E'\\uFFFD'")
66 + await execute(conn, "update snapshots set previous_snapshot_id = null where previous_snapshot_id = any(cast(:s as text[]))", s=corrupt or ["-"])
67 + await execute(conn, "delete from snapshots where id = any(cast(:s as text[]))", s=corrupt or ["-"])
68 + for sid in sensors:
69 + last = await fetch_all(conn, "select id, content_hash, normalized_hash, structural_hash from snapshots where sensor_id = :sid order by fetched_at desc limit 1", sid=sid)
70 + n = await fetch_val(conn, "select count(*) from snapshots where sensor_id = :sid", sid=sid)
71 + if last:
72 + await execute(conn, """update sensors set last_snapshot_id = :snap, last_content_hash = :ch, last_normalized_hash = :nh, last_structural_hash = :sh,
73 + snapshot_count = :n, consecutive_unchanged = 0, next_run_at = :now, etag = null, last_modified = null,
74 + config = config - 'last_structured_hash', updated_at = :now where id = :sid""",
75 + snap=last[0]["id"], ch=last[0]["content_hash"], nh=last[0]["normalized_hash"], sh=last[0]["structural_hash"], n=n, now=now, sid=sid)
76 + else:
77 + await execute(conn, """update sensors set last_snapshot_id = null, last_content_hash = null, last_normalized_hash = null, last_structural_hash = null,
78 + snapshot_count = 0, change_count = 0, meaningful_change_count = 0, event_count = 0, consecutive_unchanged = 0,
79 + next_run_at = :now, etag = null, last_modified = null, config = config - 'last_structured_hash', updated_at = :now
80 + where id = :sid""", now=now, sid=sid)
81 + await execute(conn, "update companies c set last_event_at = (select max(detected_at) from events e where e.company_id = c.id and e.status = 'active')")
82 + print(f"repaired: {len(corrupt)} snapshots removed, {len(sensors)} sensors reset and due now, events retracted, corrupt entities deleted")
83 +
84 +
85 +async def _run() -> None:
86 + try:
87 + await main()
88 + finally:
89 + await dispose()
90 +
91 +
92 +if __name__ == "__main__":
93 + asyncio.run(_run())
modified src/companyatlas/fetch.py +106 −11
@@ -10,6 +10,7 @@ import asyncio
10 10 import hashlib
11 11 import ipaddress
12 12 import logging
13 +import re
13 14 import socket
14 15 import time
15 16 from dataclasses import dataclass, field
@@ -100,6 +101,83 @@ async def validate_destination_async(url: str) -> None:
100 101 validate_destination(url, resolved_ips=ips)
101 102
102 103
104 +# ---------------------------------------------------------------------------------------------- text decoding / sanity
105 +_META_CHARSET_RE = re.compile(rb"""<meta[^>]+charset\s*=\s*["']?\s*([a-zA-Z0-9_.:-]+)""", re.IGNORECASE)
106 +_XML_ENC_RE = re.compile(rb"""^\s*<\?xml[^>]*encoding\s*=\s*["']([a-zA-Z0-9_.:-]+)["']""", re.IGNORECASE)
107 +_CONTROL_RE = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
108 +
109 +
110 +def text_quality(text: str) -> float:
111 + """Share of the text that is NOT replacement characters / C0 control bytes (1.0 = clean). Empty text is 1.0."""
112 + if not text:
113 + return 1.0
114 + bad = text.count("\ufffd") + len(_CONTROL_RE.findall(text))
115 + return max(0.0, 1.0 - bad / len(text))
116 +
117 +
118 +def declared_encodings(content: bytes, content_type: str) -> list[str]:
119 + out: list[str] = []
120 + if content.startswith(b"\xef\xbb\xbf"):
121 + out.append("utf-8-sig")
122 + elif content.startswith((b"\xff\xfe", b"\xfe\xff")):
123 + out.append("utf-16")
124 + ct = (content_type or "").lower()
125 + if "charset=" in ct:
126 + out.append(ct.split("charset=", 1)[1].split(";")[0].strip().strip('"'))
127 + head = content[:4096]
128 + m = _XML_ENC_RE.search(head) or _META_CHARSET_RE.search(head)
129 + if m:
130 + out.append(m.group(1).decode("ascii", "ignore"))
131 + seen: set[str] = set()
132 + result = []
133 + for e in out + ["utf-8", "cp1252"]:
134 + e = e.lower().replace("_", "-")
135 + if e in ("iso-8859-1", "latin-1", "latin1", "ascii", "us-ascii"):
136 + e = "cp1252" # superset; what browsers actually do
137 + if e not in seen:
138 + seen.add(e)
139 + result.append(e)
140 + return result
141 +
142 +
143 +def decode_text(content: bytes, content_type: str = "") -> str:
144 + best: tuple[float, str] | None = None
145 + for enc in declared_encodings(content, content_type):
146 + try:
147 + candidate = content.decode(enc, errors="replace")
148 + except LookupError:
149 + continue
150 + q = text_quality(candidate)
151 + if q >= 0.99:
152 + return candidate
153 + if best is None or q > best[0]:
154 + best = (q, candidate)
155 + return best[1] if best else content.decode("utf-8", errors="replace")
156 +
157 +
158 +_TEXTUAL_TYPES = ("text/", "html", "xml", "json", "javascript", "rss", "atom", "csv")
159 +
160 +
161 +def looks_binary(content: bytes, content_type: str) -> bool:
162 + """A body announced as text/HTML/XML/JSON whose bytes are not text (undecoded compression, wrong Content-Type)."""
163 + ct = (content_type or "").lower()
164 + if not any(t in ct for t in _TEXTUAL_TYPES) and ct:
165 + return False # PDFs, images… are judged by their connectors, not here
166 + sample = content[:8192]
167 + if not sample:
168 + return False
169 + if sample[:2] == b"\x1f\x8b" or sample[:4] == b"\x28\xb5\x2f\xfd": # gzip / zstd magic left undecoded
170 + return True
171 + text_bytes = sum(1 for b in sample if 32 <= b < 127 or b in (9, 10, 13) or b >= 128)
172 + high = sum(1 for b in sample if b >= 128)
173 + controls = len(sample) - text_bytes
174 + if controls / len(sample) > 0.02:
175 + return True
176 + if high / len(sample) > 0.30: # legitimate UTF-8 (CJK) has structure; verify it decodes cleanly
177 + return text_quality(sample.decode("utf-8", errors="replace")) < 0.90
178 + return False
179 +
180 +
103 181 # ---------------------------------------------------------------------------------------------- results / errors
104 182
105 183
@@ -150,14 +228,9 @@ class FetchResult:
150 228
151 229 @property
152 230 def text(self) -> str:
153 − enc = "utf-8"
154 − ct = self.content_type.lower()
155 − if "charset=" in ct:
156 − enc = ct.split("charset=", 1)[1].split(";")[0].strip().strip('"') or "utf-8"
157 − try:
158 − return self.content.decode(enc, errors="replace")
159 − except LookupError:
160 − return self.content.decode("utf-8", errors="replace")
231 + """Decoded body: BOM → HTTP charset → in-document declaration (<meta charset>, XML prolog) → UTF-8; if the chosen codec leaves
232 + more than 1 % replacement characters, the alternatives are tried and the cleanest decode wins (never garbage in, never silently)."""
233 + return decode_text(self.content, self.content_type)
161 234
162 235 @property
163 236 def is_html(self) -> bool:
@@ -289,7 +362,7 @@ class Fetcher:
289 362 self.timeout_s = timeout_s or settings.http_timeout_s
290 363 self.headers = {"User-Agent": settings.user_agent,
291 364 "Accept": "text/html,application/xhtml+xml,application/xml,application/json,application/rss+xml,text/*;q=0.9,*/*;q=0.7",
292 − "Accept-Language": "en-US,en;q=0.9,fr;q=0.6,de;q=0.4,*;q=0.2", "Accept-Encoding": "gzip, deflate, br",
365 + "Accept-Language": "en-US,en;q=0.9,fr;q=0.6,de;q=0.4,*;q=0.2",
293 366 **(headers or {})}
294 367 self._client: httpx.AsyncClient | None = None
295 368 self._http2 = http2
@@ -426,6 +499,12 @@ class Fetcher:
426 499 raise FetchError(f"body exceeds {limit} bytes", status=r.status_code, url=url, failure=FailureClass.TOO_LARGE)
427 500 chunks.append(chunk)
428 501 content = b"".join(chunks)
502 + enc = (r.headers.get("content-encoding") or "").lower()
503 + if enc and enc not in ("identity",) and enc not in ("gzip", "deflate", "br", "zstd"):
504 + raise FetchError(f"unsupported content-encoding {enc!r}", status=r.status_code, url=url, failure=FailureClass.PARSING)
505 + if content and looks_binary(content, r.headers.get("content-type", "")):
506 + raise FetchError(f"body is not text (content-encoding={enc or 'none'}, type={r.headers.get('content-type', '?')[:40]})",
507 + status=r.status_code, url=url, failure=FailureClass.PARSING)
429 508 if len(content) < min_bytes:
430 509 raise FetchError(f"short body ({len(content)} bytes)", status=r.status_code, url=url, failure=FailureClass.PARSING)
431 510 res = FetchResult(url=url, final_url=final_url, status=r.status_code, headers={k.lower(): v for k, v in r.headers.items()},
@@ -490,5 +569,21 @@ def file_result(path: str, *, url: str, content_type: str = "text/html") -> Fetc
490 569 fetched_at=datetime.now(UTC), duration_ms=0, transport="file")
491 570
492 571
493 −__all__ = ["BlockedDestination", "BlockedError", "FetchError", "FetchResult", "Fetcher", "NotModified", "classify_exception", "file_result",
494 − "governor", "looks_like_challenge", "robots", "validate_destination", "validate_destination_async"]
572 +__all__ = [
573 + "BlockedDestination",
574 + "BlockedError",
575 + "FetchError",
576 + "FetchResult",
577 + "Fetcher",
578 + "NotModified",
579 + "classify_exception",
580 + "decode_text",
581 + "file_result",
582 + "governor",
583 + "looks_binary",
584 + "looks_like_challenge",
585 + "robots",
586 + "text_quality",
587 + "validate_destination",
588 + "validate_destination_async",
589 +]
modified src/companyatlas/services/events.py +1 −0
@@ -662,6 +662,7 @@ def derive_events(change: dict[str, Any], company: dict[str, Any], sensor: dict[
662 662 if not structured_hit:
663 663 drafts += _text_diff_rules(surface, change, diff, delta)
664 664
665 + drafts = [d for d in drafts if "\ufffd" not in d.title and "\ufffd" not in (d.summary or "") and re.search(r"[^\W\d_]", d.title)]
665 666 for d in drafts:
666 667 default = EVENT_SUBTYPES.get(d.subtype, (EventType.OTHER, 0.3))[1]
667 668 d.importance = scale_importance(default, significance, d.magnitude)
modified src/companyatlas/services/pipeline.py +59 −1
@@ -19,6 +19,7 @@ import asyncio
19 19 import json
20 20 import logging
21 21 import random
22 +import re
22 23 import time
23 24 from dataclasses import dataclass, field
24 25 from datetime import UTC, datetime, timedelta
@@ -28,7 +29,16 @@ from companyatlas import archive
28 29 from companyatlas.config import settings
29 30 from companyatlas.connectors._util import is_engineering, job_fingerprint, norm_name
30 31 from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction
31 −from companyatlas.fetch import BlockedError, Fetcher, FetchError, FetchResult, NotModified, classify_exception, file_result
32 +from companyatlas.fetch import (
33 + BlockedError,
34 + Fetcher,
35 + FetchError,
36 + FetchResult,
37 + NotModified,
38 + classify_exception,
39 + file_result,
40 + text_quality,
41 +)
32 42 from companyatlas.ids import new_id
33 43 from companyatlas.sdk import connector as connectors
34 44 from companyatlas.sdk.diff import DIFF_VERSION, compare
@@ -508,6 +518,41 @@ def _delta_counts(delta: StructuredDelta) -> dict[str, int]:
508 518 return out
509 519
510 520
521 +def _clean_label(value: str | None) -> str | None:
522 + if value is None:
523 + return None
524 + v = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", value)
525 + v = re.sub(r"\s+", " ", v).strip()
526 + return v or None
527 +
528 +
529 +def _drop_corrupt_entities(ex: Extraction) -> None:
530 + """Remove typed items whose labels carry replacement characters or no letters at all (mis-decoded or empty extractions)."""
531 + def ok(label: str | None) -> bool:
532 + return bool(label) and "\ufffd" not in label and re.search(r"[^\W\d_]", label) is not None
533 +
534 + for j in ex.jobs:
535 + j.title = _clean_label(j.title) or j.title
536 + ex.jobs = [j for j in ex.jobs if ok(j.title)]
537 + for p in ex.people:
538 + p.name = _clean_label(p.name) or p.name
539 + ex.people = [p for p in ex.people if ok(p.name)]
540 + for p in ex.products:
541 + p.name = _clean_label(p.name) or p.name
542 + ex.products = [p for p in ex.products if ok(p.name)]
543 + for p in ex.plans:
544 + p.plan_name = _clean_label(p.plan_name) or p.plan_name
545 + ex.plans = [p for p in ex.plans if ok(p.plan_name)]
546 + for loc in ex.locations:
547 + loc.name = _clean_label(loc.name) or loc.name
548 + ex.locations = [loc for loc in ex.locations if ok(loc.name)]
549 + for n in ex.news:
550 + n.title = _clean_label(n.title) or n.title
551 + ex.news = [n for n in ex.news if ok(n.title) and "\ufffd" not in (n.url or "")]
552 + if ex.title:
553 + ex.title = _clean_label(ex.title)
554 +
555 +
511 556 # ------------------------------------------------------------------------------------------------------------ the run
512 557
513 558
@@ -621,6 +666,19 @@ async def run_sensor(sensor: dict[str, Any], *, fetcher: Fetcher, worker: str =
621 666 outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
622 667 return outcome
623 668
669 + # ---- corruption guard (spec: never fabricate): mis-decoded bodies never become snapshots, entities or events
670 + quality = text_quality(ex.text) if ex.text else 1.0
671 + if quality < 0.99 or (ex.title and "\ufffd" in ex.title):
672 + failure = (str(FailureClass.PARSING), f"extracted text is corrupt (quality {quality:.3f}, content-type {fetched.content_type[:40]!r})", fetched.status)
673 + await _record_failure(conn, sensor=sensor, cfg=cfg, company_id=company_id, obs_id=obs_id, failure=failure, now=now, fetch_ms=fetch_ms,
674 + connector_id=connector.connector_id, worker=worker, current=current, quality=quality, outcome=outcome,
675 + collection_method=collection_method, final_url=fetched.final_url, object_key=object_key, size=len(fetched.content),
676 + content_type=fetched.content_type)
677 + await _consume_domain_budget(conn, domain, pages)
678 + await _ledger(conn, company_id, connector.connector_id, pages, len(fetched.content))
679 + outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
680 + return outcome
681 + _drop_corrupt_entities(ex)
624 682 ex.normalized_hash = ex.normalized_hash or text_hash(ex.text)
625 683 ex.structured_hash = ex.structured_hash or _structured_hash(ex)
626 684 struct_hash = structural_hash(ex.blocks)
added tests/test_text_guards.py +38 −0
@@ -0,0 +1,38 @@
1 +"""Mis-decoded bodies never become snapshots, entities or events (2026-09-12 Brotli incident)."""
2 +from __future__ import annotations
3 +
4 +import gzip
5 +
6 +import brotli
7 +
8 +from companyatlas.fetch import decode_text, looks_binary, text_quality
9 +from companyatlas.sdk.models import ExtractedJob, ExtractedNewsItem, ExtractedPerson, Extraction
10 +from companyatlas.services.pipeline import _drop_corrupt_entities
11 +
12 +HTML = "<html><head><meta charset='utf-8'><title>Société Générale — résultats</title></head><body>Économie 日本語 Zürich</body></html>".encode()
13 +
14 +
15 +def test_binary_bodies_are_flagged() -> None:
16 + assert looks_binary(brotli.compress(HTML), "text/html; charset=utf-8")
17 + assert looks_binary(gzip.compress(HTML), "text/html")
18 + assert not looks_binary(HTML, "text/html")
19 + assert not looks_binary(b"%PDF-1.7 binary", "application/pdf") # non-textual types are left to their connectors
20 +
21 +
22 +def test_decode_prefers_clean_codec() -> None:
23 + assert text_quality(decode_text(HTML, "text/html")) == 1.0
24 + assert "bénéfices" in decode_text("Groupe Lactalis – bénéfices élevés".encode("cp1252"), "text/html")
25 + sjis = "<html><head><meta charset=shift_jis></head><body>トヨタ自動車</body></html>".encode("shift_jis")
26 + assert "トヨタ" in decode_text(sjis, "text/html")
27 + garbage = brotli.compress(HTML).decode("utf-8", errors="replace")
28 + assert text_quality(garbage) < 0.99
29 +
30 +
31 +def test_corrupt_entities_are_dropped() -> None:
32 + ex = Extraction(text="ok", blocks=[], title="News ��", jobs=[ExtractedJob(title="ML Engineer"), ExtractedJob(title="��7�c")],
33 + people=[ExtractedPerson(name="Jane Doe"), ExtractedPerson(name="1234")],
34 + news=[ExtractedNewsItem(title="Quarterly results", url="https://x.com/a"), ExtractedNewsItem(title="ng��t1]", url="https://x.com/b")])
35 + _drop_corrupt_entities(ex)
36 + assert [j.title for j in ex.jobs] == ["ML Engineer"]
37 + assert [p.name for p in ex.people] == ["Jane Doe"]
38 + assert [n.title for n in ex.news] == ["Quarterly results"]
39