usage/gate.bend

cswap-bend-cli@0.1.0.0 · 199 lines · 32 definitions

1import Base
2import cswap-bend-cli@0.1.0.0/clock.bend as Clock
3import cswap-bend-cli@0.1.0.0/fs.bend as Fs
4import cswap-bend-cli@0.1.0.0/lock.bend as Lock
5import cswap-bend-cli@0.1.0.0/b64.bend as B64
6import cswap-bend-cli@0.1.0.0/fs/write.bend as FsW
7import cswap-bend-cli@0.1.0.0/store/base.bend as SBase
8import cswap-bend-cli@0.1.0.0/usage/norm.bend as UNorm
9import cswap-bend-cli@0.1.0.0/usage/cred.bend as UCred
10import cswap-bend-cli@0.1.0.0/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
24type Gated is Data:
25  Gated{cred: Maybe<&2, String>, error: Maybe<&2, String>, consumed: Maybe<&2, String>, account: Json.Json}
26
27type Gate is Data:
28  NoGate{}
29  Gate{slot: String, email: String, root: String}
30
31def enc.path(+root: String, +slot: String, +email: String) -> String:
32  SBase.enc_path(root, slot, email)
33
34def consume.path(+root: String, +slot: String) -> String:
35  SBase.creds_dir(root) ++ "/.consume-" ++ slot ++ ".lock"
36
37def 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.
47def fp(+cred: String) -> String:
48  fp.pick(UCred.refresh_token(UCred.oauth.of(cred)), cred)
49
50def gated.err(+k: String, +consumed: Maybe<&2, String>) -> Gated:
51  Gated{None{}, Some{k}, consumed, Json.JNull{}}
52
53def gated.ok(+c: String, +consumed: Maybe<&2, String>) -> Gated:
54  Gated{Some{c}, None{}, consumed, Json.JNull{}}
55
56def gated.ok_a(+c: String, +consumed: Maybe<&2, String>, +acct: Json.Json) -> Gated:
57  Gated{Some{c}, None{}, consumed, acct}
58
59def done.g(g: Gated) -> IO(Fs.Res(Gated)):
60  IO.pure(Fs.Res(Gated), Done{g})
61
62def 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.
71def 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
78def 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
81def 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.
88def 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
95def 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
102def 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
109def 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
116def 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
123def 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
130def 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
133def busy.err(+code: U32) -> Gated:
134  gated.err(Clock.sel.s(U32.is_eq(code, 11), "consume-busy", "transient"), None{})
135
136def locked.fail(ce: U32 & String) -> Gated:
137  (c, m) = ce
138  busy.err(c)
139
140def 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
147def 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
154def 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
159def 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
163def 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
170def 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`.
182def 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
186def 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.
194def 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)