ALL

Solo para leer y verificar. No puedes copiar, usar, modificar ni redistribuir este código. Aviso completo en la página principal.

src/all_mcp/e2e_ct_watch.py

SHA-256: 3b4d4b4cd61a575805bf7243b7ceb11cf8c421f2490a30ba6afcbfede1746de0
554 líneas · 31089 bytes · Ver en GitHub

1"""ALL 2.0 WP36: the device watches Certificate Transparency for ITS OWN name (plan section 4, "Mala emisión detectable, con ancla en el dispositivo").
2
3The two other detectors (the site Worker's cron and the independent ct_watch.py) compare CT with what the control plane's registry says, so a compromised control plane
4could rewrite the registry and keep them quiet. This one cannot be rewritten from there: the only list it trusts is ``tls/spki-log`` (tls_store.py), one line for every key
5THIS computer ever made, written before the key is used and never changed. Once a day, and ten minutes after each issuance, it asks CT what certificates exist for this
6computer's own name and raises an alert for
7
8 unknown_key a certificate for the name whose key this computer never made (nobody else can have one honestly: an ACME order proves the key with a signature), and
9 bad_issuer a certificate for the name from anyone but Let's Encrypt, whatever its key.
10
11Everything else is silent: our own certificates, and their precertificates (the same key and the same issuance: CT lists the pair, Cert Spotter as one entry and crt.sh as
12two), so the pair is never counted twice and never alarms. The comparison does not look at serials, dates or the control plane.
13
14What is asked, and of whom. Cert Spotter first (``/v1/issuances`` for the exact name, wildcards that cover it included), crt.sh when Cert Spotter cannot be read whole. The
15request holds the name and fixed words and nothing else: no key, no licence, no device id, no cookie, no body; a redirect is refused. The name is public anyway (it is in CT);
16the zone is never asked about alone, so a wildcard issued for the zone shows here only through Cert Spotter's ``match_wildcards`` (the other two detectors watch the zone).
17Whole history, every time, never a cursor: a computer that was off for 100 days reads the certificates issued meanwhile (expired ones too) in its first look, and a source
18that repeats, reorders or duplicates entries changes nothing.
19
20Honesty about being blind. When this computer serves a certificate that is more than two days old, a source that is read whole must show it (it is the canary: a page that came
21back empty or an index that is behind is not an all-clear). A source that answers with a page that does not parse whole (truncated, HTML, a row without a key hash), refuses, times out or hands out
22more than this device is willing to read counts as down, and so does a crt.sh listing that needs more downloads than one run allows (the ones done are kept: the next
23run finishes). Neither source whole for ``certificates.ct_degraded_after_s`` (72 h) is ``monitoring: degraded``, and the tray says so; a watch that never looked is
24``pending``, and only a complete look that is not older than two intervals is ``ok``. Nothing here ever turns an unread CT into "all clear".
25
26State, in ``tls/ct-watch.json`` (0600 in the 0700 folder, so uninstall wipes it with the keys): the alerts (kind, key hash, first seen, acknowledged), the moments of the last
27look, and what crt.sh downloads already told. It holds no secret. ``ct_watch_status()`` is the shape ``e2e_status.build(ct=...)`` writes into ``e2e-status.json``: ``alert`` is
28the tray's persistent message (an alert nobody acknowledged: ``acknowledge()``), ``alert_open`` is the fleet block's ``ct_alert`` (also true after an acknowledgement for as
29long as the certificate that raised it is still valid: a person who saw it has not made it go away).
30
31Timing comes from contracts/policy.json (``certificates.ct_check_interval_s``, ``ct_degraded_after_s``); this file only holds the pacing of its own retries.
32"""
33from __future__ import annotations
34
35import contextlib
36import hashlib
37import json
38import logging
39import random
40import re
41import time
42import urllib.parse
43from dataclasses import dataclass, field
44from typing import Any, Callable
45
46import httpx
47from cryptography import x509
48from cryptography.hazmat.primitives import hashes, serialization
49
50from . import e2e_policy as policy
51from .tls_store import TlsStore, TlsStoreError, b64url
52
53log = logging.getLogger("all_mcp.e2e_ct_watch")
54
55STATE_FILE = "ct-watch.json"
56LOCK_FILE = "ct-watch.lock"
57SCHEMA_VERSION = 1
58CERTSPOTTER_URL = "https://api.certspotter.com/v1/issuances"
59CRTSH_URL = "https://crt.sh/"
60USER_AGENT = "all-mcp-ct-selfwatch/2"
61KINDS = ("unknown_key", "bad_issuer")
62SOURCES = ("certspotter", "crtsh")
63CANARY_AFTER_S = 172800.0 # a certificate older than this is expected to show in every CT source
64CHECK_AFTER_ISSUANCE_S = 600.0 # a new certificate has reached the logs by now (the precertificate went in before the CA issued)
65FETCH_TIMEOUT_S = 20.0
66JSON_MAX_BYTES = 4 * 1024 * 1024 # a page of CT rows for one name; more than this is not read
67PEM_MAX_BYTES = 64 * 1024
68CERTSPOTTER_MAX_PAGES = 20
69CRTSH_DOWNLOADS_PER_RUN = 12
70RETRY_FIRST_S = 900.0 # a look that failed is retried after 15 min, 1 h, 4 h, 16 h, and then at the usual interval
71RETRY_FACTOR = 4.0
72INTERVAL_JITTER = 0.1 # a device looks between 0 and 10 % after the interval, so the fleet does not ask a source at the same moment
73STALE_INTERVALS = 2 # a complete look older than this many intervals is no longer "ok"
74ALERTS_KEPT = 64
75CRTSH_KEPT = 256
76_WORD = re.compile(r"[a-z0-9_]{1,48}")
77_SPKI = re.compile(r"[A-Za-z0-9_-]{43}")
78_ROW_ID = re.compile(r"[A-Za-z0-9_.-]{1,64}")
79_CRTSH_ID = re.compile(r"[0-9]{1,20}")
80_HEX_SHA256 = re.compile(r"[0-9a-f]{64}")
81_LETS_ENCRYPT = re.compile(r"let['’]s encrypt", re.IGNORECASE)
82
83
84class CtSourceError(Exception):
85 """A source of CT that could not be read whole. ``word`` is a closed word (never text of the network) the state and the status may carry."""
86
87 def __init__(self, word: str):
88 super().__init__(word)
89 self.word = word
90
91
92# ------------------------------------------------------------------------------------------------------------------------ certificates and the comparison
93
94@dataclass(frozen=True)
95class CtCert:
96 """A certificate reduced to what the comparison needs, from whichever source. ``spki`` is the base64url SHA-256 of the SubjectPublicKeyInfo, the form of ``spki-log``."""
97 key: str
98 names: tuple[str, ...]
99 spki: str
100 issuer: str
101 not_after: int | None
102 source: str
103
104
105@dataclass(frozen=True)
106class Finding:
107 kind: str
108 spki: str
109 not_after: int | None
110
111 @property
112 def key(self) -> str:
113 return f"{self.kind}:{self.spki}"
114
115
116def normalize(name: Any) -> str:
117 return name.strip().lower().rstrip(".") if isinstance(name, str) else ""
118
119
120def covers(names: tuple[str, ...] | list[str], host: str) -> bool:
121 """Whether a certificate with these names is one for ``host``: the name itself, or a wildcard that covers it (``*.zone`` covers exactly one label)."""
122 for name in names:
123 if name == host:
124 return True
125 if name.startswith("*.") and "." in host and host.split(".", 1)[1] == name[2:]:
126 return True
127 return False
128
129
130def is_lets_encrypt(issuer: str) -> bool:
131 return _LETS_ENCRYPT.search(issuer) is not None
132
133
134def evaluate(host: str, certs: list[CtCert], known: set[str]) -> list[Finding]:
135 """What is wrong among the certificates CT showed for ``host``: against the keys THIS computer made (``known``), and against the issuer. One finding per (kind, key)."""
136 found: dict[str, Finding] = {}
137 for cert in certs:
138 if not covers(cert.names, host):
139 continue
140 if cert.spki not in known:
141 finding = Finding("unknown_key", cert.spki, cert.not_after)
142 found.setdefault(finding.key, finding)
143 if not is_lets_encrypt(cert.issuer):
144 finding = Finding("bad_issuer", cert.spki, cert.not_after)
145 found.setdefault(finding.key, finding)
146 return list(found.values())
147
148
149def _seconds(text: Any) -> int | None:
150 if not isinstance(text, str):
151 return None
152 import datetime
153 candidate = text.strip().replace(" ", "T")
154 candidate = candidate if re.search(r"(Z|[+-]\d\d:?\d\d)$", candidate) else candidate + "Z"
155 try:
156 return int(datetime.datetime.fromisoformat(candidate.replace("Z", "+00:00")).timestamp())
157 except ValueError:
158 return None
159
160
161def certs_from_certspotter(rows: Any, host: str) -> list[CtCert]:
162 """Cert Spotter ``/v1/issuances`` rows (expand=dns_names, issuer). ``pubkey_sha256`` is the hex digest of the SPKI. A row that cannot be compared (no id, no key hash) makes
163 the page unusable: skipping it would be a certificate nobody looked at."""
164 if not isinstance(rows, list):
165 raise CtSourceError("bad_body")
166 certs = []
167 for row in rows:
168 if not isinstance(row, dict) or not isinstance(row.get("id"), str) or _ROW_ID.fullmatch(row["id"]) is None:
169 raise CtSourceError("bad_body")
170 digest = str(row.get("pubkey_sha256", "")).lower()
171 if _HEX_SHA256.fullmatch(digest) is None:
172 raise CtSourceError("bad_body")
173 issuer = row.get("issuer") if isinstance(row.get("issuer"), dict) else {}
174 names = tuple(normalize(name) for name in row["dns_names"] if isinstance(name, str)) if isinstance(row.get("dns_names"), list) else (host,) # asked for this name: no list, no doubt
175 certs.append(CtCert(key=str(row.get("tbs_sha256") or row["id"]), names=names, spki=b64url(bytes.fromhex(digest)),
176 issuer=" ".join(str(part) for part in (issuer.get("friendly_name"), issuer.get("name")) if isinstance(part, str)),
177 not_after=_seconds(row.get("not_after")), source="certspotter"))
178 return certs
179
180
181def crtsh_groups(rows: Any) -> list[dict[str, Any]]:
182 """crt.sh lists an issuance twice, as a precertificate and as the final certificate: same serial, same issuer, the same key. One row of each pair is enough (the rows of one
183 group are ``ids``, so a pair already read is known by either id)."""
184 if not isinstance(rows, list):
185 raise CtSourceError("bad_body")
186 groups: dict[str, dict[str, Any]] = {}
187 for row in rows:
188 if not isinstance(row, dict) or _CRTSH_ID.fullmatch(str(row.get("id", ""))) is None:
189 raise CtSourceError("bad_body")
190 serial = str(row.get("serial_number", "")).lower()
191 identity = f"{serial}|{row.get('issuer_ca_id')}" if serial else f"id|{row['id']}"
192 group = groups.setdefault(identity, {"ids": []})
193 group["ids"].append(str(row["id"]))
194 return list(groups.values())
195
196
197def read_certificate(pem: bytes) -> tuple[tuple[str, ...], str, str, int | None]:
198 """(names, SPKI hash, issuer, not_after) of a PEM certificate or precertificate. A key this cannot hash is not hidden: its hash is of the whole certificate, which no key of
199 this computer's list can equal, so it is an unknown key."""
200 try:
201 certificate = x509.load_pem_x509_certificate(pem)
202 except ValueError:
203 raise CtSourceError("bad_certificate") from None
204 try:
205 names = tuple(certificate.extensions.get_extension_for_class(x509.SubjectAlternativeName).value.get_values_for_type(x509.DNSName))
206 except x509.ExtensionNotFound:
207 names = ()
208 if not names:
209 names = tuple(attribute.value for attribute in certificate.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME) if isinstance(attribute.value, str))
210 try:
211 spki = b64url(hashlib.sha256(certificate.public_key().public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)).digest())
212 except Exception: # noqa: BLE001 - an algorithm this library does not know
213 spki = b64url(hashlib.sha256(b"unreadable-key|" + certificate.fingerprint(hashes.SHA256())).digest())
214 try:
215 not_after: int | None = int(certificate.not_valid_after_utc.timestamp())
216 except AttributeError: # cryptography < 42
217 import datetime
218 not_after = int(certificate.not_valid_after.replace(tzinfo=datetime.timezone.utc).timestamp())
219 return tuple(normalize(name) for name in names), spki, certificate.issuer.rfc4514_string(), not_after
220
221
222# ------------------------------------------------------------------------------------------------------------------------ the two sources
223
224class CtSources:
225 """The questions, and nothing else. Every request is a GET to one of two fixed addresses with the device's own name (or a crt.sh certificate number) in the query and the
226 two fixed headers below; the name is checked against the contract's pattern first. Blocking: run it in a thread."""
227
228 def __init__(self, host: str, *, transport: httpx.BaseTransport | None = None):
229 if policy.anchored("identity.fqdn_regex").fullmatch(host) is None:
230 raise ValueError("not a device name")
231 self.host = host
232 self._http = httpx.Client(timeout=httpx.Timeout(FETCH_TIMEOUT_S), headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
233 follow_redirects=False, trust_env=False, transport=transport)
234
235 def _get(self, url: str, limit: int) -> bytes:
236 try:
237 with self._http.stream("GET", url) as response:
238 status = response.status_code
239 if status == 429:
240 raise CtSourceError("rate_limited")
241 if status != 200:
242 raise CtSourceError("redirect" if 300 <= status < 400 else f"http_{status // 100}xx")
243 body = bytearray()
244 for chunk in response.iter_bytes():
245 body.extend(chunk)
246 if len(body) > limit:
247 raise CtSourceError("too_large")
248 return bytes(body)
249 except httpx.TimeoutException:
250 raise CtSourceError("timeout") from None
251 except httpx.HTTPError:
252 raise CtSourceError("unreachable") from None
253
254 def _json(self, url: str) -> Any:
255 try:
256 return json.loads(self._get(url, JSON_MAX_BYTES).decode("utf-8"))
257 except (ValueError, UnicodeError):
258 raise CtSourceError("bad_body") from None # a page cut short, an HTML error page
259
260 def certspotter(self) -> list[CtCert]:
261 """Every issuance Cert Spotter knows for this name, from the first, page after page until an empty one."""
262 certs: dict[str, CtCert] = {}
263 after = ""
264 for _ in range(CERTSPOTTER_MAX_PAGES):
265 query = [("domain", self.host), ("include_subdomains", "false"), ("match_wildcards", "true"), ("expand", "dns_names"), ("expand", "issuer")]
266 if after:
267 query.append(("after", after))
268 rows = self._json(f"{CERTSPOTTER_URL}?{urllib.parse.urlencode(query)}")
269 if not isinstance(rows, list):
270 raise CtSourceError("bad_body")
271 if not rows:
272 return list(certs.values())
273 fresh = [cert for cert in certs_from_certspotter(rows, self.host) if cert.key not in certs]
274 if not fresh:
275 return list(certs.values()) # a server that ignores `after` and repeats a page has nothing more to say
276 certs.update((cert.key, cert) for cert in fresh)
277 after = rows[-1]["id"]
278 raise CtSourceError("too_many_pages") # never compare against part of the history
279
280 def crtsh(self, cache: dict[str, dict[str, Any]]) -> tuple[list[CtCert], dict[str, dict[str, Any]], bool]:
281 """(certificates, what was downloaded this time, complete). ``cache`` is what earlier runs downloaded, by crt.sh id: only the groups not in it cost a request."""
282 groups = crtsh_groups(self._json(f"{CRTSH_URL}?{urllib.parse.urlencode([('q', self.host), ('output', 'json')])}"))
283 certs: list[CtCert] = []
284 fresh: dict[str, dict[str, Any]] = {}
285 downloads = 0
286 complete = True
287 for group in groups:
288 entry = next((cache[i] for i in group["ids"] if i in cache), None)
289 if entry is None:
290 if downloads >= CRTSH_DOWNLOADS_PER_RUN:
291 complete = False
292 continue
293 downloads += 1
294 names, spki, issuer, not_after = read_certificate(self._get(f"{CRTSH_URL}?{urllib.parse.urlencode([('d', group['ids'][0])])}", PEM_MAX_BYTES))
295 entry = {"names": list(names), "spki": spki, "issuer": issuer, "na": not_after}
296 fresh[group["ids"][0]] = entry
297 certs.append(CtCert(key=group["ids"][0], names=tuple(entry["names"]), spki=entry["spki"], issuer=entry["issuer"], not_after=entry["na"], source="crtsh"))
298 return certs, fresh, complete
299
300
301# ------------------------------------------------------------------------------------------------------------------------ the state
302
303def word(value: Any, default: str = "") -> str:
304 return value if isinstance(value, str) and _WORD.fullmatch(value) else default
305
306
307def _int(value: Any) -> int | None:
308 return value if type(value) is int and 0 <= value < 10 ** 11 else None
309
310
311def host_tag(host: str) -> str:
312 return hashlib.sha256(b"all-ct-selfwatch-v1/" + host.encode("ascii", "replace")).hexdigest()[:16]
313
314
315def empty_state(tag: str, now: float) -> dict[str, Any]:
316 return {"v": SCHEMA_VERSION, "host": tag, "started_at": int(now), "next_check_at": None, "last_attempt_at": None, "last_ok_at": None, "fail_since": None,
317 "failures": 0, "source": "", "last_error": "", "alerts": {}, "crtsh": {}}
318
319
320def clean_state(document: Any, tag: str | None, now: float) -> dict[str, Any]:
321 """The state as read, shaped (a value of the wrong type is the default, nothing else is kept). A state of another name, or one that is not a state, starts again: CT is
322 read whole every time, so what it showed is raised again by the next look."""
323 if not isinstance(document, dict) or document.get("v") != SCHEMA_VERSION or (tag is not None and document.get("host") != tag):
324 return empty_state(tag or "", now)
325 state = empty_state(str(document.get("host", "")), now)
326 for key in ("started_at", "next_check_at", "last_attempt_at", "last_ok_at", "fail_since"):
327 state[key] = _int(document.get(key))
328 state["started_at"] = state["started_at"] if state["started_at"] is not None else int(now)
329 state["failures"] = min(_int(document.get("failures")) or 0, 1000)
330 state["source"] = word(document.get("source")) if document.get("source") in SOURCES else ""
331 state["last_error"] = word(document.get("last_error"))
332 alerts = document.get("alerts")
333 for key, alert in (alerts.items() if isinstance(alerts, dict) else ()):
334 if (isinstance(alert, dict) and alert.get("kind") in KINDS and isinstance(alert.get("spki"), str) and _SPKI.fullmatch(alert["spki"])
335 and key == f"{alert['kind']}:{alert['spki']}" and _int(alert.get("raised_at")) is not None):
336 state["alerts"][key] = {"kind": alert["kind"], "spki": alert["spki"], "raised_at": alert["raised_at"], "ack_at": _int(alert.get("ack_at")),
337 "not_after": _int(alert.get("not_after"))}
338 cache = document.get("crtsh")
339 for key, entry in (cache.items() if isinstance(cache, dict) else ()):
340 if (_CRTSH_ID.fullmatch(str(key)) and isinstance(entry, dict) and isinstance(entry.get("spki"), str) and isinstance(entry.get("issuer"), str)
341 and isinstance(entry.get("names"), list) and all(isinstance(name, str) for name in entry["names"])):
342 state["crtsh"][str(key)] = {"names": entry["names"][:100], "spki": entry["spki"][:64], "issuer": entry["issuer"][:300], "na": _int(entry.get("na"))}
343 return state
344
345
346def monitoring(state: dict[str, Any], now: float) -> str:
347 """``degraded`` once the looks have failed for the policy's 72 h, ``ok`` only for a complete look that is recent and not followed by a failure, ``pending`` otherwise
348 (never looked, a catch-up after a long sleep, a retry after one failure). Never ``ok`` by default."""
349 degraded_after = policy.value("certificates.ct_degraded_after_s")
350 interval = policy.value("certificates.ct_check_interval_s")
351 if state["fail_since"] is not None and now - state["fail_since"] >= degraded_after:
352 return "degraded"
353 if state["last_ok_at"] is not None and state["fail_since"] is None and now - state["last_ok_at"] <= STALE_INTERVALS * interval:
354 return "ok"
355 return "pending"
356
357
358def summarize(state: dict[str, Any], now: float) -> dict[str, Any]:
359 """The block ``e2e_status.build(ct=...)`` writes. ``alert``: some alert is not acknowledged. ``alert_open``: some alert is not acknowledged or its certificate is still
360 valid (or its end is not known): the fleet block's ``ct_alert``."""
361 alerts = list(state["alerts"].values())
362 waiting = [alert for alert in alerts if alert["ack_at"] is None]
363 open_ = [alert for alert in alerts if alert["ack_at"] is None or alert["not_after"] is None or alert["not_after"] > now]
364 return {"monitoring": monitoring(state, now), "alert": bool(waiting), "alert_open": bool(open_), "kinds": sorted({alert["kind"] for alert in open_}),
365 "count": len(alerts), "unacknowledged": len(waiting), "since": min((alert["raised_at"] for alert in waiting), default=None),
366 "last_ok_at": state["last_ok_at"], "last_check_at": state["last_attempt_at"], "next_check_at": state["next_check_at"], "fail_since": state["fail_since"],
367 "source": state["source"], "last_error": state["last_error"]}
368
369
370def _unavailable(error: str) -> dict[str, Any]:
371 """The block when the state cannot be read (an unsafe folder, a full disk): not green, not silent."""
372 return {"monitoring": "degraded", "alert": False, "alert_open": False, "kinds": [], "count": 0, "unacknowledged": 0, "since": None, "last_ok_at": None,
373 "last_check_at": None, "next_check_at": None, "fail_since": None, "source": "", "last_error": word(error, "storage_error")}
374
375
376@dataclass
377class CheckResult:
378 """What one look did. ``ok`` is a complete comparison; ``new`` are the alerts raised by THIS look (never the ones already standing)."""
379 ok: bool
380 source: str = ""
381 error: str = ""
382 seen: int = 0
383 new: list[Finding] = field(default_factory=list)
384
385
386# ------------------------------------------------------------------------------------------------------------------------ the watch
387
388class CtWatch:
389 def __init__(self, host: str, *, store: TlsStore | None = None, sources: CtSources | None = None, clock: Callable[[], float] = time.time,
390 jitter: Callable[[], float] = random.random):
391 self.host = host
392 self.tag = host_tag(host)
393 self.store = store or TlsStore()
394 self.sources = sources or CtSources(host)
395 self.clock = clock
396 self.jitter = jitter
397
398 # -- the state
399 def _load(self) -> dict[str, Any]:
400 raw = self.store.read_file(STATE_FILE)
401 try:
402 document = json.loads(raw.decode("utf-8")) if raw else None
403 except (ValueError, UnicodeError):
404 document = None
405 return clean_state(document, self.tag, self.clock())
406
407 def _update(self, change: Callable[[dict[str, Any]], Any]) -> Any:
408 with self.store.locked(LOCK_FILE):
409 state = self._load()
410 outcome = change(state)
411 self.store.write_file(STATE_FILE, json.dumps(state, sort_keys=True, separators=(",", ":")).encode("utf-8"))
412 return outcome
413
414 def status(self) -> dict[str, Any]:
415 """The block for ``e2e-status.json``; one cheap read of the state file."""
416 try:
417 return summarize(self._load(), self.clock())
418 except TlsStoreError as error:
419 return _unavailable(error.word)
420
421 def due_in(self) -> float:
422 """Seconds until the next look (0 when it is due, or when the state cannot be read: the look itself says why)."""
423 try:
424 state = self._load()
425 except TlsStoreError:
426 return 0.0
427 now = self.clock()
428 interval = float(policy.value("certificates.ct_check_interval_s"))
429 due = state["next_check_at"]
430 if due is None:
431 return 0.0
432 return max(0.0, min(due - now, interval * (1 + INTERVAL_JITTER))) # a clock that went back cannot push the look further than one interval
433
434 def nudge(self, after_s: float = CHECK_AFTER_ISSUANCE_S) -> None:
435 """A certificate was just issued: look in ``after_s`` seconds unless a look is due sooner."""
436 def change(state: dict[str, Any]) -> None:
437 soon = int(self.clock() + after_s)
438 state["next_check_at"] = soon if state["next_check_at"] is None else min(state["next_check_at"], soon)
439 with contextlib.suppress(TlsStoreError, OSError):
440 self._update(change)
441
442 def acknowledge(self) -> int:
443 return acknowledge(self.store, self.clock)
444
445 def _canary(self) -> str | None:
446 """The key of the certificate this computer serves, once it is old enough to be in CT everywhere: a source that is read whole shows it. One that does not has an
447 incomplete view (a page that came back empty, an index that is behind), and an empty or short answer is never taken for an all-clear."""
448 try:
449 active = self.store.active()
450 except TlsStoreError:
451 return None
452 if active is None or self.clock() - active.info.not_before < CANARY_AFTER_S:
453 return None
454 return active.spki
455
456 # -- one look
457 def check(self) -> CheckResult:
458 """Ask CT for this name and compare. Blocking. Never raises for what a source or the disk does: the outcome is in the result and in the state."""
459 try:
460 known = {row["spki_sha256"] for row in self.store.spki_log() if _SPKI.fullmatch(str(row.get("spki_sha256", "")))}
461 cache = self._load()["crtsh"]
462 except TlsStoreError as error:
463 return CheckResult(False, error=word(error.word, "storage_error"))
464 canary = self._canary()
465 certs: list[CtCert] = []
466 fresh: dict[str, dict[str, Any]] = {}
467 ok, source, errors = False, "", []
468 for name in SOURCES:
469 try:
470 if name == "certspotter":
471 got, complete = self.sources.certspotter(), True
472 else:
473 got, fresh, complete = self.sources.crtsh(cache)
474 except CtSourceError as error:
475 errors.append(error.word)
476 continue
477 except Exception as error: # noqa: BLE001 - a bug in a parser is a source that cannot be read, never a crashed watch
478 log.warning("CT source %s failed: %s", name, type(error).__name__)
479 errors.append("source_error")
480 continue
481 certs.extend(got) # what any source showed is compared, even when the look as a whole is not complete
482 if not complete:
483 errors.append("incomplete")
484 elif canary is not None and canary not in {cert.spki for cert in got}:
485 errors.append("canary_missing")
486 else:
487 ok, source = True, name
488 break
489 findings = evaluate(self.host, certs, known)
490 error = "" if ok else (errors[-1] if errors else "source_error")
491 result = CheckResult(ok, source, error, len(certs or []))
492
493 def record(state: dict[str, Any]) -> None:
494 now = int(self.clock())
495 interval = float(policy.value("certificates.ct_check_interval_s"))
496 for finding in findings:
497 if finding.key not in state["alerts"]:
498 state["alerts"][finding.key] = {"kind": finding.kind, "spki": finding.spki, "raised_at": now, "ack_at": None, "not_after": finding.not_after}
499 result.new.append(finding)
500 if len(state["alerts"]) > ALERTS_KEPT: # a flood of foreign certificates cannot grow the file: what nobody has seen stays, the oldest acknowledged go
501 ranked = sorted(state["alerts"].items(), key=lambda item: (item[1]["ack_at"] is None, item[1]["raised_at"]))
502 state["alerts"] = dict(ranked[-ALERTS_KEPT:])
503 state["crtsh"] = dict(list({**state["crtsh"], **fresh}.items())[-CRTSH_KEPT:])
504 state["last_attempt_at"] = now
505 if ok:
506 state.update(last_ok_at=now, fail_since=None, failures=0, source=source, last_error="")
507 state["next_check_at"] = int(now + interval * (1 + INTERVAL_JITTER * self.jitter()))
508 else:
509 state["failures"] = min(state["failures"] + 1, 1000)
510 state["fail_since"] = state["fail_since"] if state["fail_since"] is not None else now
511 state["last_error"] = word(error, "source_error")
512 state["next_check_at"] = int(now + min(interval, RETRY_FIRST_S * RETRY_FACTOR ** (state["failures"] - 1)))
513 try:
514 self._update(record)
515 except (TlsStoreError, OSError) as failure:
516 result.ok, result.error = False, word(getattr(failure, "word", ""), "storage_error") # the alerts of this look are in the result, the caller logs them
517 for finding in result.new:
518 log.error("CT shows a certificate for this computer's name that it did not make (%s)", finding.kind)
519 return result
520
521
522def acknowledge(store: TlsStore | None = None, clock: Callable[[], float] = time.time) -> int:
523 """The person has seen the alerts: the tray's message goes away until a NEW alert is raised. Returns how many were acknowledged. Callable from any process (the tray, a
524 command): the state is changed under its own lock."""
525 store = store or TlsStore()
526 with store.locked(LOCK_FILE):
527 raw = store.read_file(STATE_FILE)
528 try:
529 document = json.loads(raw.decode("utf-8")) if raw else None
530 except (ValueError, UnicodeError):
531 document = None
532 state = clean_state(document, None, clock())
533 count = 0
534 for alert in state["alerts"].values():
535 if alert["ack_at"] is None:
536 alert["ack_at"] = int(clock())
537 count += 1
538 if count:
539 store.write_file(STATE_FILE, json.dumps(state, sort_keys=True, separators=(",", ":")).encode("utf-8"))
540 return count
541
542
543def stored_status(host: str, *, store: TlsStore | None = None, clock: Callable[[], float] = time.time) -> dict[str, Any] | None:
544 """What the state file says without a watch running (the child is idle or the kill switch is on): None when there is no state, so nothing is said that was not seen.
545 An alert that stands is shown in every state of the encrypted path."""
546 store = store or TlsStore()
547 try:
548 raw = store.read_file(STATE_FILE)
549 document = json.loads(raw.decode("utf-8")) if raw else None
550 except (TlsStoreError, ValueError, UnicodeError):
551 return None
552 if not isinstance(document, dict) or document.get("host") != host_tag(host):
553 return None
554 return summarize(clean_state(document, host_tag(host), clock()), clock())