-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbox_selfcheck.py
More file actions
1873 lines (1686 loc) · 82.4 KB
/
Copy pathbox_selfcheck.py
File metadata and controls
1873 lines (1686 loc) · 82.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""Runs ON the box. Asserts the shop's invariants and publishes a verdict the world can fetch.
WHY THE DIRECTION IS INVERTED, because this is the design decision worth defending.
The obvious shape is a CI job that reaches into the box and inspects it. That shape is impractical
here, and the reasons are measured rather than assumed: port 22 is blocked network-wide from the
operator's location (control: `github.com:22` times out while `:443` opens in about a second), and
Compute Instance Run Command reports `desired-state: ENABLED` while being absent from the live
plugin list, so the agent never instantiated it.
NOT every inbound route is shut, and that distinction matters because the opposite claim sends a
reader away from the one that works. OCI Bastion plus Cloud Shell REACHES the node: Bastion runs,
it accepts the ed25519 key the node authorises (the RSA-only constraint belongs to Cloud Shell's
FIPS OpenSSH, not to Bastion), and sessions have been created and used through it. What that route
is not is CHEAP or unattended: it needs a browser, a session that expires, and a human-ish hop.
So the box checks ITSELF and pushes the verdict outward through the Cloudflare tunnel it already
runs for the x402 gate. Nothing needs to reach in, and a checker anywhere fetches one JSON
document over plain HTTPS.
THAT FETCHER DOES NOT EXIST YET. `deploy/make_invariants.py` writes the manifest this file
consumes, and this file computes a verdict, but nothing schedules the run and nothing retrieves
the result: `box_selfcheck` appears in zero of the six workflows and in no unit here. Until both
halves exist the inversion is a design that works when invoked by hand rather than a gate, and
describing it otherwise is the failure it was written to prevent.
That inversion is strictly better than the inbound design, not merely a workaround for a blocked
port: a verdict computed on the box can see the deployed bytes, the live memory and the running
services, which an external prober cannot see at all.
THE THREE TIERS, because they need genuinely different mechanisms and conflating them is what
produced tonight's drift.
CODE AND SKILLS must be byte-identical to a named commit. Compared by sha256 against the
manifest written at deploy time. A hand-edit is drift by definition.
CONFIG is SUPPOSED to differ, because the box holds real endpoints and real keys.
Diffing it produces noise that trains people to ignore the checker. Only the
network-bearing fields are asserted.
STATE cannot be synced at all. brain.db is supposed to diverge. What it must never
do is carry a funds-critical constant, so the assertion is a PROHIBITION on
content rather than a comparison against anything.
FAILS CLOSED, deliberately, and this is the one place that choice is not obvious. A missing
invariants file, an unreadable skill or an absent manifest all return NOT OK rather than being
skipped. A checker that cannot see its subject and reports green is worse than no checker, because
the green is then quoted as evidence. Every check that cannot run says so in its own detail line.
SELF-TEST. `--self-test` builds a synthetic tree where every invariant is violated and requires
each check to FAIL, then a clean tree and requires each to PASS. A checker never shown to fail has
not been shown to work, and this one exists precisely because three green checkers missed an
eleven-day skill drift.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
ZC = Path(os.environ.get("ZEROCLAW_HOME", str(Path.home() / ".zeroclaw")))
# Written by `deploy/make_invariants.py` from the repo, then copied here at deploy time. Absent
# means we cannot know what SHOULD be here, which is a failure rather than a skip -- and it was
# absent for this file's whole life until that generator existed, so every check below was
# unrunnable on the box whatever its logic said.
INVARIANTS = ZC / "SHOP-INVARIANTS.json"
DEPLOYED_SHA = ZC / "DEPLOYED_SHA"
VERDICT_DEFAULT = ZC / "state" / "box-selfcheck.json"
# A base58 Solana address, used to find any mint-shaped token in a file. Deliberately broad: the
# check is "is there an address here that is not the configured one", so over-matching is safe and
# under-matching is not.
B58 = re.compile(r"\b[1-9A-HJ-NP-Za-km-z]{32,44}\b")
# How many mint-scan findings the detail line SAMPLES. The line is served publicly and read in a
# terminal, so it cannot carry an unbounded list. The cap is a sample size, never the count: the
# total is always reported alongside, because a bounded list without its total is unmeasurable and
# reads as complete.
MINT_FINDINGS_SHOWN = 8
# A line that FORBIDS a network necessarily names it, so the network-prose check has to tell a
# prohibition apart from an assertion to the customer. Deliberately a small, boring marker set:
# this is a semantic distinction and regex does those badly, so the honest posture is a narrow
# list plus a stated ceiling rather than a clever pattern.
#
# CEILING, so nobody reads a green as stronger than it is: a prohibition phrased with none of these
# markers still reads as an assertion and FALSE-POSITIVES, and an assertion that happens to carry
# one of them elsewhere in the same line FALSE-NEGATIVES. The strong instrument is the emitted
# message, not the template; this check is the cheap file-side backstop for the case where the
# value is right and the sentence under it is wrong.
PROHIBITION_RE = re.compile(
r"\b(never|not|no longer|don'?t|do NOT|must not|cannot|can'?t|avoid|forbid\w*|"
r"prohibit\w*|refus\w*|reject\w*|wrong|stale|incorrect|instead of|rather than|"
r"no longer accurate|used to)\b",
re.IGNORECASE,
)
def sha256_file(p: Path) -> str:
h = hashlib.sha256()
with p.open("rb") as fh:
for chunk in iter(lambda: fh.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def load_invariants() -> dict | None:
try:
return json.loads(INVARIANTS.read_text(encoding="utf-8"))
except Exception:
return None
class Result:
def __init__(self) -> None:
self.checks: list[dict] = []
# Structured payload a check wants PUBLISHED alongside its verdict, keyed by name.
# A detail line is one string read in a terminal; a unit's definition is a nested
# document a reviewer greps. Carrying it here rather than stuffing it into `detail`
# keeps `ok` computed from checks alone and keeps redaction in one place
# (`build_verdict` walks this too, so nothing reaches disk unredacted).
self.data: dict = {}
def add(self, name: str, ok: bool, detail: str) -> None:
self.checks.append({"name": name, "ok": bool(ok), "detail": detail})
def attach(self, name: str, payload: object) -> None:
self.data[name] = payload
@property
def ok(self) -> bool:
return bool(self.checks) and all(c["ok"] for c in self.checks)
def is_commit_sha(value: object) -> bool:
"""A full 40-character hex commit id, and nothing that merely resembles one.
A LENGTH TEST IS NOT ENOUGH HERE, and the counterexample is already in this repo:
`make_invariants.py` writes the literal string `"unknown"` into `repo_commit` when
`git rev-parse HEAD` fails, and `"unknown"` is exactly seven characters. A `len >= 7` guard
therefore accepts the one value that means "no commit at all" and republishes it as a
corroborated vintage, which is the exact failure this whole check exists to remove, wearing
a different disguise. Requiring 40 hex digits rejects it on both counts.
"""
return (
isinstance(value, str)
and len(value) == 40
and all(c in "0123456789abcdef" for c in value.lower())
)
def check_vintage_agreement(inv: dict, r: Result) -> None:
"""TWO RECORDS OF THE DEPLOY VINTAGE EXIST. Assert they still agree.
`SHOP-INVARIANTS.json` carries `repo_commit`, written by `make_invariants.py` from
`git rev-parse HEAD` in the SAME run that computed every hash in `files`. It is therefore
the authoritative vintage by construction: it names the commit the manifest actually
verifies against. `DEPLOYED_SHA` is a separate file that nothing in this repo writes, so it
is maintained by hand at deploy time.
A hand-maintained label beside a generated one drifts, and this one did. On 2026-08-16 the
box served `deployed_sha` from 2026-08-06 while the files it was running had been deployed
on 2026-08-15: the manifest was green, correctly, because the hashes had been regenerated,
and the published label was nine days stale. Every reader of that endpoint, human or gate,
was handed the wrong baseline, and the correct one was sitting unread in the same directory.
Reported rather than silently preferred, because the two disagreeing is itself the finding:
it means a deploy updated one record and not the other, and whatever did that will do it
again. `repo_commit` is what `deployed_sha` now carries, so a consumer gets the value the
hashes belong to.
"""
generated = inv.get("repo_commit")
label = None
try:
label = DEPLOYED_SHA.read_text(encoding="utf-8").strip() or None
except OSError:
label = None
if not is_commit_sha(generated):
r.add(
"deploy-vintage",
False,
f"the invariants file carries no usable repo_commit ({generated!r}), so the "
f"commit its hashes belong to cannot be named",
)
return
# A DIRTY TREE AT GENERATION MEANS THE COMMIT DOES NOT IDENTIFY THE CONTENT, which
# `make_invariants.py` already warns about at generation time and nothing carried onto the
# box. The hashes are still right, because they were taken from the files themselves; what is
# wrong is the NAME attached to them, which is this check's whole subject.
if inv.get("repo_dirty") is True:
r.add(
"deploy-vintage",
False,
f"repo_commit {generated[:12]} was generated from a DIRTY tree, so it names a "
f"commit whose content is not what was deployed. The hashes are still authoritative; "
f"the commit is not. Redeploy from a clean checkout.",
)
return
if label is None:
# Not a failure. The generated record is the authoritative one and it is present; the
# hand file is the redundant copy, and its absence removes the thing that can drift.
r.add(
"deploy-vintage",
True,
f"repo_commit {generated[:12]}; no hand-written DEPLOYED_SHA to disagree with it",
)
return
agree = label.startswith(generated[:12]) or generated.startswith(label[:12])
r.add(
"deploy-vintage",
agree,
f"repo_commit {generated[:12]} and DEPLOYED_SHA {label[:12]} agree"
if agree
else (
f"repo_commit {generated[:12]} but DEPLOYED_SHA says {label[:12]}. The hashes "
f"below belong to the FIRST; the second is a hand-written label that a deploy "
f"updated the files without updating. Trust repo_commit and correct the file."
),
)
def check_manifest(inv: dict, r: Result) -> None:
"""CODE AND SKILLS: every tracked file byte-identical to the commit it was deployed from."""
files = inv.get("files") or {}
if not files:
r.add(
"manifest", False, "invariants file lists no files, so nothing was compared"
)
return
bad = []
for rel, want in sorted(files.items()):
p = ZC / rel
if not p.is_file():
bad.append(f"{rel}: MISSING")
continue
got = sha256_file(p)
if got != want:
bad.append(f"{rel}: {got[:12]} != {want[:12]}")
r.add(
"manifest",
not bad,
f"{len(files)} file(s) compared; "
+ ("all match" if not bad else "; ".join(bad)),
)
def _read_target(rel: str, p: Path, findings: list[str]) -> str | None:
"""Bytes of one scan target as text, or None with the reason recorded as a finding.
Binary-safe by design: decoding with replacement makes brain.db scannable the same way
strings(1) would scan it, which also reaches freelist pages and deleted rows that a SQL
query over the live tables cannot see. That reach is the reason this stays a byte scan
rather than becoming a set of sqlite queries.
"""
if not p.is_file():
findings.append(f"{rel}: MISSING")
return None
try:
return p.read_bytes().decode("utf-8", "replace")
except Exception as exc:
findings.append(f"{rel}: unreadable ({exc})")
return None
def check_mint_prohibition(inv: dict, r: Result) -> None:
"""No wrong mint may reach a customer, asserted with a DIFFERENT POLARITY PER TIER.
THE TWO TIERS ARE NOT THE SAME PROBLEM, and running one mechanism across both is what made
this the only permanently-red check on the box.
DEPLOYED FILES (`mint_scan`) skills, SOPs and scripts. We write them, we hash them, and
measurement across every tracked file under skills/ and
sops/ found ZERO foreign base58 tokens outside test
fixtures. So "any address that is not the configured one"
is affordable here and it is the strong form: it catches a
mint nobody has ever seen, including a typo and a
substituted merchant. Unchanged.
AGENT STATE (`state_scan`) brain.db. Conversational memory legitimately accumulates
arbitrary addresses, because a Solana Pay reference key is
a FRESH RANDOM ADDRESS PER ORDER and the agent records it.
An allowlist here is unbounded BY CONSTRUCTION: the set of
legitimate tokens grows with every sale, so the check can
only ever go redder. Measured 2026-08-16 on the live box:
27 distinct unexpected tokens, each appearing exactly ONCE
in real content (the 54 occurrences are the memories_fts
index mirroring memories, and this scanner already dedupes
per file with set(), so the mirror was never the noise).
Every one was a one-off reference key.
So state is asserted as a PROHIBITION ON KNOWN-BAD VALUES, which is what this check has been
named all along. The list is `retired_mints`: finite, auditable, derived by the generator
from the same mint-to-network table that refuses to guess a network, so a mint cannot be
retired in one place and live in the other.
WHAT THIS GIVES UP, stated rather than hidden: an UNKNOWN wrong mint sitting only in agent
memory is no longer caught. That is a real reduction and it is the price of the check being
able to pass at all. It is bounded by what stayed strong: the deployed files keep the
allowlist, `code-pins` still requires both constants in the script that builds every link,
and `network-prose` still reads the sentence under the value. A mint the agent invents has
to survive pay_link.py's pin before it reaches a customer.
THE CONTROL THAT DECIDED THE DESIGN: the 2026-08-06 incident must still be caught. That
incident's mint is the devnet USDC mint, it is retired, and a brain.db row carrying it is
flagged by the denylist exactly as it was by the allowlist. Both directions are driven in
--self-test against a synthetic sqlite database with the live table shape.
FAILS CLOSED on an empty denylist while state targets exist: a prohibition with nothing to
prohibit reads every database as clean, which is the silent green this whole file exists to
refuse.
"""
mint = inv.get("mint")
merchant = inv.get("merchant")
if not mint:
r.add(
"mint-prohibition",
False,
"no mint configured, so nothing could be asserted",
)
return
allowed = {mint, merchant} | set(inv.get("allowed_addresses") or [])
allowed.discard(None)
known_other = set(inv.get("known_other") or [])
# Retired values are compared case-sensitively and exactly; base58 is case-significant.
retired = {t for t in (inv.get("retired_mints") or []) if t}
retired.discard(mint)
file_targets = list(inv.get("mint_scan") or [])
state_targets = list(inv.get("state_scan") or [])
if not file_targets and not state_targets:
r.add("mint-prohibition", False, "no scan targets configured")
return
if state_targets and not retired:
r.add(
"mint-prohibition",
False,
f"{len(state_targets)} state target(s) configured with an EMPTY retired_mints "
"list, so the prohibition would read every database as clean",
)
return
findings: list[str] = []
scanned = 0
for rel in file_targets:
text = _read_target(rel, ZC / rel, findings)
if text is None:
continue
scanned += 1
for tok in set(B58.findall(text)):
if tok in allowed or tok in known_other:
continue
findings.append(f"{rel}: unexpected address {tok[:10]}..")
for rel in state_targets:
text = _read_target(rel, ZC / rel, findings)
if text is None:
continue
scanned += 1
present = set(B58.findall(text))
for tok in sorted(retired & present):
findings.append(f"{rel}: RETIRED mint {tok[:10]}.. present in agent state")
# THE CAP MUST NAME WHAT IT DROPPED. This listed the first 8 findings and never the total, so
# a reader could not tell 8 from 8,000 and the check was permanently unmeasurable: the question
# "is this scan noisy in steady state" had no answer available from its own output, and the
# verdict file carries the same string, so reading it on the box gave the same 8. A silent
# truncation reads as completeness. The count goes first, before the sample, because the count
# is the part a decision is made on.
uniq = sorted(set(findings))
if not uniq:
detail = (
f"{scanned} target(s) scanned "
f"({len(file_targets)} file allowlist, {len(state_targets)} state denylist over "
f"{len(retired)} retired mint(s)); nothing prohibited found"
)
else:
shown = uniq[:MINT_FINDINGS_SHOWN]
more = len(uniq) - len(shown)
detail = (
f"{scanned} target(s) scanned; {len(uniq)} finding(s): "
+ "; ".join(shown)
+ (f"; ... and {more} more not shown" if more else "")
)
r.add("mint-prohibition", not findings, detail)
def check_network_prose(inv: dict, r: Result) -> None:
"""STATE-adjacent: the skill must not tell a customer the wrong network.
Separate from the mint check on purpose. On 2026-08-06 the mint was right and the sentence
under it said devnet, so a value-only check reported clean while the customer was misinformed.
"""
want = (inv.get("network") or "").lower()
forbid = {"devnet", "testnet", "localnet"} - {want}
if not want:
r.add("network-prose", False, "no network configured")
return
bad = []
checked = 0
for rel in inv.get("prose_scan") or []:
p = ZC / rel
if not p.is_file():
bad.append(f"{rel}: MISSING")
continue
checked += 1
text = p.read_text(encoding="utf-8", errors="replace")
for lineno, line in enumerate(text.splitlines(), 1):
low = line.lower()
hits = [w for w in sorted(forbid) if w in low]
if not hits:
continue
# A PROHIBITION HAS TO NAME WHAT IT FORBIDS, so a whole-file count of the forbidden
# word scores the CORRECTED file worse than one that never mentioned the hazard. That
# is not a strict check, it is an inverted one: this gate was red before the fix, red
# after, red forever, and `Result.ok` is all-of so it pinned the entire verdict to
# DRIFTED. Skipping prohibition lines is what makes the check able to pass at all.
if PROHIBITION_RE.search(line):
continue
bad.append(f"{rel}:{lineno}: {', '.join(repr(w) for w in hits)}")
r.add(
"network-prose",
not bad,
f"{checked} file(s) checked against network={want}; "
+ ("clean" if not bad else "; ".join(bad)),
)
def check_pins(inv: dict, r: Result) -> None:
"""CODE: the last script before a customer sees an address must carry both pins."""
bad = []
checked = 0
for rel in inv.get("pinned_scripts") or []:
p = ZC / rel
if not p.is_file():
bad.append(f"{rel}: MISSING")
continue
checked += 1
src = p.read_text(encoding="utf-8", errors="replace")
for field, value in (
("merchant", inv.get("merchant")),
("mint", inv.get("mint")),
):
if not value or value not in src:
bad.append(f"{rel}: {field} pin absent")
r.add(
"code-pins",
not bad and checked > 0,
f"{checked} script(s) checked; "
+ ("both pins present" if not bad else "; ".join(bad)),
)
# --------------------------------------------------------------------------------------------
# UNIT DEFINITIONS. Published so reviewing a unit never requires a shell on the box.
# --------------------------------------------------------------------------------------------
#
# THE GAP THIS CLOSES. `deploy/deploy-targets.json` names six units. Two of them,
# `zc-announce` and `zc-selfcheck`, are committed here and reviewable by anyone with the repo.
# The other four are not, and one of those four is `x402-feed-gate.service` -- the unit for the
# component that takes money. Its ExecStart, the account it runs as, and the environment it
# loads exist only on the box, so nobody can review them and no gate here can see them drift.
#
# THE DIRECTION IS THE SAME INVERSION THIS FILE ALREADY ARGUES FOR. Reading those units means a
# shell, and every remote-hands route into the node is currently shut: Cloud Shell is over its
# monthly tenancy limit, outbound 22 is blocked from the operator's network, Bastion rides SSH
# and lands back on that block, and Run Command is absent from the node's agent plugins. So the
# box PUBLISHES its own unit definitions through the tunnel it already runs, and review becomes
# one HTTPS fetch rather than an interactive session nobody can currently open.
#
# STRUCTURE IS PUBLISHED AND VALUES ARE NOT, because the verdict is served publicly at
# `/selfcheck` and a unit file can carry a credential in an `Environment=` line. The split:
#
# published ExecStart and the other Exec* lines, User, Group, WorkingDirectory, Type,
# Restart, the hardening directives, the timer schedule, the ordering deps.
# names only `Environment=` contributes VARIABLE NAMES and never a value.
# path only `EnvironmentFile=` contributes the PATH. Its contents are never read; this
# file does not open it, which is stronger than reading and filtering it.
# dropped everything else, counted but not emitted.
#
# AN ALLOWLIST RATHER THAN A DENYLIST, and the concrete reason is `SetCredential=name:value`.
# It carries a secret separated by a COLON, so every value-stripping rule keyed on `=` misses it
# entirely and a denylist that nobody thought to extend would ship it to a public URL. Under an
# allowlist an unrecognised directive is invisible by default and the failure is a reviewer
# asking for one more field, which is recoverable. `LoadCredential=` is left out for the same
# reason even though it is path-shaped. The cost of this choice is real and stated: a directive
# worth publishing stays dark until someone adds it here, and the dropped COUNT is what makes
# that visible rather than silent.
UNIT_DIRECTIVES_PUBLISHED = frozenset(
{
# [Unit]
"Description",
"Documentation",
"After",
"Before",
"Requires",
"Wants",
"PartOf",
"BindsTo",
"ConditionPathExists",
# [Service] identity and execution -- the custody-relevant half.
"Type",
"User",
"Group",
"WorkingDirectory",
"ExecStart",
"ExecStartPre",
"ExecStartPost",
"ExecReload",
"ExecStop",
"ExecStopPost",
"Restart",
"RestartSec",
"TimeoutStartSec",
"TimeoutStopSec",
"RuntimeMaxSec",
"StandardOutput",
"StandardError",
"SyslogIdentifier",
"PassEnvironment",
# [Service] hardening. None of these can hold a secret and all of them are exactly
# what a security reviewer opens the unit to find out.
"NoNewPrivileges",
"PrivateTmp",
"PrivateDevices",
"ProtectSystem",
"ProtectHome",
"ProtectKernelTunables",
"ProtectControlGroups",
"ReadWritePaths",
"ReadOnlyPaths",
"InaccessiblePaths",
"CapabilityBoundingSet",
"AmbientCapabilities",
"RestrictAddressFamilies",
"SystemCallFilter",
"MemoryMax",
"LimitNOFILE",
# [Timer]
"OnCalendar",
"OnBootSec",
"OnStartupSec",
"OnUnitActiveSec",
"OnActiveSec",
"AccuracySec",
"RandomizedDelaySec",
"Persistent",
"Unit",
# [Install]
"WantedBy",
"RequiredBy",
"Also",
}
)
UNIT_DIRECTIVES_NAMES_ONLY = frozenset({"Environment"})
UNIT_DIRECTIVES_PATH_ONLY = frozenset({"EnvironmentFile"})
# systemd's own rule for an environment variable name. A token that does not match is not a
# name, so it is DROPPED rather than published: the alternative is emitting a fragment of a
# value under the label "name", which is the leak this whole split exists to prevent.
ENV_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
# Caps, following this file's existing rule for the mint scan: a bounded list ALWAYS reports its
# true total, because a sample without its denominator reads as complete and is unmeasurable.
UNIT_DIRECTIVES_SHOWN = 40
UNIT_ENV_NAMES_SHOWN = 24
# A `key=value` token anywhere inside a published directive value. Deliberately keyed on the
# `=` alone rather than on a key that looks secret-shaped: a denylist of secret-looking keys
# fails open on the first name nobody predicted, and `ExecStart` is the one published directive
# whose value is attacker-shaped in the sense that matters -- it can carry an inline environment
# assignment (`/usr/bin/env API_TOKEN=... prog`) or a credential flag (`--token=...`).
#
# THE COST IS STATED RATHER THAN HIDDEN: a benign `--port=8402` publishes as `--port=<redacted>`,
# so a reviewer sees which flags exist and not what they are set to. That is a question they can
# ask; a leaked key is not a question anyone gets to ask afterwards. A value worth reviewing
# belongs in the write-up or in a positional argument, both of which survive this untouched.
ASSIGNMENT_TOKEN_RE = re.compile(r"^([^\s=]+)=(.+)$")
def scrub_assignment_values(value: str) -> str:
"""Replace the value half of every `key=value` token, keeping the key.
SHLEX, NOT WHITESPACE, and this reverses a deliberate earlier choice whose stated reason was
measured to be false. That reason was: "splitting on whitespace can only ever produce MORE
tokens, and more tokens means more redaction, so the conservative direction is the default."
More tokens means LESS redaction, because a fragment carrying no `=` is published VERBATIM.
Quoted, space-containing values are valid systemd `Exec*` syntax, and on them a whitespace
split tears the secret and publishes its tail to an endpoint anyone can fetch:
ExecStart=/bin/env API_KEY="sk live 9f3a2b" /usr/bin/gate
whitespace -> ExecStart=<redacted> API_KEY=<redacted> live 9f3a2b" /usr/bin/gate
shlex -> ExecStart=<redacted> API_KEY=<redacted> /usr/bin/gate
Measured on three shapes before the change, all three leaked. shlex MERGING a quoted token is
exactly what fixes it rather than a complication to reason about: `API_KEY="sk live 9f3a2b"`
becomes ONE token, which then matches as an assignment and is redacted whole.
This makes it agree with `environment_names` below, which already used shlex and already
carried a comment calling the two "the exact inverse ... worth stating so neither gets made
consistent with the other". They now agree because the argument for splitting them was wrong,
not because consistency is tidy.
Unbalanced quoting makes shlex raise, and the answer matches the neighbour's: publish NOTHING
from the line rather than guess where a value ends. Original spacing is not preserved and
nothing downstream re-executes this string.
"""
try:
tokens = shlex.split(value)
except ValueError:
return "<redacted: unbalanced quoting, whole value withheld>"
out = []
for tok in tokens:
m = ASSIGNMENT_TOKEN_RE.match(tok)
out.append(f"{m.group(1)}=<redacted>" if m else tok)
return " ".join(out)
def environment_names(value: str) -> tuple[list[str], int]:
"""Variable NAMES from an `Environment=` value. Returns (names, dropped).
shlex IS the right splitter here and whitespace is not, which is the exact inverse of
`scrub_assignment_values` above and worth stating so neither gets "made consistent" with the
other. `Environment="GREETING=hello there" MODE=fast` is two assignments, and a whitespace
split would read `there` as a third token whose name half is `there` -- a fragment of a VALUE
published under the label "name". shlex keeps the quoted assignment whole.
Unbalanced quoting makes shlex raise, and the answer to that is to publish NOTHING from the
line and count it, never to guess at where the values end.
"""
try:
tokens = shlex.split(value)
except ValueError:
return [], 1
names: list[str] = []
dropped = 0
for tok in tokens:
name = tok.split("=", 1)[0]
if ENV_NAME_RE.fullmatch(name):
names.append(name)
else:
dropped += 1
return names, dropped
def parse_unit_definition(text: str) -> dict:
"""Parse `systemctl cat` output into the publishable shape. PURE, so the self-test drives it.
Split out for the same reason `unit_verdict` is: the systemd call cannot run in a synthetic
tree, and a redaction rule that is only ever exercised through a subprocess is a rule nobody
can prove fails. Every control below drives this function on planted text.
`systemctl cat` emits `# <absolute path>` before each fragment and before every drop-in, and
the drop-ins are exactly where an override hides, so the paths are collected as `sources`.
Every other comment line is discarded, which is not incidental: the units in this repo carry
long rationale comments, and an allowlist over directives drops all of them for free.
"""
sources: list[str] = []
directives: list[str] = []
env_names: list[str] = []
env_files: list[str] = []
dropped = 0
section = ""
pending = ""
for raw in text.splitlines():
line = raw.strip()
if pending:
line = f"{pending} {line}"
pending = ""
# A directive may continue onto the next physical line. Joining FIRST means a
# continuation of an `Environment=` line is classified as one assignment rather than
# arriving as an orphan token whose key half is part of a value.
if line.endswith("\\"):
pending = line[:-1].rstrip()
continue
if not line:
continue
if line.startswith("#") or line.startswith(";"):
body = line[1:].strip()
if body.startswith("/") and " " not in body:
sources.append(body)
continue
if line.startswith("[") and line.endswith("]"):
section = line[1:-1]
continue
if "=" not in line:
dropped += 1
continue
key, _, value = line.partition("=")
key, value = key.strip(), value.strip()
if key in UNIT_DIRECTIVES_NAMES_ONLY:
names, bad = environment_names(value)
env_names.extend(names)
dropped += bad
elif key in UNIT_DIRECTIVES_PATH_ONLY:
env_files.append(value)
elif key in UNIT_DIRECTIVES_PUBLISHED:
prefix = f"[{section}] " if section else ""
directives.append(f"{prefix}{key}={scrub_assignment_values(value)}")
else:
dropped += 1
if pending:
# A trailing continuation with nothing after it. Counted rather than parsed.
dropped += 1
unique_names = sorted(set(env_names))
return {
"sources": sources,
"directives": directives[:UNIT_DIRECTIVES_SHOWN],
"directives_total": len(directives),
"environment_names": unique_names[:UNIT_ENV_NAMES_SHOWN],
"environment_names_total": len(unique_names),
"environment_files": env_files,
"dropped_directives": dropped,
}
def systemctl_cat(unit: str) -> tuple[bool, str]:
"""(ok, text-or-reason). Isolated so `collect_unit_definitions` can be driven without systemd."""
try:
out = subprocess.run(
["systemctl", "--user", "cat", unit],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=15,
)
except Exception as exc:
return False, f"error:{exc}"
if out.returncode != 0:
first = (out.stderr or "").strip().splitlines()
return False, first[0] if first else f"rc={out.returncode}"
return True, out.stdout or ""
def check_unit_definitions(inv: dict, r: Result, reader=None) -> None:
"""Publish each configured unit's DEFINITION, and report the count with its denominator.
THE DIVISION OF LABOUR WITH `check_services` IS DELIBERATE, so nobody merges them later.
`check_services` owns LIVENESS and fails closed on a unit it cannot read, which is correct:
a unit that is missing or dead is already red there. This check owns PUBLICATION, and an
individual unreadable unit does not turn it red -- doing so would report the same defect
twice under two names while adding nothing a reader can act on. What DOES turn it red is
reading none of them, because that is the instrument failing rather than the subject: no
systemctl on PATH, the wrong `--user` scope, a container with no session bus. Zero readable
out of six and zero configured are different failures and both say so in their own line.
NOT VERIFIED AGAINST THE REAL BOX. Deploys are frozen until after the live demo on
2026-08-20 20:00 EST, so every case below runs against planted `systemctl cat` text and
nothing here has met a live unit. On the first deploy after the freeze lifts, check three
things against `/selfcheck` before trusting the output:
1. the count reads `6 of 6` -- anything less names a unit that is not installed under the
`--user` scope, which is a real finding rather than a checker bug;
2. `environment_files` lists a path per secret-bearing unit and `environment_names` lists
the `X402_*` names, which together are the evidence that the split worked on real input;
3. no value appears anywhere in the published payload. Fetch it and grep it for a value you
know is set on the box. That grep is the only check that matters, and it cannot be run
from here.
4. THE PUBLISHED INVARIANT COUNT MOVES, and two surfaces quote it. This adds one row, so
`verify-proof.py` starts printing "on all 8 invariants" where it prints 7 today, and
`notes/DEMO-RUNBOOK.md` pins the 7 in two places. Neither is wrong now -- 7 is the truth
until this reaches the box -- so they are swept AT DEPLOY, not before. The runbook is
gitignored, so no gate in this repo can see that drift and nothing but this line will
raise it.
AND ONE THING NOT TO MISREAD ON THAT FIRST FETCH: no unit here will publish a `User=` line,
and its absence is the truth rather than a redaction. `User=` is a system-manager directive
that the service manager REFUSES inside a user unit, and every unit in this list is a
`--user` unit -- `grep -rn 'User=' deploy/*.service` finds none for that reason. The account
these run as is fixed by whose session owns the manager, so "what does it run as" is answered
by the scope plus the lingering account, never by a directive. `User` stays in the allowlist
anyway, because a unit that ever moves to the system manager should publish it immediately.
"""
units = inv.get("units") or []
read = reader or systemctl_cat
defs: dict = {}
unreadable: list[str] = []
for unit in units:
ok, payload = read(unit)
if ok:
defs[unit] = parse_unit_definition(payload)
else:
unreadable.append(f"{unit}: {payload}")
r.attach("unit_definitions", defs)
total = len(units)
if total == 0:
r.add(
"service-definitions",
False,
"0 of 0 unit(s) readable; no units configured, so nothing was published and "
"this is NOT a pass",
)
return
detail = f"{len(defs)} of {total} unit(s) readable"
if unreadable:
detail += "; not published: " + "; ".join(unreadable)
else:
detail += "; every definition published"
r.add("service-definitions", bool(defs), detail)
def unit_verdict(
unit: str, active_state: str, result: str, utype: str
) -> tuple[bool, str]:
"""Pure decision, split out from the systemd call SO THE SELF-TEST CAN DRIVE IT.
THIS WAS INVERTED UNTIL 2026-08-06 AND THE INVERSION WAS INVISIBLE, because the self-test set
`units = []` and skipped the only check that had it. The old expression allowed `inactive` for
a `.timer` and forbade it for everything else, which is backwards in both directions:
stopped zc-feed.timer -> PASSED the feed is the submission's lead claim, and its
death was the exact thing this gate existed to catch
finished oneshot -> FAILED the healthy steady state of zc-feed.service
The correct model needs the unit's TYPE, which `is-active` alone cannot supply:
.timer a loaded timer waiting to fire reports ACTIVE. So `inactive` means stopped and
whatever it drives is dead. Nothing else is acceptable.
oneshot finishes and reports inactive; healthy only when Result=success. A crashed one
reports inactive too, which is why Result rather than ActiveState decides it.
anything a daemon (simple/notify/forking) must be running. `inactive` is dead even when
else it was stopped cleanly, so Result=success must NOT rescue it here.
"""
if unit.endswith(".timer"):
ok = active_state == "active"
return ok, "" if ok else f"{unit}={active_state} (timer not scheduled)"
if active_state in ("active", "activating"):
return True, ""
if utype == "oneshot" and active_state == "inactive" and result == "success":
return True, ""
return (
False,
f"{unit}={active_state}/{result or 'unknown'} type={utype or 'unknown'}",
)
def check_services(inv: dict, r: Result) -> None:
"""Liveness. A correct configuration on a dead service is not a working shop."""
units = inv.get("units") or []
if not units:
r.add("services", False, "no units configured")
return
bad = []
for unit in units:
try:
out = subprocess.run(
[
"systemctl",
"--user",
"show",
unit,
"-p",
"ActiveState",
"-p",
"Result",
"-p",
"Type",
],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=15,
)
kv = dict(
line.split("=", 1)
for line in (out.stdout or "").splitlines()
if "=" in line
)
except Exception as exc:
# Fail CLOSED: an unreadable unit is not a passing unit.
bad.append(f"{unit}=error:{exc}")
continue
ok, why = unit_verdict(
unit,
kv.get("ActiveState", ""),
kv.get("Result", ""),
kv.get("Type", ""),
)
if not ok:
bad.append(why)
r.add("services", not bad, "all healthy" if not bad else "; ".join(bad))
def run_checks() -> Result:
r = Result()
inv = load_invariants()
if inv is None:
r.add(
"invariants",
False,
f"cannot read {INVARIANTS}; nothing was checked, and this is NOT a pass",
)
return r
r.add("invariants", True, f"loaded {INVARIANTS.name}")
check_vintage_agreement(inv, r)
check_manifest(inv, r)
check_mint_prohibition(inv, r)
check_network_prose(inv, r)
check_pins(inv, r)
check_services(inv, r)
check_unit_definitions(inv, r)
return r
def redact(text: str) -> str:
"""Strip the two things a PUBLICLY SERVED detail line must never carry.
The verdict is fetched over plain HTTPS by anyone, so a detail line is public copy rather than
a local log line. Two classes have to go, and only two:
the home directory an absolute path carries the account name on any box that is not this
one. `$HOME/.zeroclaw/x` becomes `~/.zeroclaw/x`, which is the form the
reproduction doc uses anyway, so nothing legible is lost.
a chat recipient a WhatsApp JID is a phone number. Nothing here needs to name it, and
`announce_settlements.sh` already avoids carrying one for the same reason.
DELIBERATELY NARROW, because the obvious wider version would gut the checker. Base58 tokens are
NOT redacted: the merchant address and the USDC mint are exactly what the mint and manifest
checks assert, they are public constants published in the write-up, and a verdict that hides
them cannot state which mint it found. Over-redaction here would leave a green checker saying
nothing, which is the failure mode this file's docstring already warns about.
"""
home = str(Path.home())
for form in (home, home.replace("\\", "/")):
if form and form != "/":
text = text.replace(form, "~")
return re.sub(r"\b\d+@(?:g\.us|s\.whatsapp\.net)", "<recipient>", text)
def redact_tree(obj: object) -> object:
"""`redact` over a nested payload, so an attached document gets the same treatment as a line.
Unit definitions arrive as nested lists of strings full of absolute paths -- `systemctl cat`
names every fragment by its full path and a `WorkingDirectory` is one by definition. Walking
the structure rather than redacting at each producer keeps ONE redaction site, so a future
attachment inherits it instead of needing to remember it.
"""
if isinstance(obj, str):
return redact(obj)
if isinstance(obj, dict):
return {k: redact_tree(v) for k, v in obj.items()}
if isinstance(obj, list):
return [redact_tree(v) for v in obj]
return obj
def build_verdict(r: Result) -> dict:
# `deployed_sha` NOW CARRIES THE GENERATED VINTAGE, not the hand-written label. It used to
# publish DEPLOYED_SHA alone, and on 2026-08-16 that served a commit nine days older than
# the files the same payload was certifying. The name is kept because it is already read by
# `scripts/verify-proof.py` and by anyone who has opened the endpoint; what changes is that
# it now names the commit the hashes in this verdict actually belong to. The hand file is
# still published beside it so a disagreement stays visible rather than being papered over,
# and `check_vintage_agreement` turns that disagreement red.
label = "unknown"
try:
label = DEPLOYED_SHA.read_text(encoding="utf-8").strip() or "unknown"
except OSError:
pass
inv = load_invariants() or {}
generated = inv.get("repo_commit")
# Same strictness as check_vintage_agreement, deliberately sharing one predicate: publishing
# the "unknown" sentinel under `deployed_sha_source: repo_commit` would label a non-commit as
# the corroborated baseline, which is worse than publishing the hand label it replaced.
have_generated = is_commit_sha(generated)
sha = generated if have_generated else label
return {
"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"generated_at_epoch": int(time.time()),
"deployed_sha": sha,
"deployed_sha_source": (
"repo_commit"
if have_generated
else "DEPLOYED_SHA (no repo_commit available)"
),
"deployed_sha_label": label,
"ok": r.ok,
# Redacted HERE rather than at the serving layer, so what lands on disk is already safe.
# The gate that serves this is a dumb file reader; putting the defense in the writer means