1	# ezhttp/src/wire: one request out and one response back, over
2	# bend-kit-wire's sockets. Its effects park on the runtime's event loop, so
3	# exchanges run side by side under IO.fork. bend-kit-dns resolves the host;
4	# TLS is bend-kit-wire's (OpenSSL 3 loaded at run time, BEND_LIBSSL to point
5	# at it).
6	#
7	# An exchange's answer is a status on its first line, then text: "0" and the
8	# raw response, or a non-zero status and the reason (wirespec.bend
9	# wire.fail). Every decision is wirespec.bend's: the deadline (wire.ms.of),
10	# the octets sent (wire.payload), what a failed call is reported as
11	# (wire.fail) and when the response is whole (wire.read.next), so a server
12	# that keeps the connection open after a complete response is not waited on
13	# (SPEC.md HTTP-CLI-1, 6, 10, 13; HTTP-TRUST-3, 5).
14	import Base
15	import bend-kit-wire@0.4.6.1/wire.bend as W
16	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17	import bend-kit-dns@0.6.0.2/dns.bend as Dns
18	import ./wirespec.bend as Spec
19	
20	# the most octets one receive asks for
21	def wire.max() -> U32:
22	  65536
23	
24	# the most receives one exchange makes
25	def wire.reads() -> Nat:
26	  65536n
27	
28	# --- the socket calls, plain or over TLS ---
29	
30	# close a socket, ending its TLS session when it has one
31	def wire.close(secure: Bool, s: Socket) -> IO(Unit):
32	  match secure:
33	    case True{}:
34	      W.tls.close(s)
35	    case False{}:
36	      Socket.close(s)
37	
38	# the next octets a socket has received, packed four to a word
39	def wire.recv(secure: Bool, s: Socket, ms: U32)
40	  -> IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
41	  match secure:
42	    case True{}:
43	      W.tls.recv.words(s, wire.max(), ms)
44	    case False{}:
45	      W.recv.words(s, wire.max(), ms)
46	
47	# a send.words.deadline answer without its count of octets accepted
48	def wire.sent(r: Socket & (U32 & Result<&1, &1, U32 & String, Unit>))
49	  -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
50	  (s, (n, res)) = r
51	  IO.pure(Socket & Result<&1, &1, U32 & String, Unit>, (s, res))
52	
53	# send octets, the plain send bounded by the deadline
54	def wire.send(secure: Bool, s: Socket, +len: U32, buf: Array<U32>, ms: U32)
55	  -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
56	  match secure:
57	    case True{}:
58	      W.tls.send.words(s, len, buf)
59	    case False{}:
60	      IO.bind(Socket & (U32 & Result<&1, &1, U32 & String, Unit>),
61	        Socket & Result<&1, &1, U32 & String, Unit>,
62	        W.send.words.deadline(s, len, buf, ms), wire.sent)
63	
64	# the TLS handshake with the host, when the exchange is secure
65	def wire.shake(secure: Bool, s: Socket, host: String, ms: U32)
66	  -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
67	  match secure:
68	    case True{}:
69	      W.tls.connect(s, host, ms)
70	    case False{}:
71	      IO.pure(Socket & Result<&1, &1, U32 & String, Unit>, (s, Done{Unit{}}))
72	
73	# --- reading the answer ---
74	
75	# a read that answered: its answer
76	def wire.end.of(r: Spec.Read) -> String:
77	  match r:
78	    case Spec.Heard{answer}:
79	      answer
80	    case Spec.Again{acc}:
81	      Spec.wire.spent()
82	
83	# the held text went in: now the close
84	def wire.end.more(r: Spec.Read) -> String:
85	  match r:
86	    case Spec.Heard{answer}:
87	      answer
88	    case Spec.Again{acc}:
89	      wire.end.of(Spec.wire.read.next(acc, "0\n"))
90	
91	# what the server's close leaves: the text still held, then the close
92	def wire.end(acc: Spec.Got, +text: String, none: Bool) -> String:
93	  match none:
94	    case True{}:
95	      wire.end.of(Spec.wire.read.next(acc, "0\n"))
96	    case False{}:
97	      wire.end.more(Spec.wire.read.next(acc, "0\n" ++ text))
98	
99	# a judged read: read again, or close and answer
100	def wire.loop.on(+secure: Bool, s: Socket, r: Spec.Read,
101	  more: Socket -> Spec.Got -> IO(String)) -> IO(String):
102	  match r:
103	    case Spec.Again{acc}:
104	      more(s)(acc)
105	    case Spec.Heard{answer}:
106	      IO.bind(Unit, String, wire.close(secure, s),
107	        _u => IO.pure(String, answer))
108	
109	# decoded text once it is known whether it is empty: read again when the
110	# octets finished no character, else judge the text
111	def wire.loop.judge(+secure: Bool, s: Socket, acc: Spec.Got, +text: String,
112	  held: List<&2, U32>, none: Bool,
113	  more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
114	  match none:
115	    case True{}:
116	      more(s)(held)(acc)
117	    case False{}:
118	      wire.loop.on(secure, s, Spec.wire.read.next(acc, "0\n" ++ text),
119	        s2 => a2 => more(s2)(held)(a2))
120	
121	# octets that arrived, decoded with what was held
122	def wire.loop.text(+secure: Bool, s: Socket, acc: Spec.Got,
123	  t: String & List<&2, U32>,
124	  more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
125	  (+text, held) = t
126	  wire.loop.judge(secure, s, acc, text, held, String.is_empty(text), more)
127	
128	# the server's close: what was held, decoded as it is, then the close
129	def wire.loop.closed(+secure: Bool, s: Socket, held: List<&2, U32>,
130	  acc: Spec.Got) -> IO(String):
131	  +text = Spec.wire.utf8.flush(held)
132	  IO.bind(Unit, String, wire.close(secure, s),
133	    _u => IO.pure(String, wire.end(acc, text, String.is_empty(text))))
134	
135	# a receive's octets, or the server's close
136	def wire.loop.got(+secure: Bool, s: Socket, held: List<&2, U32>,
137	  acc: Spec.Got, +len: U32, buf: Array<U32>, closed: Bool,
138	  more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
139	  match closed:
140	    case True{}:
141	      wire.loop.closed(secure, s, held, acc)
142	    case False{}:
143	      wire.loop.text(secure, s, acc,
144	        Spec.wire.utf8.feed(held, Bytes.to_string(Bytes.Bytes{len, buf})),
145	        more)
146	
147	# a receive's outcome: a failure is the answer
148	def wire.loop.un(+secure: Bool, +host: String, +port: U32, +ms: U32,
149	  held: List<&2, U32>, acc: Spec.Got,
150	  r: Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>,
151	  more: Socket -> List<&2, U32> -> Spec.Got -> IO(String)) -> IO(String):
152	  (s, res) = r
153	  match res:
154	    case Fail{(n, why)}:
155	      IO.bind(Unit, String, wire.close(secure, s),
156	        _u => IO.pure(String,
157	          Spec.wire.fail(Spec.Receive{}, host, port, ms, n, why)))
158	    case Done{(+len, buf)}:
159	      wire.loop.got(secure, s, held, acc, len, buf, U32.is_zero(len), more)
160	
161	# receive until wirespec.bend says the answer is whole
162	def wire.loop(fuel: Nat, +secure: Bool, s: Socket, +host: String, +port: U32,
163	  +ms: U32, held: List<&2, U32>, acc: Spec.Got) -> IO(String):
164	  match fuel:
165	    case 0n:
166	      IO.bind(Unit, String, wire.close(secure, s),
167	        _u => IO.pure(String, Spec.wire.spent()))
168	    case 1n+p:
169	      IO.bind(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>, String,
170	        wire.recv(secure, s, ms),
171	        r => wire.loop.un(secure, host, port, ms, held, acc, r,
172	          s2 => h2 => a2 => wire.loop(p, secure, s2, host, port, ms, h2, a2)))
173	
174	# --- one exchange ---
175	
176	# after the send: read the answer, or the send's failure
177	def wire.after.send(+secure: Bool, +host: String, +port: U32, +ms: U32,
178	  +method: String, r: Socket & Result<&1, &1, U32 & String, Unit>)
179	  -> IO(String):
180	  (s, res) = r
181	  match res:
182	    case Fail{(n, why)}:
183	      IO.bind(Unit, String, wire.close(secure, s),
184	        _u => IO.pure(String,
185	          Spec.wire.fail(Spec.Send{}, host, port, ms, n, why)))
186	    case Done{_u}:
187	      wire.loop(wire.reads(), secure, s, host, port, ms, [],
188	        Spec.wire.got(method, []))
189	
190	# send the octets an exchange carries, then read its answer
191	def wire.put(+secure: Bool, s: Socket, +host: String, +port: U32, +ms: U32,
192	  +method: String, payload: Bytes.Bytes) -> IO(String):
193	  Bytes.Bytes{+len, buf} = payload
194	  IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
195	    wire.send(secure, s, len, buf, ms),
196	    r => wire.after.send(secure, host, port, ms, method, r))
197	
198	# after the handshake: send, or the handshake's failure (the socket closed
199	# plainly, as no TLS session was made)
200	def wire.after.shake(+secure: Bool, +host: String, +port: U32, +ms: U32,
201	  +method: String, payload: String,
202	  r: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(String):
203	  (s, res) = r
204	  match res:
205	    case Fail{(n, why)}:
206	      IO.bind(Unit, String, Socket.close(s),
207	        _u => IO.pure(String,
208	          Spec.wire.fail(Spec.Handshake{}, host, port, ms, n, why)))
209	    case Done{_u}:
210	      wire.put(secure, s, host, port, ms, method, Bytes.from_string(payload))
211	
212	# after the connect: the handshake, or the connect's failure
213	def wire.after.connect(+secure: Bool, +host: String, +port: U32, +ms: U32,
214	  +method: String, payload: String,
215	  r: Result<&1, &1, U32 & String, Socket>) -> IO(String):
216	  match r:
217	    case Fail{(n, why)}:
218	      IO.pure(String, Spec.wire.fail(Spec.Connect{}, host, port, ms, n, why))
219	    case Done{s}:
220	      IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
221	        wire.shake(secure, s, host, ms),
222	        r2 => wire.after.shake(secure, host, port, ms, method, payload, r2))
223	
224	# a connect that failed tries the next address; one that worked is kept
225	def wire.dial.next(r: Result<&1, &1, U32 & String, Socket>,
226	  more: Unit -> IO(Result<&1, &1, U32 & String, Socket>))
227	  -> IO(Result<&1, &1, U32 & String, Socket>):
228	  match r:
229	    case Done{s}:
230	      IO.pure(Result<&1, &1, U32 & String, Socket>, Done{s})
231	    case Fail{e}:
232	      more(Unit{})
233	
234	# connect to each address in turn until one connects; the last one's
235	# failure when none does
236	def wire.dial(rest: List<&2, String>, +first: String, +port: U32, +ms: U32)
237	  -> IO(Result<&1, &1, U32 & String, Socket>):
238	  match rest:
239	    case []:
240	      W.connect(first, port, ms)
241	    case ip <> t:
242	      IO.bind(Result<&1, &1, U32 & String, Socket>,
243	        Result<&1, &1, U32 & String, Socket>,
244	        W.connect(first, port, ms),
245	        r => wire.dial.next(r, _u => wire.dial(t, ip, port, ms)))
246	
247	# after the resolve: connect, or the resolve's failure (no address at all is
248	# ENOENT, as bend-kit-dns reports a name with none)
249	def wire.after.resolve(+secure: Bool, +host: String, +port: U32, +ms: U32,
250	  +method: String, payload: String,
251	  r: Result<&1, &1, U32 & String, List<&2, String>>) -> IO(String):
252	  match r:
253	    case Fail{(n, why)}:
254	      IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, n, why))
255	    case Done{[]}:
256	      IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, 2, ""))
257	    case Done{ip <> t}:
258	      IO.bind(Result<&1, &1, U32 & String, Socket>, String,
259	        wire.dial(t, ip, port, ms),
260	        c => wire.after.connect(secure, host, port, ms, method, payload, c))
261	
262	# the OS resolver's addresses, NUL-separated, as a list
263	def wire.looked(r: Result<&1, &1, U32 & String, String>)
264	  -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
265	  match r:
266	    case Fail{e}:
267	      IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Fail{e})
268	    case Done{text}:
269	      IO.pure(Result<&1, &1, U32 & String, List<&2, String>>,
270	        Done{Dns.lookup.split(text)})
271	
272	# a host's addresses: a numeric address or localhost as it is, else the OS
273	# resolver's answer
274	def wire.resolve(+host: String, literal: List<&2, String>)
275	  -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
276	  match literal:
277	    case []:
278	      IO.bind(Result<&1, &1, U32 & String, String>,
279	        Result<&1, &1, U32 & String, List<&2, String>>,
280	        Dns.lookup.all(String.to_lower(host)), wire.looked)
281	    case ip <> t:
282	      IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Done{ip <> t})
283	
284	# a request with this method, sent to a host as wirespec.bend says, with
285	# its tail of octets after it, every call bounded by the deadline; its whole
286	# answer
287	def wire.run(+secure: Bool, +ms: U32, +host: String, +port: U32,
288	  +method: String, req: String, tail: List<&2, U32>) -> IO(String):
289	  IO.bind(Result<&1, &1, U32 & String, List<&2, String>>, String,
290	    wire.resolve(host, Dns.resolve.literal.all(host)),
291	    r => wire.after.resolve(secure, host, port, ms, method,
292	      Spec.wire.payload(req, tail), r))
293	
294	# a request sent to a host, its tail of octets after it, and its whole answer
295	# read back, with the default deadline
296	def wire.talk(secure: Bool, host: String, port: U32, method: String,
297	  req: String, tail: List<&2, U32>) -> IO(String):
298	  wire.run(secure, Spec.wire.ms.of(None{}), host, port, method, req, tail)
299	
300	# the same exchange, bounded by a timeout in milliseconds
301	def wire.talk.timed(secure: Bool, ms: U32, host: String, port: U32,
302	  method: String, req: String, tail: List<&2, U32>) -> IO(String):
303	  wire.run(secure, Spec.wire.ms.of(Some{ms}), host, port, method, req, tail)