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_cert.py

SHA-256: 8626594491f9433fb48af59a6045ac632bfeafaa5d6949f16ccf8c2b296e3ba9
1530 lines · 84856 bytes · View on GitHub

1	"""ALL 1.1 end-to-end experiment (plan-all-1.1.md, 2.9.1): the device's OWN TLS certificate.
2
3	The private key is generated here, on the device, written 0600 under ``state_dir()/tls/`` and never leaves it. Only a CSR (public key and a
4	name) goes to the certificate authority, over ACME HTTP-01 (RFC 8555): the CA asks ``http://<name>/.well-known/acme-challenge/<token>`` on
5	port 80, the tunnel server forwards that plain HTTP request to this device through the tunnel, and the device answers it
6	(e2e_tunnel_v1.py, ChallengeStore). The CA never sees the key, and neither does the tunnel server.
7
8	Pieces
9	* ChallengeStore the tokens the ACME responder of the tunnel may answer (thread-safe, short-lived).
10	* AcmeClient a minimal RFC 8555 client (account, order, http-01, finalize, download) written with ``httpx`` and ``cryptography``, which
11	the agent already depends on: the experiment adds no package to uv.lock (no ``acme``, no ``josepy``).
12	* CertManager key + CSR + issue + renew + store. ``ensure()`` reuses a stored certificate until two thirds of its life have passed
13	(60 days of 90, the plan's rule: the ratio, not a day count, so shorter lifetimes keep working) and then asks for a new one.
14
15	The CA is one of: ``staging`` and ``production`` (Let's Encrypt), a directory URL (tests, Pebble), or ``selfsigned`` (a local development
16	mode with no network at all: the runner's ``--local`` and the tests). A certificate is stored together with the CA it came from, so a
17	staging certificate is never served as if it were a production one and the other way round.
18
19	ALL 2.0 adds DeviceCertManager (below): a NEW key for every issuance, kept in a generation folder (tls_store.py), DNS-01 through the control plane's
20	broker (reserve first, confirm after), ARI, the CA's own clock, rate-limit back-off, revokeCert at uninstall. HTTP-01 and CertManager stay for the
21	experiment's runner and tunnel v1 only.
22
23	Accepting the CA's terms of service is a decision of the person running the experiment: an account is created only with agree_tos=True
24	(the runner's --agree-tos). Nothing here runs at import time and nothing here touches the network unless issue() is called.
25	"""
26	from __future__ import annotations
27
28	import asyncio
29	import contextlib
30	import datetime
31	import hashlib
32	import json
33	import logging
34	import os
35	import re
36	import secrets
37	import threading
38	import time
39	from dataclasses import dataclass, replace
40	from pathlib import Path
41	from typing import Any, Callable
42	from urllib.parse import urlsplit
43	from email.utils import parsedate_to_datetime
44
45	import httpx
46	from cryptography import x509
47	from cryptography.hazmat.primitives import hashes, serialization
48	from cryptography.hazmat.primitives.asymmetric import ec, rsa
49	from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
50	from cryptography.x509.oid import NameOID
51
52	from . import state_io
53	from .config import state_dir
54	from ._device_storage import LockTimeout
55	from .tls_store import (CertInfo, Generation, TlsStore, TlsStoreError, b64url, certificate_policy, chain_matches_key, # noqa: F401 (some are re-exported)
56	inspect_chain, key_pem, leaf_der, parse_chain, policy as tls_policy, spki_sha256)
57
58	log = logging.getLogger("all_mcp.e2e_cert")
59
60	ACME_DIRECTORIES = {
61	"staging": "https://acme-staging-v02.api.letsencrypt.org/directory",
62	"production": "https://acme-v02.api.letsencrypt.org/directory",
63	}
64	SELF_SIGNED = "selfsigned"
65	USER_AGENT = "all-mcp-e2e-experiment/1.1 (+https://all-mcp.fyi)"
66	KEY_TYPES = ("ec", "rsa")
67	CHALLENGE_TTL_SECONDS = 1800.0
68	TOKEN_RE = re.compile(r"[A-Za-z0-9_-]{1,128}")
69	HOSTNAME_RE = re.compile(r"(?=.{1,253}$)([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)(\.[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)+")
70	MAX_ACME_BODY = 256 * 1024
71
72
73	class AcmeError(Exception):
74	"""The CA refused or the flow could not finish. ``type`` is the ACME problem type (urn:ietf:params:acme:error:...), when there was one."""
75
76	def __init__(self, message: str, *, type: str = "", status: int = 0, retry_after: float \| None = None):
77	super().__init__(message)
78	self.type = type
79	self.status = status
80	self.retry_after = retry_after
81
82
83	# ---------------------------------------------------------------------------------------------------------------- files, keys, CSR
84
85	def tls_dir() -> Path:
86	return state_dir() / "tls"
87
88
89	def _is_windows() -> bool:
90	return os.name == "nt"
91
92
93	def _windows_owner_only(path: Path, *, directory: bool) -> None:
94	"""On Windows ``os.chmod`` and ``mkdir(mode=)`` restrict nothing: the folder or file would keep the ACL it inherits (Users can read it). Cut the
95	inheritance and grant only the current user and SYSTEM (the service may run as either): ``icacls /inheritance:r /grant:r``. A failure is raised:
96	a private key must not be left readable because the lock-down did not work."""
97	import subprocess
98	flags = 0x08000000 # CREATE_NO_WINDOW
99	who = subprocess.run(["whoami", "/user", "/fo", "csv", "/nh"], capture_output=True, text=True, timeout=15, creationflags=flags)
100	match = re.search(r"S-1-[0-9-]+", who.stdout or "")
101	if who.returncode != 0 or not match:
102	raise OSError("could not tell which user this is, so the key's folder was not locked to it")
103	rights = "(OI)(CI)F" if directory else "F"
104	done = subprocess.run(["icacls", str(path), "/inheritance:r", "/grant:r", f"{match.group(0)}:{rights}", f"S-1-5-18:{rights}"],
105	capture_output=True, text=True, timeout=30, creationflags=flags)
106	if done.returncode != 0:
107	raise OSError(f"icacls could not lock {path.name} to its owner: {(done.stdout or done.stderr or '').strip()[:200]}")
108
109
110	def _private_dir(path: Path) -> Path:
111	"""The folder is created 0700 (on Windows: owner-only ACL); one that already exists keeps its mode (it may not be ours to change), the files inside are private anyway."""
112	made = [folder for folder in (path, *path.parents) if not folder.exists()] # mkdir(parents=True) makes tls/ as well as gen-<n>/: both are ours
113	path.mkdir(parents=True, exist_ok=True, mode=0o700)
114	for folder in reversed(made):
115	if _is_windows():
116	_windows_owner_only(folder, directory=True)
117	elif folder == path:
118	os.chmod(folder, 0o700) # mkdir's mode is cut by the umask
119	return path
120
121
122	def write_private(path: Path, data: bytes) -> None:
123	"""Write ``data`` to ``path`` atomically and readable by this user only from the first byte: the file is CREATED 0600, so there is no moment
124	in which a private key sits on disk with wider permissions (a chmod after a write would leave one)."""
125	_private_dir(path.parent)
126	temporary = path.with_name(f".{path.name}.{secrets.token_hex(4)}.tmp")
127	descriptor = os.open(temporary, os.O_WRONLY \| os.O_CREAT \| os.O_EXCL, 0o600)
128	try:
129	with os.fdopen(descriptor, "wb") as stream:
130	stream.write(data)
131	stream.flush()
132	os.fsync(stream.fileno())
133	if _is_windows():
134	_windows_owner_only(temporary, directory=False)
135	state_io.replace_retrying(temporary, path)
136	finally:
137	try:
138	os.unlink(temporary)
139	except FileNotFoundError:
140	pass
141
142
143	def ensure_private_mode(path: Path) -> None:
144	"""A key file that is readable by group or others (copied, restored from a backup, a bad umask) is tightened to 0600 and logged."""
145	if _is_windows():
146	_windows_owner_only(path, directory=False) # no mode bits to read there: set the ACL again, it is idempotent
147	return
148	mode = path.stat().st_mode & 0o777
149	if mode & 0o077:
150	log.warning("%s was readable by others (mode %o): set to 0600", path.name, mode)
151	os.chmod(path, 0o600)
152
153
154	def generate_key(key_type: str = "ec"):
155	if key_type == "ec":
156	return ec.generate_private_key(ec.SECP256R1())
157	if key_type == "rsa":
158	return rsa.generate_private_key(public_exponent=65537, key_size=2048)
159	raise ValueError(f"key type must be one of {KEY_TYPES}")
160
161
162	def load_or_create_key(path: Path, key_type: str = "ec"):
163	"""The device key. Created once (0600) and kept: a restart reuses it, so the stored certificate keeps matching it."""
164	if path.exists():
165	ensure_private_mode(path)
166	return serialization.load_pem_private_key(path.read_bytes(), password=None)
167	key = generate_key(key_type)
168	write_private(path, key_pem(key))
169	return key
170
171
172	def make_csr(key, hostname: str) -> bytes:
173	"""DER CSR for one name. It carries the public key and the name, nothing else; the CA never receives the private key."""
174	subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostname)] if len(hostname) <= 64 else [])
175	builder = (x509.CertificateSigningRequestBuilder().subject_name(subject)
176	.add_extension(x509.SubjectAlternativeName([x509.DNSName(hostname)]), critical=False))
177	return builder.sign(key, hashes.SHA256()).public_bytes(serialization.Encoding.DER)
178
179
180	def self_signed(key, hostname: str, *, days: int = 30, now: float \| None = None) -> bytes:
181	"""A self-signed certificate (PEM) for local development and tests. Never what ChatGPT trusts: only the runner's --local mode uses it."""
182	moment = datetime.datetime.fromtimestamp(now if now is not None else time.time(), datetime.timezone.utc)
183	name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostname[:64])])
184	certificate = (x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key())
185	.serial_number(x509.random_serial_number())
186	.not_valid_before(moment - datetime.timedelta(minutes=5)).not_valid_after(moment + datetime.timedelta(days=days))
187	.add_extension(x509.SubjectAlternativeName([x509.DNSName(hostname)]), critical=False)
188	.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
189	.sign(key, hashes.SHA256()))
190	return certificate.public_bytes(serialization.Encoding.PEM)
191
192
193	# ---------------------------------------------------------------------------------------------------------------- the responder's tokens
194
195	class ChallengeStore:
196	"""HTTP-01 answers the device's port-80 responder may give: token -> key authorization. Shared between the thread that runs ACME and the
197	event loop that answers the CA, so it is locked; entries die after CHALLENGE_TTL_SECONDS even if nobody removes them."""
198
199	def __init__(self, ttl: float = CHALLENGE_TTL_SECONDS, clock: Callable[[], float] = time.monotonic):
200	self._ttl = ttl
201	self._clock = clock
202	self._lock = threading.Lock()
203	self._items: dict[str, tuple[str, float]] = {}
204
205	def publish(self, token: str, key_authorization: str) -> None:
206	if not TOKEN_RE.fullmatch(token):
207	raise ValueError("invalid ACME token")
208	with self._lock:
209	self._items[token] = (key_authorization, self._clock() + self._ttl)
210
211	def unpublish(self, token: str) -> None:
212	with self._lock:
213	self._items.pop(token, None)
214
215	def get(self, token: str) -> str \| None:
216	with self._lock:
217	row = self._items.get(token)
218	if row is None:
219	return None
220	if row[1] <= self._clock():
221	del self._items[token]
222	return None
223	return row[0]
224
225	def __len__(self) -> int:
226	with self._lock:
227	return len(self._items)
228
229
230	# ---------------------------------------------------------------------------------------------------------------- ACME (RFC 8555)
231
232	def jwk_of(public_key: ec.EllipticCurvePublicKey) -> dict[str, str]:
233	numbers = public_key.public_numbers()
234	return {"crv": "P-256", "kty": "EC", "x": b64url(numbers.x.to_bytes(32, "big")), "y": b64url(numbers.y.to_bytes(32, "big"))}
235
236
237	def jwk_thumbprint(jwk: dict[str, str]) -> str:
238	"""RFC 7638: SHA-256 of the required members in lexicographic order, no whitespace."""
239	canonical = json.dumps({k: jwk[k] for k in ("crv", "kty", "x", "y")}, sort_keys=True, separators=(",", ":"))
240	return b64url(hashlib.sha256(canonical.encode("ascii")).digest())
241
242
243	def key_authorization(token: str, jwk: dict[str, str]) -> str:
244	return f"{token}.{jwk_thumbprint(jwk)}"
245
246
247	def _es256(key: ec.EllipticCurvePrivateKey, signing_input: bytes) -> bytes:
248	r, s = decode_dss_signature(key.sign(signing_input, ec.ECDSA(hashes.SHA256())))
249	return r.to_bytes(32, "big") + s.to_bytes(32, "big") # JWS wants the raw pair, not DER
250
251
252	def _retry_after(response: httpx.Response) -> float \| None:
253	"""Retry-After is a number of seconds or an HTTP date (then measured from the CA's own Date header, not from this computer's clock)."""
254	value = response.headers.get("retry-after", "").strip()
255	if value.isascii() and value.isdigit(): # str.isdigit() is also true for "²", which float() refuses
256	return float(value)
257	moment, now = _http_date(value), _http_date(response.headers.get("date", ""))
258	return max(0.0, moment - now) if moment is not None and now is not None else None
259
260
261	def _http_date(value: str) -> float \| None:
262	try:
263	return parsedate_to_datetime(value).timestamp() if value else None
264	except (TypeError, ValueError):
265	return None
266
267
268	def _problem(response: httpx.Response) -> AcmeError:
269	kind, detail = "", ""
270	try:
271	document = response.json()
272	if isinstance(document, dict):
273	kind = str(document.get("type") or "")[:200]
274	detail = str(document.get("detail") or "")[:300]
275	except ValueError:
276	pass
277	return AcmeError(f"ACME {response.status_code} {kind.rsplit(':', 1)[-1] or 'error'}: {detail or response.reason_phrase}".strip(),
278	type=kind, status=response.status_code, retry_after=_retry_after(response))
279
280
281	class AcmeClient:
282	"""RFC 8555 over a synchronous httpx client. One instance, one thread at a time (CertManager calls it from a worker thread so the event loop
283	stays free to answer the CA through the tunnel while it validates)."""
284
285	def __init__(self, directory_url: str, account_key: ec.EllipticCurvePrivateKey, *, http: httpx.Client \| None = None, email: str \| None = None,
286	agree_tos: bool = False, account_url: str \| None = None, on_account: Callable[[str], None] \| None = None,
287	sleep: Callable[[float], None] = time.sleep, clock: Callable[[], float] = time.monotonic,
288	poll_interval: float = 2.0, poll_timeout: float = 180.0, on_date: Callable[[float], None] \| None = None):
289	if urlsplit(directory_url).scheme not in {"https", "http"}:
290	raise ValueError("invalid ACME directory URL")
291	self.directory_url = directory_url
292	self._key = account_key
293	self._jwk = jwk_of(account_key.public_key())
294	self._http = http or httpx.Client(timeout=httpx.Timeout(30.0), headers={"User-Agent": USER_AGENT}, follow_redirects=False)
295	self._email = email
296	self._agree = agree_tos
297	self.account_url = account_url
298	self._on_account = on_account
299	self._sleep = sleep
300	self._clock = clock
301	self._poll_interval = poll_interval
302	self._poll_timeout = poll_timeout
303	self._on_date = on_date
304	self.last_date: float \| None = None # the CA's own clock, from the Date header of its last answer
305	# True from the moment an order of this client may have a certificate (its authorizations are valid, so the CSR is sent or the order is already valid) until
306	# the next obtain(): an order that failed before that moment can have no certificate, and the broker is told so (HttpBroker.abandon).
307	self.order_may_be_issued = False
308	self._directory: dict \| None = None
309	self._nonce: str \| None = None
310
311	# -- transport
312	def _seen(self, response: httpx.Response) -> None:
313	moment = _http_date(response.headers.get("date", ""))
314	if moment is not None:
315	self.last_date = moment
316	if self._on_date:
317	self._on_date(moment)
318
319	def _get(self, url: str) -> httpx.Response:
320	response = self._http.get(url)
321	self._seen(response)
322	if response.status_code >= 400:
323	raise _problem(response)
324	return response
325
326	def directory(self) -> dict:
327	if self._directory is None:
328	document = self._get(self.directory_url).json()
329	for name in ("newNonce", "newAccount", "newOrder"):
330	if not isinstance(document, dict) or not isinstance(document.get(name), str):
331	raise AcmeError(f"ACME directory has no {name}")
332	self._directory = document
333	return self._directory
334
335	def _fresh_nonce(self) -> str:
336	response = self._http.head(self.directory()["newNonce"])
337	self._seen(response)
338	nonce = response.headers.get("replay-nonce")
339	if response.status_code >= 400 or not nonce:
340	raise AcmeError("ACME server gave no nonce", status=response.status_code)
341	return nonce
342
343	def _post(self, url: str, payload: dict \| None, *, use_account: bool = True, accept: str \| None = None) -> httpx.Response:
344	"""A signed POST. ``payload`` None is POST-as-GET. A badNonce answer carries a fresh nonce and is retried, as the RFC asks."""
345	for attempt in range(4):
346	nonce, self._nonce = self._nonce or self._fresh_nonce(), None
347	protected: dict = {"alg": "ES256", "nonce": nonce, "url": url}
348	if use_account:
349	if not self.account_url:
350	raise AcmeError("ACME account not created yet")
351	protected["kid"] = self.account_url
352	else:
353	protected["jwk"] = self._jwk
354	protected64 = b64url(json.dumps(protected, separators=(",", ":")).encode())
355	payload64 = "" if payload is None else b64url(json.dumps(payload, separators=(",", ":")).encode())
356	signature = _es256(self._key, f"{protected64}.{payload64}".encode("ascii"))
357	body = json.dumps({"protected": protected64, "payload": payload64, "signature": b64url(signature)})
358	headers = {"Content-Type": "application/jose+json"}
359	if accept:
360	headers["Accept"] = accept
361	response = self._http.post(url, content=body, headers=headers)
362	self._seen(response)
363	if len(response.content) > MAX_ACME_BODY:
364	raise AcmeError("ACME answer too large")
365	self._nonce = response.headers.get("replay-nonce") or None
366	if response.status_code < 400:
367	return response
368	problem = _problem(response)
369	if problem.type.endswith(":badNonce") and attempt < 3:
370	continue
371	raise problem
372	raise AcmeError("ACME badNonce repeated")
373
374	# -- account
375	def ensure_account(self) -> str:
376	if self.account_url:
377	return self.account_url
378	if not self._agree:
379	raise AcmeError("the CA's terms of service were not accepted (agree_tos=False): nothing was sent to the CA")
380	payload: dict = {"termsOfServiceAgreed": True}
381	if self._email:
382	payload["contact"] = [f"mailto:{self._email}"]
383	response = self._post(self.directory()["newAccount"], payload, use_account=False)
384	location = response.headers.get("location")
385	if not location:
386	raise AcmeError("ACME account has no Location")
387	self.account_url = location
388	if self._on_account:
389	self._on_account(location)
390	return location
391
392	# -- order
393	def _poll(self, url: str, done: tuple[str, ...], what: str) -> dict:
394	deadline = self._clock() + self._poll_timeout
395	while True:
396	document = self._post(url, None).json()
397	status = document.get("status")
398	if status in done:
399	return document
400	if status == "invalid":
401	detail = ""
402	for challenge in document.get("challenges") or []:
403	error = challenge.get("error") or {}
404	if error:
405	detail = f" ({str(error.get('type', '')).rsplit(':', 1)[-1]}: {str(error.get('detail', ''))[:200]})"
406	raise AcmeError(f"ACME {what} became invalid{detail}", type="invalid")
407	if self._clock() >= deadline:
408	raise AcmeError(f"ACME {what} did not finish in {self._poll_timeout:.0f}s (status {status})", type="timeout")
409	self._sleep(self._poll_interval)
410
411	def _authorize(self, url: str, publish: Callable[[str, str], None], unpublish: Callable[[str], None], challenge_type: str = "http-01") -> None:
412	authorization = self._post(url, None).json()
413	if authorization.get("status") == "valid":
414	return
415	challenge = next((c for c in authorization.get("challenges") or [] if c.get("type") == challenge_type), None)
416	if challenge is None:
417	raise AcmeError(f"the CA offered no {challenge_type} challenge")
418	token = str(challenge.get("token") or "")
419	if not TOKEN_RE.fullmatch(token):
420	raise AcmeError("the CA sent an invalid challenge token")
421	authorized = key_authorization(token, self._jwk)
422	# http-01 publishes the key authorization itself; dns-01 publishes its SHA-256 (RFC 8555 8.4), the 43-character TXT value
423	proof = authorized if challenge_type == "http-01" else b64url(hashlib.sha256(authorized.encode("ascii")).digest())
424	try:
425	publish(token, proof)
426	self._post(str(challenge["url"]), {}) # "I am ready": the CA now looks (through the tunnel, or at the TXT record)
427	self._poll(url, ("valid",), "authorization")
428	finally:
429	unpublish(token)
430
431	def obtain(self, hostname: str, csr_der: bytes, publish: Callable[[str, str], None], unpublish: Callable[[str], None], *,
432	challenge: str = "http-01", profile: str \| None = None, replaces: str \| None = None) -> bytes:
433	"""Order, prove control of ``hostname``, finalize with the CSR and return the PEM chain (leaf first). ``publish(token, proof)`` puts the proof where
434	the CA will look (http-01: the key authorization on the responder; dns-01: the TXT value) and ``unpublish(token)`` takes it away, always.
435	``profile`` and ``replaces`` (the ARI certificate id of the certificate this order renews) go into the order when given."""
436	self.ensure_account()
437	self.order_may_be_issued = False
438	order_request: dict = {"identifiers": [{"type": "dns", "value": hostname}]}
439	if profile:
440	order_request["profile"] = profile
441	if replaces:
442	order_request["replaces"] = replaces
443	response = self._post(self.directory()["newOrder"], order_request)
444	order, order_url = response.json(), response.headers.get("location")
445	if not order_url:
446	raise AcmeError("ACME order has no Location")
447	for authorization_url in order.get("authorizations") or []:
448	self._authorize(str(authorization_url), publish, unpublish, challenge)
449	order = self._poll(order_url, ("ready", "valid"), "order")
450	self.order_may_be_issued = True
451	if order.get("status") != "valid":
452	self._post(str(order["finalize"]), {"csr": b64url(csr_der)})
453	order = self._poll(order_url, ("valid",), "order")
454	certificate_url = order.get("certificate")
455	if not certificate_url:
456	raise AcmeError("ACME order is valid but has no certificate")
457	pem = self._post(str(certificate_url), None, accept="application/pem-certificate-chain").content
458	parse_chain(pem)
459	return pem
460
461	# -- ALL 2.0
462	def sync_time(self) -> float \| None:
463	"""The CA's clock now: one HEAD on its nonce endpoint, read from the Date header. None when the CA sends no Date."""
464	self._nonce = self._fresh_nonce()
465	return self.last_date
466
467	def offered_profile(self, preferred: str, fallback: str) -> str \| None:
468	"""``preferred`` when the directory lists it among its profiles, else ``fallback`` when listed, else None (no profile in the order)."""
469	profiles = (self.directory().get("meta") or {}).get("profiles")
470	if isinstance(profiles, dict):
471	for name in (preferred, fallback):
472	if name in profiles:
473	return name
474	return None
475
476	def renewal_info(self, certificate_id: str) -> "RenewalWindow \| None":
477	"""ARI (RFC 9773): when the CA suggests renewing the certificate. None when the CA has no ARI or does not know the certificate."""
478	base = self.directory().get("renewalInfo")
479	if not isinstance(base, str):
480	return None
481	try:
482	response = self._get(base.rstrip("/") + "/" + certificate_id)
483	except AcmeError as exc:
484	if exc.status == 404:
485	return None
486	raise
487	document = response.json()
488	window = document.get("suggestedWindow") if isinstance(document, dict) else None
489	try:
490	start, end = _rfc3339(window["start"]), _rfc3339(window["end"])
491	except (TypeError, KeyError, ValueError):
492	raise AcmeError("the CA's renewal information is malformed") from None
493	if end <= start: # RFC 9773 4.2: an end that equals or precedes the start is invalid
494	raise AcmeError("the CA's renewal window ends before it starts")
495	return RenewalWindow(start=start, end=end, retry_after=_retry_after(response))
496
497	def revoke(self, certificate_der: bytes, reason: int = 5) -> None:
498	"""revokeCert signed with the ACCOUNT key (RFC 8555 7.6). Reason 5 is cessationOfOperation (uninstall). A certificate that is already revoked
499	is not an error."""
500	url = self.directory().get("revokeCert")
501	if not isinstance(url, str):
502	raise AcmeError("the CA offers no revokeCert")
503	try:
504	self._post(url, {"certificate": b64url(certificate_der), "reason": reason})
505	except AcmeError as exc:
506	if not exc.type.endswith(":alreadyRevoked"):
507	raise
508
509
510	def _rfc3339(text: str) -> float:
511	return datetime.datetime.fromisoformat(str(text).replace("Z", "+00:00")).timestamp()
512
513
514	@dataclass(frozen=True)
515	class RenewalWindow:
516	start: float
517	end: float
518	retry_after: float \| None
519
520
521	def certificate_id(pem: bytes) -> str:
522	"""The ARI certificate identifier: base64url of the issuer's key identifier (the leaf's Authority Key Identifier), a dot, and base64url of the
523	serial number as DER integer content (RFC 9773 4.1). It is the value of ``replaces`` and the tail of the renewalInfo URL."""
524	leaf = parse_chain(pem)[0]
525	try:
526	key_id = leaf.extensions.get_extension_for_class(x509.AuthorityKeyIdentifier).value.key_identifier
527	except x509.ExtensionNotFound:
528	key_id = None
529	if not key_id:
530	raise ValueError("the certificate has no Authority Key Identifier")
531	serial = leaf.serial_number
532	return f"{b64url(key_id)}.{b64url(serial.to_bytes((serial.bit_length() + 8) // 8, 'big'))}"
533
534
535	# ---------------------------------------------------------------------------------------------------------------- the device's certificate
536
537	@dataclass(frozen=True)
538	class CertState:
539	chain_path: Path
540	key_path: Path
541	info: CertInfo
542	ca: str
543	issued_now: bool
544	generation: int = 0 # the tls/gen-<n>/ it lives in (DeviceCertManager); 0 for the 1.1 flat layout
545
546	@property
547	def renew_at(self) -> float:
548	return self.info.renew_at
549
550
551	class CertManager:
552	"""LEGACY (the 1.1 experiment: one key kept for good, flat files in tls/, HTTP-01 through the tunnel; the tunnel v1 tests still build on it). The 2.0
553	device uses DeviceCertManager below, which keeps the same call names (ensure, current, due, needs_terms, issue) over generations.
554
555	Key, CSR, issue, renew and store the device's certificate. Everything lives in ``directory`` (default state_dir()/tls/):
556	device.key (0600), fullchain.pem, certificate.json (which CA, which name), acme-account-<ca>.key and .json."""
557
558	def __init__(self, hostname: str, *, ca: str = "staging", directory: Path \| None = None, key_type: str = "ec", email: str \| None = None,
559	agree_tos: bool = False, challenges: ChallengeStore \| None = None, http: httpx.Client \| None = None,
560	clock: Callable[[], float] = time.time, sleep: Callable[[float], None] = time.sleep,
561	poll_interval: float = 2.0, poll_timeout: float = 180.0):
562	hostname = hostname.strip().lower().rstrip(".")
563	if not HOSTNAME_RE.fullmatch(hostname):
564	raise ValueError("invalid host name")
565	if key_type not in KEY_TYPES:
566	raise ValueError(f"key type must be one of {KEY_TYPES}")
567	self.hostname = hostname
568	self.ca = ca
569	self.directory_url = ACME_DIRECTORIES.get(ca, ca if ca != SELF_SIGNED else "")
570	if ca != SELF_SIGNED and not self.directory_url.startswith(("https://", "http://")):
571	raise ValueError("ca must be staging, production, selfsigned or an ACME directory URL")
572	self.dir = directory or tls_dir()
573	self.key_type = key_type
574	self.email = email
575	self.agree_tos = agree_tos
576	self.challenges = challenges if challenges is not None else ChallengeStore() # an EMPTY store is falsy (__len__): never `or`
577	self._http = http
578	self._clock = clock
579	self._sleep = sleep
580	self._poll_interval = poll_interval
581	self._poll_timeout = poll_timeout
582	self.key_path = self.dir / "device.key"
583	self.chain_path = self.dir / "fullchain.pem"
584	self.meta_path = self.dir / "certificate.json"
585	self._ca_tag = ca if ca in ACME_DIRECTORIES or ca == SELF_SIGNED else "custom-" + hashlib.sha256(ca.encode()).hexdigest()[:8]
586	self._lock = threading.Lock()
587
588	# -- what is stored
589	def _stored_ca(self) -> str \| None:
590	try:
591	document = json.loads(self.meta_path.read_text(encoding="utf-8"))
592	return str(document.get("ca")) if isinstance(document, dict) else None
593	except (OSError, ValueError):
594	return None
595
596	def current(self) -> CertState \| None:
597	"""The stored certificate when it is usable now: it covers the name, matches the key, comes from the configured CA and has not expired.
598	Anything else is None (and a new one is issued)."""
599	try:
600	pem = self.chain_path.read_bytes()
601	key = serialization.load_pem_private_key(self.key_path.read_bytes(), password=None)
602	info = inspect_chain(pem)
603	if (self.hostname not in info.names or not chain_matches_key(pem, key) or self._stored_ca() != self._ca_tag
604	or not info.not_before - 600 <= self._clock() < info.not_after):
605	return None
606	ensure_private_mode(self.key_path)
607	return CertState(self.chain_path, self.key_path, info, self._ca_tag, False)
608	except (OSError, ValueError):
609	return None
610
611	def due(self, state: CertState \| None = None) -> bool:
612	state = state or self.current()
613	return state is None or self._clock() >= state.renew_at
614
615	def needs_terms(self) -> bool:
616	"""True when the next certificate needs a NEW account at the CA, i.e. someone has to accept the CA's terms first (never for selfsigned or when a
617	stored certificate is still good, and never when an account of this CA already exists here)."""
618	if self.ca == SELF_SIGNED:
619	return False
620	state = self.current()
621	if state is not None and not self.due(state):
622	return False
623	return not (self.dir / f"acme-account-{self._ca_tag}.json").exists()
624
625	# -- issuing
626	def _account(self) -> AcmeClient:
627	key_path = self.dir / f"acme-account-{self._ca_tag}.key"
628	url_path = self.dir / f"acme-account-{self._ca_tag}.json"
629	_private_dir(self.dir)
630	account_key = load_or_create_key(key_path, "ec")
631	url = None
632	try:
633	document = json.loads(url_path.read_text(encoding="utf-8"))
634	url = str(document["url"]) if isinstance(document, dict) and document.get("url") else None
635	except (OSError, ValueError, KeyError):
636	pass
637
638	def remember(location: str) -> None:
639	write_private(url_path, json.dumps({"url": location, "directory": self.directory_url}).encode())
640
641	return AcmeClient(self.directory_url, account_key, http=self._http, email=self.email, agree_tos=self.agree_tos, account_url=url,
642	on_account=remember, sleep=self._sleep, poll_interval=self._poll_interval, poll_timeout=self._poll_timeout)
643
644	def issue(self) -> CertState:
645	"""Always asks for a new certificate (ensure() decides when). The device key is created on first use and kept."""
646	with self._lock:
647	_private_dir(self.dir)
648	key = load_or_create_key(self.key_path, self.key_type)
649	if self.ca == SELF_SIGNED:
650	pem = self_signed(key, self.hostname, now=self._clock())
651	else:
652	log.info("asking %s for a certificate for %s", urlsplit(self.directory_url).netloc, self.hostname)
653	csr = make_csr(key, self.hostname)
654	try:
655	pem = self._account().obtain(self.hostname, csr, self.challenges.publish, self.challenges.unpublish)
656	except AcmeError as exc:
657	if not exc.type.endswith(":accountDoesNotExist"):
658	raise
659	# the CA no longer knows our account (Let's Encrypt staging is reset now and then): forget it and register again
660	log.warning("the CA does not know the stored account any more: registering a new one")
661	(self.dir / f"acme-account-{self._ca_tag}.json").unlink(missing_ok=True)
662	pem = self._account().obtain(self.hostname, csr, self.challenges.publish, self.challenges.unpublish)
663	info = inspect_chain(pem)
664	if self.hostname not in info.names or not chain_matches_key(pem, key):
665	raise AcmeError("the certificate does not match this device's key and name")
666	write_private(self.chain_path, pem)
667	write_private(self.meta_path, json.dumps({"ca": self._ca_tag, "hostname": self.hostname, "issued_at": int(self._clock()),
668	"not_after": int(info.not_after)}).encode())
669	return CertState(self.chain_path, self.key_path, info, self._ca_tag, True)
670
671	def ensure(self) -> CertState:
672	"""The certificate to serve with: the stored one while it is not due for renewal, otherwise a new one. A stored certificate that
673	is still valid but due is replaced only if the new one is obtained: a failed renewal raises and the old one keeps working until it expires."""
674	state = self.current()
675	if state is not None and not self.due(state):
676	return state
677	try:
678	return self.issue()
679	except Exception:
680	if state is not None:
681	expires = datetime.datetime.fromtimestamp(state.info.not_after, datetime.timezone.utc).isoformat()
682	log.warning("renewal failed; the current certificate stays in use until %s", expires)
683	raise
684
685
686	# ================================================================================================================ ALL 2.0
687	# The device's certificate manager, version 2: DNS-01 through the control plane's broker, a new key for every issuance in a generation folder
688	# (tls_store.py), ARI, the CA's clock for every decision, a persisted attempt counter with back-off, rate limits that keep the current certificate.
689	# Plan 2.0 sections 2 and 4. Every number comes from contracts/policy.json (tls_store.policy()).
690
691	BACKOFF_CEILING_S = 86400.0 # consecutive failures wait 6 h, 12 h, 24 h and then 24 h (the plan names the 6 h floor and the doubling)
692	SOFT_RETRY_FLOOR_S = 30.0 # a failure that cost nothing (no connection, a 5xx before any reservation) is retried soon: 30 s, doubling to 30 min
693	SOFT_RETRY_CEILING_S = 1800.0
694	BLOCK_CEILING_S = 7 * 86400.0 # no answer of a server can keep this device quiet for longer than this
695	CLOCK_SYNC_EVERY_S = 600.0
696	DIRECTORY_TTL_S = 86400.0
697	DAY_S = 86400.0
698	STATE_VERSION = 1
699	BROKER_BODY_MAX = 65536
700	WORD_RE = re.compile(r"[a-z_]{1,48}")
701	RESERVATION_RE = re.compile(r"[0-9a-f]{32}")
702	SPKI_RE = re.compile(r"[A-Za-z0-9_-]{43}")
703	ACME_PATH = "/v1/devices/acme/"
704
705
706	# ---------------------------------------------------------------------------------------------------------------- the broker
707
708	class BrokerError(Exception):
709	"""The broker (or the way to it) refused. ``word`` is the closed word of its error answer ("label_retired", "dns_rate_limited", ...) or one of
710	ours: broker_unreachable, credentials_unavailable, bad_answer. ``status`` is the HTTP status (0: no answer at all)."""
711
712	def __init__(self, word: str, *, status: int = 0, retry_after: float \| None = None):
713	super().__init__(word)
714	self.word = word
715	self.status = status
716	self.retry_after = retry_after
717
718	@property
719	def soft(self) -> bool:
720	"""A failure that says nothing against this device: nobody answered, the control plane is busy or down, or it asks to wait."""
721	return self.status in (0, 429) or self.status >= 500
722
723
724	@dataclass(frozen=True)
725	class Reservation:
726	id: str
727	klass: str
728	expires_at: float
729	server_time: float
730
731
732	@dataclass(frozen=True)
733	class Deferral:
734	"""Over budget is never an error: come back at ``enroll_after`` (the broker's jitter is already in it)."""
735	enroll_after: float
736	server_time: float
737
738
739	@dataclass(frozen=True)
740	class DnsAnswer:
741	name: str
742	type: str
743	ttl: int
744
745
746	class HttpBroker:
747	"""The device's side of POST /v1/devices/acme/{reserve,dns,confirm,release}. Every call carries the device's own
748	control session and proof (``headers(url, method)``, device_credentials.control_headers by default). The answers are checked against the
749	contract's shapes; an answer that does not fit is ``bad_answer``, never trusted."""
750
751	def __init__(self, base_url: str, *, headers: Callable[[str, str], dict[str, str]] \| None = None, http: httpx.Client \| None = None,
752	timeout: float = 20.0):
753	if urlsplit(base_url).scheme not in {"https", "http"}:
754	raise ValueError("invalid control plane URL")
755	self.base = base_url.rstrip("/")
756	self._headers = headers or _control_headers
757	self._http = http or httpx.Client(timeout=httpx.Timeout(timeout), headers={"User-Agent": USER_AGENT}, follow_redirects=False)
758
759	def _call(self, method: str, operation: str, body: dict \| None = None) -> dict:
760	url = f"{self.base}{ACME_PATH}{operation}"
761	try:
762	headers = {"Accept": "application/json", **self._headers(url, method)}
763	except Exception: # the credentials are not there (yet): nothing was sent, and nothing about it is worth more than a closed word
764	raise BrokerError("credentials_unavailable") from None
765	content = None
766	if body is not None:
767	headers["Content-Type"] = "application/json"
768	content = json.dumps(body, separators=(",", ":")).encode("utf-8")
769	try:
770	response = self._http.request(method, url, content=content, headers=headers)
771	except httpx.HTTPError:
772	raise BrokerError("broker_unreachable") from None
773	if len(response.content) > BROKER_BODY_MAX:
774	raise BrokerError("bad_answer", status=response.status_code)
775	try:
776	document = response.json()
777	except ValueError:
778	document = None
779	if response.status_code >= 400:
780	word = document.get("error") if isinstance(document, dict) else None
781	wait = document.get("retry_after_s") if isinstance(document, dict) else None
782	raise BrokerError(word if isinstance(word, str) and WORD_RE.fullmatch(word) else "broker_error", status=response.status_code,
783	retry_after=float(wait) if type(wait) is int else _retry_after(response))
784	if not isinstance(document, dict) or document.get("ok") is not True:
785	raise BrokerError("bad_answer", status=response.status_code)
786	return document
787
788	@staticmethod
789	def _seconds(document: dict, key: str) -> float:
790	value = document.get(key)
791	if type(value) is not int or value < 0:
792	raise BrokerError("bad_answer")
793	return float(value)
794
795	def reserve(self, spki_sha256: str, purpose: str) -> Reservation \| Deferral:
796	document = self._call("POST", "reserve", {"v": 1, "spki_sha256": spki_sha256, "purpose": purpose})
797	stamp = self._seconds(document, "server_time")
798	if document.get("status") == "deferred":
799	return Deferral(self._seconds(document, "enroll_after"), stamp)
800	reservation_id, klass = document.get("reservation_id"), document.get("class")
801	if (document.get("status") != "granted" or not isinstance(reservation_id, str) or not RESERVATION_RE.fullmatch(reservation_id)
802	or klass not in tls_policy()["enums"]["budget_class"]):
803	raise BrokerError("bad_answer")
804	return Reservation(reservation_id, klass, self._seconds(document, "expires_at"), stamp)
805
806	def _dns(self, reservation_id: str, body: dict) -> DnsAnswer:
807	document = self._call("POST", "dns", {"v": 1, "reservation_id": reservation_id, **body})
808	name, ttl = document.get("name"), document.get("ttl")
809	if not isinstance(name, str) or document.get("type") != "TXT" or type(ttl) is not int or not 1 <= ttl <= 3600:
810	raise BrokerError("bad_answer")
811	return DnsAnswer(name, "TXT", ttl)
812
813	def set_txt(self, reservation_id: str, value: str) -> DnsAnswer:
814	return self._dns(reservation_id, {"action": "set", "value": value})
815
816	def clear_txt(self, reservation_id: str) -> DnsAnswer:
817	return self._dns(reservation_id, {"action": "clear"})
818
819	def confirm(self, reservation_id: str, spki_sha256: str, not_before: int, not_after: int) -> float:
820	document = self._call("POST", "confirm", {"v": 1, "reservation_id": reservation_id, "spki_sha256": spki_sha256, "not_before": int(not_before),
821	"not_after": int(not_after)})
822	return self._seconds(document, "server_time")
823
824	def abandon(self, reservation_id: str) -> None:
825	"""The order for this reservation ended before the CSR was sent: no certificate can exist for its key, and the broker frees the slot."""
826	self._call("POST", "abandon", {"v": 1, "reservation_id": reservation_id})
827
828	def release(self, reason: str) -> None:
829	self._call("POST", "release", {"v": 1, "reason": reason})
830
831	def status(self) -> dict:
832	return self._call("GET", "status")
833
834
835	def _control_headers(url: str, method: str) -> dict[str, str]:
836	from .device_credentials import control_headers
837	return control_headers(url, method)
838
839
840	def control_plane_url(values: dict[str, str] \| None = None) -> str:
841	from .config import runtime_values
842	values = runtime_values() if values is None else values
843	return (values.get("ALL_CONTROL_PLANE_URL") or values.get("ALL_AI_PROXY_URL") or "https://api.all-mcp.fyi").rstrip("/")
844
845
846	# ---------------------------------------------------------------------------------------------------------------- the CA's clock
847
848	class CaClock:
849	"""Time as the CA tells it. The device's own clock can be a year off (a dead battery, a wrong time zone, a VM resumed from an old snapshot); a
850	certificate that "expired" on that clock must not be renewed over and over. ``observe`` takes the CA's Date header (or the broker's server_time);
851	from then on ``now`` is that moment plus the time elapsed on the MONOTONIC clock, so a later jump of the wall clock changes nothing."""
852
853	def __init__(self, wall: Callable[[], float] = time.time, monotonic: Callable[[], float] = time.monotonic):
854	self._wall = wall
855	self._monotonic = monotonic
856	self._remote: float \| None = None
857	self._at = 0.0
858
859	def observe(self, remote: float) -> None:
860	self._remote, self._at = float(remote), self._monotonic()
861
862	def observe_if_stale(self, remote: float, maximum_age: float) -> None:
863	"""A second source (the broker's server_time, whole seconds) fills in only when the CA's own Date is not recent."""
864	if self.age() >= maximum_age:
865	self.observe(remote)
866
867	@property
868	def synced(self) -> bool:
869	return self._remote is not None
870
871	def age(self) -> float:
872	"""Seconds since the last observation (infinite before the first)."""
873	return self._monotonic() - self._at if self._remote is not None else float("inf")
874
875	def now(self) -> float:
876	return self._remote + (self._monotonic() - self._at) if self._remote is not None else self._wall()
877
878	@property
879	def offset(self) -> float \| None:
880	"""The CA's time minus this computer's clock, in seconds; None while the CA has not been heard."""
881	return self.now() - self._wall() if self._remote is not None else None
882
883
884	# ---------------------------------------------------------------------------------------------------------------- the manager
885
886	@dataclass(frozen=True)
887	class PassResult:
888	"""What one pass of the manager did. ``action``: valid (nothing to do), issued, adopted (a certificate a crash left unnamed was put in service),
889	waiting (due, but not allowed yet: ``reason`` and ``wait_until``), failed (``error`` is the closed word), terms (the CA's terms are not accepted
890	yet, nothing was sent), offline (the CA could not be reached to learn the time), busy (another pass runs), unsafe (the storage is not trusted)."""
891	action: str
892	generation: int \| None = None
893	wait_until: float \| None = None
894	reason: str = ""
895	error: str = ""
896
897	@property
898	def issued(self) -> bool:
899	return self.action in ("issued", "adopted")
900
901
902	class CertUnavailable(AcmeError):
903	"""ensure() has no certificate to give: none was ever issued and the pass could not issue one now (``retry_after`` says how long to wait)."""
904
905
906	def _snake(word: str) -> str:
907	return re.sub(r"(?<!^)(?=[A-Z])", "_", word).lower()
908
909
910	def error_word(exc: BaseException) -> str:
911	"""The closed word that stands for a failure in the status file and the fleet heartbeat: never text from the network."""
912	if isinstance(exc, BrokerError):
913	return exc.word
914	if isinstance(exc, AcmeError):
915	tail = _snake(exc.type.rsplit(":", 1)[-1]) if exc.type else "acme_error"
916	return {"invalid": "validation_failed", "timeout": "acme_timeout"}.get(tail, tail if WORD_RE.fullmatch(tail) else "acme_error")
917	if isinstance(exc, TlsStoreError):
918	return exc.word
919	if isinstance(exc, httpx.HTTPError):
920	return "ca_unreachable"
921	if isinstance(exc, OSError):
922	return "storage_error"
923	return "internal_error"
924
925
926	class DeviceCertManager:
927	"""The device's TLS certificate, from first issuance to uninstall.
928
929	``run_once()`` is one decision pass, safe to call at any time, from any process (a second one at the same moment answers ``busy``). It never
930	raises for an expected condition and never touches the serving certificate unless a new one is complete and validated. The rules (plan section 4):
931
932	* the CA decides what time it is (its Date header); the device's clock only breaks ties when the CA cannot be reached;
933	* a certificate is due when ARI says so (checked twice a day), or when a third of its life is left (the backstop, with or without ARI), or when it
934	has expired;
935	* an attempt is counted (and written to disk) when a reservation is granted; at least min_attempt_gap_s between attempts, back-off doubling after
936	failures, at most max_issuances_per_day successes in 24 hours; a rate limit or an ``enroll_after`` keeps the current certificate and waits;
937	* every issuance: new EC P-256 key -> its SPKI hash reserved at the broker BEFORE the order -> DNS-01 through the broker -> chain written beside
938	the key in tls/gen-<n>/ -> validated as a pair -> pointer swapped atomically -> issuance confirmed at the broker.
939
940	Modes: ``dns-01`` (a broker is given: production), ``http-01`` (``dev_http01=True`` and no broker: the experiment's runner through its tunnel) and
941	``selfsigned`` (ca="selfsigned": local development and tests, no network).
942	"""
943
944	def __init__(self, hostname: str, *, ca: str = "staging", directory: Path \| None = None, broker=None, email: str \| None = None,
945	agree_tos: bool = False, dev_http01: bool = False, challenges: ChallengeStore \| None = None, http: httpx.Client \| None = None,
946	clock: Callable[[], float] = time.time, monotonic: Callable[[], float] = time.monotonic,
947	sleep: Callable[[float], None] = time.sleep, poll_interval: float = 2.0, poll_timeout: float = 180.0, dns_settle: float = 5.0,
948	key_type: str = "ec", faults: Callable[[str], None] \| None = None, rng: Callable[[], float] = secrets.SystemRandom().random):
949	hostname = hostname.strip().lower().rstrip(".")
950	if not HOSTNAME_RE.fullmatch(hostname):
951	raise ValueError("invalid host name")
952	if key_type != "ec":
953	raise ValueError("device keys are EC P-256")
954	self.hostname = hostname
955	self.ca = ca
956	self.directory_url = ACME_DIRECTORIES.get(ca, ca if ca != SELF_SIGNED else "")
957	if ca != SELF_SIGNED and not self.directory_url.startswith(("https://", "http://")):
958	raise ValueError("ca must be staging, production, selfsigned or an ACME directory URL")
959	if ca == SELF_SIGNED:
960	self.mode = "selfsigned"
961	elif broker is not None:
962	if dev_http01:
963	raise ValueError("a broker and HTTP-01 together: choose one (HTTP-01 is for development only)")
964	if not re.fullmatch(tls_policy()["identity"]["fqdn_regex"], hostname):
965	raise ValueError("a device host is one label under the devices zone")
966	self.mode = "dns-01"
967	elif dev_http01:
968	self.mode = "http-01"
969	else:
970	raise ValueError("a broker is required: DNS-01 through the control plane is the way a device gets its certificate "
971	"(HTTP-01 exists only for development: dev_http01=True)")
972	self.broker = broker
973	self.email = email
974	self.agree_tos = agree_tos
975	self.challenges = challenges if challenges is not None else ChallengeStore() # an EMPTY store is falsy (__len__): never `or`
976	self.store = TlsStore(directory, faults=faults)
977	self.dir = self.store.dir
978	self.clock = CaClock(clock, monotonic)
979	self._wall = clock
980	self._http = http
981	self._sleep = sleep
982	self._poll_interval = poll_interval
983	self._poll_timeout = poll_timeout
984	self._dns_settle = dns_settle
985	self._rng = rng
986	self._policy = certificate_policy()
987	self._ca_tag = ca if ca in ACME_DIRECTORIES or ca == SELF_SIGNED else "custom-" + hashlib.sha256(ca.encode()).hexdigest()[:8]
988	self._client_cache: AcmeClient \| None = None
989	self._mutex = threading.Lock() # two passes of the SAME process; across processes it is the file lock
990
991	# -- paths a consumer wants (the active generation's; before any issuance they point where the first one will not be)
992	@property
993	def chain_path(self) -> Path:
994	generation = self._active()
995	return generation.chain_path if generation else self.dir / "fullchain.pem"
996
997	@property
998	def key_path(self) -> Path:
999	generation = self._active()
1000	return generation.key_path if generation else self.dir / "device.key"
1001
1002	def _active(self) -> Generation \| None:
1003	try:
1004	return self.store.active()
1005	except TlsStoreError:
1006	return None
1007
1008	# -- the CA
1009	def _account_files(self) -> tuple[str, str]:
1010	return f"acme-account-{self._ca_tag}.key", f"acme-account-{self._ca_tag}.json"
1011
1012	def _account_url(self) -> str \| None:
1013	raw = self.store.read_file(self._account_files()[1])
1014	try:
1015	document = json.loads(raw.decode("utf-8")) if raw else None
1016	except (ValueError, UnicodeError):
1017	return None
1018	return str(document["url"]) if isinstance(document, dict) and document.get("url") else None
1019
1020	def _acme(self) -> AcmeClient:
1021	"""The CA client for this pass. The account key is created on first use (0600, in tls/) and only ever used to register and to revoke."""
1022	if self._client_cache is not None:
1023	return self._client_cache
1024	key_file, url_file = self._account_files()
1025	raw = self.store.read_file(key_file)
1026	if raw:
1027	account_key = serialization.load_pem_private_key(raw, password=None)
1028	else:
1029	account_key = generate_key("ec")
1030	self.store.write_file(key_file, key_pem(account_key))
1031
1032	def remember(location: str) -> None:
1033	self.store.write_file(url_file, json.dumps({"url": location, "directory": self.directory_url}).encode())
1034
1035	self._client_cache = AcmeClient(self.directory_url, account_key, http=self._http, email=self.email, agree_tos=self.agree_tos,
1036	account_url=self._account_url(), on_account=remember, sleep=self._sleep, poll_interval=self._poll_interval,
1037	poll_timeout=self._poll_timeout, on_date=self.clock.observe)
1038	return self._client_cache
1039
1040	def needs_terms(self) -> bool:
1041	"""True when the next certificate needs a NEW account at the CA, i.e. someone has to accept the CA's terms first."""
1042	if self.mode == "selfsigned" or self._account_url() is not None:
1043	return False
1044	return self._due_reason(self._active(), self.clock.now()) is not None
1045
1046	# -- the pass
1047	def run_once(self, *, force: bool = False) -> PassResult:
1048	"""One decision pass (see the class). ``force`` skips only the question "is it due"; every wait still applies."""
1049	if not self._mutex.acquire(blocking=False):
1050	return PassResult("busy")
1051	try:
1052	with self.store.exclusive(0.0):
1053	return self._pass(force)
1054	except LockTimeout:
1055	return PassResult("busy")
1056	except TlsStoreError as exc:
1057	if exc.word == "storage_error": # the disk, not the folder: full, read-only, an I/O error
1058	log.warning("the TLS folder could not be written (%s)", exc.word)
1059	return PassResult("failed", error=exc.word)
1060	log.warning("the TLS folder is not trusted (%s): nothing was changed", exc.word)
1061	with contextlib.suppress(Exception):
1062	self.store.write_status(self._status_document(None, {}, self.clock.now(), "unsafe", error=exc.word))
1063	return PassResult("unsafe", error=exc.word)
1064	finally:
1065	self._mutex.release()
1066
1067	def _pass(self, force: bool) -> PassResult:
1068	self._client_cache = None
1069	self.store.key_storage() # writes the descriptor on first use; refuses a descriptor that says something this version cannot honor
1070	state = self._load_state()
1071	if self.mode != "selfsigned" and self._account_url() is None and not self.agree_tos:
1072	# A new account needs the person's agreement to the CA's terms (the installer's flag): until then nothing at all is sent to the CA
1073	active = self._active()
1074	self._save(state, active, self.clock.now(), "waiting" if active is None else "valid", reason="terms_required")
1075	return PassResult("terms", generation=active.number if active else None, reason="terms_required")
1076	trusted = self._sync_clock()
1077	now = self.clock.now()
1078	adopted = self._adopt(now, state)
1079	for name in self.store.gc():
1080	log.info("removed %s", name)
1081	active = self.store.active() # storage the checks refuse raises here: nothing is repaired or replaced behind a planted link or a foreign file
1082	self._flush_confirm(state, active)
1083	if self.mode != "selfsigned" and active is not None and not trusted:
1084	self._save(state, active, now, "valid", reason="ca_unreachable")
1085	return PassResult("offline", generation=active.number, reason="ca_unreachable")
1086	if active is not None and self.mode != "selfsigned":
1087	self._refresh_ari(state, active, now)
1088	reason = "forced" if force else self._due_reason(active, now, state)
1089	if reason is None:
1090	self._save(state, active, now, "valid")
1091	return PassResult("adopted" if adopted else "valid", generation=active.number if active else None)
1092	wait = self._wait(state, now) if self.mode != "selfsigned" else None
1093	if wait is not None:
1094	self._save(state, active, now, "waiting", reason=wait[1], wait_until=wait[0])
1095	return PassResult("waiting", generation=active.number if active else None, wait_until=wait[0], reason=wait[1])
1096	return self._issue(state, active, reason, now)
1097
1098	# -- the clock
1099	def _sync_clock(self) -> bool:
1100	"""Learn the CA's time (one HEAD request) unless it was learned a moment ago. False when it could not be learned and was never learned."""
1101	if self.mode == "selfsigned" or self.clock.age() < CLOCK_SYNC_EVERY_S:
1102	return True
1103	try:
1104	self._acme().sync_time()
1105	except (AcmeError, httpx.HTTPError, ValueError) as exc:
1106	log.info("the CA's clock could not be read (%s)", error_word(exc))
1107	return self.clock.synced and self.clock.age() < CLOCK_SYNC_EVERY_S * 6
1108
1109	# -- state
1110	def _load_state(self) -> dict[str, Any]:
1111	try:
1112	state = self.store.read_state()
1113	except TlsStoreError as exc:
1114	if exc.word != "file_invalid":
1115	raise
1116	state = None
1117	corrupted = True
1118	else:
1119	corrupted = False
1120	state = dict(state) if isinstance(state, dict) and state.get("v") == STATE_VERSION else {"v": STATE_VERSION}
1121	if corrupted: # a record that cannot be read must not reset the counters to "never tried": the gap starts now
1122	state["last_attempt"] = self.clock.now()
1123	for key, default in (("attempt_count", 0), ("failures", 0), ("soft_failures", 0)):
1124	state[key] = state[key] if type(state.get(key)) is int and state[key] >= 0 else default
1125	return state
1126
1127	def _persist(self, state: dict[str, Any]) -> None:
1128	self.store.write_state(state)
1129
1130	# -- deciding
1131	def _due_reason(self, active: Generation \| None, now: float, state: dict[str, Any] \| None = None) -> str \| None:
1132	if active is None:
1133	return "none"
1134	if active.ca != self._ca_tag or self.hostname not in active.info.names:
1135	return "other_ca"
1136	if now >= active.info.not_after:
1137	return "expired"
1138	if now >= active.info.renew_at:
1139	return "backstop"
1140	ari = (state or {}).get("ari") or {}
1141	if ari.get("fp") == active.info.fingerprint and "start" in ari and now >= ari["start"] + ari["frac"] * (ari["end"] - ari["start"]):
1142	return "ari"
1143	return None
1144
1145	def _refresh_ari(self, state: dict[str, Any], active: Generation, now: float) -> None:
1146	"""Ask the CA when it suggests renewing, twice a day. A CA without ARI, a failure or a malformed answer changes nothing: the backstop stays."""
1147	ari = state.get("ari") if isinstance(state.get("ari"), dict) else {}
1148	same = ari.get("fp") == active.info.fingerprint
1149	if same and now < ari.get("next_check", 0):
1150	return
1151	if now >= active.info.not_after:
1152	return # RFC 9773 4.3: never ask about a certificate that has expired
1153	interval = DAY_S / self._policy["ari_checks_per_day"]
1154	record: dict[str, Any] = {"fp": active.info.fingerprint, "next_check": now + interval, "frac": ari["frac"] if same and "frac" in ari else self._rng()}
1155	try:
1156	window = self._acme().renewal_info(certificate_id(active.chain_pem))
1157	except (AcmeError, httpx.HTTPError, ValueError) as exc:
1158	log.info("ARI: %s", error_word(exc))
1159	record.update({key: ari[key] for key in ("checked_at", "start", "end") if same and key in ari}) # keep the last window it gave
1160	else:
1161	if window is not None:
1162	record.update(checked_at=now, start=window.start, end=window.end, next_check=now + max(interval, min(window.retry_after or 0.0, DAY_S))) # the policy's twice a day is the floor; a day the ceiling (RFC 9773 4.3.2)
1163	state["ari"] = record
1164	self._persist(state)
1165
1166	def _backoff(self, failures: int) -> float:
1167	gap = float(self._policy["min_attempt_gap_s"])
1168	return min(BACKOFF_CEILING_S, gap * 2 ** max(0, failures - 1)) if failures else gap
1169
1170	def _wait(self, state: dict[str, Any], now: float) -> tuple[float, str] \| None:
1171	"""The earliest moment an issuance is allowed again and why, or None when one is allowed now."""
1172	waits: list[tuple[float, str]] = []
1173	blocked = state.get("blocked_until")
1174	if type(blocked) in (int, float) and blocked > now:
1175	waits.append((min(float(blocked), now + BLOCK_CEILING_S), str(state.get("blocked_reason") or "blocked")))
1176	soft = state.get("soft_until")
1177	if type(soft) in (int, float) and soft > now:
1178	waits.append((min(float(soft), now + BLOCK_CEILING_S), "retry_soon"))
1179	last = state.get("last_attempt")
1180	if type(last) in (int, float):
1181	if last > now: # the clock went back: the gap starts again from now, so a wrong clock can delay but never loop
1182	last = state["last_attempt"] = now
1183	gap = self._backoff(state.get("failures", 0))
1184	if last + gap > now:
1185	waits.append((last + gap, "backoff" if state.get("failures") else "gap"))
1186	issued = [min(float(t), now) for t in state.get("issuances", []) if type(t) in (int, float) and t > now - DAY_S]
1187	if len(issued) >= self._policy["max_issuances_per_day"]:
1188	waits.append((min(issued) + DAY_S, "daily_cap"))
1189	return max(waits) if waits else None
1190
1191	# -- issuing
1192	def _issue(self, state: dict[str, Any], active: Generation \| None, reason: str, now: float) -> PassResult:
1193	self._save(state, active, now, "issuing", reason=reason)
1194	number: int \| None = None
1195	attempted = False
1196	reservation: Reservation \| None = None
1197	client: AcmeClient \| None = None
1198	try:
1199	if self.mode == "selfsigned":
1200	number, key = self.store.new_generation()
1201	chain = self_signed(key, self.hostname, now=now)
1202	profile = None
1203	else:
1204	client = self._acme()
1205	client.ensure_account() # a new-account limit (per IP) ends here, before anything was reserved
1206	number, key = self.store.new_generation()
1207	spki = spki_sha256(key.public_key())
1208	if self.mode == "dns-01":
1209	answer = self.broker.reserve(spki, "renewal" if state.get("confirmed_once") else "first")
1210	self.clock.observe_if_stale(answer.server_time, CLOCK_SYNC_EVERY_S)
1211	now = self.clock.now()
1212	if isinstance(answer, Deferral):
1213	self.store.discard(number)
1214	state["blocked_until"], state["blocked_reason"] = min(answer.enroll_after, now + BLOCK_CEILING_S), "enroll_after"
1215	self._save(state, active, now, "waiting", reason="enroll_after", wait_until=state["blocked_until"])
1216	return PassResult("waiting", generation=active.number if active else None, wait_until=state["blocked_until"], reason="enroll_after")
1217	reservation = answer
1218	client.order_may_be_issued = False
1219	state["attempt_count"] += 1
1220	state["last_attempt"] = now
1221	state["soft_failures"], state["soft_until"] = 0, None
1222	self._persist(state) # counted BEFORE the order: a crash from here on still leaves the attempt on record
1223	attempted = True
1224	chain, profile = self._order(client, key, reservation, active, now)
1225	self.store.hit("after_ca")
1226	now = self.clock.now()
1227	generation = self.store.write_chain(number, chain, self.hostname, {"ca": self._ca_tag, "profile": profile, "issued_at": int(now),
1228	"reason": reason})
1229	if not (generation.info.not_before - 600 <= now < generation.info.not_after):
1230	raise TlsStoreError("chain_invalid", "the CA's certificate is not valid now")
1231	# Written BEFORE the swap, so a crash on either side of it knows what to confirm and that the CA counted this issuance (the daily cap
1232	# errs on the side of counting an issuance that never went into service).
1233	issued = [t for t in state.get("issuances", []) if type(t) in (int, float) and now - DAY_S < t <= now]
1234	state.update(issuances=[*issued, now], failures=0, blocked_until=None, blocked_reason=None, soft_until=None, soft_failures=0, last_error=None,
1235	ari=None)
1236	if self.mode == "dns-01":
1237	state["pending_confirm"] = {"reservation_id": reservation.id, "spki": generation.spki, "gen": number,
1238	"not_before": int(generation.info.not_before), "not_after": int(generation.info.not_after)}
1239	self._persist(state)
1240	self.store.hit("before_swap")
1241	self.store.activate(number)
1242	self.store.hit("after_swap")
1243	except (AcmeError, BrokerError, TlsStoreError, httpx.HTTPError, OSError) as exc:
1244	self._abandon(reservation, client)
1245	return self._failed(state, active, exc, number, attempted)
1246	except Exception as exc: # a bug must not take the process down, and must not repeat at once either
1247	log.warning("certificate pass failed: %s", type(exc).__name__, exc_info=True)
1248	self._abandon(reservation, client)
1249	return self._failed(state, active, exc, number, attempted)
1250	# from here the new certificate is serving: confirming and the status file only, nothing below may undo it
1251	active = self._active()
1252	self._flush_confirm(state, active)
1253	self._save(state, active, now, "valid")
1254	return PassResult("issued", generation=number)
1255
1256	def _abandon(self, reservation: Reservation \| None, client: AcmeClient \| None) -> None:
1257	"""Tell the broker that the order of this reservation failed before the CSR was sent, so that the slot it keeps for an order that may have been issued is
1258	freed (a validation that failed, a CA that was down). An order that got as far as the CSR is not abandoned: a certificate may exist, and only the CT
1259	reconciliation can lift that hold. Best effort: a broker that cannot be reached leaves the slot to the reconciliation."""
1260	if reservation is None or client is None or client.order_may_be_issued:
1261	return
1262	try:
1263	self.broker.abandon(reservation.id)
1264	except BrokerError as exc:
1265	log.info("the abandoned order could not be reported (%s)", exc.word)
1266
1267	def _order(self, client: AcmeClient, key, reservation: Reservation \| None, active: Generation \| None, now: float) -> tuple[bytes, str \| None]:
1268	policy_c = self._policy
1269	profile = client.offered_profile(policy_c["profile"], policy_c["profile_fallback"])
1270	replaces = None
1271	if active is not None and active.ca == self._ca_tag and now < active.info.not_after and "renewalInfo" in client.directory():
1272	with contextlib.suppress(ValueError):
1273	replaces = certificate_id(active.chain_pem)
1274	csr = make_csr(key, self.hostname)
1275	challenge = "dns-01" if self.mode == "dns-01" else "http-01"
1276	publish, unpublish = (self._txt_hooks(reservation) if reservation is not None else (self.challenges.publish, self.challenges.unpublish))
1277	while True:
1278	try:
1279	return client.obtain(self.hostname, csr, publish, unpublish, challenge=challenge, profile=profile, replaces=replaces), profile
1280	except AcmeError as exc:
1281	if replaces and (exc.type.endswith(":alreadyReplaced") or exc.status in (400, 403, 404, 409)
1282	and not exc.type.endswith((":rateLimited", ":badNonce", ":invalidProfile", ":accountDoesNotExist"))):
1283	# alreadyReplaced (409): another order already replaces that certificate. Boulder also REFUSES the whole order (it does not fall back to a
1284	# plain one) when `replaces` is malformed, names a certificate it does not know, or one issued to another account (an account that was
1285	# registered again): all of them renew as a plain order, once.
1286	replaces = None
1287	elif profile and exc.type.endswith(":invalidProfile"):
1288	profile = None
1289	elif exc.type.endswith(":accountDoesNotExist") and self._account_url():
1290	# the CA no longer knows our account (a test CA was reset): forget it and register again, which needs the terms to be accepted
1291	log.warning("the CA does not know the stored account any more: registering a new one")
1292	self.store.write_file(self._account_files()[1], b"{}")
1293	self._client_cache = None
1294	client = self._acme()
1295	else:
1296	raise
1297
1298	def _txt_hooks(self, reservation: Reservation):
1299	"""publish/unpublish for DNS-01 through the broker. The answer is CHECKED: the broker must say it wrote a TXT at _acme-challenge.<our host>;
1300	anything else (a record of another name, another type) aborts the order and takes back what was written."""
1301	expected = "_acme-challenge." + self.hostname
1302	ttl = self._policy["txt_ttl_s"]
1303
1304	def publish(_token: str, proof: str) -> None:
1305	if not SPKI_RE.fullmatch(proof):
1306	raise BrokerError("bad_answer")
1307	answer = self.broker.set_txt(reservation.id, proof)
1308	if answer.name != expected or answer.type != "TXT" or answer.ttl != ttl:
1309	raise BrokerError("dns_mismatch")
1310	if self._dns_settle:
1311	self._sleep(self._dns_settle)
1312
1313	def unpublish(_token: str) -> None:
1314	try:
1315	self.broker.clear_txt(reservation.id)
1316	except BrokerError as exc: # the broker also clears it on confirm and when the reservation times out
1317	log.info("the TXT could not be cleared now (%s)", exc.word)
1318
1319	return publish, unpublish
1320
1321	def _failed(self, state: dict[str, Any], active: Generation \| None, exc: BaseException, number: int \| None, attempted: bool) -> PassResult:
1322	"""Whatever went wrong, the serving certificate was not touched. The generation that never served is removed, the failure is counted (when an
1323	order had been started) and the next attempt waits: at least the gap, doubling, and as long as a rate limit or the broker asks."""
1324	now = self.clock.now()
1325	word = error_word(exc)
1326	if number is not None:
1327	with contextlib.suppress(TlsStoreError):
1328	self.store.discard(number)
1329	state["last_error"] = word
1330	wait = 0.0
1331	# Boulder: a rate limit is 429 rateLimited; the per-IP request limits of its load balancers are a bare 503 with Retry-After
1332	rate_limited = isinstance(exc, AcmeError) and (exc.type.endswith(":rateLimited") or exc.status == 429
1333	or (exc.status == 503 and exc.retry_after is not None))
1334	if attempted:
1335	state["failures"] = state.get("failures", 0) + 1
1336	elif not rate_limited and not (isinstance(exc, BrokerError) and not exc.soft):
1337	state["soft_failures"] = state.get("soft_failures", 0) + 1
1338	state["soft_until"] = now + min(SOFT_RETRY_CEILING_S, SOFT_RETRY_FLOOR_S * 2 ** (state["soft_failures"] - 1))
1339	retry = getattr(exc, "retry_after", None)
1340	if rate_limited or (isinstance(exc, BrokerError) and (exc.status == 429 or exc.word in ("label_retired", "license_inactive"))):
1341	# a limit (the CA's, a new-account-per-IP one, the broker's) is a wait, never a hard failure: the certificate in use keeps serving
1342	wait = retry if retry else float(self._policy["min_attempt_gap_s"])
1343	state["blocked_until"] = now + min(wait, BLOCK_CEILING_S)
1344	state["blocked_reason"] = "rate_limited" if rate_limited else word
1345	elif isinstance(exc, BrokerError) and not exc.soft and not attempted:
1346	state["blocked_until"], state["blocked_reason"] = now + float(self._policy["min_attempt_gap_s"]), word
1347	log.info("issuance did not complete (%s)", word)
1348	waiting = self._wait(state, now)
1349	until = waiting[0] if waiting else None
1350	self._save(state, active, now, "waiting" if active is None or now < active.info.not_after else "expired", reason=word, wait_until=until)
1351	return PassResult("failed", generation=active.number if active else None, error=word, wait_until=until)
1352
1353	# -- confirming and adopting
1354	def _flush_confirm(self, state: dict[str, Any], active: Generation \| None) -> None:
1355	pending = state.get("pending_confirm")
1356	if not pending or self.broker is None:
1357	state.pop("pending_confirm", None)
1358	return
1359	if (not isinstance(pending, dict) or not isinstance(pending.get("reservation_id"), str) or not isinstance(pending.get("spki"), str)
1360	or type(pending.get("not_before")) is not int or type(pending.get("not_after")) is not int):
1361	pending = {} # a record that is not ours to read: dropped below with the ones that have nothing to confirm
1362	if active is None or pending.get("gen") != active.number:
1363	state.pop("pending_confirm", None) # the certificate it was for never went into service: there is nothing to confirm
1364	self._persist(state)
1365	return
1366	try:
1367	stamp = self.broker.confirm(pending["reservation_id"], pending["spki"], pending["not_before"], pending["not_after"])
1368	self.clock.observe_if_stale(stamp, CLOCK_SYNC_EVERY_S)
1369	except BrokerError as exc:
1370	if exc.soft:
1371	log.info("the confirmation will be sent later (%s)", exc.word)
1372	return
1373	log.warning("the broker refused the confirmation (%s)", exc.word) # an expired or closed reservation keeps its slot without it
1374	state.pop("pending_confirm", None)
1375	state["confirmed_once"] = True
1376	self._persist(state)
1377
1378	def _adopt(self, now: float, state: dict[str, Any]) -> Generation \| None:
1379	"""A crash between "chain written" and "pointer swapped" leaves a complete, valid generation that nothing names. If this device made its key and
1380	it is newer than the serving one, put it in service: the CA already counted it and the broker may already hold its reservation."""
1381	if self.mode == "selfsigned":
1382	return None
1383	known = {row["spki_sha256"] for row in self.store.spki_log()}
1384	active = self.store.active()
1385	for generation in self.store.resumable(self.hostname):
1386	if generation.ca != self._ca_tag or generation.spki not in known or generation.info.not_after <= now:
1387	continue
1388	if active is None or generation.info.not_before > active.info.not_before:
1389	self.store.activate(generation.number)
1390	log.info("generation %s was complete but not in service: now it is", generation.number)
1391	return generation
1392	break
1393	return None
1394
1395	# -- the status file
1396	def _status_document(self, active: Generation \| None, state: dict[str, Any], now: float, word: str, *, reason: str = "", error: str = "",
1397	wait_until: float \| None = None) -> dict[str, Any]:
1398	offset = self.clock.offset
1399	try:
1400	pointer = self.store.pointer()
1401	except TlsStoreError:
1402	pointer = None
1403	ari = state.get("ari") or {}
1404	document: dict[str, Any] = {
1405	"v": 1, "state": word, "reason": reason or None, "mode": self.mode, "host": self.hostname, "ca": self._ca_tag,
1406	"updated_at": int(now), "generation": active.number if active else None, "previous_generation": pointer.get("previous") if pointer else None,
1407	"served": bool(pointer and pointer["served"]), "spki_sha256": active.spki if active else None,
1408	"not_before": int(active.info.not_before) if active else None, "not_after": int(active.info.not_after) if active else None,
1409	"profile": active.meta.get("profile") if active else None,
1410	"attempts_total": state.get("attempt_count", 0), "failures": state.get("failures", 0),
1411	"last_attempt_at": int(state["last_attempt"]) if type(state.get("last_attempt")) in (int, float) else None,
1412	"next_attempt_at": int(wait_until) if wait_until else None,
1413	"last_error": error or state.get("last_error") or None,
1414	"clock_offset_s": round(offset, 1) if offset is not None else None,
1415	"clock_skew": bool(offset is not None and abs(offset) > tls_policy()["authorization"]["node_skew_max_s"]),
1416	"ari_window": [int(ari["start"]), int(ari["end"])] if "start" in ari else None,
1417	"confirm_pending": bool(state.get("pending_confirm")),
1418	}
1419	return document
1420
1421	def _save(self, state: dict[str, Any], active: Generation \| None, now: float, word: str, *, reason: str = "", wait_until: float \| None = None) -> None:
1422	self._persist(state)
1423	status = self._status_document(active, state, now, word, reason=reason, wait_until=wait_until)
1424	try:
1425	status["key_storage"] = self.store.key_storage()
1426	except TlsStoreError as exc:
1427	status["key_storage"] = {"error": exc.word}
1428	self.store.write_status(status)
1429
1430	def status(self) -> dict[str, Any] \| None:
1431	try:
1432	return self.store.read_status()
1433	except TlsStoreError:
1434	return None
1435
1436	# -- what the 1.1 runner and the tunnel client already call
1437	def current(self) -> CertState \| None:
1438	"""The serving certificate when it is usable now: it covers the name, comes from the configured CA and is within its validity."""
1439	generation = self._active()
1440	if generation is None or generation.ca != self._ca_tag or self.hostname not in generation.info.names:
1441	return None
1442	if not generation.info.not_before - 600 <= self.clock.now() < generation.info.not_after:
1443	return None
1444	return CertState(generation.chain_path, generation.key_path, generation.info, self._ca_tag, False, generation.number)
1445
1446	def due(self, state: CertState \| None = None) -> bool:
1447	return self._due_reason(self._active(), self.clock.now(), self._load_state_quiet()) is not None
1448
1449	def _load_state_quiet(self) -> dict[str, Any]:
1450	try:
1451	return self.store.read_state() or {}
1452	except TlsStoreError:
1453	return {}
1454
1455	def ensure(self, *, force: bool = False) -> CertState:
1456	result = self.run_once(force=force)
1457	state = self.current()
1458	if state is None:
1459	wait = max(0.0, (result.wait_until or 0.0) - self.clock.now()) if result.wait_until else None
1460	raise CertUnavailable(f"no certificate yet ({result.reason or result.error or result.action})", type=result.error, retry_after=wait)
1461	return replace(state, issued_now=result.issued)
1462
1463	def issue(self) -> CertState:
1464	return self.ensure(force=True)
1465
1466	def next_wake(self) -> float:
1467	"""Seconds until a pass can change anything: the earliest of the renewal moment, the next ARI check, the end of a wait, a confirmation to
1468	retry. Between 60 s and an hour, so a changed clock or a changed file is noticed."""
1469	now = self.clock.now()
1470	state = self._load_state_quiet()
1471	active = self._active()
1472	moments = [now + 3600.0]
1473	if active is not None:
1474	moments.append(active.info.renew_at)
1475	ari = state.get("ari") or {}
1476	if ari.get("fp") == active.info.fingerprint:
1477	moments.append(float(ari.get("next_check", now + 3600.0)))
1478	if "start" in ari:
1479	moments.append(ari["start"] + ari["frac"] * (ari["end"] - ari["start"]))
1480	if state.get("pending_confirm"):
1481	moments.append(now + 300.0)
1482	wait = self._wait(state, now)
1483	if wait is not None and (active is None or self._due_reason(active, now, state) is not None):
1484	moments.append(wait[0])
1485	return max(60.0, min(m for m in moments) - now) if len(moments) > 1 else 3600.0
1486
1487	# -- uninstall
1488	def uninstall(self, *, revoke: bool = True) -> dict[str, Any]:
1489	"""Plan section 4, in this order: the keys and certificates go FIRST, at once, with no network; then revokeCert, signed with the ACME account key
1490	(read into memory before the folder was removed), for every certificate that was still in service; best effort. Releasing the label and the TXT
1491	at the broker is the installer's next step (HttpBroker.release("uninstall")), after this one."""
1492	evidence: dict[str, Any] = {"wiped": False, "revoked": 0, "revoke_errors": []}
1493	leaves: list[bytes] = []
1494	account_key = account_url = None
1495	with contextlib.suppress(TlsStoreError):
1496	for number in self.store.generations():
1497	generation = self.store.generation(number)
1498	if generation is not None and generation.ca == self._ca_tag:
1499	leaves.append(leaf_der(generation.chain_pem))
1500	raw = self.store.read_file(self._account_files()[0])
1501	account_key = serialization.load_pem_private_key(raw, password=None) if raw else None
1502	account_url = self._account_url()
1503	self.store.destroy()
1504	evidence["wiped"] = not self.dir.exists() or not any(self.dir.iterdir())
1505	if not revoke or self.mode == "selfsigned" or not leaves or account_key is None or not account_url:
1506	return evidence
1507	client = AcmeClient(self.directory_url, account_key, http=self._http, account_url=account_url, sleep=self._sleep)
1508	for der in leaves:
1509	try:
1510	client.revoke(der)
1511	evidence["revoked"] += 1
1512	except (AcmeError, httpx.HTTPError) as exc:
1513	evidence["revoke_errors"].append(error_word(exc))
1514	return evidence
1515
1516
1517	async def renew_then_connect(manager: DeviceCertManager, connect: Callable[[], Any], *, wait: float \| None = None) -> PassResult:
1518	"""The order a device starts in: the certificate FIRST, the tunnel SECOND. A computer that was off for 100 days wakes with an expired certificate; the
1519	tunnel can be up in a second, but ChatGPT would meet a certificate that fails, so the pass runs first (at most ``wait`` seconds, by default
1520	policy.certificates.wait_after_install_s) and the tunnel connects afterwards whatever the pass said: a certificate that is not due, one that is
1521	waiting for a rate limit and one that could not be issued still leave the connection worth making (the control channel is not the certificate)."""
1522	limit = float(tls_policy()["certificates"]["wait_after_install_s"]) if wait is None else wait
1523	try:
1524	result = await asyncio.wait_for(asyncio.to_thread(manager.run_once), limit)
1525	except asyncio.TimeoutError:
1526	result = PassResult("failed", error="pass_timeout")
1527	outcome = connect()
1528	if asyncio.iscoroutine(outcome):
1529	await outcome
1530	return result