1	import Base
2	import ./json.bend as J
3	import ./utf8.bend as U
4	import ./base64.bend as B
5	import ./http.bend as H
6	import ./net.bend as N
7	import ./core.bend as C
8	import ./config.bend as K
9	import ./client.bend as Cl
10	import ./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	
22	type Out is Data:
23	  Out{status: J.Json, stdout: String, stderr: String}
24	
25	# Output
26	# ------
27	
28	# Whether the command exited 0.
29	def 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).
36	def 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	
41	def Exec.stdout(o: Out) -> String:
42	  match o:
43	    case Out{st, a, b}:
44	      a
45	
46	def Exec.stderr(o: Out) -> String:
47	  match o:
48	    case Out{st, a, b}:
49	      b
50	
51	# Request
52	# -------
53	
54	def 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	
61	def Exec.flag(on: Bool) -> String:
62	  match on:
63	    case True{}:
64	      "&stdin=true"
65	    case False{}:
66	      ""
67	
68	def 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	
74	def 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	
80	def 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.
90	type Scan is Data:
91	  Scan{head: List<&2, U32>, matched: U32, rest: List<&2, U32>, found: Bool}
92	
93	def 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.
103	def 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	
107	def 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	
117	def 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	
122	def Scan.found(s: Scan) -> Bool:
123	  match s:
124	    case Scan{h, m, r, f}:
125	      f
126	
127	type 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	
131	def 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.
137	def 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.
164	type 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	
168	def 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	
180	def 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	
185	def 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	
197	def 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.
204	def 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	
209	def 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	
216	def 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	
221	def 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	
226	def 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	
231	def Acc.closed(a: Acc) -> Bool:
232	  match a:
233	    case Acc{st, o, e, s, ch, cl}:
234	      cl
235	
236	def 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	
241	def 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	
249	def 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	
255	type 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	
259	def 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.
265	def 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	
291	def Exec.Res() -> Type:
292	  Result<&2, &2, Cl.KErr, Out>
293	
294	def 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	
300	def 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	
307	def 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.
316	def 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	
320	def 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	
329	def 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).
338	def 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	
342	def 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	
350	def 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	
358	def 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	
367	def 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	
377	def 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.
389	def 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))