1use super::*;
4
5pub struct TlsConfig {
15 pub cert: Vec<u8>,
16 pub key: Vec<u8>,
17}
18
19pub(super) fn serve_http(
20 port: u16,
21 handler_name: String,
22 program: Arc<Program>,
23 policy: Policy,
24 tls: Option<TlsConfig>,
25 opts: ServeOpts,
26) -> Result<Value, String> {
27 match tls {
28 None => serve_http_plain(port, handler_name, program, policy, opts),
29 Some(cfg) => serve_http_tls_legacy(port, handler_name, program, policy, cfg),
30 }
31}
32
33pub(super) fn serve_http_plain(
43 port: u16,
44 handler_name: String,
45 program: Arc<Program>,
46 policy: Policy,
47 opts: ServeOpts,
48) -> Result<Value, String> {
49 use http_body_util::BodyExt as _;
50 use hyper::server::conn::http1;
51 use hyper::service::service_fn;
52 use hyper_util::rt::{TokioExecutor, TokioIo};
53 use hyper_util::server::conn::auto;
54 use tokio::net::TcpListener as TokioTcpListener;
55
56 let inline_vm = opts.inline_vm;
57 let http2 = opts.http2;
58 let host = opts.host.clone();
59 let rt = tokio::runtime::Builder::new_multi_thread()
60 .enable_all()
61 .build()
62 .map_err(|e| format!("net.serve: tokio runtime: {e}"))?;
63 rt.block_on(async move {
64 let listener = TokioTcpListener::bind((host.as_str(), port))
65 .await
66 .map_err(|e| format!("net.serve bind {host}:{port}: {e}"))?;
67 eprintln!(
68 "net.serve: listening on http://{host}:{port}{}{}",
69 if inline_vm { " (inline-vm)" } else { "" },
70 if http2 { " (http1+http2)" } else { "" }
71 );
72 loop {
73 let (stream, _) = listener
74 .accept()
75 .await
76 .map_err(|e| format!("net.serve accept: {e}"))?;
77 let io = TokioIo::new(stream);
78 let program = Arc::clone(&program);
79 let policy = policy.clone();
80 let handler_name = handler_name.clone();
81 tokio::spawn(async move {
82 let program2 = Arc::clone(&program);
83 let policy2 = policy.clone();
84 let handler_name2 = handler_name.clone();
85 let svc = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
86 let program = Arc::clone(&program2);
87 let policy = policy2.clone();
88 let handler_name = handler_name2.clone();
89 async move {
90 let (parts, body) = req.into_parts();
91 let body_bytes = body
92 .collect()
93 .await
94 .map(|c| c.to_bytes())
95 .unwrap_or_default();
96 let result = if inline_vm {
97 let lex_req = build_request_value_parts(&parts, &body_bytes);
101 let handler = DefaultHandler::new(policy)
102 .with_program(Arc::clone(&program));
103 let mut vm = Vm::with_handler(&program, Box::new(handler));
104 let r = vm.call(&handler_name, vec![lex_req]);
105 Ok(r.map(|v| unpack_response(&mut vm, &v)))
108 } else {
109 tokio::task::spawn_blocking(move || {
110 let lex_req = build_request_value_parts(&parts, &body_bytes);
111 let handler = DefaultHandler::new(policy)
112 .with_program(Arc::clone(&program));
113 let mut vm = Vm::with_handler(&program, Box::new(handler));
114 let r = vm.call(&handler_name, vec![lex_req]);
115 r.map(|v| unpack_response(&mut vm, &v))
116 })
117 .await
118 };
119 Ok::<_, std::convert::Infallible>(match result {
120 Ok(Ok(unpacked)) => build_hyper_response(unpacked),
121 Ok(Err(e)) => error_response(500, &format!("internal error: {e}")),
122 Err(e) => error_response(500, &format!("task panicked: {e}")),
123 })
124 }
125 });
126 let result = if http2 {
127 auto::Builder::new(TokioExecutor::new())
128 .serve_connection(io, svc)
129 .await
130 .map_err(|e| e.to_string())
131 } else {
132 http1::Builder::new()
133 .serve_connection(io, svc)
134 .await
135 .map_err(|e| e.to_string())
136 };
137 if let Err(e) = result {
138 eprintln!("net.serve: connection error: {e}");
139 }
140 });
141 }
142 })
143}
144
145pub(super) fn serve_http_tls_legacy(
147 port: u16,
148 handler_name: String,
149 program: Arc<Program>,
150 policy: Policy,
151 cfg: TlsConfig,
152) -> Result<Value, String> {
153 let ssl = tiny_http::SslConfig {
154 certificate: cfg.cert,
155 private_key: cfg.key,
156 };
157 let server = tiny_http::Server::https(("0.0.0.0", port), ssl)
158 .map_err(|e| format!("net.serve_tls bind {port}: {e}"))?;
159 eprintln!("net.serve: listening on https://0.0.0.0:{port}");
160 for req in server.incoming_requests() {
161 let program = Arc::clone(&program);
162 let policy = policy.clone();
163 let handler_name = handler_name.clone();
164 std::thread::spawn(move || handle_request_tls(req, program, policy, handler_name));
165 }
166 Ok(Value::Unit)
167}
168
169pub(super) fn handle_request_tls(
170 mut req: tiny_http::Request,
171 program: Arc<Program>,
172 policy: Policy,
173 handler_name: String,
174) {
175 let lex_req = build_request_value_tiny(&mut req);
176 let handler = DefaultHandler::new(policy).with_program(Arc::clone(&program));
177 let mut vm = Vm::with_handler(&program, Box::new(handler));
178 match vm.call(&handler_name, vec![lex_req]) {
179 Ok(resp) => {
180 let (status, body, headers) = unpack_response(&mut vm, &resp);
186 respond_with_body_tls(req, status, body, headers);
187 }
188 Err(e) => {
189 let response = tiny_http::Response::from_string(format!("internal error: {e}"))
190 .with_status_code(500);
191 let _ = req.respond(response);
192 }
193 }
194}
195
196pub(super) fn serve_http_fn(
201 port: u16,
202 closure: Value,
203 program: Arc<Program>,
204 policy: Policy,
205 opts: ServeOpts,
206) -> Result<Value, String> {
207 use http_body_util::BodyExt as _;
208 use hyper::server::conn::http1;
209 use hyper::service::service_fn;
210 use hyper_util::rt::{TokioExecutor, TokioIo};
211 use hyper_util::server::conn::auto;
212 use tokio::net::TcpListener as TokioTcpListener;
213
214 let inline_vm = opts.inline_vm;
215 let http2 = opts.http2;
216 let host = opts.host.clone();
217 let rt = tokio::runtime::Builder::new_multi_thread()
218 .enable_all()
219 .build()
220 .map_err(|e| format!("net.serve_fn: tokio runtime: {e}"))?;
221 rt.block_on(async move {
222 let listener = TokioTcpListener::bind((host.as_str(), port))
223 .await
224 .map_err(|e| format!("net.serve_fn bind {host}:{port}: {e}"))?;
225 eprintln!(
226 "net.serve_fn: listening on http://{host}:{port}{}{}",
227 if inline_vm { " (inline-vm)" } else { "" },
228 if http2 { " (http1+http2)" } else { "" }
229 );
230 loop {
231 let (stream, _) = listener
232 .accept()
233 .await
234 .map_err(|e| format!("net.serve_fn accept: {e}"))?;
235 let io = TokioIo::new(stream);
236 let program = Arc::clone(&program);
237 let policy = policy.clone();
238 let closure = closure.clone();
239 tokio::spawn(async move {
240 let program2 = Arc::clone(&program);
241 let policy2 = policy.clone();
242 let closure2 = closure.clone();
243 let svc = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
244 let program = Arc::clone(&program2);
245 let policy = policy2.clone();
246 let closure = closure2.clone();
247 async move {
248 let (parts, body) = req.into_parts();
249 let body_bytes = body
250 .collect()
251 .await
252 .map(|c| c.to_bytes())
253 .unwrap_or_default();
254 let result = if inline_vm {
255 let lex_req = build_request_value_parts(&parts, &body_bytes);
256 let handler = DefaultHandler::new(policy)
257 .with_program(Arc::clone(&program));
258 let mut vm = Vm::with_handler(&program, Box::new(handler));
259 let scope = vm.enter_request_scope();
266 let r = vm.invoke_closure_value(closure, vec![lex_req]);
267 let r = r.map(|v| unpack_response(&mut vm, &v));
271 vm.exit_request_scope(scope);
272 Ok(r)
273 } else {
274 tokio::task::spawn_blocking(move || {
275 let lex_req = build_request_value_parts(&parts, &body_bytes);
276 let handler = DefaultHandler::new(policy)
277 .with_program(Arc::clone(&program));
278 let mut vm = Vm::with_handler(&program, Box::new(handler));
279 let scope = vm.enter_request_scope();
280 let r = vm.invoke_closure_value(closure, vec![lex_req]);
281 let r = r.map(|v| unpack_response(&mut vm, &v));
282 vm.exit_request_scope(scope);
283 r
284 })
285 .await
286 };
287 Ok::<_, std::convert::Infallible>(match result {
288 Ok(Ok(unpacked)) => build_hyper_response(unpacked),
289 Ok(Err(e)) => error_response(500, &format!("internal error: {e}")),
290 Err(e) => error_response(500, &format!("task panicked: {e}")),
291 })
292 }
293 });
294 let result = if http2 {
295 auto::Builder::new(TokioExecutor::new())
296 .serve_connection(io, svc)
297 .await
298 .map_err(|e| e.to_string())
299 } else {
300 http1::Builder::new()
301 .serve_connection(io, svc)
302 .await
303 .map_err(|e| e.to_string())
304 };
305 if let Err(e) = result {
306 eprintln!("net.serve_fn: connection error: {e}");
307 }
308 });
309 }
310 })
311}
312
313#[derive(Clone, Debug)]
317pub(crate) enum RouteSeg {
318 Literal(String),
319 Param(String),
322}
323
324pub(super) fn compile_path_pattern(pat: &str) -> Result<Vec<RouteSeg>, String> {
328 if pat.is_empty() {
329 return Err("path pattern must be non-empty (use \"/\" for the root)".into());
330 }
331 if !pat.starts_with('/') {
332 return Err(format!("path pattern must start with '/' (got {pat:?})"));
333 }
334 let mut segs = Vec::new();
335 for raw in pat.split('/') {
336 if let Some(name) = raw.strip_prefix(':') {
337 if name.is_empty() {
338 return Err(format!(
339 ":-segment in pattern {pat:?} must have a name (e.g. :id)"
340 ));
341 }
342 segs.push(RouteSeg::Param(name.to_string()));
343 } else {
344 segs.push(RouteSeg::Literal(raw.to_string()));
345 }
346 }
347 Ok(segs)
348}
349
350pub(super) fn match_path_pattern(
356 segs: &[RouteSeg],
357 path: &str,
358) -> Option<std::collections::BTreeMap<lex_bytecode::MapKey, Value>> {
359 let path_segs: Vec<&str> = path.split('/').collect();
360 if path_segs.len() != segs.len() {
361 return None;
362 }
363 let mut params = std::collections::BTreeMap::new();
364 for (pat, p) in segs.iter().zip(path_segs.iter()) {
365 match pat {
366 RouteSeg::Literal(lit) => {
367 if lit != p {
368 return None;
369 }
370 }
371 RouteSeg::Param(name) => {
372 params.insert(
373 lex_bytecode::MapKey::Str(name.clone()),
374 Value::Str((*p).into()),
375 );
376 }
377 }
378 }
379 Some(params)
380}
381
382pub(super) fn decode_routes_arg(
387 v: Value,
388) -> Result<Vec<(String, Vec<RouteSeg>, Value)>, String> {
389 let list = match v {
390 Value::List(xs) => xs,
391 _ => return Err("net.serve_routed: routes must be a List".into()),
392 };
393 let mut out = Vec::with_capacity(list.len());
394 for (i, item) in list.into_iter().enumerate() {
395 let tup = match item {
396 Value::Tuple(xs) if xs.len() == 3 => xs,
397 other => return Err(format!(
398 "net.serve_routed: route #{i} must be a (method, pattern, handler) 3-tuple, got {other:?}"
399 )),
400 };
401 let mut it = tup.into_iter();
402 let method_raw = match it.next() {
403 Some(Value::Str(s)) => s.to_string(),
404 _ => return Err(format!("net.serve_routed: route #{i} method must be Str")),
405 };
406 let method = if method_raw == "*" { method_raw } else { method_raw.to_uppercase() };
408 let pattern = match it.next() {
409 Some(Value::Str(s)) => s.to_string(),
410 _ => return Err(format!("net.serve_routed: route #{i} path-pattern must be Str")),
411 };
412 let segs = compile_path_pattern(&pattern)
413 .map_err(|e| format!("net.serve_routed: route #{i} ({pattern:?}): {e}"))?;
414 let closure = match it.next() {
415 Some(c @ Value::Closure { .. }) => c,
416 _ => return Err(format!("net.serve_routed: route #{i} handler must be a closure")),
417 };
418 out.push((method, segs, closure));
419 }
420 Ok(out)
421}
422
423pub(crate) fn dispatch_route<'a>(
428 routes: &'a [(String, Vec<RouteSeg>, Value)],
429 req_method: &str,
430 req_path: &str,
431) -> Option<(&'a Value, std::collections::BTreeMap<lex_bytecode::MapKey, Value>)> {
432 let req_method_upper = req_method.to_ascii_uppercase();
433 for (m, segs, closure) in routes {
434 if m != "*" && m != &req_method_upper {
435 continue;
436 }
437 if let Some(params) = match_path_pattern(segs, req_path) {
438 return Some((closure, params));
439 }
440 }
441 None
442}
443
444pub(crate) fn stamp_path_params(
448 req: &mut Value,
449 params: std::collections::BTreeMap<lex_bytecode::MapKey, Value>,
450) {
451 if let Value::Record { fields: rec, .. } = req {
452 rec.insert("path_params".into(), Value::Map(params));
453 }
454}
455
456pub(super) fn serve_http_routed(
462 port: u16,
463 routes: Vec<(String, Vec<RouteSeg>, Value)>,
464 fallback: Value,
465 program: Arc<Program>,
466 policy: Policy,
467 opts: ServeOpts,
468) -> Result<Value, String> {
469 use http_body_util::BodyExt as _;
470 use hyper::server::conn::http1;
471 use hyper::service::service_fn;
472 use hyper_util::rt::{TokioExecutor, TokioIo};
473 use hyper_util::server::conn::auto;
474 use tokio::net::TcpListener as TokioTcpListener;
475
476 let inline_vm = opts.inline_vm;
477 let http2 = opts.http2;
478 let host = opts.host.clone();
479 let routes = Arc::new(routes);
480 let rt = tokio::runtime::Builder::new_multi_thread()
481 .enable_all()
482 .build()
483 .map_err(|e| format!("net.serve_routed: tokio runtime: {e}"))?;
484 rt.block_on(async move {
485 let listener = TokioTcpListener::bind((host.as_str(), port))
486 .await
487 .map_err(|e| format!("net.serve_routed bind {host}:{port}: {e}"))?;
488 eprintln!(
489 "net.serve_routed: listening on http://{host}:{port} ({} routes{}{})",
490 routes.len(),
491 if inline_vm { ", inline-vm" } else { "" },
492 if http2 { ", http1+http2" } else { "" }
493 );
494 loop {
495 let (stream, _) = listener
496 .accept()
497 .await
498 .map_err(|e| format!("net.serve_routed accept: {e}"))?;
499 let io = TokioIo::new(stream);
500 let program = Arc::clone(&program);
501 let policy = policy.clone();
502 let routes = Arc::clone(&routes);
503 let fallback = fallback.clone();
504 tokio::spawn(async move {
505 let program2 = Arc::clone(&program);
506 let policy2 = policy.clone();
507 let routes2 = Arc::clone(&routes);
508 let fallback2 = fallback.clone();
509 let svc = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
510 let program = Arc::clone(&program2);
511 let policy = policy2.clone();
512 let routes = Arc::clone(&routes2);
513 let fallback = fallback2.clone();
514 async move {
515 let (parts, body) = req.into_parts();
516 let body_bytes = body
517 .collect()
518 .await
519 .map(|c| c.to_bytes())
520 .unwrap_or_default();
521 let method = parts.method.as_str().to_string();
522 let path = match parts.uri.path() {
523 "" => "/".to_string(),
524 p => p.to_string(),
525 };
526 let result = if inline_vm {
527 let mut lex_req = build_request_value_parts(&parts, &body_bytes);
528 let (closure, params) = match dispatch_route(&routes, &method, &path) {
529 Some((c, p)) => (c.clone(), p),
530 None => (fallback.clone(), std::collections::BTreeMap::new()),
531 };
532 stamp_path_params(&mut lex_req, params);
533 let handler = DefaultHandler::new(policy)
534 .with_program(Arc::clone(&program));
535 let mut vm = Vm::with_handler(&program, Box::new(handler));
536 let r = vm.invoke_closure_value(closure, vec![lex_req]);
537 Ok(r.map(|v| unpack_response(&mut vm, &v)))
540 } else {
541 tokio::task::spawn_blocking(move || {
542 let mut lex_req = build_request_value_parts(&parts, &body_bytes);
543 let (closure, params) = match dispatch_route(&routes, &method, &path) {
544 Some((c, p)) => (c.clone(), p),
545 None => (fallback.clone(), std::collections::BTreeMap::new()),
546 };
547 stamp_path_params(&mut lex_req, params);
548 let handler = DefaultHandler::new(policy)
549 .with_program(Arc::clone(&program));
550 let mut vm = Vm::with_handler(&program, Box::new(handler));
551 let r = vm.invoke_closure_value(closure, vec![lex_req]);
552 r.map(|v| unpack_response(&mut vm, &v))
553 })
554 .await
555 };
556 Ok::<_, std::convert::Infallible>(match result {
557 Ok(Ok(unpacked)) => build_hyper_response(unpacked),
558 Ok(Err(e)) => error_response(500, &format!("internal error: {e}")),
559 Err(e) => error_response(500, &format!("task panicked: {e}")),
560 })
561 }
562 });
563 let result = if http2 {
564 auto::Builder::new(TokioExecutor::new())
565 .serve_connection(io, svc)
566 .await
567 .map_err(|e| e.to_string())
568 } else {
569 http1::Builder::new()
570 .serve_connection(io, svc)
571 .await
572 .map_err(|e| e.to_string())
573 };
574 if let Err(e) = result {
575 eprintln!("net.serve_routed: connection error: {e}");
576 }
577 });
578 }
579 })
580}
581
582pub(super) fn env_inline_vm() -> bool {
587 match std::env::var("LEX_NET_INLINE_VM") {
588 Ok(v) => {
589 let s = v.trim().to_ascii_lowercase();
590 s == "1" || s == "true"
591 }
592 Err(_) => false,
593 }
594}
595
596#[derive(Debug, Clone)]
602pub(crate) struct ServeOpts {
603 pub(crate) http2: bool,
604 pub(crate) inline_vm: bool,
605 pub(crate) host: String,
606}
607
608impl ServeOpts {
609 pub(super) fn from_env() -> Self {
613 Self {
614 http2: env_http2(),
615 inline_vm: env_inline_vm(),
616 host: "0.0.0.0".to_string(),
617 }
618 }
619
620 pub(super) fn lex_defaults() -> Self {
625 Self {
626 http2: false,
627 inline_vm: false,
628 host: "0.0.0.0".to_string(),
629 }
630 }
631
632 pub(super) fn to_value(&self) -> Value {
634 let mut rec = indexmap::IndexMap::new();
635 rec.insert("http2".to_string(), Value::Bool(self.http2));
636 rec.insert("inline_vm".to_string(), Value::Bool(self.inline_vm));
637 rec.insert("host".to_string(), Value::Str(self.host.clone().into()));
638 Value::record_dynamic(rec)
639 }
640}
641
642pub(super) fn decode_serve_opts(v: &Value) -> Result<ServeOpts, String> {
647 let rec = match v {
648 Value::Record { fields: r, .. } => r,
649 other => return Err(format!("opts must be a Record, got {other:?}")),
650 };
651 let http2 = match rec.get("http2") {
652 Some(Value::Bool(b)) => *b,
653 _ => return Err("opts.http2 must be Bool".into()),
654 };
655 let inline_vm = match rec.get("inline_vm") {
656 Some(Value::Bool(b)) => *b,
657 _ => return Err("opts.inline_vm must be Bool".into()),
658 };
659 let host = match rec.get("host") {
660 Some(Value::Str(s)) => s.to_string(),
661 _ => return Err("opts.host must be Str".into()),
662 };
663 Ok(ServeOpts { http2, inline_vm, host })
664}
665
666pub(super) fn make_tls_config_value(cert_pem: Vec<u8>, key_pem: Vec<u8>) -> Value {
676 let mut rec = indexmap::IndexMap::new();
677 rec.insert("cert".into(), Value::Bytes(cert_pem));
678 rec.insert("key".into(), Value::Bytes(key_pem));
679 Value::record_dynamic(rec)
680}
681
682#[cfg(feature = "quic")]
683pub(super) fn decode_tls_config(v: &Value) -> Result<crate::quic::QuicTls, String> {
684 let rec = match v {
685 Value::Record { fields: r, .. } => r,
686 other => return Err(format!("TlsConfig: expected Record, got {other:?}")),
687 };
688 let cert = match rec.get("cert") {
689 Some(Value::Bytes(b)) => b.to_vec(),
690 _ => return Err("TlsConfig.cert: must be Bytes".into()),
691 };
692 let key = match rec.get("key") {
693 Some(Value::Bytes(b)) => b.to_vec(),
694 _ => return Err("TlsConfig.key: must be Bytes".into()),
695 };
696 Ok(crate::quic::QuicTls { cert_pem: cert, key_pem: key })
697}
698
699pub(super) fn dispatch_tls_from_pem_files(
700 handler: &DefaultHandler,
701 args: Vec<Value>,
702) -> Result<Value, String> {
703 let cert_path = expect_str(args.first())?.to_string();
704 let key_path = expect_str(args.get(1))?.to_string();
705 let cert_resolved = handler.resolve_read_path(&cert_path);
706 let key_resolved = handler.resolve_read_path(&key_path);
707 if !handler.policy.allow_fs_read.is_empty() {
708 let allowed = |p: &std::path::Path| -> bool {
709 handler.policy.allow_fs_read.iter().any(|a| p.starts_with(a))
710 };
711 if !allowed(&cert_resolved) {
712 return Ok(err(Value::Str(
713 format!("tls.from_pem_files: cert `{cert_path}` outside --allow-fs-read").into(),
714 )));
715 }
716 if !allowed(&key_resolved) {
717 return Ok(err(Value::Str(
718 format!("tls.from_pem_files: key `{key_path}` outside --allow-fs-read").into(),
719 )));
720 }
721 }
722 let cert = match std::fs::read(&cert_resolved) {
723 Ok(b) => b,
724 Err(e) => return Ok(err(Value::Str(format!("read cert {cert_path}: {e}").into()))),
725 };
726 let key = match std::fs::read(&key_resolved) {
727 Ok(b) => b,
728 Err(e) => return Ok(err(Value::Str(format!("read key {key_path}: {e}").into()))),
729 };
730 Ok(ok(make_tls_config_value(cert, key)))
731}
732
733#[cfg(feature = "quic")]
734pub(super) fn dispatch_tls_self_signed(args: Vec<Value>) -> Result<Value, String> {
735 let hostname = expect_str(args.first())?.to_string();
736 match crate::quic::self_signed_pem(&hostname) {
737 Ok((cert, key)) => Ok(ok(make_tls_config_value(cert, key))),
738 Err(e) => Ok(err(Value::Str(format!("tls.self_signed: {e}").into()))),
739 }
740}
741
742#[cfg(not(feature = "quic"))]
743pub(super) fn dispatch_tls_self_signed(_args: Vec<Value>) -> Result<Value, String> {
744 Ok(err(Value::Str(
745 "tls.self_signed: lex-runtime was compiled without the `quic` feature (needed for rcgen)".into(),
746 )))
747}
748
749impl DefaultHandler {
750 #[cfg(feature = "quic")]
751 pub(super) fn dispatch_serve_quic_named(&self, args: Vec<Value>) -> Result<Value, String> {
752 let port = match args.first() {
753 Some(Value::Int(n)) if (0..=65535).contains(n) => *n as u16,
754 _ => return Err("net.serve_quic(port, tls, handler): port must be Int 0..=65535".into()),
755 };
756 let tls = decode_tls_config(args.get(1)
757 .ok_or_else(|| "net.serve_quic(port, tls, handler): missing tls".to_string())?)?;
758 let handler_name = expect_str(args.get(2))?.to_string();
759 let program = self.program.clone()
760 .ok_or_else(|| "net.serve_quic requires a Program reference; use DefaultHandler::with_program".to_string())?;
761 let policy = self.policy.clone();
762 crate::quic::serve_http3_named(port, handler_name, tls, program, policy, ServeOpts::from_env())
763 }
764
765 #[cfg(feature = "quic")]
766 pub(super) fn dispatch_serve_quic_fn(&self, args: Vec<Value>) -> Result<Value, String> {
767 let port = match args.first() {
768 Some(Value::Int(n)) if (0..=65535).contains(n) => *n as u16,
769 _ => return Err("net.serve_quic_fn(port, tls, handler): port must be Int 0..=65535".into()),
770 };
771 let tls = decode_tls_config(args.get(1)
772 .ok_or_else(|| "net.serve_quic_fn(port, tls, handler): missing tls".to_string())?)?;
773 let closure = match args.into_iter().nth(2) {
774 Some(c @ Value::Closure { .. }) => c,
775 _ => return Err("net.serve_quic_fn(port, tls, handler): handler must be a closure".into()),
776 };
777 let program = self.program.clone()
778 .ok_or_else(|| "net.serve_quic_fn requires a Program reference; use DefaultHandler::with_program".to_string())?;
779 let policy = self.policy.clone();
780 crate::quic::serve_http3_fn(port, closure, tls, program, policy, ServeOpts::from_env())
781 }
782
783 #[cfg(feature = "quic")]
784 pub(super) fn dispatch_serve_quic_routed(&self, args: Vec<Value>) -> Result<Value, String> {
785 let port = match args.first() {
786 Some(Value::Int(n)) if (0..=65535).contains(n) => *n as u16,
787 _ => return Err("net.serve_quic_routed(port, tls, routes, fallback): port must be Int 0..=65535".into()),
788 };
789 let tls = decode_tls_config(args.get(1)
790 .ok_or_else(|| "net.serve_quic_routed(port, tls, routes, fallback): missing tls".to_string())?)?;
791 let routes_val = args.get(2).cloned()
792 .ok_or_else(|| "net.serve_quic_routed(port, tls, routes, fallback): missing routes".to_string())?;
793 let fallback = match args.into_iter().nth(3) {
794 Some(c @ Value::Closure { .. }) => c,
795 _ => return Err("net.serve_quic_routed(port, tls, routes, fallback): fallback must be a closure".into()),
796 };
797 let routes = decode_routes_arg(routes_val)?;
798 let program = self.program.clone()
799 .ok_or_else(|| "net.serve_quic_routed requires a Program reference; use DefaultHandler::with_program".to_string())?;
800 let policy = self.policy.clone();
801 crate::quic::serve_http3_routed(port, routes, fallback, tls, program, policy, ServeOpts::from_env())
802 }
803
804 #[cfg(not(feature = "quic"))]
805 pub(super) fn dispatch_serve_quic_named(&self, _args: Vec<Value>) -> Result<Value, String> {
806 Err("net.serve_quic: lex-runtime was compiled without the `quic` feature (needed for quinn + h3)".into())
807 }
808 #[cfg(not(feature = "quic"))]
809 pub(super) fn dispatch_serve_quic_fn(&self, _args: Vec<Value>) -> Result<Value, String> {
810 Err("net.serve_quic_fn: lex-runtime was compiled without the `quic` feature (needed for quinn + h3)".into())
811 }
812 #[cfg(not(feature = "quic"))]
813 pub(super) fn dispatch_serve_quic_routed(&self, _args: Vec<Value>) -> Result<Value, String> {
814 Err("net.serve_quic_routed: lex-runtime was compiled without the `quic` feature (needed for quinn + h3)".into())
815 }
816}
817
818pub(super) fn env_http2() -> bool {
828 match std::env::var("LEX_NET_HTTP2") {
829 Ok(v) => {
830 let s = v.trim().to_ascii_lowercase();
831 s == "1" || s == "true"
832 }
833 Err(_) => false,
834 }
835}
836
837pub(crate) fn build_request_value_parts(
839 parts: &hyper::http::request::Parts,
840 body: &bytes::Bytes,
841) -> Value {
842 let method = parts.method.as_str().to_string();
843 let path = parts.uri.path().to_string();
851 let query = parts.uri.query().map(str::to_string).unwrap_or_default();
852 let mut headers_map = std::collections::BTreeMap::new();
853 for (name, val) in &parts.headers {
854 if let Ok(v) = val.to_str() {
855 headers_map.insert(
856 lex_bytecode::MapKey::Str(name.as_str().to_ascii_lowercase()),
857 Value::Str(v.to_string().into()),
858 );
859 }
860 }
861 let body_str = String::from_utf8_lossy(body).into_owned();
862 let mut rec = indexmap::IndexMap::new();
863 rec.insert("method".into(), Value::Str(method.into()));
864 rec.insert("path".into(), Value::Str(path.into()));
865 rec.insert("query".into(), Value::Str(query.into()));
866 rec.insert("body".into(), Value::Str(body_str.into()));
867 rec.insert("headers".into(), Value::Map(headers_map));
868 rec.insert("path_params".into(), Value::Map(std::collections::BTreeMap::new()));
869 Value::record_dynamic(rec)
870}
871
872pub(super) fn build_request_value_tiny(req: &mut tiny_http::Request) -> Value {
874 let method = format!("{:?}", req.method()).to_uppercase();
875 let url = req.url().to_string();
876 let (path, query) = match url.split_once('?') {
877 Some((p, q)) => (p.to_string(), q.to_string()),
878 None => (url, String::new()),
879 };
880 let mut headers_map = std::collections::BTreeMap::new();
881 for h in req.headers() {
882 headers_map.insert(
883 lex_bytecode::MapKey::Str(h.field.as_str().as_str().to_ascii_lowercase()),
884 Value::Str(h.value.as_str().to_string().into()),
885 );
886 }
887 let mut body = String::new();
888 let _ = req.as_reader().read_to_string(&mut body);
889 let mut rec = indexmap::IndexMap::new();
890 rec.insert("method".into(), Value::Str(method.into()));
891 rec.insert("path".into(), Value::Str(path.into()));
892 rec.insert("query".into(), Value::Str(query.into()));
893 rec.insert("body".into(), Value::Str(body.into()));
894 rec.insert("headers".into(), Value::Map(headers_map));
895 rec.insert("path_params".into(), Value::Map(std::collections::BTreeMap::new()));
896 Value::record_dynamic(rec)
897}
898
899pub(crate) fn unpack_response(vm: &mut Vm, v: &Value) -> UnpackedResponse {
900 if !matches!(v, Value::Record { .. } | Value::ArenaRecord { .. }) {
906 return (
907 500,
908 ResponseBodyOut::Str(format!("handler returned non-record: {v:?}")),
909 vec![],
910 );
911 }
912
913 let status = vm.get_record_field(v, "status").and_then(|s| match s {
914 Value::Int(n) => Some(n as u16),
915 _ => None,
916 }).unwrap_or(200);
917
918 let body = match vm.get_record_field(v, "body") {
922 Some(Value::Variant { name, mut args }) if args.len() == 1 => {
923 let inner = args.pop().unwrap();
924 match (name.as_str(), inner) {
925 ("BodyStr", Value::Str(s)) => ResponseBodyOut::Str(s.to_string()),
927 ("BodyStream", iter_v) => {
928 let drained = materialize_lazy_iter(vm, iter_v);
929 ResponseBodyOut::TextChunks(drain_iter_str(&drained))
930 }
931 ("BodyBytes", iter_v) => {
932 let drained = materialize_lazy_iter(vm, iter_v);
933 ResponseBodyOut::BytesChunks(drain_iter_bytes(&drained))
934 }
935 _ => ResponseBodyOut::Str(String::new()),
936 }
937 }
938 Some(Value::Str(s)) => ResponseBodyOut::Str(s.to_string()),
944 _ => ResponseBodyOut::Str(String::new()),
945 };
946
947 let headers: Vec<(String, String)> = match vm.get_record_field(v, "headers") {
948 Some(Value::Map(hmap)) => hmap.iter().filter_map(|(k, val)| {
949 if let (lex_bytecode::MapKey::Str(name), Value::Str(s)) = (k, val) {
950 Some((name.clone(), s.to_string()))
951 } else {
952 None
953 }
954 }).collect(),
955 _ => vec![],
956 };
957
958 (status, body, headers)
959}
960
961pub(super) type HyperRespBody =
962 http_body_util::combinators::BoxBody<bytes::Bytes, std::convert::Infallible>;
963
964pub(super) fn build_hyper_response(
974 (status, body, headers): UnpackedResponse,
975) -> hyper::Response<HyperRespBody> {
976 use http_body_util::BodyExt as _;
977 let boxed_body: HyperRespBody = match body {
978 ResponseBodyOut::Str(s) => {
979 http_body_util::Full::new(bytes::Bytes::from(s.into_bytes())).boxed()
980 }
981 ResponseBodyOut::TextChunks(chunks) | ResponseBodyOut::BytesChunks(chunks) => {
982 HyperChunkedBody::from(chunks).boxed()
983 }
984 };
985 let mut builder = hyper::Response::builder().status(status);
986 for (name, val) in headers {
987 builder = builder.header(name, val);
988 }
989 builder
990 .body(boxed_body)
991 .unwrap_or_else(|_| error_response(500, "response build error"))
992}
993
994pub(super) fn error_response(status: u16, msg: &str) -> hyper::Response<HyperRespBody> {
995 use http_body_util::BodyExt as _;
996 hyper::Response::builder()
997 .status(status)
998 .body(
999 http_body_util::Full::new(bytes::Bytes::from(msg.to_owned()))
1000 .boxed(),
1001 )
1002 .unwrap_or_else(|_| {
1003 use http_body_util::BodyExt as _;
1004 hyper::Response::new(http_body_util::Empty::new().map_err(|e| match e {}).boxed())
1005 })
1006}
1007
1008pub(super) struct HyperChunkedBody {
1011 pub(super) chunks: std::collections::VecDeque<Vec<u8>>,
1012}
1013
1014impl From<Vec<Vec<u8>>> for HyperChunkedBody {
1015 fn from(chunks: Vec<Vec<u8>>) -> Self {
1016 Self {
1017 chunks: chunks.into_iter().filter(|c| !c.is_empty()).collect(),
1018 }
1019 }
1020}
1021
1022impl hyper::body::Body for HyperChunkedBody {
1023 type Data = bytes::Bytes;
1024 type Error = std::convert::Infallible;
1025
1026 fn poll_frame(
1027 mut self: std::pin::Pin<&mut Self>,
1028 _cx: &mut std::task::Context<'_>,
1029 ) -> std::task::Poll<Option<Result<hyper::body::Frame<Self::Data>, Self::Error>>> {
1030 match self.chunks.pop_front() {
1031 Some(chunk) => std::task::Poll::Ready(Some(Ok(hyper::body::Frame::data(
1032 bytes::Bytes::from(chunk),
1033 )))),
1034 None => std::task::Poll::Ready(None),
1035 }
1036 }
1037}
1038
1039pub(super) fn respond_with_body_tls(
1043 req: tiny_http::Request,
1044 status: u16,
1045 body: ResponseBodyOut,
1046 headers: Vec<(String, String)>,
1047) {
1048 let tiny_headers: Vec<tiny_http::Header> = headers
1049 .into_iter()
1050 .filter_map(|(name, val)| format!("{name}: {val}").parse::<tiny_http::Header>().ok())
1051 .collect();
1052 match body {
1053 ResponseBodyOut::Str(s) => {
1054 let mut response = tiny_http::Response::from_string(s).with_status_code(status);
1055 for h in tiny_headers {
1056 response.add_header(h);
1057 }
1058 let _ = req.respond(response);
1059 }
1060 ResponseBodyOut::TextChunks(chunks) | ResponseBodyOut::BytesChunks(chunks) => {
1061 let reader = ChunkReader::new(chunks);
1062 let response = tiny_http::Response::new(
1063 tiny_http::StatusCode(status),
1064 tiny_headers,
1065 reader,
1066 None,
1067 None,
1068 );
1069 let _ = req.respond(response);
1070 }
1071 }
1072}
1073
1074pub(crate) type UnpackedResponse = (u16, ResponseBodyOut, Vec<(String, String)>);
1084
1085pub(crate) enum ResponseBodyOut {
1086 Str(String),
1087 TextChunks(Vec<Vec<u8>>),
1091 BytesChunks(Vec<Vec<u8>>),
1094}
1095
1096pub(super) fn drain_iter_str(v: &Value) -> Vec<Vec<u8>> {
1109 match v {
1110 Value::Variant { name, args }
1111 if name == "__IterEager" && args.len() == 2 =>
1112 {
1113 if let (Value::List(items), Value::Int(idx)) = (&args[0], &args[1]) {
1114 items.iter().skip(*idx as usize).filter_map(|item| {
1115 if let Value::Str(s) = item { Some(s.as_bytes().to_vec()) } else { None }
1116 }).collect()
1117 } else {
1118 Vec::new()
1119 }
1120 }
1121 _ => Vec::new(),
1122 }
1123}
1124
1125pub(super) fn drain_iter_bytes(v: &Value) -> Vec<Vec<u8>> {
1129 match v {
1130 Value::Variant { name, args }
1131 if name == "__IterEager" && args.len() == 2 =>
1132 {
1133 if let (Value::List(items), Value::Int(idx)) = (&args[0], &args[1]) {
1134 items.iter().skip(*idx as usize).filter_map(|item| {
1135 if let Value::List(ints) = item {
1136 Some(ints.iter().filter_map(|i| match i {
1137 Value::Int(n) => Some((*n & 0xff) as u8),
1138 _ => None,
1139 }).collect::<Vec<u8>>())
1140 } else {
1141 None
1142 }
1143 }).collect()
1144 } else {
1145 Vec::new()
1146 }
1147 }
1148 _ => Vec::new(),
1149 }
1150}
1151
1152pub(super) fn materialize_lazy_iter(vm: &mut Vm, v: Value) -> Value {
1165 let mut current = v;
1166 let mut items: Vec<Value> = Vec::new();
1167 loop {
1168 match current {
1169 Value::Variant { name, args } if name == "__IterLazy" && args.len() == 2 => {
1170 let seed = args[0].clone();
1171 let step = args[1].clone();
1172 match vm.invoke_closure_value(step.clone(), vec![seed]) {
1173 Ok(Value::Variant { name: opt, args: opt_args })
1174 if opt == "None" =>
1175 {
1176 let _ = opt_args;
1177 break;
1178 }
1179 Ok(Value::Variant { name: opt, args: opt_args })
1180 if opt == "Some" && opt_args.len() == 1 =>
1181 {
1182 if let Value::Tuple(pair) = &opt_args[0] {
1183 if pair.len() == 2 {
1184 items.push(pair[0].clone());
1185 current = Value::Variant {
1186 name: "__IterLazy".to_string(),
1187 args: vec![pair[1].clone(), step],
1188 };
1189 continue;
1190 }
1191 }
1192 break;
1194 }
1195 _ => break,
1196 }
1197 }
1198 other => {
1201 if items.is_empty() {
1202 return other;
1203 }
1204 let _ = other;
1207 break;
1208 }
1209 }
1210 }
1211 Value::Variant {
1212 name: "__IterEager".to_string(),
1213 args: vec![
1214 Value::List(items.into_iter().collect()),
1215 Value::Int(0),
1216 ],
1217 }
1218}
1219
1220
1221pub(super) struct ChunkReader {
1227 pub(super) chunks: std::collections::VecDeque<Vec<u8>>,
1228 pub(super) cursor: usize,
1229}
1230
1231impl ChunkReader {
1232 pub(super) fn new(chunks: Vec<Vec<u8>>) -> Self {
1233 Self {
1234 chunks: chunks.into_iter().filter(|c| !c.is_empty()).collect(),
1235 cursor: 0,
1236 }
1237 }
1238}
1239
1240impl std::io::Read for ChunkReader {
1241 fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
1242 loop {
1243 let Some(front) = self.chunks.front() else {
1244 return Ok(0);
1245 };
1246 let remaining = &front[self.cursor..];
1247 if remaining.is_empty() {
1248 self.chunks.pop_front();
1249 self.cursor = 0;
1250 continue;
1251 }
1252 let n = remaining.len().min(buf.len());
1253 buf[..n].copy_from_slice(&remaining[..n]);
1254 self.cursor += n;
1255 if self.cursor >= front.len() {
1256 self.chunks.pop_front();
1257 self.cursor = 0;
1258 }
1259 return Ok(n);
1260 }
1261 }
1262}
1263
1264#[cfg(test)]
1271mod unpack_response_tests {
1272 use super::*;
1273 use std::sync::Arc;
1274 use indexmap::IndexMap;
1275 use lex_bytecode::{Const, Op, Program, Value};
1276 use lex_bytecode::program::{Function, ZERO_BODY_HASH};
1277 use lex_bytecode::vm::Vm;
1278
1279 fn build_arena_response_program() -> Arc<Program> {
1284 let constants = vec![
1285 Const::FieldName("status".into()), Const::FieldName("body".into()), Const::Int(200), Const::VariantName("BodyStr".into()), Const::Str("hello".into()), ];
1291 let mut function_names = IndexMap::new();
1292 function_names.insert("handler".to_string(), 0);
1293 Arc::new(Program {
1294 constants,
1295 functions: vec![Function {
1296 name: "handler".into(),
1297 arity: 0,
1298 locals_count: 0,
1299 code: vec![
1300 Op::PushConst(2), Op::PushConst(4), Op::MakeVariant { name_idx: 3, arity: 1 }, Op::AllocArenaRecord { shape_idx: 0, field_count: 2 }, Op::Return,
1305 ],
1306 effects: vec![],
1307 body_hash: ZERO_BODY_HASH,
1308 refinements: vec![],
1309 field_ic_sites: 0,
1310 }],
1311 function_names,
1312 module_aliases: IndexMap::new(),
1313 entry: Some(0),
1314 record_shapes: vec![vec![0, 1]], })
1316 }
1317
1318 #[test]
1324 fn unpack_response_reads_arena_record_via_slab() {
1325 let p = build_arena_response_program();
1326 let mut vm = Vm::new(&p);
1327 let scope = vm.enter_request_scope();
1328
1329 let resp = vm.invoke(p.function_names["handler"], vec![]).unwrap();
1330 assert!(matches!(resp, Value::ArenaRecord { .. }),
1333 "expected ArenaRecord (slab path), got {resp:?}");
1334
1335 let (status, body, headers) = unpack_response(&mut vm, &resp);
1336 vm.exit_request_scope(scope);
1337
1338 assert_eq!(status, 200);
1339 assert!(headers.is_empty());
1340 match body {
1341 ResponseBodyOut::Str(s) => assert_eq!(s, "hello"),
1342 _ => panic!("expected BodyStr"),
1343 }
1344 }
1345
1346 #[test]
1351 fn unpack_response_reads_heap_record() {
1352 let p = build_arena_response_program();
1353 let mut vm = Vm::new(&p);
1354
1355 let resp = vm.invoke(p.function_names["handler"], vec![]).unwrap();
1357 assert!(matches!(resp, Value::Record { .. }),
1358 "expected heap Record (fallback path), got {resp:?}");
1359
1360 let (status, body, headers) = unpack_response(&mut vm, &resp);
1361 assert_eq!(status, 200);
1362 assert!(headers.is_empty());
1363 match body {
1364 ResponseBodyOut::Str(s) => assert_eq!(s, "hello"),
1365 _ => panic!("expected BodyStr"),
1366 }
1367 }
1368
1369 #[test]
1372 fn unpack_response_falls_back_to_500_on_non_record() {
1373 let p = build_arena_response_program();
1374 let mut vm = Vm::new(&p);
1375 let v = Value::Int(7);
1376 let (status, _body, _headers) = unpack_response(&mut vm, &v);
1377 assert_eq!(status, 500);
1378 }
1379}