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