pub mod middleware;
pub mod openapi;
#[cfg(test)]
mod httpapi_test;
use std::collections::HashSet;
use std::sync::Arc;
use axum::extract::Request;
use axum::http::{Method, StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::{Router, delete, get, head, options, patch, post, put};
use serde_json::{Map, Value};
use crate::config::Config;
use crate::ctx::Ctx;
use crate::errors;
use crate::registry::Registry;
use crate::spec::entry::Entry;
pub const MAX_BODY_BYTES: usize = 1 << 20;
pub fn router(reg: &Registry, ctx: Arc<Ctx>) -> errors::Result<Router> {
router_with(
reg,
ctx,
std::collections::HashMap::new(),
false,
&crate::cli::bin_name(),
crate::version::version(),
)
}
pub(crate) fn router_with(
reg: &Registry,
ctx: Arc<Ctx>,
defaults: std::collections::HashMap<String, String>,
no_server_headers: bool,
app_name: &str,
app_version: &str,
) -> errors::Result<Router> {
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
router_inner(reg, ctx, defaults, no_server_headers, app_name, app_version)
}))
.unwrap_or_else(|p| {
let msg = p
.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| p.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "route registration panic".to_string());
Err(errors::Error::new(
errors::Kind::Internal,
format!("httpapi: route conflicts with an existing route ({msg})"),
))
})
}
fn router_inner(
reg: &Registry,
ctx: Arc<Ctx>,
defaults: std::collections::HashMap<String, String>,
no_server_headers: bool,
app_name: &str,
app_version: &str,
) -> errors::Result<Router> {
let mut r: Router = Router::new();
let mut seen: HashSet<(String, String)> = HashSet::new();
for e in reg.all() {
if e.http.skip || e.cli.daemon {
continue; }
if e.http.path.is_empty() {
continue; }
for method in http_methods(&e) {
if !seen.insert((method.clone(), e.http.path.clone())) {
return Err(errors::Error::new(
errors::Kind::Internal,
format!(
"httpapi: route {:?} {:?} conflicts with an existing route",
method, e.http.path
),
));
}
let h = handle_entry_c(
e.clone(),
Arc::clone(&ctx),
Arc::new(defaults.clone()),
no_server_headers,
);
let mr = method_router(&method, h)?;
r = r.route(&e.http.path, mr);
}
}
r = r
.route("/healthz", get(healthz))
.route(
"/openapi.json",
get(openapi::openapi_handler(reg, app_name, app_version)),
)
.fallback(missing_route);
Ok(r)
}
pub(crate) fn http_methods(e: &crate::spec::Entry) -> Vec<String> {
let m = e.http.method.trim();
if m.is_empty() {
return vec!["GET".to_string(), "POST".to_string()];
}
m.split(',')
.map(|s| s.trim().to_uppercase())
.filter(|s| !s.is_empty())
.collect()
}
async fn missing_route(req: Request) -> Response {
let lang = req
.headers()
.get(header::ACCEPT_LANGUAGE)
.and_then(|v| v.to_str().ok())
.and_then(crate::lang::parse_accept_language);
let msg = match lang {
Some(l) => crate::lang::t_lang(l, "http.err_not_found"),
None => crate::lang::t("http.err_not_found"),
};
write_error(StatusCode::NOT_FOUND, &msg)
}
fn method_router<H, T>(method: &str, h: H) -> errors::Result<axum::routing::MethodRouter>
where
H: axum::handler::Handler<T, ()>,
T: 'static,
{
Ok(match method {
"GET" => get(h),
"POST" => post(h),
"PUT" => put(h),
"PATCH" => patch(h),
"DELETE" => delete(h),
"HEAD" => head(h),
"OPTIONS" => options(h),
other => {
return Err(errors::Error::new(
errors::Kind::Internal,
format!("httpapi: unsupported HTTP method {other:?}"),
));
}
})
}
#[derive(Clone)]
pub struct EntryHandler {
pub(crate) entry: Arc<Entry>,
pub(crate) ctx: Arc<Ctx>,
pub(crate) defaults: Arc<std::collections::HashMap<String, String>>,
pub(crate) no_server_headers: bool,
}
impl axum::handler::Handler<(), ()> for EntryHandler {
type Future = std::pin::Pin<Box<dyn std::future::Future<Output = Response> + Send>>;
fn call(self, req: Request, _state: ()) -> Self::Future {
Box::pin(async move {
handle_request(
&self.entry,
&self.ctx,
&self.defaults,
self.no_server_headers,
req,
)
.await
})
}
}
pub fn handler_for(entry: Arc<Entry>) -> EntryHandler {
handle_entry_c(
entry,
Arc::new(Ctx::new()),
Arc::new(std::collections::HashMap::new()),
false,
)
}
fn handle_entry_c(
entry: Arc<Entry>,
ctx: Arc<Ctx>,
defaults: Arc<std::collections::HashMap<String, String>>,
no_server_headers: bool,
) -> EntryHandler {
EntryHandler {
entry,
ctx,
defaults,
no_server_headers,
}
}
async fn handle_request(
entry: &Arc<Entry>,
ctx: &Arc<Ctx>,
defaults: &std::collections::HashMap<String, String>,
no_server_headers: bool,
req: Request,
) -> Response {
let (parts, body) = req.into_parts();
let method = parts.method.clone();
let path = parts.uri.path().to_string();
let query = parts.uri.query().unwrap_or_default().to_string();
let headers = parts.headers.clone();
let req_lang = headers
.get(header::ACCEPT_LANGUAGE)
.and_then(|v| v.to_str().ok())
.and_then(crate::lang::parse_accept_language);
let localize = |key: &str| match req_lang {
Some(l) => crate::lang::t_lang(l, key),
None => crate::lang::t(key),
};
let mut m: Map<String, Value> = Map::new();
for (k, v) in entry.http_defaults() {
m.insert(k, v);
}
for (k, v) in defaults.iter() {
if !m.contains_key(k) {
m.insert(k.clone(), Value::String(v.clone()));
}
}
let body_bytes: Vec<u8> = if method != Method::GET && method != Method::HEAD {
collect_body(body, MAX_BODY_BYTES).await
} else {
Vec::new()
};
if !body_bytes.is_empty() {
let json_declared = headers
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.map(|v| v.contains("application/json"))
.unwrap_or(false);
match serde_json::from_slice::<serde_json::Value>(&body_bytes) {
Ok(Value::Object(parsed)) => {
for (k, v) in parsed {
m.insert(k, v);
}
}
Ok(_) => {
return write_error(StatusCode::BAD_REQUEST, &localize("http.err_invalid_json"));
}
Err(_) if json_declared => {
return write_error(StatusCode::BAD_REQUEST, &localize("http.err_invalid_json"));
}
Err(_) => {
}
}
}
let mut form_parsed: Option<Vec<(String, String)>> = None;
for f in &entry.root.children {
if f.skip {
if f.http.location == "header"
&& let Some(v) = headers.get(http_name(f)).and_then(|v| v.to_str().ok())
&& !v.is_empty()
{
m.insert(f.name.clone(), Value::String(v.to_string()));
}
continue;
}
match f.http.location.as_str() {
"" | "query" => {
if let Some(vs) = query_values(&query, &f.json_name) {
if is_string_slice(f) {
m.insert(
f.json_name.clone(),
Value::Array(vs.iter().map(|s| Value::String(s.clone())).collect()),
);
} else if let Some(first) = vs.first() {
m.insert(f.json_name.clone(), Value::String(first.clone()));
}
}
}
"header" => {
if let Some(v) = headers.get(http_name(f)).and_then(|v| v.to_str().ok())
&& !v.is_empty()
{
m.insert(f.json_name.clone(), Value::String(v.to_string()));
}
}
"path" => {
if let Some(v) = path_param(&entry.http.path, &path, http_name(f)) {
m.insert(f.json_name.clone(), Value::String(v));
}
}
"form" => {
if form_parsed.is_none() {
form_parsed = Some(crate::httpapi::parse_form_query(&String::from_utf8_lossy(
&body_bytes,
)));
}
if let Some((_, v)) = form_parsed
.as_ref()
.unwrap()
.iter()
.find(|(k, _)| k == &f.json_name)
{
m.insert(f.json_name.clone(), Value::String(v.clone()));
}
}
_ => {}
}
}
let per_req_ctx = match req_lang {
Some(l) => Arc::new(ctx.with_language(l.as_str())),
None => Arc::clone(ctx),
};
let started = std::time::Instant::now();
let mut resp = match (entry.invoke)(&per_req_ctx, &m) {
Ok(out) => {
let s = serde_json::to_string_pretty(&out).unwrap_or_else(|_| "null".to_string());
json_response(StatusCode::OK, &s)
}
Err(e) => write_coded_error(&e),
};
if !no_server_headers {
let h = resp.headers_mut();
if let Ok(v) = header::HeaderValue::from_str(&entry.name) {
let _ = h.insert(header::HeaderName::from_static("x-xyz-command"), v);
}
let dur = started.elapsed().as_millis().to_string();
if let Ok(v) = header::HeaderValue::from_str(&dur) {
let _ = h.insert(header::HeaderName::from_static("x-xyz-duration-ms"), v);
}
}
resp
}
pub(crate) fn http_name(f: &crate::spec::FieldMeta) -> &str {
if !f.http.name.is_empty() {
return f.http.name.as_str();
}
if !f.json_name.is_empty() {
return f.json_name.as_str();
}
f.name.as_str()
}
fn is_string_slice(f: &crate::spec::FieldMeta) -> bool {
matches!(f.kind, crate::spec::FieldKind::Slice)
}
pub(crate) fn parse_form_query(s: &str) -> Vec<(String, String)> {
s.split('&')
.filter(|p| !p.is_empty())
.map(|pair| match pair.split_once('=') {
Some((k, v)) => (pct_decode(k), pct_decode(v)),
None => (pct_decode(pair), String::new()),
})
.collect()
}
fn query_values(query: &str, key: &str) -> Option<Vec<String>> {
let pairs = parse_form_query(query);
let hit: Vec<String> = pairs
.iter()
.filter(|(k, _)| k == key)
.map(|(_, v)| v.clone())
.collect();
if hit.is_empty() { None } else { Some(hit) }
}
fn path_param(template: &str, path: &str, name: &str) -> Option<String> {
let tsegs: Vec<&str> = template.split('/').collect();
let psegs: Vec<&str> = path.split('/').collect();
if tsegs.len() != psegs.len() {
return None;
}
for (t, p) in tsegs.iter().zip(psegs.iter()) {
if let Some(inner) = t.strip_prefix('{').and_then(|s| s.strip_suffix('}'))
&& inner == name
{
return Some(pct_decode(p));
}
}
None
}
fn pct_decode(s: &str) -> String {
let bytes = s.as_bytes();
let mut out = Vec::with_capacity(bytes.len());
let mut i = 0;
while i < bytes.len() {
if bytes[i] == b'%' && i + 2 < bytes.len() + 1 {
let h = hex_val(bytes.get(i + 1));
let l = hex_val(bytes.get(i + 2));
if let (Some(h), Some(l)) = (h, l) {
out.push(h * 16 + l);
i += 3;
continue;
}
}
if bytes[i] == b'+' {
out.push(b' ');
} else {
out.push(bytes[i]);
}
i += 1;
}
String::from_utf8_lossy(&out).into_owned()
}
fn hex_val(b: Option<&u8>) -> Option<u8> {
match b.copied()? {
b'0'..=b'9' => Some(b.unwrap() - b'0'),
b'a'..=b'f' => Some(b.unwrap() - b'a' + 10),
b'A'..=b'F' => Some(b.unwrap() - b'A' + 10),
_ => None,
}
}
fn write_coded_error(e: &errors::Error) -> Response {
let status =
StatusCode::from_u16(errors::error_status(e)).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
let mut body = errors::error_body(e);
if body.error.is_empty() {
body.error = status.canonical_reason().unwrap_or("error").to_string();
}
let s = serde_json::to_string(&body).unwrap_or_else(|_| "{\"error\":\"\"}".to_string());
json_response(status, &s)
}
fn write_error(status: StatusCode, msg: &str) -> Response {
let msg = if msg.is_empty() {
status.canonical_reason().unwrap_or("error").to_string()
} else {
msg.to_string()
};
let body = serde_json::to_string(&serde_json::json!({ "error": msg }))
.unwrap_or_else(|_| "{\"error\":\"\"}".to_string());
json_response(status, &body)
}
fn json_response(status: StatusCode, body: &str) -> Response {
let mut b = body.to_string();
b.push('\n');
(
status,
[(header::CONTENT_TYPE, "application/json; charset=utf-8")],
b,
)
.into_response()
}
async fn healthz() -> Response {
json_response(StatusCode::OK, "{\"status\":\"ok\"}")
}
async fn collect_body(body: axum::body::Body, cap: usize) -> Vec<u8> {
use futures_util::StreamExt as _;
let mut out = Vec::new();
let mut stream = body.into_data_stream();
while let Some(Ok(chunk)) = stream.next().await {
let rem = cap.saturating_sub(out.len());
if rem == 0 {
break;
}
out.extend_from_slice(&chunk[..chunk.len().min(rem)]);
}
out
}
pub(crate) fn normalize_addr(addr: &str) -> String {
if let Some(rest) = addr.strip_prefix(':')
&& !rest.contains(':')
{
return format!("0.0.0.0:{rest}");
}
addr.to_string()
}
pub(crate) fn serve(
ctx: &Ctx,
reg: &Registry,
args: &[String],
cfg: Config,
mount_mcp: bool,
) -> i32 {
let cfg = crate::builtins::parse_serve_args(args, cfg);
let sctx = Arc::new(ctx.clone());
let mut router = match router_with(
reg,
Arc::clone(&sctx),
cfg.channel_defaults.clone(),
cfg.no_server_headers,
&cfg.resolved_name(),
&cfg.resolved_version(),
) {
Ok(r) => r,
Err(e) => {
crate::logx::errorf(format_args!("{e}"));
return 2;
}
};
let mut mcp_note = String::new();
#[cfg(feature = "mcp")]
{
if mount_mcp
&& let Some(m) = crate::mcp::mountable(
reg,
&crate::mcp::Options {
name: cfg.resolved_name(),
version: cfg.resolved_version(),
bearer_tokens: cfg.bearer_tokens.clone(),
default_ctx: Some(sctx.clone()),
defaults: cfg.channel_defaults.clone(),
..Default::default()
},
)
{
router = router.nest_service("/mcp", m);
mcp_note = " + /mcp".to_string();
}
}
if !cfg.cors_origins.is_empty() {
crate::logx::debugf(format_args!(
"{}",
crate::lang::tf("log.cors_on", &[&format!("{:?}", cfg.cors_origins)])
));
}
let router = middleware::apply(
router,
cfg.bearer_tokens.clone(),
cfg.cors_origins.clone(),
cfg.timeout,
);
let router = router.layer(axum::middleware::from_fn(middleware::server_context_mw(
cfg.resolved_name(),
cfg.resolved_version(),
cfg.response_headers.clone(),
!cfg.no_server_headers,
)));
let tls_on = !cfg.cert_file.is_empty() || !cfg.key_file.is_empty();
if tls_on && (cfg.cert_file.is_empty() || cfg.key_file.is_empty()) {
crate::logx::errorf(format_args!("TLS requires both --tls-cert and --tls-key"));
return 2;
}
let scheme = if tls_on { "https" } else { "http" };
crate::logx::infof(format_args!(
"{}",
crate::lang::tf("log.serve_listening", &[scheme, &cfg.addr, &mcp_note])
));
let rt = match tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
{
Ok(rt) => rt,
Err(e) => {
crate::logx::errorf(format_args!("{e}"));
return 1;
}
};
rt.block_on(async move {
let listener =
match tokio::net::TcpListener::bind(&crate::httpapi::normalize_addr(&cfg.addr)).await {
Ok(l) => l,
Err(e) => {
crate::logx::errorf(format_args!("{e}"));
return 1;
}
};
let handle = axum_server::Handle::new();
let cert_file = cfg.cert_file.clone();
let key_file = cfg.key_file.clone();
let serve = {
let router = router.clone();
let handle = handle.clone();
move || async move {
if tls_on {
let tls_config = match load_rustls_config(&cert_file, &key_file) {
Ok(c) => c,
Err(e) => {
crate::logx::errorf(format_args!("{e}"));
return 1;
}
};
let rustls_cfg = axum_server::tls_rustls::RustlsConfig::from_config(
std::sync::Arc::new(tls_config),
);
let server = match axum_server::from_tcp_rustls(
listener.into_std().unwrap(),
rustls_cfg,
) {
Ok(s) => s,
Err(e) => {
crate::logx::errorf(format_args!("{e}"));
return 1;
}
};
let _ = server
.handle(handle.clone())
.serve(router.clone().into_make_service())
.await;
} else {
let server = match axum_server::from_tcp(listener.into_std().unwrap()) {
Ok(s) => s,
Err(e) => {
crate::logx::errorf(format_args!("{e}"));
return 1;
}
};
let _ = server
.handle(handle.clone())
.serve(router.clone().into_make_service())
.await;
};
0
}
};
let _ = &serve;
let server_handle = handle.clone();
let serve_fut = tokio::spawn(async move { serve().await });
let cancel_fut = async move {
sctx.cancelled_async().await;
server_handle.graceful_shutdown(Some(std::time::Duration::from_secs(5)));
};
tokio::select! {
join_res = serve_fut => join_res.unwrap_or(1),
_ = cancel_fut => {
crate::logx::infof(format_args!("{}", crate::lang::t("log.graceful")));
0
}
}
})
}
#[cfg(feature = "http-stack")]
fn load_rustls_config(cert_file: &str, key_file: &str) -> errors::Result<rustls::ServerConfig> {
use std::io::BufReader;
let certs = rustls_pemfile::certs(&mut BufReader::new(
std::fs::File::open(cert_file).map_err(errors::Error::from)?,
))
.collect::<Result<Vec<_>, _>>()
.map_err(|e| errors::Error::new(errors::Kind::Internal, format!("cert parse: {e}")))?;
let key = rustls_pemfile::private_key(&mut BufReader::new(
std::fs::File::open(key_file).map_err(errors::Error::from)?,
))
.map_err(|e| errors::Error::new(errors::Kind::Internal, format!("key parse: {e}")))?
.ok_or_else(|| {
errors::Error::new(errors::Kind::Internal, "no private key found".to_string())
})?;
let _ = rustls::crypto::ring::default_provider().install_default();
let config = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certs, key)
.map_err(|e| errors::Error::new(errors::Kind::Internal, format!("tls config: {e}")))?;
Ok(config)
}