nx_tools_argv_idem_gate.nx source
↩ module page · 101 lines · 5534 B
1// Actual ta_mcp_call integration in a newly created private /tmp directory.
2// Never modifies production allowlists, keyfiles or ledgers. Refuses reused/noncanonical isolation.
3import "nx_tools_api.nx"
4import "nx_gate_verdict.nx"
5
6const TAI_PRIVATE_MODE: i64 = 448
7const TAI_TOKEN_CAP: i64 = 1024
8const TAI_BODY_CAP: i64 = 4096
9const TAI_REPLY_CAP: i64 = 1048576
10
11func tai_exists(path: *u8) -> i64 {
12 let fd: i64 = sys_openat_rd(path)
13 if fd < 0 { return 0 }
14 sys_close(fd); return 1
15}
16func tai_equal(a: *u8, b: *u8) -> i64 {
17 var i: i64 = 0
18 while a[i] == b[i] { if a[i] == (0 as u8) { return 1 } i = i + 1 }
19 return 0
20}
21func tai_finish(ctr: *i64) -> i64 {
22 let rc: i64 = gv_verdict("TOOLS-ARGV-IDEM" as *u8, ctr, "malformed keyed argv is refused by the actual MCP caller and its new claim is released in isolated storage" as *u8)
23 sys_exit(rc); return rc
24}
25func tai_call(body: *u8, n: i64, reply: *u8) -> i64 {
26 let req: *u8 = "POST /mcp HTTP/1.1\r\nHost: gate.invalid\r\n\r\n" as *u8
27 let o: i64 = ta_cat(reply, 0, "{\"jsonrpc\":\"2.0\",\"id\":1" as *u8)
28 return ta_mcp_call("knowledge/unused-gate-registry-" as *u8, req, ta_slen(req), body, n, reply, o)
29}
30
31func main() -> i64 {
32 let ctr: *i64 = gv_ctr()
33 gv_head("TOOLS-ARGV-IDEM" as *u8)
34 let dir: *u8 = sys_mmap(SYS_PATH_MAX)
35 let cwd: *u8 = sys_mmap(SYS_PATH_MAX)
36 var p: i64 = ta_cat(dir, 0, "/tmp/nx-argv-idem-" as *u8)
37 p = ta_catn(dir, p, sys_now_ms()); dir[p] = 0 as u8
38 let made: i64 = sys_mkdir(dir, TAI_PRIVATE_MODE)
39 gv_check("fresh private fixture directory created" as *u8, made == 0, ctr)
40 if made != 0 { return tai_finish(ctr) }
41 let moved: i64 = sys_chdir(dir)
42 gv_check("fixture chdir succeeds" as *u8, moved == 0, ctr)
43 if moved != 0 { return tai_finish(ctr) }
44 let cn: i64 = sys_getcwd(cwd, SYS_PATH_MAX)
45 var isolated: i64 = 0
46 if cn > 0 { isolated = tai_equal(dir, cwd) }
47 gv_check("canonical cwd is the new fixture directory" as *u8, isolated, ctr)
48 if isolated != 1 { return tai_finish(ctr) }
49 var clean: i64 = 1
50 if tai_exists("tools_cap_secret.key" as *u8) == 1 { clean = 0 }
51 if tai_exists("tool_allowlist.conf" as *u8) == 1 { clean = 0 }
52 if tai_exists("tool_backends.conf" as *u8) == 1 { clean = 0 }
53 if tai_exists(TI_LEDGER) == 1 { clean = 0 }
54 gv_check("fixture has no key allowlist backend or ledger" as *u8, clean, ctr)
55 if clean != 1 { return tai_finish(ctr) }
56 let jobs: i64 = sys_mkdir(TI_DIR, TAI_PRIVATE_MODE)
57 gv_check("isolated claim directory created" as *u8, jobs == 0, ctr)
58 if jobs != 0 { return tai_finish(ctr) }
59 let tool: *u8 = "argv_unicode_gate_no_dispatch" as *u8
60 let key: *u8 = "argv-unicode-invalid-fixture" as *u8
61 let token: *u8 = sys_mmap(TAI_TOKEN_CAP)
62 let now: i64 = sys_now_realtime_sec()
63 // Match the current verified cap fixture's bounded validity window and issue API.
64 let tn: i64 = capt_issue(TA_CAP_SECRET, ta_slen(TA_CAP_SECRET), tool, ta_slen(tool), now + 3600, 1001, token, TAI_TOKEN_CAP)
65 gv_check("gate-only capability minted" as *u8, tn > 0, ctr)
66 if tn <= 0 { return tai_finish(ctr) }
67 let body: *u8 = sys_mmap(TAI_BODY_CAP)
68 var bn: i64 = ta_cat(body, 0, "{\"name\":\"" as *u8)
69 bn = ta_cat(body, bn, tool)
70 bn = ta_cat(body, bn, "\",\"_cap\":\"" as *u8)
71 bn = ta_catb(body, bn, token, tn)
72 bn = ta_cat(body, bn, "\",\"_idem\":\"" as *u8)
73 bn = ta_cat(body, bn, key)
74 bn = ta_cat(body, bn, "\",\"arguments\":{\"argv\":[\"\\ud800\"]}}" as *u8)
75 let reply: *u8 = sys_mmap(TAI_REPLY_CAP)
76 let hex: *u8 = sys_mmap(TI_HEX_CH + 8)
77 let row: *u8 = sys_mmap(TI_ROW_CAP)
78 let claim: *u8 = sys_mmap(TI_PATH_CAP)
79 let first: i64 = ti_classify(TI_LEDGER, TI_DIR, key, ta_slen(key), hex, row, TI_ROW_CAP, now)
80 gv_check("control actually reserves a new key" as *u8, first == TI_NEW, ctr)
81 if first != TI_NEW { return tai_finish(ctr) }
82 ti_claim_path(TI_DIR, hex, claim)
83 gv_check("control claim exists before actual caller" as *u8, tai_exists(claim), ctr)
84 let held: i64 = tai_call(body, bn, reply)
85 gv_check("actual caller observes held key as INFLIGHT" as *u8, ta_indexof(reply, held, "INFLIGHT" as *u8) >= 0, ctr)
86 gv_check("control claim can be released" as *u8, ti_release(TI_DIR, hex) == 1, ctr)
87 let refused: i64 = tai_call(body, bn, reply)
88 gv_check("actual caller refuses malformed argv with invalid params" as *u8, ta_indexof(reply, refused, "-32602" as *u8) >= 0, ctr)
89 gv_check("actual refusal leaves no claim" as *u8, tai_exists(claim) == 0, ctr)
90 let retry: i64 = tai_call(body, bn, reply)
91 gv_check("same keyed request reaches validation again" as *u8, ta_indexof(reply, retry, "-32602" as *u8) >= 0, ctr)
92 gv_check("same key does not remain INFLIGHT" as *u8, ta_indexof(reply, retry, "INFLIGHT" as *u8) < 0, ctr)
93 gv_check("retry refusal again leaves no claim" as *u8, tai_exists(claim) == 0, ctr)
94 gv_check("invalid input creates no execution outcome ledger" as *u8, tai_exists(TI_LEDGER) == 0, ctr)
95 let again: i64 = ti_classify(TI_LEDGER, TI_DIR, key, ta_slen(key), hex, row, TI_ROW_CAP, now)
96 gv_check("key remains reservable after both caller refusals" as *u8, again == TI_NEW, ctr)
97 if again == TI_NEW { ti_release(TI_DIR, hex) }
98 gv_check("allowlist was never created" as *u8, tai_exists("tool_allowlist.conf" as *u8) == 0, ctr)
99 gv_check("keyfile was never created" as *u8, tai_exists("tools_cap_secret.key" as *u8) == 0, ctr)
100 return tai_finish(ctr)
101}