For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.

src/all_mcp/e2e_ct_watch.py

SHA-256: 3b4d4b4cd61a575805bf7243b7ceb11cf8c421f2490a30ba6afcbfede1746de0
554 lines · 31089 bytes · View on 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
3	The 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
4	could 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
5	THIS 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
6	computer'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
11	Everything 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
12	two), so the pair is never counted twice and never alarms. The comparison does not look at serials, dates or the control plane.
13
14	What 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
15	request 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);
16	the 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).
17	Whole 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
18	that repeats, reorders or duplicates entries changes nothing.
19
20	Honesty 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
21	back 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
22	more 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
23	run 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
26	State, 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
27	look, 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
28	the 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
29	long as the certificate that raised it is still valid: a person who saw it has not made it go away).
30
31	Timing 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	"""
33	from __future__ import annotations
34
35	import contextlib
36	import hashlib
37	import json
38	import logging
39	import random
40	import re
41	import time
42	import urllib.parse
43	from dataclasses import dataclass, field
44	from typing import Any, Callable
45
46	import httpx
47	from cryptography import x509
48	from cryptography.hazmat.primitives import hashes, serialization
49
50	from . import e2e_policy as policy
51	from .tls_store import TlsStore, TlsStoreError, b64url
52
53	log = logging.getLogger("all_mcp.e2e_ct_watch")
54
55	STATE_FILE = "ct-watch.json"
56	LOCK_FILE = "ct-watch.lock"
57	SCHEMA_VERSION = 1
58	CERTSPOTTER_URL = "https://api.certspotter.com/v1/issuances"
59	CRTSH_URL = "https://crt.sh/"
60	USER_AGENT = "all-mcp-ct-selfwatch/2"
61	KINDS = ("unknown_key", "bad_issuer")
62	SOURCES = ("certspotter", "crtsh")
63	CANARY_AFTER_S = 172800.0 # a certificate older than this is expected to show in every CT source
64	CHECK_AFTER_ISSUANCE_S = 600.0 # a new certificate has reached the logs by now (the precertificate went in before the CA issued)
65	FETCH_TIMEOUT_S = 20.0
66	JSON_MAX_BYTES = 4 * 1024 * 1024 # a page of CT rows for one name; more than this is not read
67	PEM_MAX_BYTES = 64 * 1024
68	CERTSPOTTER_MAX_PAGES = 20
69	CRTSH_DOWNLOADS_PER_RUN = 12
70	RETRY_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
71	RETRY_FACTOR = 4.0
72	INTERVAL_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
73	STALE_INTERVALS = 2 # a complete look older than this many intervals is no longer "ok"
74	ALERTS_KEPT = 64
75	CRTSH_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
84	class 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)
95	class 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)
106	class 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
116	def normalize(name: Any) -> str:
117	return name.strip().lower().rstrip(".") if isinstance(name, str) else ""
118
119
120	def 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
130	def is_lets_encrypt(issuer: str) -> bool:
131	return _LETS_ENCRYPT.search(issuer) is not None
132
133
134	def 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
149	def _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
161	def 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
181	def 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
197	def 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
224	class 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
303	def word(value: Any, default: str = "") -> str:
304	return value if isinstance(value, str) and _WORD.fullmatch(value) else default
305
306
307	def _int(value: Any) -> int \| None:
308	return value if type(value) is int and 0 <= value < 10 ** 11 else None
309
310
311	def host_tag(host: str) -> str:
312	return hashlib.sha256(b"all-ct-selfwatch-v1/" + host.encode("ascii", "replace")).hexdigest()[:16]
313
314
315	def 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
320	def 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
346	def 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
358	def 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
370	def _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
377	class 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
388	class 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
522	def 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
543	def 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())