src/exec.bend

bend-kube-runtime@0.2.0.0 · 393 lines · 50 definitions

1import Base
2import bend-kube-runtime@0.2.0.0/src/json.bend as J
3import bend-kube-runtime@0.2.0.0/src/utf8.bend as U
4import bend-kube-runtime@0.2.0.0/src/base64.bend as B
5import bend-kube-runtime@0.2.0.0/src/http.bend as H
6import bend-kube-runtime@0.2.0.0/src/net.bend as N
7import bend-kube-runtime@0.2.0.0/src/core.bend as C
8import bend-kube-runtime@0.2.0.0/src/config.bend as K
9import bend-kube-runtime@0.2.0.0/src/client.bend as Cl
10import bend-kube-runtime@0.2.0.0/src/ws.bend as W
11
12# Exec
13# ====
14#
15# Runs a command in a pod's container, as `kubectl exec` does: a WebSocket
16# upgrade of GET .../pods/NAME/exec speaking v5.channel.k8s.io. Every
17# message carries its channel in its first byte: 0 stdin, 1 stdout,
18# 2 stderr, 3 the final Status; a client closes stdin with the message
19# 255 0, which v5 added (v4 cannot close stdin, so it is not offered).
20# The command's whole output is collected: exec suits short commands.
21
22type Out is Data:
23  Out{status: J.Json, stdout: String, stderr: String}
24
25# Output
26# ------
27
28# Whether the command exited 0.
29def Exec.ok(o: Out) -> Bool:
30  match o:
31    case Out{st, a, b}:
32      String.eq(J.Json.str_or(J.Json.at(st, ["status"]), ""), "Success")
33
34# Why it did not: the Status message (the exit code, or why it could not
35# run).
36def Exec.why(o: Out) -> String:
37  match o:
38    case Out{st, a, b}:
39      J.Json.str_or(J.Json.at(st, ["message"]), "")
40
41def Exec.stdout(o: Out) -> String:
42  match o:
43    case Out{st, a, b}:
44      a
45
46def Exec.stderr(o: Out) -> String:
47  match o:
48    case Out{st, a, b}:
49      b
50
51# Request
52# -------
53
54def Exec.args(cmd: List<&2, String>) -> String:
55  match cmd:
56    case Nil{}:
57      ""
58    case Con{c, t}:
59      "&command=" ++ H.Http.encode(c) ++ Exec.args(t)
60
61def Exec.flag(on: Bool) -> String:
62  match on:
63    case True{}:
64      "&stdin=true"
65    case False{}:
66      ""
67
68def Exec.path(ns: String, pod: String, container: String, cmd: List<&2, String>, stdin: Bool) ->
69  String:
70  "/api/v1/namespaces/" ++ H.Http.encode(ns) ++ "/pods/" ++ H.Http.encode(pod) ++
71    "/exec?container=" ++ H.Http.encode(container) ++ "&stdout=true&stderr=true" ++
72    Exec.flag(stdin) ++ Exec.args(cmd)
73
74def Exec.key(+k: U32) -> String:
75  B.B64.encode([W.Ws.byte(k, 0n), W.Ws.byte(k, 1n), W.Ws.byte(k, 2n), W.Ws.byte(k, 3n),
76    W.Ws.byte(k, 3n), W.Ws.byte(k, 2n), W.Ws.byte(k, 1n), W.Ws.byte(k, 0n),
77    W.Ws.byte(k, 1n), W.Ws.byte(k, 3n), W.Ws.byte(k, 0n), W.Ws.byte(k, 2n),
78    W.Ws.byte(k, 2n), W.Ws.byte(k, 0n), W.Ws.byte(k, 3n), W.Ws.byte(k, 1n)])
79
80def Exec.req(path: String, +k: U32) -> H.Req:
81  H.Req{"GET", path, [H.Header{"Upgrade", "websocket"}, H.Header{"Connection", "Upgrade"},
82    H.Header{"Sec-WebSocket-Version", "13"}, H.Header{"Sec-WebSocket-Key", Exec.key(k)},
83    H.Header{"Sec-WebSocket-Protocol", "v5.channel.k8s.io"}], ""}
84
85# The upgrade's head
86# ------------------
87
88# Bytes up to the blank line that ends the head (reversed), how much of
89# CR LF CR LF matched so far, and the bytes after it.
90type Scan is Data:
91  Scan{head: List<&2, U32>, matched: U32, rest: List<&2, U32>, found: Bool}
92
93def Scan.next.go(+m: U32, hit: Bool, cr: Bool) -> U32:
94  match hit cr:
95    case True{} c:
96      (m + 1 : U32)
97    case False{} True{}:
98      1
99    case False{} False{}:
100      0
101
102# The next count of CR LF CR LF matched, after byte b.
103def Scan.next(+m: U32, +b: U32) -> U32:
104  Scan.next.go(m,
105    U32.is_eq(b, U32.mod(U32.shrn(168626701, Nat.mul(8n, U32.to_nat(m))), 256)), U32.is_eq(b, 13))
106
107def Scan.feed.go(bs: List<&2, U32>, +head: List<&2, U32>, +m: U32, +rest: List<&2, U32>,
108  found: Bool) -> Scan:
109  match bs found:
110    case Nil{} f:
111      Scan{head, m, rest, f}
112    case Con{b, t} True{}:
113      Scan{head, m, List.append(&2, U32, rest, Con{b, t}), True{}}
114    case Con{+b, t} False{}:
115      Scan.feed.go(t, Con{b, head}, Scan.next(m, b), rest, U32.is_eq(Scan.next(m, b), 4))
116
117def Scan.feed(s: Scan, bs: List<&2, U32>) -> Scan:
118  match s:
119    case Scan{head, m, rest, f}:
120      Scan.feed.go(bs, head, m, rest, f)
121
122def Scan.found(s: Scan) -> Bool:
123  match s:
124    case Scan{h, m, r, f}:
125      f
126
127type HEv is Type:
128  HGot{sock: Socket, r: Result<&1, &1, U32 & String, List<&2, U32>>}
129  HChk{sock: Socket, found: Bool, eof: Bool}
130
131def HEv.got(p: Socket & Result<&1, &1, U32 & String, List<&2, U32>>) -> HEv:
132  match p:
133    case (s, r):
134      HGot{s, r}
135
136# Reads until the head is whole.
137def Exec.head(fuel: Nat, ev: HEv, +sc: Scan, +ms: U32) ->
138  IO(Socket & Result<&2, &2, Cl.KErr, Scan>):
139  match fuel ev:
140    case 0n HGot{s, r}:
141      IO.pure(Socket & Result<&2, &2, Cl.KErr, Scan>, (s, Fail{Cl.KNet{0, "exec: head too long"}}))
142    case 0n HChk{s, f, e}:
143      IO.pure(Socket & Result<&2, &2, Cl.KErr, Scan>, (s, Fail{Cl.KNet{0, "exec: head too long"}}))
144    case 1n+f HGot{s, Done{+bs}}:
145      +next = Scan.feed(sc, bs)
146      Exec.head(f, HChk{s, Scan.found(next), List.is_empty(&2, U32, bs)}, next, ms)
147    case 1n+f HGot{s, Fail{(c, m)}}:
148      IO.pure(Socket & Result<&2, &2, Cl.KErr, Scan>, (s, Fail{Cl.KNet{c, "exec: recv: " ++ m}}))
149    case 1n+f HChk{s, True{}, e}:
150      IO.pure(Socket & Result<&2, &2, Cl.KErr, Scan>, (s, Done{sc}))
151    case 1n+f HChk{s, False{}, True{}}:
152      IO.pure(Socket & Result<&2, &2, Cl.KErr, Scan>,
153        (s, Fail{Cl.KNet{0, "exec: connection closed before the upgrade"}}))
154    case 1n+f HChk{s, False{}, False{}}:
155      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>,
156        Socket & Result<&2, &2, Cl.KErr, Scan>, N.Tls.recv(s, 65536, ms),
157        p => Exec.head(f, HEv.got(p), sc, ms))
158
159# The messages
160# ------------
161
162# Output so far (each reversed), the channel of the message a continuation
163# frame extends, whether the server closed, and the decoder.
164type Acc is Data:
165  Acc{st: W.WSt, out: List<&2, U32>, err: List<&2, U32>, status: List<&2, U32>, chan: U32,
166    closed: Bool}
167
168def Acc.put.go(st: W.WSt, +chan: U32, bs: List<&2, U32>, +o: List<&2, U32>, +e: List<&2, U32>,
169  +s: List<&2, U32>, closed: Bool, one: Bool, two: Bool, three: Bool) -> Acc:
170  match one two three:
171    case True{} _ _:
172      Acc{st, List.append(&2, U32, List.reverse(&2, U32, bs), o), e, s, chan, closed}
173    case False{} True{} _:
174      Acc{st, o, List.append(&2, U32, List.reverse(&2, U32, bs), e), s, chan, closed}
175    case False{} False{} True{}:
176      Acc{st, o, e, List.append(&2, U32, List.reverse(&2, U32, bs), s), chan, closed}
177    case False{} False{} False{}:
178      Acc{st, o, e, s, chan, closed}
179
180def Acc.put(st: W.WSt, +chan: U32, bs: List<&2, U32>, +o: List<&2, U32>, +e: List<&2, U32>,
181  +s: List<&2, U32>, closed: Bool) -> Acc:
182  Acc.put.go(st, chan, bs, o, e, s, closed, U32.is_eq(chan, 1), U32.is_eq(chan, 2),
183    U32.is_eq(chan, 3))
184
185def Acc.frame.kind(st: W.WSt, +o: List<&2, U32>, +e: List<&2, U32>, +s: List<&2, U32>, +ch: U32,
186  cl: Bool, data: List<&2, U32>, cont: Bool, msg: Bool, close: Bool) -> Acc:
187  match data cont msg close:
188    case d True{} m c:
189      Acc.put(st, ch, d, o, e, s, cl)
190    case Con{+c, t} False{} True{} x:
191      Acc.put(st, c, t, o, e, s, cl)
192    case d False{} False{} True{}:
193      Acc{st, o, e, s, ch, True{}}
194    case d False{} m c:
195      Acc{st, o, e, s, ch, cl}
196
197def Acc.frame.go(st: W.WSt, +o: List<&2, U32>, +e: List<&2, U32>, +s: List<&2, U32>, +ch: U32,
198  cl: Bool, +op: U32, data: List<&2, U32>) -> Acc:
199  Acc.frame.kind(st, o, e, s, ch, cl, data, U32.is_zero(op), Bool.or(U32.is_eq(op, 1),
200    U32.is_eq(op, 2)), U32.is_eq(op, 8))
201
202# One frame: a data frame starts with its channel; a continuation extends
203# the last one; close ends the stream; pings and pongs are dropped.
204def Acc.frame(a: Acc, f: W.Frame) -> Acc:
205  match a f:
206    case Acc{st, +o, +e, +s, +ch, cl} W.Frame{fin, +op, data}:
207      Acc.frame.go(st, o, e, s, ch, cl, op, data)
208
209def Acc.frames(fs: List<&2, W.Frame>, a: Acc) -> Acc:
210  match fs:
211    case Nil{}:
212      a
213    case Con{f, t}:
214      Acc.frames(t, Acc.frame(a, f))
215
216def Acc.with(a: Acc, st: W.WSt) -> Acc:
217  match a:
218    case Acc{s0, o, e, s, ch, cl}:
219      Acc{st, o, e, s, ch, cl}
220
221def Acc.fed(a: Acc, f: W.Fed) -> Acc:
222  match f:
223    case W.Fed{st, fs}:
224      Acc.frames(fs, Acc.with(a, st))
225
226def Acc.feed(a: Acc, bs: List<&2, U32>) -> Acc:
227  match a:
228    case Acc{+st, o, e, s, ch, cl}:
229      Acc.fed(Acc{st, o, e, s, ch, cl}, W.Ws.feed(st, bs))
230
231def Acc.closed(a: Acc) -> Bool:
232  match a:
233    case Acc{st, o, e, s, ch, cl}:
234      cl
235
236def Acc.error(a: Acc) -> Maybe<&2, String>:
237  match a:
238    case Acc{st, o, e, s, ch, cl}:
239      W.Ws.error(st)
240
241def Acc.status(s: Result<&2, &2, String, J.Json>) -> J.Json:
242  match s:
243    case Done{j}:
244      j
245    case Fail{m}:
246      J.Json.obj([J.KV{"status", J.JStr{"Failure"}},
247        J.KV{"message", J.JStr{"exec ended without a status"}}])
248
249def Acc.out(a: Acc) -> Out:
250  match a:
251    case Acc{st, o, e, s, ch, cl}:
252      Out{Acc.status(J.Json.parse(U.Utf8.decode_all(List.reverse(&2, U32, s)))),
253        U.Utf8.decode_all(List.reverse(&2, U32, o)), U.Utf8.decode_all(List.reverse(&2, U32, e))}
254
255type XEv is Type:
256  XGot{sock: Socket, r: Result<&1, &1, U32 & String, List<&2, U32>>}
257  XChk{sock: Socket, eof: Bool, closed: Bool, bad: Maybe<&2, String>}
258
259def XEv.got(p: Socket & Result<&1, &1, U32 & String, List<&2, U32>>) -> XEv:
260  match p:
261    case (s, r):
262      XGot{s, r}
263
264# Reads messages until the server closes or the stream ends.
265def Exec.read(fuel: Nat, ev: XEv, +acc: Acc, +ms: U32) -> IO(Socket & Result<&2, &2, Cl.KErr, Out>):
266  match fuel ev:
267    case 0n XGot{s, r}:
268      IO.pure(Socket & Result<&2, &2, Cl.KErr, Out>, (s, Fail{Cl.KNet{0, "exec: output too long"}}))
269    case 0n XChk{s, e, c, b}:
270      IO.pure(Socket & Result<&2, &2, Cl.KErr, Out>, (s, Fail{Cl.KNet{0, "exec: output too long"}}))
271    case 1n+f XGot{s, Done{+bs}}:
272      +next = Acc.feed(acc, bs)
273      Exec.read(f, XChk{s, List.is_empty(&2, U32, bs), Acc.closed(next), Acc.error(next)}, next,
274        ms)
275    case 1n+f XGot{s, Fail{(c, m)}}:
276      IO.pure(Socket & Result<&2, &2, Cl.KErr, Out>, (s, Fail{Cl.KNet{c, "exec: recv: " ++ m}}))
277    case 1n+f XChk{s, e, c, Some{why}}:
278      IO.pure(Socket & Result<&2, &2, Cl.KErr, Out>, (s, Fail{Cl.KNet{0, "exec: " ++ why}}))
279    case 1n+f XChk{s, True{}, c, None{}}:
280      IO.pure(Socket & Result<&2, &2, Cl.KErr, Out>, (s, Done{Acc.out(acc)}))
281    case 1n+f XChk{s, False{}, True{}, None{}}:
282      IO.pure(Socket & Result<&2, &2, Cl.KErr, Out>, (s, Done{Acc.out(acc)}))
283    case 1n+f XChk{s, False{}, False{}, None{}}:
284      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>,
285        Socket & Result<&2, &2, Cl.KErr, Out>, N.Tls.recv(s, 65536, ms),
286        p => Exec.read(f, XEv.got(p), acc, ms))
287
288# Running
289# -------
290
291def Exec.Res() -> Type:
292  Result<&2, &2, Cl.KErr, Out>
293
294def Exec.close(p: Socket & Result<&2, &2, Cl.KErr, Out>) -> IO(Result<&2, &2, Cl.KErr, Out>):
295  match p:
296    case (s, r):
297      IO.bind(Unit, Result<&2, &2, Cl.KErr, Out>, N.Tls.close(s),
298        u => IO.pure(Result<&2, &2, Cl.KErr, Out>, r))
299
300def Exec.listen(s: Socket, rest: List<&2, U32>, +ms: U32) ->
301  IO(Result<&2, &2, Cl.KErr, Out>):
302  IO.bind(Socket & Result<&2, &2, Cl.KErr, Out>, Result<&2, &2, Cl.KErr, Out>,
303    Exec.read(4294967295n, XChk{s, False{}, False{}, None{}},
304      Acc.feed(Acc{W.WHead{}, Nil{}, Nil{}, Nil{}, 1, False{}}, rest), ms),
305    Exec.close)
306
307def Exec.sent(p: Socket & Result<&1, &1, U32 & String, Unit>, rest: List<&2, U32>, +ms: U32) ->
308  IO(Result<&2, &2, Cl.KErr, Out>):
309  match p:
310    case (s, Done{u}):
311      Exec.listen(s, rest, ms)
312    case (s, Fail{(c, m)}):
313      Exec.close((s, Fail{Cl.KNet{c, "exec: send: " ++ m}}))
314
315# The input, then the close of stdin, in one write.
316def Exec.input(+text: String, +k: U32) -> List<&2, U32>:
317  List.append(&2, U32, W.Ws.frame(W.Ws.binary(), Con{0, U.Utf8.encode(text)}, k),
318    W.Ws.frame(W.Ws.binary(), [255, 0], U32.xor(k, 1515870810)))
319
320def Exec.upgraded(s: Socket, rest: List<&2, U32>, stdin: Maybe<&2, String>, +k: U32, +ms: U32) ->
321  IO(Result<&2, &2, Cl.KErr, Out>):
322  match stdin:
323    case None{}:
324      Exec.listen(s, rest, ms)
325    case Some{text}:
326      IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, Result<&2, &2, Cl.KErr, Out>,
327        N.Tls.send_bytes(s, Exec.input(text, k), ms), p => Exec.sent(p, rest, ms))
328
329def Exec.message(+text: String, j: Result<&2, &2, String, J.Json>) -> String:
330  match j:
331    case Done{s}:
332      J.Json.str_or(J.Json.at(s, ["message"]), text)
333    case Fail{m}:
334      text
335
336# A refused upgrade: the status, and the message of the Status body the
337# head's recv brought along (or the bytes as they are).
338def Exec.refused(+code: U32, reason: String, +rest: List<&2, U32>) -> Cl.KErr:
339  Cl.KApi{C.Status{code, reason,
340    Exec.message(String.trim(U.Utf8.decode_all(rest)), J.Json.parse(U.Utf8.decode_all(rest)))}}
341
342def Exec.answer.go(s: Socket, +code: U32, reason: String, rest: List<&2, U32>,
343  stdin: Maybe<&2, String>, +k: U32, +ms: U32, up: Bool) -> IO(Result<&2, &2, Cl.KErr, Out>):
344  match up:
345    case True{}:
346      Exec.upgraded(s, rest, stdin, k, ms)
347    case False{}:
348      Exec.close((s, Fail{Exec.refused(code, reason, rest)}))
349
350def Exec.answer(s: Socket, h: Maybe<&2, H.Head>, rest: List<&2, U32>, stdin: Maybe<&2, String>,
351  +k: U32, +ms: U32) -> IO(Result<&2, &2, Cl.KErr, Out>):
352  match h:
353    case Some{H.Head{+code, reason, hs}}:
354      Exec.answer.go(s, code, reason, rest, stdin, k, ms, U32.is_eq(code, 101))
355    case None{}:
356      Exec.close((s, Fail{Cl.KNet{0, "exec: a malformed answer to the upgrade"}}))
357
358def Exec.headed(p: Socket & Result<&2, &2, Cl.KErr, Scan>, stdin: Maybe<&2, String>, +k: U32,
359  +ms: U32) -> IO(Result<&2, &2, Cl.KErr, Out>):
360  match p:
361    case (s, Done{Scan{head, m, rest, f}}):
362      Exec.answer(s, H.Http.parse_head(U.Utf8.decode_all(List.reverse(&2, U32, head))), rest,
363        stdin, k, ms)
364    case (s, Fail{e}):
365      Exec.close((s, Fail{e}))
366
367def Exec.opened(t: Result<&1, &1, U32 & String, N.Tls>, stdin: Maybe<&2, String>, +k: U32,
368  +ms: U32) -> IO(Result<&2, &2, Cl.KErr, Out>):
369  match t:
370    case Done{N.Tls{s}}:
371      IO.bind(Socket & Result<&2, &2, Cl.KErr, Scan>, Result<&2, &2, Cl.KErr, Out>,
372        Exec.head(4096n, HChk{s, False{}, False{}}, Scan{Nil{}, 0, Nil{}, False{}}, ms),
373        p => Exec.headed(p, stdin, k, ms))
374    case Fail{(c, m)}:
375      IO.pure(Result<&2, &2, Cl.KErr, Out>, Fail{Cl.KNet{c, m}})
376
377def Exec.keyed(+cfg: K.Config, path: String, stdin: Maybe<&2, String>, +ms: U32,
378  r: Result<&1, &1, U32 & String, U32>) -> IO(Result<&2, &2, Cl.KErr, Out>):
379  match r:
380    case Done{+k}:
381      IO.bind(Result<&1, &1, U32 & String, N.Tls>, Result<&2, &2, Cl.KErr, Out>,
382        Cl.Conn.open(cfg, Exec.req(path, k), False{}), t => Exec.opened(t, stdin, k, ms))
383    case Fail{(c, m)}:
384      IO.pure(Result<&2, &2, Cl.KErr, Out>, Fail{Cl.KNet{c, "exec: no random key: " ++ m}})
385
386# Runs cmd in container of pod ns/name, writing stdin to it (None: no
387# stdin), and collects its output; ms bounds each wait for the server
388# (0: none). Done means the exec ran: Exec.ok says how the command ended.
389def Exec.run(+cfg: K.Config, ns: String, pod: String, container: String, cmd: List<&2, String>,
390  +stdin: Maybe<&2, String>, +ms: U32) -> IO(Result<&2, &2, Cl.KErr, Out>):
391  IO.bind(Result<&1, &1, U32 & String, U32>, Result<&2, &2, Cl.KErr, Out>, IO.random_u32(),
392    r => Exec.keyed(cfg, Exec.path(ns, pod, container, cmd, Maybe.is_some(&2, String, stdin)),
393      stdin, ms, r))