1	import Base
2	import ../clock.bend as Clock
3	import ../fs.bend as Fs
4	import ../lock.bend as Lock
5	import ../b64.bend as B64
6	import ../fs/write.bend as FsW
7	import ../store/base.bend as SBase
8	import ./norm.bend as UNorm
9	import ./cred.bend as UCred
10	import ../json.bend as Json
11	
12	# ---- the refresh consume gate ----------------------------------------------
13	#
14	# A refresh token is single-use, and the Python cswap shares the slots, so the
15	# POST runs under the same per-slot flock the reference takes
16	# (credentials/.consume-<slot>.lock): re-read the slot's .enc under it, never
17	# consume a superseded snapshot, POST, write the rotated credential back, then
18	# release.
19	
20	# cred: the slot's now-current credential (adopted or rotated); error: the
21	# refresh failure or `consume-busy` / `store-unmirrored`; consumed: the bytes
22	# the gate POSTed or adopted (a strike binds to those, not the snapshot).
23	
24	type Gated is Data:
25	  Gated{cred: Maybe<&2, String>, error: Maybe<&2, String>, consumed: Maybe<&2, String>, account: Json.Json}
26	
27	type Gate is Data:
28	  NoGate{}
29	  Gate{slot: String, email: String, root: String}
30	
31	def enc.path(+root: String, +slot: String, +email: String) -> String:
32	  SBase.enc_path(root, slot, email)
33	
34	def consume.path(+root: String, +slot: String) -> String:
35	  SBase.creds_dir(root) ++ "/.consume-" ++ slot ++ ".lock"
36	
37	def fp.pick(+m: Maybe<&2, String>, +cred: String) -> String:
38	  match m:
39	    case Some{r}:
40	      "rt:" ++ r
41	    case None{}:
42	      "full:" ++ cred
43	
44	# Lineage identity, as the reference's credential_fingerprint (refresh token
45	# when there is one, else the whole credential), kept as text: it is only ever
46	# compared for equality, so no hash is needed.
47	def fp(+cred: String) -> String:
48	  fp.pick(UCred.refresh_token(UCred.oauth.of(cred)), cred)
49	
50	def gated.err(+k: String, +consumed: Maybe<&2, String>) -> Gated:
51	  Gated{None{}, Some{k}, consumed, Json.JNull{}}
52	
53	def gated.ok(+c: String, +consumed: Maybe<&2, String>) -> Gated:
54	  Gated{Some{c}, None{}, consumed, Json.JNull{}}
55	
56	def gated.ok_a(+c: String, +consumed: Maybe<&2, String>, +acct: Json.Json) -> Gated:
57	  Gated{Some{c}, None{}, consumed, acct}
58	
59	def done.g(g: Gated) -> IO(Fs.Res(Gated)):
60	  IO.pure(Fs.Res(Gated), Done{g})
61	
62	def stored(r: Fs.Res(String)) -> Maybe<&2, String>:
63	  match r:
64	    case Fail{e}:
65	      None{}
66	    case Done{t}:
67	      UNorm.nonempty(B64.decode(t))
68	
69	# The world moved past the caller's snapshot and the current generation is
70	# fresh: the refresh the caller wanted has happened.
71	def adopt.pick(+cur: Maybe<&2, String>, +snap: Maybe<&2, String>, +ex: Bool) -> Bool:
72	  match cur snap:
73	    case Some{a} Some{b}:
74	      Bool.not(String.eq(a, b)) && Bool.not(ex)
75	    case _ _:
76	      False{}
77	
78	def adopt(+cur: String, +snapshot: String, +now: U32) -> Bool:
79	  adopt.pick(UCred.access_token(UCred.oauth.of(cur)), UCred.access_token(UCred.oauth.of(snapshot)), UCred.expired(UCred.oauth.of(cur), now))
80	
81	def write.back(+path: String, +old_text: String, +new: String, +acct: Json.Json) -> IO(Fs.Res(Gated)):
82	  IO.bind(Fs.Res(Unit), Fs.Res(Gated), FsW.write_atomic(path ++ ".prev", old_text, "600"),
83	    u => IO.bind(Fs.Res(Unit), Fs.Res(Gated), FsW.write_atomic(path, B64.encode(new), "600"),
84	      w => done.g(gated.ok_a(new, Some{new}, acct))))
85	
86	# CAS on the lineage: write the successor only when the store still holds the
87	# generation we consumed; a moved store keeps its newer lineage.
88	def cas.pick(+same: Bool, +path: String, +text: String, +new: String, +acct: Json.Json, +now_cred: String) -> IO(Fs.Res(Gated)):
89	  match same:
90	    case True{}:
91	      write.back(path, text, new, acct)
92	    case False{}:
93	      done.g(gated.ok(now_cred, Some{now_cred}))
94	
95	def cas.store(+m: Maybe<&2, String>, +path: String, +text: String, +cur: String, +new: String, +acct: Json.Json) -> IO(Fs.Res(Gated)):
96	  match m:
97	    case None{}:
98	      done.g(gated.ok_a(new, Some{cur}, acct))
99	    case Some{+now_cred}:
100	      cas.pick(String.eq(fp(now_cred), fp(cur)), path, text, new, acct, now_cred)
101	
102	def act.post(r: Result<&2, &2, String, UCred.Granted>, +path: String, +text: String, +cur: String) -> IO(Fs.Res(Gated)):
103	  match r:
104	    case Fail{k}:
105	      done.g(gated.err(k, Some{cur}))
106	    case Done{UCred.Granted{new, acct}}:
107	      IO.bind(Fs.Res(String), Fs.Res(Gated), Fs.read_file(path), r2 => cas.store(stored(r2), path, text, cur, new, acct))
108	
109	def act.decide(+adopt: Bool, +path: String, +text: String, +cur: String) -> IO(Fs.Res(Gated)):
110	  match adopt:
111	    case True{}:
112	      done.g(gated.ok(cur, Some{cur}))
113	    case False{}:
114	      IO.bind(Result<&2, &2, String, UCred.Granted>, Fs.Res(Gated), UCred.refresh_acct(cur), r => act.post(r, path, text, cur))
115	
116	def act.cur(+m: Maybe<&2, String>, +path: String, +text: String, +snapshot: String) -> IO(Fs.Res(Gated)):
117	  match m:
118	    case None{}:
119	      done.g(gated.err("transient", None{}))
120	    case Some{+cur}:
121	      IO.bind(U32, Fs.Res(Gated), Clock.now_s(), now => act.decide(adopt(cur, snapshot, now), path, text, cur))
122	
123	def act.read(r: Fs.Res(String), +path: String, +snapshot: String) -> IO(Fs.Res(Gated)):
124	  match r:
125	    case Fail{e}:
126	      done.g(gated.err("transient", None{}))
127	    case Done{+text}:
128	      act.cur(stored(Done{text}), path, text, snapshot)
129	
130	def gate.act(+path: String, +snapshot: String) -> IO(Fs.Res(Gated)):
131	  IO.bind(Fs.Res(String), Fs.Res(Gated), Fs.read_file(path), r => act.read(r, path, snapshot))
132	
133	def busy.err(+code: U32) -> Gated:
134	  gated.err(Clock.sel.s(U32.is_eq(code, 11), "consume-busy", "transient"), None{})
135	
136	def locked.fail(ce: U32 & String) -> Gated:
137	  (c, m) = ce
138	  busy.err(c)
139	
140	def locked.res(r: Fs.Res(Gated)) -> Gated:
141	  match r:
142	    case Done{g}:
143	      g
144	    case Fail{e}:
145	      locked.fail(e)
146	
147	def timeout.of(r: Result<&1, &1, U32 & String, String>) -> U32:
148	  match r:
149	    case Fail{e}:
150	      10000
151	    case Done{v}:
152	      Maybe.default(&2, U32, U32.read(v), 10000)
153	
154	def gated.locked(+slot: String, +email: String, +snapshot: String, +root: String, +ms: U32) -> IO(Gated):
155	  IO.bind(Fs.Res(Gated), Gated,
156	    Lock.with_lock(Gated, consume.path(root, slot), ms, gate.act(enc.path(root, slot, email), snapshot)),
157	    r => IO.pure(Gated, locked.res(r)))
158	
159	def gated.go(+slot: String, +email: String, +snapshot: String, +root: String) -> IO(Gated):
160	  IO.bind(Result<&1, &1, U32 & String, String>, Gated, IO.get_env("CSWAP_CONSUME_TIMEOUT_MS"),
161	    t => gated.locked(slot, email, snapshot, root, timeout.of(t)))
162	
163	def gated.env2(+set: Bool, +slot: String, +email: String, +snapshot: String, +root: String) -> IO(Gated):
164	  match set:
165	    case True{}:
166	      IO.pure(Gated, gated.err("store-unmirrored", None{}))
167	    case False{}:
168	      gated.go(slot, email, snapshot, root)
169	
170	def gated.env(r: Result<&1, &1, U32 & String, String>, +slot: String, +email: String, +snapshot: String, +root: String) -> IO(Gated):
171	  match r:
172	    case Fail{e}:
173	      gated.go(slot, email, snapshot, root)
174	    case Done{+v}:
175	      gated.env2(Bool.not(String.is_empty(v)), slot, email, snapshot, root)
176	
177	# consume_backup_grant: refresh the slot's credential under its consume lock.
178	# `snapshot` is the caller's copy; the gate consumes the freshest copy it can
179	# read from <root>/credentials/.creds-<slot>-<email>.enc, or adopts a fresher fresh one
180	# without any POST. Busy past the timeout (10 s, or
181	# CSWAP_CONSUME_TIMEOUT_MS) is `consume-busy`.
182	def refresh_gated(+slot: String, +email: String, +snapshot: String, +root: String) -> IO(Gated):
183	  IO.bind(Result<&1, &1, U32 & String, String>, Gated, IO.get_env("CLAUDE_SECURESTORAGE_CONFIG_DIR"),
184	    r => gated.env(r, slot, email, snapshot, root))
185	
186	def plain.pick(r: Result<&2, &2, String, String>, +cred: String) -> Gated:
187	  match r:
188	    case Done{c}:
189	      gated.ok(c, Some{cred})
190	    case Fail{k}:
191	      gated.err(k, Some{cred})
192	
193	# Gate or direct POST, by the caller's choice.
194	def refresh_via(g: Gate, +cred: String) -> IO(Gated):
195	  match g:
196	    case NoGate{}:
197	      IO.bind(Result<&2, &2, String, String>, Gated, UCred.refresh(cred), r => IO.pure(Gated, plain.pick(r, cred)))
198	    case Gate{slot, email, root}:
199	      refresh_gated(slot, email, cred, root)