Menu
AkurAI-Build
publicLatest change e5121cbf4cf5b44c00928705f6f3f3988d273833 - server: surface vector capacity/validation errors in the 400 body (closes #7) by Olafur Bui
use std::{
borrow::Cow,
collections::{BTreeMap, HashSet},
fs,
io::Read,
net::SocketAddr,
path::{Path, PathBuf},
sync::Arc,
time::Duration,
};
use crate::artifact::{self, CacheClass, Manifest, Resolution};
use anyhow::{Context, Result, ensure};
use axum::{
Json, Router,
body::{Body, Bytes},
extract::{DefaultBodyLimit, OriginalUri, Request, State},
http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri, header},
middleware::{self, Next},
response::{Html, IntoResponse, Response},
routing::{get, post, put},
};
use minijinja::{AutoEscape, Environment, context};
use percent_encoding::percent_decode_str;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use subtle::ConstantTimeEq;
use tower::limit::GlobalConcurrencyLimitLayer;
use tower_http::{compression::CompressionLayer, timeout::TimeoutLayer, trace::TraceLayer};
use tracing::{error, info, warn};
use zeroize::Zeroizing;
use crate::db::{
Database, VectorInputError, VectorMatch, validate_vector_id, validate_vector_search,
validate_vector_upsert,
};
use crate::fsguard;
const MAX_BODY_BYTES: usize = 1024 * 1024;
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
const MAX_CONCURRENT_REQUESTS: usize = 1024;
#[derive(Clone)]
pub struct ServeOptions {
pub address: SocketAddr,
pub pages: PathBuf,
pub public: PathBuf,
pub tenant: String,
pub model: String,
pub development: bool,
pub api_token: Arc<Zeroizing<String>>,
pub public_origin: Option<String>,
}
#[derive(Clone)]
struct AppState {
database: Database,
pages: Arc<PageRouter>,
pages_dir: PathBuf,
tenant: String,
model: String,
development: bool,
api_token: Arc<Zeroizing<String>>,
public_origin: Option<String>,
loopback: bool,
public_dir: PathBuf,
public_assets: Arc<BTreeMap<String, StaticFile>>,
search_gate: Arc<tokio::sync::Semaphore>,
write_gate: Arc<tokio::sync::Semaphore>,
}
#[derive(Clone)]
struct StaticState {
manifest: Arc<Manifest>,
files: Arc<BTreeMap<String, StaticFile>>,
}
#[derive(Clone)]
struct StaticFile {
bytes: Bytes,
etag: String,
}
pub struct PageRouter {
pages: Vec<Page>,
environment: Environment<'static>,
not_found: Option<String>,
internal_error: Option<String>,
}
#[derive(Debug, Clone)]
struct Page {
template_name: String,
pattern: String,
segments: Vec<Segment>,
}
#[derive(Debug, Clone, Eq, PartialEq)]
enum Segment {
Static(String),
Parameter(String),
CatchAll(String),
OptionalCatchAll(String),
}
#[derive(Debug)]
struct PageMatch {
template_name: String,
params: BTreeMap<String, String>,
}
#[derive(Debug, Deserialize)]
struct VectorUpsert {
id: String,
content: String,
embedding: Vec<f32>,
}
#[derive(Debug, Deserialize)]
struct VectorSearch {
embedding: Vec<f32>,
#[serde(default = "default_search_limit")]
limit: usize,
}
#[derive(Debug, Serialize)]
struct SearchResponse {
matches: Vec<VectorMatch>,
}
#[derive(Debug, Serialize)]
struct Message<'a> {
message: &'a str,
}
struct ApiError {
status: StatusCode,
public: Cow<'static, str>,
}
impl ApiError {
fn unauthorized() -> Self {
Self {
status: StatusCode::UNAUTHORIZED,
public: Cow::Borrowed("unauthorized"),
}
}
fn bad_request(public: &'static str) -> Self {
Self {
status: StatusCode::BAD_REQUEST,
public: Cow::Borrowed(public),
}
}
fn too_many_requests() -> Self {
Self {
status: StatusCode::TOO_MANY_REQUESTS,
public: Cow::Borrowed("another vector search is already running"),
}
}
fn internal(error: anyhow::Error) -> Self {
error!(error = ?error, "request failed");
Self {
status: StatusCode::INTERNAL_SERVER_ERROR,
public: Cow::Borrowed("internal server error"),
}
}
fn database(error: anyhow::Error) -> Self {
match error.downcast::<VectorInputError>() {
Ok(error) => Self {
status: StatusCode::BAD_REQUEST,
public: Cow::Owned(error.to_string()),
},
Err(error) => Self::internal(error),
}
}
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
(
self.status,
Json(Message {
message: &self.public,
}),
)
.into_response()
}
}
pub async fn serve(database: Database, options: ServeOptions) -> Result<()> {
ensure!(
options.pages.is_dir(),
"pages directory does not exist: {}",
options.pages.display()
);
ensure!(
options.public.is_dir(),
"public directory does not exist: {}",
options.public.display()
);
let public_assets = preload_public_files(&options.public)?;
let public_origin = options
.public_origin
.as_deref()
.map(validate_public_origin)
.transpose()?;
if !options.address.ip().is_loopback()
&& !public_origin
.as_deref()
.is_some_and(|origin| origin.starts_with("https://"))
{
warn!(
address = %options.address,
"public HTTP listener has no HTTPS public origin; keep it on a trusted network or add a TLS proxy"
);
}
let page_router = PageRouter::scan(&options.pages)?;
ensure!(
!page_router.pages.is_empty(),
"no .html pages found in {}",
options.pages.display()
);
let state = AppState {
database,
pages: Arc::new(page_router),
pages_dir: options.pages.clone(),
tenant: options.tenant,
model: options.model,
development: options.development,
api_token: options.api_token,
public_origin,
loopback: options.address.ip().is_loopback(),
public_dir: options.public,
public_assets: Arc::new(public_assets),
search_gate: Arc::new(tokio::sync::Semaphore::new(1)),
write_gate: Arc::new(tokio::sync::Semaphore::new(1)),
};
let app = build_app(state);
let listener = tokio::net::TcpListener::bind(options.address)
.await
.with_context(|| format!("bind {}", options.address))?;
info!(address = %listener.local_addr()?, "bunfork listening");
axum::serve(listener, app)
.with_graceful_shutdown(shutdown_signal())
.await
.context("serve HTTP")
}
fn build_app(state: AppState) -> Router {
native_status_routes()
.route(
"/api/vectors/{id}",
put(upsert_vector).delete(delete_vector),
)
.route("/api/vectors/search", post(search_vectors))
.route("/assets/{*path}", get(public_asset))
.fallback(render_page)
.with_state(state)
.layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
.layer(CompressionLayer::new())
.layer(TimeoutLayer::with_status_code(
StatusCode::REQUEST_TIMEOUT,
REQUEST_TIMEOUT,
))
.layer(TraceLayer::new_for_http())
.layer(middleware::from_fn(reject_hidden_request_paths))
.layer(middleware::from_fn(security_headers))
.layer(GlobalConcurrencyLimitLayer::new(MAX_CONCURRENT_REQUESTS))
}
pub async fn serve_static(
address: SocketAddr,
root: &Path,
manifest_path: &Path,
development: bool,
) -> Result<()> {
let root = fs::canonicalize(root).context("resolve static artifact root")?;
let manifest = artifact::read_manifest(manifest_path)?;
artifact::verify(&root, &manifest)?;
let files = preload_static_files(&root, &manifest)?;
if development {
warn!("static dev mode reuses the frozen artifact; rebuild and restart to see changes");
}
if !address.ip().is_loopback() {
warn!(
address = %address,
"public HTTP listener has no built-in TLS; keep it on a trusted network or add a TLS proxy"
);
}
let state = StaticState {
manifest: Arc::new(manifest),
files: Arc::new(files),
};
let app = static_status_routes()
.fallback(static_request)
.with_state(state)
.layer(TimeoutLayer::with_status_code(
StatusCode::REQUEST_TIMEOUT,
REQUEST_TIMEOUT,
))
.layer(TraceLayer::new_for_http())
.layer(middleware::from_fn(static_security_headers))
.layer(GlobalConcurrencyLimitLayer::new(MAX_CONCURRENT_REQUESTS));
let listener = tokio::net::TcpListener::bind(address)
.await
.with_context(|| format!("bind {address}"))?;
info!(address = %listener.local_addr()?, mode = "static", "bunfork listening");
axum::serve(listener, app)
.with_graceful_shutdown(shutdown_signal())
.await
.context("serve static HTTP")
}
fn preload_static_files(root: &Path, manifest: &Manifest) -> Result<BTreeMap<String, StaticFile>> {
let mut loaded = BTreeMap::new();
let mut total = 0_u64;
for (relative, record) in &manifest.files {
total = total
.checked_add(record.bytes)
.context("static artifact size overflow")?;
ensure!(
total <= artifact::MAX_TOTAL_BYTES,
"static artifact exceeds the 512 MiB in-memory serving limit"
);
let bytes = read_static_file(&root.join(relative), relative, record)?;
loaded.insert(
relative.clone(),
StaticFile {
bytes: Bytes::from(bytes),
etag: format!("\"{}\"", record.sha256),
},
);
}
Ok(loaded)
}
fn preload_public_files(root: &Path) -> Result<BTreeMap<String, StaticFile>> {
let inventory = artifact::inventory(root).context("inventory native public files")?;
let mut loaded = BTreeMap::new();
let mut total = 0_u64;
for (relative, record) in inventory {
total = total
.checked_add(record.bytes)
.context("native public asset size overflow")?;
ensure!(
total <= artifact::MAX_TOTAL_BYTES,
"native public assets exceed the 512 MiB in-memory serving limit"
);
let bytes = read_static_file(&root.join(&relative), &relative, &record)?;
loaded.insert(
relative,
StaticFile {
bytes: Bytes::from(bytes),
etag: format!("W/\"{}\"", record.sha256),
},
);
}
Ok(loaded)
}
fn read_static_file(path: &Path, label: &str, record: &artifact::FileRecord) -> Result<Vec<u8>> {
let mut file = fsguard::open_nofollow(path, false)
.with_context(|| format!("open admitted static file {label}"))?;
let metadata = file.metadata()?;
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
ensure!(
metadata.nlink() == 1,
"static file became hard-linked: {label}"
);
}
ensure!(
metadata.len() == record.bytes,
"static file size changed: {label}"
);
let mut bytes = Vec::with_capacity(usize::try_from(record.bytes)?);
file.read_to_end(&mut bytes)?;
let digest = format!("{:x}", Sha256::digest(&bytes));
ensure!(
digest == record.sha256,
"static file digest changed: {label}"
);
fsguard::ensure_path_matches_file(path, &file, label)?;
Ok(bytes)
}
async fn static_request(
State(state): State<StaticState>,
method: Method,
OriginalUri(uri): OriginalUri,
headers: HeaderMap,
) -> Response {
if method != Method::GET && method != Method::HEAD {
let mut response = StatusCode::METHOD_NOT_ALLOWED.into_response();
response
.headers_mut()
.insert(header::ALLOW, HeaderValue::from_static("GET, HEAD"));
return response;
}
let resolution = match artifact::resolve(&state.manifest, uri.path()) {
Ok(resolution) => resolution,
Err(_) => return StatusCode::BAD_REQUEST.into_response(),
};
match resolution {
Resolution::File {
relative_path,
status,
cache,
}
| Resolution::Fallback {
relative_path,
status,
cache,
} => static_file_response(&state, &method, &headers, &relative_path, status, cache),
Resolution::Redirect {
mut location,
status,
} => {
if let Some(query) = uri.query() {
location.push('?');
location.push_str(query);
}
let status = StatusCode::from_u16(status).unwrap_or(StatusCode::PERMANENT_REDIRECT);
let mut response = status.into_response();
match HeaderValue::from_str(&location) {
Ok(location) => {
response.headers_mut().insert(header::LOCATION, location);
response
}
Err(_) => StatusCode::BAD_REQUEST.into_response(),
}
}
Resolution::NotFound { status } => StatusCode::from_u16(status)
.unwrap_or(StatusCode::NOT_FOUND)
.into_response(),
}
}
fn static_file_response(
state: &StaticState,
method: &Method,
request_headers: &HeaderMap,
relative: &str,
status: u16,
cache: CacheClass,
) -> Response {
let (served_path, encoding) = negotiated_static_path(state, relative, request_headers);
let Some(file) = state.files.get(&served_path) else {
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
};
let status = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
if request_headers
.get(header::IF_NONE_MATCH)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| if_none_match_matches(value, &file.etag))
{
let mut response = StatusCode::NOT_MODIFIED.into_response();
let headers = response.headers_mut();
if let Ok(etag) = HeaderValue::from_str(&file.etag) {
headers.insert(header::ETAG, etag);
}
headers.insert(header::VARY, HeaderValue::from_static("Accept-Encoding"));
headers.insert(
header::CACHE_CONTROL,
HeaderValue::from_static(static_cache_control(relative, cache)),
);
if let Some(encoding) = encoding {
headers.insert(header::CONTENT_ENCODING, HeaderValue::from_static(encoding));
}
return response;
}
let body = if method == Method::HEAD {
Body::empty()
} else {
Body::from(file.bytes.clone())
};
let mut response = Response::new(body);
*response.status_mut() = status;
let headers = response.headers_mut();
headers.insert(
header::CONTENT_TYPE,
HeaderValue::from_static(content_type(relative)),
);
if let Ok(length) = HeaderValue::from_str(&file.bytes.len().to_string()) {
headers.insert(header::CONTENT_LENGTH, length);
}
if let Ok(etag) = HeaderValue::from_str(&file.etag) {
headers.insert(header::ETAG, etag);
}
if let Some(encoding) = encoding {
headers.insert(header::CONTENT_ENCODING, HeaderValue::from_static(encoding));
}
headers.insert(header::VARY, HeaderValue::from_static("Accept-Encoding"));
headers.insert(
header::CACHE_CONTROL,
HeaderValue::from_static(static_cache_control(relative, cache)),
);
response
}
/// RFC 9110 If-None-Match evaluation: honor `*` and use weak comparison
/// (ignore `W/` prefixes on both sides) so weak-tagged revalidations from
/// caches and proxies still produce 304s.
fn if_none_match_matches(header_value: &str, etag: &str) -> bool {
if header_value.trim() == "*" {
return true;
}
let opaque = etag.strip_prefix("W/").unwrap_or(etag);
header_value.split(',').any(|tag| {
let tag = tag.trim();
tag.strip_prefix("W/").unwrap_or(tag) == opaque
})
}
fn static_cache_control(relative: &str, cache: CacheClass) -> &'static str {
if cache == CacheClass::Document {
"no-cache"
} else if has_immutable_namespace(relative) {
"public, max-age=31536000, immutable"
} else {
"public, max-age=0, must-revalidate"
}
}
fn negotiated_static_path(
state: &StaticState,
relative: &str,
headers: &HeaderMap,
) -> (String, Option<&'static str>) {
let accepted = headers
.get(header::ACCEPT_ENCODING)
.and_then(|value| value.to_str().ok())
.unwrap_or_default();
let br_path = format!("{relative}.br");
let gzip_path = format!("{relative}.gz");
let br_quality = encoding_quality(accepted, "br");
let gzip_quality = encoding_quality(accepted, "gzip");
let br_available = br_quality > 0 && state.files.contains_key(&br_path);
let gzip_available = gzip_quality > 0 && state.files.contains_key(&gzip_path);
if br_available && (!gzip_available || br_quality >= gzip_quality) {
return (br_path, Some("br"));
}
if gzip_available {
return (gzip_path, Some("gzip"));
}
(relative.to_owned(), None)
}
fn encoding_quality(header: &str, wanted: &str) -> u16 {
let mut exact = None;
let mut wildcard = None;
for item in header.split(',') {
let mut parts = item.trim().split(';');
let name = parts.next().unwrap_or_default().trim();
let mut quality = 1000;
for parameter in parts {
let Some((key, value)) = parameter.trim().split_once('=') else {
continue;
};
if key.trim().eq_ignore_ascii_case("q") {
quality = parse_quality(value).unwrap_or(0);
}
}
if name.eq_ignore_ascii_case(wanted) {
exact = Some(quality);
} else if name == "*" {
wildcard = Some(quality);
}
}
exact.or(wildcard).unwrap_or(0)
}
fn parse_quality(value: &str) -> Option<u16> {
let value = value.trim();
if value == "0" {
return Some(0);
}
if value == "1" {
return Some(1000);
}
if let Some(fraction) = value.strip_prefix("0.")
&& fraction.len() <= 3
&& fraction.bytes().all(|byte| byte.is_ascii_digit())
{
let parsed = fraction.parse::<u16>().ok()?;
return Some(parsed * 10_u16.pow(u32::try_from(3 - fraction.len()).ok()?));
}
if let Some(fraction) = value.strip_prefix("1.")
&& fraction.len() <= 3
&& fraction.bytes().all(|byte| byte == b'0')
{
return Some(1000);
}
None
}
fn has_immutable_namespace(path: &str) -> bool {
if path.starts_with("_app/immutable/") {
return true;
}
let Some(next) = path.strip_prefix("_next/static/") else {
return false;
};
let namespace = next.split('/').next().unwrap_or_default();
matches!(
namespace,
"chunks" | "css" | "image" | "immutable" | "media" | "pages" | "runtime"
)
}
fn content_type(path: &str) -> &'static str {
// Precompressed `.br`/`.gz` siblings are only relabeled when served
// through content negotiation (the caller passes the logical path).
// A direct request for `app.js.br` falls through to octet-stream so
// raw brotli bytes are never labeled as the underlying type.
match Path::new(path)
.extension()
.and_then(|extension| extension.to_str())
.unwrap_or_default()
.to_ascii_lowercase()
.as_str()
{
"html" | "htm" => "text/html; charset=utf-8",
"css" => "text/css; charset=utf-8",
"js" | "mjs" | "cjs" => "text/javascript; charset=utf-8",
"json" | "map" => "application/json; charset=utf-8",
"txt" => "text/plain; charset=utf-8",
"xml" => "application/xml; charset=utf-8",
"svg" => "image/svg+xml",
"png" => "image/png",
"jpg" | "jpeg" => "image/jpeg",
"gif" => "image/gif",
"webp" => "image/webp",
"avif" => "image/avif",
"ico" => "image/x-icon",
"woff" => "font/woff",
"woff2" => "font/woff2",
"ttf" => "font/ttf",
"otf" => "font/otf",
"wasm" => "application/wasm",
"pdf" => "application/pdf",
"webmanifest" => "application/manifest+json",
"mp4" | "m4v" => "video/mp4",
"webm" => "video/webm",
"mp3" => "audio/mpeg",
"ogg" | "oga" => "audio/ogg",
"wav" => "audio/wav",
_ => "application/octet-stream",
}
}
async fn static_security_headers(request: Request, next: Next) -> Response {
let mut response = next.run(request).await;
let headers = response.headers_mut();
headers.insert(
header::X_CONTENT_TYPE_OPTIONS,
HeaderValue::from_static("nosniff"),
);
headers.insert(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY"));
headers.insert(
header::REFERRER_POLICY,
HeaderValue::from_static("no-referrer"),
);
headers.insert(
header::CONTENT_SECURITY_POLICY,
HeaderValue::from_static(
"default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; font-src 'self'; connect-src 'self'; base-uri 'self'; object-src 'none'; frame-ancestors 'none'; form-action 'self'",
),
);
headers.insert(
HeaderName::from_static("permissions-policy"),
HeaderValue::from_static("camera=(), geolocation=(), microphone=()"),
);
headers.insert(
HeaderName::from_static("cross-origin-opener-policy"),
HeaderValue::from_static("same-origin"),
);
headers.insert(
HeaderName::from_static("cross-origin-resource-policy"),
HeaderValue::from_static("same-origin"),
);
response
}
impl PageRouter {
pub fn scan(root: &Path) -> Result<Self> {
ensure!(
root.is_dir(),
"pages directory does not exist: {}",
root.display()
);
let mut files = Vec::new();
scan_directory(root, root, &mut files)?;
files.sort();
let templates = load_templates(root)?;
let mut environment = template_environment(&templates)?;
let not_found = add_special_page(&mut environment, &templates, root, "_404.html")?;
let internal_error = add_special_page(&mut environment, &templates, root, "_500.html")?;
let mut pages = Vec::new();
let mut effective_patterns = HashSet::new();
for source in files {
let relative = source
.strip_prefix(root)
.context("page escaped pages directory")?;
let segments = parse_route(relative)?;
let pattern = display_pattern(&segments);
let effective = effective_pattern(&segments);
ensure!(
effective_patterns.insert(effective),
"duplicate page route {pattern}"
);
let template_name = format!("page:{pattern}.html");
ensure!(
!templates.contains_key(&template_name),
"generated page template name {template_name} (from {}) collides with a loaded shared template; rename the shared template or the page",
source.display()
);
environment
.add_template_owned(template_name.clone(), read_page(&source)?)
.with_context(|| format!("compile page template {}", source.display()))?;
pages.push(Page {
template_name,
pattern,
segments,
});
}
pages.sort_by(compare_pages);
Ok(Self {
pages,
environment,
not_found,
internal_error,
})
}
pub fn route_count(&self) -> usize {
self.pages.len()
}
fn match_path(&self, path: &str) -> Result<Option<PageMatch>> {
ensure!(!path.contains('\0'), "request path contains a null byte");
ensure!(
!path.contains("//"),
"request path contains an empty segment"
);
let raw_segments = path
.trim_matches('/')
.split('/')
.filter(|segment| !segment.is_empty())
.collect::<Vec<_>>();
let mut decoded = Vec::with_capacity(raw_segments.len());
for raw in raw_segments {
let segment = percent_decode_str(raw)
.decode_utf8()
.context("request path is not valid UTF-8")?
.into_owned();
ensure!(segment != "." && segment != "..", "unsafe request path");
ensure!(
!segment.contains(['/', '\\', '\0']),
"unsafe encoded request path segment"
);
decoded.push(segment);
}
for page in &self.pages {
if let Some(params) = match_segments(&page.segments, &decoded) {
return Ok(Some(PageMatch {
template_name: page.template_name.clone(),
params,
}));
}
}
Ok(None)
}
}
async fn health() -> impl IntoResponse {
Json(Message { message: "ok" })
}
async fn native_ready(State(state): State<AppState>) -> Response {
let database = state.database.clone();
let result = match tokio::task::spawn_blocking(move || database.check_ready()).await {
Ok(result) => result,
Err(error) => Err(error.into()),
};
readiness_response(result)
}
fn readiness_response(result: Result<()>) -> Response {
match result {
Ok(()) => Json(Message { message: "ok" }).into_response(),
Err(error) => {
warn!(?error, "database readiness probe failed");
(
StatusCode::SERVICE_UNAVAILABLE,
Json(Message {
message: "unavailable",
}),
)
.into_response()
}
}
}
fn health_routes<S>() -> Router<S>
where
S: Clone + Send + Sync + 'static,
{
Router::new()
.route("/api/health", get(health))
.route("/_bunfork/health", get(health))
}
fn native_status_routes() -> Router<AppState> {
health_routes()
.route("/api/ready", get(native_ready))
.route("/_bunfork/ready", get(native_ready))
}
fn static_status_routes<S>() -> Router<S>
where
S: Clone + Send + Sync + 'static,
{
health_routes()
.route("/api/ready", get(health))
.route("/_bunfork/ready", get(health))
}
async fn public_asset(
State(state): State<AppState>,
method: Method,
OriginalUri(uri): OriginalUri,
headers: HeaderMap,
) -> Response {
let relative = match decode_public_asset_path(uri.path()) {
Ok(relative) => relative,
Err(_) => return StatusCode::BAD_REQUEST.into_response(),
};
// Dev mode re-reads public assets per request so edits are visible
// without a restart, matching the page rescan semantics.
let dev_assets;
let assets: &BTreeMap<String, StaticFile> = if state.development {
let public_dir = state.public_dir.clone();
match tokio::task::spawn_blocking(move || preload_public_files(&public_dir)).await {
Ok(Ok(reloaded)) => {
dev_assets = reloaded;
&dev_assets
}
Ok(Err(error)) => return ApiError::internal(error).into_response(),
Err(error) => return ApiError::internal(error.into()).into_response(),
}
} else {
state.public_assets.as_ref()
};
let Some(file) = assets.get(&relative) else {
return StatusCode::NOT_FOUND.into_response();
};
if headers
.get(header::IF_NONE_MATCH)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| if_none_match_matches(value, &file.etag))
{
let mut response = StatusCode::NOT_MODIFIED.into_response();
if let Ok(etag) = HeaderValue::from_str(&file.etag) {
response.headers_mut().insert(header::ETAG, etag);
}
response.headers_mut().insert(
header::CACHE_CONTROL,
HeaderValue::from_static(static_cache_control(
&format!("assets/{relative}"),
CacheClass::Asset,
)),
);
return response;
}
let body = if method == Method::HEAD {
Body::empty()
} else {
Body::from(file.bytes.clone())
};
let mut response = Response::new(body);
let response_headers = response.headers_mut();
response_headers.insert(
header::CONTENT_TYPE,
HeaderValue::from_static(content_type(&relative)),
);
if let Ok(length) = HeaderValue::from_str(&file.bytes.len().to_string()) {
response_headers.insert(header::CONTENT_LENGTH, length);
}
if let Ok(etag) = HeaderValue::from_str(&file.etag) {
response_headers.insert(header::ETAG, etag);
}
response_headers.insert(
header::CACHE_CONTROL,
HeaderValue::from_static(static_cache_control(
&format!("assets/{relative}"),
CacheClass::Asset,
)),
);
response
}
fn decode_public_asset_path(path: &str) -> Result<String> {
let raw = path
.strip_prefix("/assets/")
.context("request is outside the public asset namespace")?;
ensure!(!raw.is_empty() && !raw.contains("//"), "invalid asset path");
let mut decoded = Vec::new();
for segment in raw.split('/') {
let segment = percent_decode_str(segment)
.decode_utf8()
.context("asset path is not valid UTF-8")?
.into_owned();
ensure!(
!segment.is_empty()
&& segment != "."
&& segment != ".."
&& !segment.starts_with('.')
&& !segment.contains(['/', '\\', '\0']),
"unsafe asset path"
);
decoded.push(segment);
}
Ok(decoded.join("/"))
}
async fn upsert_vector(
State(state): State<AppState>,
headers: HeaderMap,
axum::extract::Path(id): axum::extract::Path<String>,
Json(payload): Json<VectorUpsert>,
) -> Result<impl IntoResponse, ApiError> {
authorize(&headers, &state)?;
validate_same_origin(&headers, &state)?;
if id != payload.id {
return Err(ApiError::bad_request("path and body vector ids differ"));
}
validate_vector_upsert(
&state.tenant,
&state.model,
&payload.id,
&payload.content,
&payload.embedding,
)
.map_err(ApiError::database)?;
// Queue writers on an async semaphore instead of stacking parked
// blocking-pool threads on the connection mutex.
let permit = state
.write_gate
.clone()
.acquire_owned()
.await
.map_err(|error| ApiError::internal(error.into()))?;
let database = state.database.clone();
let tenant = state.tenant.clone();
let model = state.model.clone();
tokio::task::spawn_blocking(move || {
let _permit = permit;
database.upsert_vector(
&tenant,
&model,
&payload.id,
&payload.content,
&payload.embedding,
)
})
.await
.map_err(|error| ApiError::internal(error.into()))?
.map_err(ApiError::database)?;
Ok((StatusCode::CREATED, Json(Message { message: "stored" })))
}
async fn search_vectors(
State(state): State<AppState>,
headers: HeaderMap,
Json(payload): Json<VectorSearch>,
) -> Result<Json<SearchResponse>, ApiError> {
authorize(&headers, &state)?;
validate_same_origin(&headers, &state)?;
validate_vector_search(
&state.tenant,
&state.model,
&payload.embedding,
payload.limit,
)
.map_err(ApiError::database)?;
let permit = state
.search_gate
.clone()
.try_acquire_owned()
.map_err(|_| ApiError::too_many_requests())?;
let database = state.database.clone();
let tenant = state.tenant.clone();
let model = state.model.clone();
let matches = tokio::task::spawn_blocking(move || {
// Hold the permit inside the blocking task so a client disconnect
// cannot release the gate while the scan is still running.
let _permit = permit;
database.search_vectors(&tenant, &model, &payload.embedding, payload.limit)
})
.await
.map_err(|error| ApiError::internal(error.into()))?
.map_err(ApiError::database)?;
Ok(Json(SearchResponse { matches }))
}
async fn delete_vector(
State(state): State<AppState>,
headers: HeaderMap,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<StatusCode, ApiError> {
authorize(&headers, &state)?;
validate_same_origin(&headers, &state)?;
validate_vector_id(&state.tenant, &state.model, &id).map_err(ApiError::database)?;
let permit = state
.write_gate
.clone()
.acquire_owned()
.await
.map_err(|error| ApiError::internal(error.into()))?;
let database = state.database.clone();
let tenant = state.tenant.clone();
let model = state.model.clone();
let deleted = tokio::task::spawn_blocking(move || {
let _permit = permit;
database.delete_vector(&tenant, &model, &id)
})
.await
.map_err(|error| ApiError::internal(error.into()))?
.map_err(ApiError::database)?;
if deleted {
Ok(StatusCode::NO_CONTENT)
} else {
Err(ApiError {
status: StatusCode::NOT_FOUND,
public: Cow::Borrowed("vector not found"),
})
}
}
async fn render_page(
State(state): State<AppState>,
method: Method,
OriginalUri(uri): OriginalUri,
) -> Response {
if method != Method::GET && method != Method::HEAD {
let mut response = StatusCode::METHOD_NOT_ALLOWED.into_response();
response
.headers_mut()
.insert(header::ALLOW, HeaderValue::from_static("GET, HEAD"));
return response;
}
let router = if state.development {
let pages_dir = state.pages_dir.clone();
match tokio::task::spawn_blocking(move || PageRouter::scan(&pages_dir)).await {
Ok(Ok(router)) => Arc::new(router),
Ok(Err(error)) => return ApiError::internal(error).into_response(),
Err(error) => return ApiError::internal(error.into()).into_response(),
}
} else {
Arc::clone(&state.pages)
};
let query = match parse_query(&uri) {
Ok(query) => query,
Err(_) => return (StatusCode::BAD_REQUEST, Html("<h1>Bad request</h1>")).into_response(),
};
let matched = match router.match_path(uri.path()) {
Ok(Some(matched)) => matched,
Ok(None) => {
if let Some(template_name) = &router.not_found {
return render_native_template(
&router,
template_name,
BTreeMap::new(),
&query,
&uri,
&state,
StatusCode::NOT_FOUND,
)
.unwrap_or_else(|error| ApiError::internal(error).into_response());
}
return (StatusCode::NOT_FOUND, Html("<h1>Not found</h1>")).into_response();
}
Err(_) => return (StatusCode::BAD_REQUEST, Html("<h1>Bad request</h1>")).into_response(),
};
match render_native_template(
&router,
&matched.template_name,
matched.params,
&query,
&uri,
&state,
StatusCode::OK,
) {
Ok(response) => response,
Err(error) => {
error!(error = ?error, "page rendering failed");
if let Some(template_name) = &router.internal_error {
render_native_template(
&router,
template_name,
BTreeMap::new(),
&BTreeMap::new(),
&uri,
&state,
StatusCode::INTERNAL_SERVER_ERROR,
)
.unwrap_or_else(|error| ApiError::internal(error).into_response())
} else {
ApiError::internal(error).into_response()
}
}
}
}
fn render_native_template(
router: &PageRouter,
template_name: &str,
params: BTreeMap<String, String>,
query: &BTreeMap<String, Vec<String>>,
uri: &Uri,
state: &AppState,
status: StatusCode,
) -> Result<Response> {
let template = router.environment.get_template(template_name)?;
let html = template.render(context! {
params => params,
query => query,
pathname => uri.path(),
tenant => state.tenant,
model => state.model,
})?;
Ok((status, Html(html)).into_response())
}
fn parse_query(uri: &Uri) -> Result<BTreeMap<String, Vec<String>>> {
let Some(query) = uri.query() else {
return Ok(BTreeMap::new());
};
ensure!(query.len() <= 8192, "query string exceeds 8 KiB");
let mut values = BTreeMap::<String, Vec<String>>::new();
for (index, pair) in query.split('&').enumerate() {
ensure!(index < 64, "query string has too many fields");
if pair.is_empty() {
continue;
}
let (key, value) = pair.split_once('=').unwrap_or((pair, ""));
let key = decode_query_component(key)?;
let value = decode_query_component(value)?;
ensure!(!key.is_empty(), "query field name cannot be empty");
ensure!(key.len() <= 256, "query field name is too long");
ensure!(value.len() <= 4096, "query field value is too long");
values.entry(key).or_default().push(value);
}
Ok(values)
}
fn decode_query_component(value: &str) -> Result<String> {
let bytes = value.as_bytes();
for index in 0..bytes.len() {
if bytes[index] == b'%' {
ensure!(
index + 2 < bytes.len()
&& bytes[index + 1].is_ascii_hexdigit()
&& bytes[index + 2].is_ascii_hexdigit(),
"query contains malformed percent encoding"
);
}
}
percent_decode_str(&value.replace('+', " "))
.decode_utf8()
.context("query is not valid UTF-8")
.map(|value| value.into_owned())
}
fn authorize(headers: &HeaderMap, state: &AppState) -> Result<(), ApiError> {
let Some(value) = headers.get(header::AUTHORIZATION) else {
return Err(ApiError::unauthorized());
};
let Ok(value) = value.to_str() else {
return Err(ApiError::unauthorized());
};
let Some(provided) = value.strip_prefix("Bearer ") else {
return Err(ApiError::unauthorized());
};
let expected = state.api_token.as_str();
let valid = provided.len() == expected.len()
&& bool::from(provided.as_bytes().ct_eq(expected.as_bytes()));
if valid {
Ok(())
} else {
Err(ApiError::unauthorized())
}
}
fn validate_same_origin(headers: &HeaderMap, state: &AppState) -> Result<(), ApiError> {
let Some(origin) = headers.get(header::ORIGIN) else {
return Ok(());
};
let origin = origin
.to_str()
.map_err(|_| ApiError::bad_request("invalid origin header"))?;
if let Some(expected) = &state.public_origin {
return if origin == expected {
Ok(())
} else {
Err(ApiError::bad_request("cross-origin mutation rejected"))
};
}
// Without a configured public origin the Host header is the only
// reference, and both Origin and Host are attacker-supplied on a
// non-loopback listener; comparing them is self-referential. Restrict
// the fallback to loopback binds and fail closed everywhere else.
if !state.loopback {
return Err(ApiError::bad_request(
"browser mutations require --public-origin on non-loopback listeners",
));
}
let Some(host) = headers.get(header::HOST) else {
return Err(ApiError::bad_request("missing host header"));
};
let host = host
.to_str()
.map_err(|_| ApiError::bad_request("invalid host header"))?;
if origin == format!("http://{host}") || origin == format!("https://{host}") {
Ok(())
} else {
Err(ApiError::bad_request("cross-origin mutation rejected"))
}
}
fn validate_public_origin(origin: &str) -> Result<String> {
let uri: Uri = origin.parse().context("invalid public origin")?;
let scheme = uri
.scheme_str()
.context("public origin requires a scheme")?;
let authority = uri
.authority()
.context("public origin requires an authority")?;
ensure!(
matches!(scheme, "http" | "https")
&& matches!(uri.path(), "" | "/")
&& uri.query().is_none(),
"public origin must contain only an http(s) scheme and authority"
);
Ok(format!("{scheme}://{authority}"))
}
async fn security_headers(request: Request, next: Next) -> Response {
let mut response = next.run(request).await;
let headers = response.headers_mut();
headers.insert(
header::X_CONTENT_TYPE_OPTIONS,
HeaderValue::from_static("nosniff"),
);
headers.insert(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY"));
headers.insert(
header::REFERRER_POLICY,
HeaderValue::from_static("no-referrer"),
);
headers.insert(
header::CONTENT_SECURITY_POLICY,
HeaderValue::from_static(
"default-src 'self'; base-uri 'none'; object-src 'none'; frame-ancestors 'none'; form-action 'self'",
),
);
headers.insert(
HeaderName::from_static("permissions-policy"),
HeaderValue::from_static("camera=(), geolocation=(), microphone=()"),
);
headers.insert(
HeaderName::from_static("cross-origin-opener-policy"),
HeaderValue::from_static("same-origin"),
);
headers.insert(
HeaderName::from_static("cross-origin-resource-policy"),
HeaderValue::from_static("same-origin"),
);
response
}
async fn reject_hidden_request_paths(request: Request, next: Next) -> Response {
if has_hidden_segment(request.uri().path()) {
return StatusCode::NOT_FOUND.into_response();
}
next.run(request).await
}
fn has_hidden_segment(path: &str) -> bool {
path.split('/').any(|segment| {
percent_decode_str(segment)
.decode_utf8()
.is_ok_and(|segment| segment.starts_with('.'))
})
}
fn scan_directory(root: &Path, current: &Path, pages: &mut Vec<PathBuf>) -> Result<()> {
let mut entries = fs::read_dir(current)
.with_context(|| format!("read pages directory {}", current.display()))?
.collect::<std::io::Result<Vec<_>>>()?;
entries.sort_by_key(|entry| entry.file_name());
for entry in entries {
let file_type = entry.file_type()?;
ensure!(!file_type.is_symlink(), "page tree cannot contain symlinks");
let path = entry.path();
let name = entry.file_name();
let name = name.to_string_lossy();
ensure!(
!name.starts_with('.'),
"page tree cannot contain hidden entries"
);
if name.starts_with('_') {
continue;
}
if file_type.is_dir() {
scan_directory(root, &path, pages)?;
} else {
ensure!(file_type.is_file(), "page tree contains a special file");
if path.extension().and_then(|extension| extension.to_str()) == Some("html") {
ensure!(path.starts_with(root), "page escaped pages directory");
pages.push(path);
}
}
}
Ok(())
}
fn parse_route(relative: &Path) -> Result<Vec<Segment>> {
let mut components = relative
.components()
.map(|component| component.as_os_str().to_string_lossy().into_owned())
.collect::<Vec<_>>();
let file = components.pop().context("page path has no file name")?;
let stem = file
.strip_suffix(".html")
.context("page must end in .html")?;
if stem != "index" {
components.push(stem.to_owned());
}
let mut segments = Vec::new();
for component in components {
if component.starts_with('(') && component.ends_with(')') {
ensure!(component.len() > 2, "empty route group");
continue;
}
segments.push(parse_segment(&component)?);
}
for (index, segment) in segments.iter().enumerate() {
if matches!(segment, Segment::CatchAll(_) | Segment::OptionalCatchAll(_)) {
ensure!(
index + 1 == segments.len(),
"catch-all route must be terminal"
);
}
}
let mut parameters = HashSet::new();
for segment in &segments {
let name = match segment {
Segment::Parameter(name)
| Segment::CatchAll(name)
| Segment::OptionalCatchAll(name) => name,
Segment::Static(_) => continue,
};
ensure!(parameters.insert(name), "route repeats parameter {name}");
}
Ok(segments)
}
fn read_page(path: &Path) -> Result<String> {
let mut file = fsguard::open_nofollow(path, false)
.with_context(|| format!("open page without following symlinks {}", path.display()))?;
let metadata = file.metadata()?;
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
ensure!(metadata.nlink() == 1, "page cannot be hard-linked");
}
ensure!(metadata.len() <= 1024 * 1024, "page exceeds 1 MiB");
let mut source = String::new();
file.read_to_string(&mut source)
.with_context(|| format!("read UTF-8 page {}", path.display()))?;
Ok(source)
}
fn load_templates(root: &Path) -> Result<BTreeMap<String, Arc<str>>> {
let Some(app_root) = root.parent() else {
return Ok(BTreeMap::new());
};
let template_root = app_root.join("templates");
match fs::symlink_metadata(&template_root) {
Ok(metadata) => ensure!(
metadata.is_dir() && !metadata.file_type().is_symlink(),
"template root must be a real directory"
),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(BTreeMap::new()),
Err(error) => return Err(error.into()),
}
let mut paths = Vec::new();
scan_directory(&template_root, &template_root, &mut paths)?;
let mut templates = BTreeMap::new();
for path in paths {
let relative = path
.strip_prefix(&template_root)
.context("template escaped template root")?;
let name = relative
.components()
.map(|component| {
component
.as_os_str()
.to_str()
.context("template path must be UTF-8")
})
.collect::<Result<Vec<_>>>()?
.join("/");
ensure!(
templates
.insert(name.clone(), Arc::from(read_page(&path)?))
.is_none(),
"duplicate template {name}"
);
}
Ok(templates)
}
fn add_special_page(
environment: &mut Environment<'static>,
templates: &BTreeMap<String, Arc<str>>,
root: &Path,
name: &str,
) -> Result<Option<String>> {
let path = root.join(name);
match fs::symlink_metadata(&path) {
Ok(_) => {
let template_name = format!("page:{name}");
ensure!(
!templates.contains_key(&template_name),
"generated page template name {template_name} (from {}) collides with a loaded shared template; rename the shared template or the page",
path.display()
);
environment
.add_template_owned(template_name.clone(), read_page(&path)?)
.with_context(|| format!("compile page template {}", path.display()))?;
Ok(Some(template_name))
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(error.into()),
}
}
fn template_environment(templates: &BTreeMap<String, Arc<str>>) -> Result<Environment<'static>> {
let mut environment = Environment::new();
environment.set_auto_escape_callback(|name| {
if name.ends_with(".html") {
AutoEscape::Html
} else {
AutoEscape::None
}
});
for (name, source) in templates {
environment.add_template_owned(name.clone(), source.to_string())?;
}
Ok(environment)
}
fn parse_segment(value: &str) -> Result<Segment> {
ensure!(!value.is_empty(), "route contains an empty segment");
if let Some(name) = value
.strip_prefix("[[...")
.and_then(|name| name.strip_suffix("]]"))
{
validate_parameter(name)?;
return Ok(Segment::OptionalCatchAll(name.to_owned()));
}
if let Some(name) = value
.strip_prefix("[...")
.and_then(|name| name.strip_suffix(']'))
{
validate_parameter(name)?;
return Ok(Segment::CatchAll(name.to_owned()));
}
if let Some(name) = value
.strip_prefix('[')
.and_then(|name| name.strip_suffix(']'))
{
validate_parameter(name)?;
return Ok(Segment::Parameter(name.to_owned()));
}
ensure!(
!value.contains(['[', ']']) && value != "." && value != "..",
"invalid static route segment: {value}"
);
Ok(Segment::Static(value.to_owned()))
}
fn validate_parameter(name: &str) -> Result<()> {
ensure!(
!name.is_empty()
&& name
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_'),
"invalid route parameter: {name}"
);
Ok(())
}
fn match_segments(pattern: &[Segment], path: &[String]) -> Option<BTreeMap<String, String>> {
let mut params = BTreeMap::new();
let mut path_index = 0;
for segment in pattern {
match segment {
Segment::Static(expected) => {
if path.get(path_index)? != expected {
return None;
}
path_index += 1;
}
Segment::Parameter(name) => {
params.insert(name.clone(), path.get(path_index)?.clone());
path_index += 1;
}
Segment::CatchAll(name) => {
if path_index >= path.len() {
return None;
}
params.insert(name.clone(), path[path_index..].join("/"));
path_index = path.len();
}
Segment::OptionalCatchAll(name) => {
params.insert(name.clone(), path[path_index..].join("/"));
path_index = path.len();
}
}
}
(path_index == path.len()).then_some(params)
}
fn display_pattern(segments: &[Segment]) -> String {
if segments.is_empty() {
return "/".to_owned();
}
format!(
"/{}",
segments
.iter()
.map(|segment| match segment {
Segment::Static(value) => value.clone(),
Segment::Parameter(name) => format!("[{name}]"),
Segment::CatchAll(name) => format!("[...{name}]"),
Segment::OptionalCatchAll(name) => format!("[[...{name}]]"),
})
.collect::<Vec<_>>()
.join("/")
)
}
fn effective_pattern(segments: &[Segment]) -> String {
segments
.iter()
.map(|segment| match segment {
Segment::Static(value) => format!("s:{value}"),
Segment::Parameter(_) => "p".to_owned(),
Segment::CatchAll(_) => "c".to_owned(),
Segment::OptionalCatchAll(_) => "o".to_owned(),
})
.collect::<Vec<_>>()
.join("/")
}
fn compare_pages(left: &Page, right: &Page) -> std::cmp::Ordering {
let rank = |segment: &Segment| match segment {
Segment::Static(_) => 0_u8,
Segment::Parameter(_) => 1,
Segment::CatchAll(_) => 2,
Segment::OptionalCatchAll(_) => 3,
};
left.segments
.iter()
.map(rank)
.cmp(right.segments.iter().map(rank))
.then_with(|| right.segments.len().cmp(&left.segments.len()))
.then_with(|| left.pattern.cmp(&right.pattern))
}
pub fn validate_public_tree(root: &Path) -> Result<()> {
artifact::inventory(root).context("validate native public files")?;
Ok(())
}
fn default_search_limit() -> usize {
10
}
async fn shutdown_signal() {
let ctrl_c = async {
if let Err(error) = tokio::signal::ctrl_c().await {
error!(?error, "failed to install Ctrl-C handler");
std::future::pending::<()>().await;
}
};
#[cfg(unix)]
let terminate = async {
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
Ok(mut signal) => {
signal.recv().await;
}
Err(error) => {
error!(?error, "failed to install SIGTERM handler");
std::future::pending::<()>().await;
}
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
() = ctrl_c => {},
() = terminate => {},
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::{body::Body, http::Method};
use tower::ServiceExt as _;
#[tokio::test]
async fn static_status_routes_are_stable_and_get_only() -> Result<()> {
for path in [
"/api/health",
"/api/ready",
"/_bunfork/health",
"/_bunfork/ready",
] {
let response = static_status_routes::<()>()
.oneshot(Request::builder().uri(path).body(Body::empty())?)
.await?;
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
axum::body::to_bytes(response.into_body(), 1024).await?,
r#"{"message":"ok"}"#
);
let response = static_status_routes::<()>()
.oneshot(
Request::builder()
.method(Method::POST)
.uri(path)
.body(Body::empty())?,
)
.await?;
assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
}
Ok(())
}
#[tokio::test]
async fn native_readiness_maps_database_failure_to_unavailable() -> Result<()> {
let response = readiness_response(Err(anyhow::anyhow!("database unavailable")));
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(
axum::body::to_bytes(response.into_body(), 1024).await?,
r#"{"message":"unavailable"}"#
);
let response = readiness_response(Ok(()));
assert_eq!(response.status(), StatusCode::OK);
Ok(())
}
#[test]
fn static_routes_precede_dynamic_and_catch_all() -> Result<()> {
let directory = tempfile::tempdir()?;
let pages = directory.path();
fs::create_dir_all(pages.join("docs"))?;
fs::write(pages.join("index.html"), "home")?;
fs::write(pages.join("[slug].html"), "dynamic")?;
fs::write(pages.join("about.html"), "about")?;
fs::write(pages.join("docs/[...path].html"), "docs")?;
let router = PageRouter::scan(pages)?;
let about = router.match_path("/about")?.context("about route")?;
assert_eq!(about.template_name, "page:/about.html");
assert_eq!(
router
.environment
.get_template(&about.template_name)?
.render(context! {})?,
"about"
);
let dynamic = router.match_path("/hello")?.context("dynamic route")?;
assert_eq!(
dynamic.params.get("slug").map(String::as_str),
Some("hello")
);
let docs = router.match_path("/docs/a/b")?.context("docs route")?;
assert_eq!(docs.params.get("path").map(String::as_str), Some("a/b"));
Ok(())
}
#[test]
fn page_route_rejects_shared_template_namespace_collision() -> Result<()> {
let directory = tempfile::tempdir()?;
let pages = directory.path().join("pages");
let templates = directory.path().join("templates").join("page:");
fs::create_dir_all(&pages)?;
fs::create_dir_all(&templates)?;
fs::write(pages.join("about.html"), "page about")?;
fs::write(templates.join("about.html"), "shared about")?;
let Err(error) = PageRouter::scan(&pages) else {
panic!("collision must be rejected");
};
let message = format!("{error:#}");
assert!(
message.contains("page:/about.html"),
"error should name the colliding template: {message}"
);
assert!(
message.contains("collides with a loaded shared template"),
"error should describe the collision: {message}"
);
Ok(())
}
#[test]
fn hidden_public_files_and_pages_are_rejected() -> Result<()> {
let directory = tempfile::tempdir()?;
let public = directory.path().join("public");
let pages = directory.path().join("pages");
fs::create_dir_all(&public)?;
fs::create_dir_all(&pages)?;
fs::write(public.join(".bunfork.key"), "secret")?;
fs::write(pages.join(".secret.html"), "secret")?;
assert!(validate_public_tree(&public).is_err());
assert!(PageRouter::scan(&pages).is_err());
assert!(has_hidden_segment("/assets/.bunfork.key"));
assert!(has_hidden_segment("/assets/%2ebunfork.token"));
Ok(())
}
#[test]
fn public_origin_is_canonicalized() -> Result<()> {
assert_eq!(
validate_public_origin("https://example.com/")?,
"https://example.com"
);
assert!(validate_public_origin("https://example.com/path").is_err());
assert!(validate_public_origin("javascript:alert(1)").is_err());
Ok(())
}
#[test]
fn immutable_caching_requires_a_framework_semantic_namespace() {
assert!(has_immutable_namespace("_app/immutable/start.C544yXVy.js"));
assert!(has_immutable_namespace("_app/immutable/chunks/Bjy-W4x2.js"));
assert!(has_immutable_namespace(
"_next/static/chunks/472-abcdef123456.js"
));
assert!(!has_immutable_namespace(
"_next/static/service-worker/sw.js"
));
assert!(!has_immutable_namespace(
"_next/static/build-id/_buildManifest.js"
));
assert!(!has_immutable_namespace("_nuxt/entry.D7oq8EHe.js"));
assert!(!has_immutable_namespace("assets/index-GUtueyEN.js"));
assert!(!has_immutable_namespace("assets/app-MarketingHero.js"));
}
fn static_test_state() -> StaticState {
let contents = [
("app.js", "console.log('hi')"),
("app.js.br", "brotli-bytes"),
("app.js.gz", "gzip-bytes"),
];
let mut files = BTreeMap::new();
let mut records = BTreeMap::new();
for (path, bytes) in contents {
files.insert(
path.to_owned(),
StaticFile {
bytes: Bytes::from_static(bytes.as_bytes()),
etag: format!("\"{:x}\"", Sha256::digest(bytes.as_bytes())),
},
);
records.insert(
path.to_owned(),
artifact::FileRecord {
sha256: format!("{:x}", Sha256::digest(bytes.as_bytes())),
bytes: bytes.len() as u64,
},
);
}
let manifest = Manifest {
schema: "bunfork-static-v1".to_owned(),
root: ".".to_owned(),
mode: artifact::Mode::Mpa,
base: "/".to_owned(),
fallback: None,
trailing_slash: artifact::TrailingSlash::Directory,
precompressed: true,
immutable: true,
files: records,
};
StaticState {
manifest: Arc::new(manifest),
files: Arc::new(files),
}
}
fn header_str(response: &Response, name: HeaderName) -> Option<&str> {
response
.headers()
.get(name)
.and_then(|value| value.to_str().ok())
}
#[test]
fn static_negotiation_selects_encoding_and_scopes_etags() -> Result<()> {
let state = static_test_state();
let br_etag = format!("\"{:x}\"", Sha256::digest(b"brotli-bytes"));
let mut headers = HeaderMap::new();
headers.insert(
header::ACCEPT_ENCODING,
HeaderValue::from_static("br, gzip;q=0.5"),
);
let response = static_file_response(
&state,
&Method::GET,
&headers,
"app.js",
200,
CacheClass::Asset,
);
assert_eq!(header_str(&response, header::CONTENT_ENCODING), Some("br"));
assert_eq!(
header_str(&response, header::CONTENT_TYPE),
Some("text/javascript; charset=utf-8")
);
assert_eq!(header_str(&response, header::VARY), Some("Accept-Encoding"));
assert_eq!(header_str(&response, header::ETAG), Some(br_etag.as_str()));
let mut headers = HeaderMap::new();
headers.insert(header::ACCEPT_ENCODING, HeaderValue::from_static("gzip"));
let response = static_file_response(
&state,
&Method::GET,
&headers,
"app.js",
200,
CacheClass::Asset,
);
assert_eq!(
header_str(&response, header::CONTENT_ENCODING),
Some("gzip")
);
let response = static_file_response(
&state,
&Method::GET,
&HeaderMap::new(),
"app.js",
200,
CacheClass::Asset,
);
assert!(response.headers().get(header::CONTENT_ENCODING).is_none());
// Revalidating with the encoding-specific ETag yields an encoded 304.
let mut headers = HeaderMap::new();
headers.insert(header::ACCEPT_ENCODING, HeaderValue::from_static("br"));
headers.insert(header::IF_NONE_MATCH, HeaderValue::from_str(&br_etag)?);
let response = static_file_response(
&state,
&Method::GET,
&headers,
"app.js",
200,
CacheClass::Asset,
);
assert_eq!(response.status(), StatusCode::NOT_MODIFIED);
assert_eq!(header_str(&response, header::CONTENT_ENCODING), Some("br"));
// Direct sibling requests are plain binary downloads, never relabeled.
assert_eq!(content_type("app.js.br"), "application/octet-stream");
assert_eq!(content_type("app.js.gz"), "application/octet-stream");
Ok(())
}
proptest::proptest! {
#[test]
fn decoded_public_asset_paths_are_traversal_free(path in "/assets/.{0,48}") {
if let Ok(decoded) = decode_public_asset_path(&path) {
let traversal_free = !decoded.is_empty()
&& decoded.split('/').all(|segment| {
!segment.is_empty()
&& segment != ".."
&& !segment.starts_with('.')
&& !segment.contains('\\')
});
proptest::prop_assert!(traversal_free);
}
}
#[test]
fn query_parsing_never_panics(query in ".{0,64}") {
if let Ok(uri) = format!("/?{query}").parse::<Uri>() {
let _ = parse_query(&uri);
}
}
}
#[test]
fn if_none_match_follows_rfc_9110_weak_comparison() {
assert!(if_none_match_matches("*", "\"abc\""));
assert!(if_none_match_matches("W/\"abc\"", "\"abc\""));
assert!(if_none_match_matches("\"abc\"", "W/\"abc\""));
assert!(if_none_match_matches("\"x\", \"abc\"", "\"abc\""));
assert!(!if_none_match_matches("\"abcd\"", "\"abc\""));
}
#[test]
fn encoding_quality_honors_zero_wildcards_and_preference() {
assert_eq!(encoding_quality("br;q=0.0, gzip;q=0.8", "br"), 0);
assert_eq!(encoding_quality("br;q=0.2, gzip;q=0.8", "gzip"), 800);
assert_eq!(encoding_quality("*;q=0.4, br;q=0", "br"), 0);
assert_eq!(encoding_quality("*;q=0.4, br;q=0", "gzip"), 400);
assert_eq!(encoding_quality("gzip;q=bogus", "gzip"), 0);
}
#[test]
fn native_public_assets_are_preloaded_and_paths_are_strict() -> Result<()> {
let directory = tempfile::tempdir()?;
fs::write(directory.path().join("app.css"), "original")?;
let assets = preload_public_files(directory.path())?;
fs::write(directory.path().join("app.css"), "changed")?;
let asset = assets.get("app.css").context("preloaded asset")?;
assert_eq!(asset.bytes, "original");
assert_eq!(
asset.etag,
format!("W/\"{:x}\"", Sha256::digest(b"original"))
);
assert_eq!(
decode_public_asset_path("/assets/icons/logo.svg")?,
"icons/logo.svg"
);
assert!(decode_public_asset_path("/assets/%2e%2e/secret").is_err());
assert!(decode_public_asset_path("/assets/icons%2flogo.svg").is_err());
assert!(decode_public_asset_path("/assets/.token").is_err());
Ok(())
}
#[test]
fn routes_queries_and_templates_reject_ambiguous_input() -> Result<()> {
let directory = tempfile::tempdir()?;
let pages = directory.path();
fs::create_dir_all(pages.join("[id]"))?;
fs::write(pages.join("[id]/[id].html"), "duplicate")?;
assert!(PageRouter::scan(pages).is_err());
fs::remove_dir_all(pages.join("[id]"))?;
fs::write(pages.join("index.html"), "{% if %}")?;
assert!(PageRouter::scan(pages).is_err());
fs::write(pages.join("index.html"), "valid")?;
let router = PageRouter::scan(pages)?;
assert!(router.match_path("/%2e%2e/secret").is_err());
assert!(router.match_path("/a%2fb").is_err());
let uri: Uri = "/?tag=a&tag=b&q=hello+world".parse()?;
let query = parse_query(&uri)?;
assert_eq!(
query.get("tag"),
Some(&vec!["a".to_owned(), "b".to_owned()])
);
assert_eq!(query.get("q"), Some(&vec!["hello world".to_owned()]));
assert!(parse_query(&"/?bad=%zz".parse()?).is_err());
Ok(())
}
const TEST_KEY: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
const TEST_TOKEN: &str = "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210";
fn test_app(pages: &Path, loopback: bool) -> Result<Router> {
let database = crate::db::Database::memory(TEST_KEY)?;
database.migrate()?;
let router = PageRouter::scan(pages)?;
let state = AppState {
database,
pages: Arc::new(router),
pages_dir: pages.to_path_buf(),
tenant: "tenant".to_owned(),
model: "model".to_owned(),
development: false,
api_token: Arc::new(Zeroizing::new(TEST_TOKEN.to_owned())),
public_origin: None,
loopback,
public_dir: PathBuf::new(),
public_assets: Arc::new(BTreeMap::new()),
search_gate: Arc::new(tokio::sync::Semaphore::new(1)),
write_gate: Arc::new(tokio::sync::Semaphore::new(1)),
};
Ok(build_app(state))
}
#[tokio::test]
async fn vector_capacity_error_message_reaches_the_400_body() -> Result<()> {
let error = ApiError::database(
VectorInputError(
"vector capacity (10000) reached for this tenant, model, and dimension; \
delete vectors before adding more"
.to_owned(),
)
.into(),
);
assert_eq!(error.status, StatusCode::BAD_REQUEST);
let response = error.into_response();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let bytes = axum::body::to_bytes(response.into_body(), 64 * 1024).await?;
let json: serde_json::Value = serde_json::from_slice(&bytes)?;
assert!(
json["message"]
.as_str()
.is_some_and(|message| message.contains("vector capacity")),
"expected vector capacity message in body, got {json}"
);
Ok(())
}
fn upsert_request(token: Option<&str>, origin: Option<&str>) -> Result<Request> {
let mut builder = Request::builder()
.method(Method::PUT)
.uri("/api/vectors/note-1")
.header(header::CONTENT_TYPE, "application/json");
if let Some(token) = token {
builder = builder.header(header::AUTHORIZATION, format!("Bearer {token}"));
}
if let Some(origin) = origin {
builder = builder
.header(header::ORIGIN, origin)
.header(header::HOST, "localhost:3100");
}
Ok(builder.body(Body::from(
r#"{"id":"note-1","content":"hello","embedding":[1.0,0.0]}"#,
))?)
}
#[tokio::test]
async fn vector_api_requires_bearer_auth_and_round_trips() -> Result<()> {
let directory = tempfile::tempdir()?;
fs::write(directory.path().join("index.html"), "home")?;
let app = test_app(directory.path(), true)?;
let response = app.clone().oneshot(upsert_request(None, None)?).await?;
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
let wrong = "0".repeat(64);
let response = app
.clone()
.oneshot(upsert_request(Some(&wrong), None)?)
.await?;
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
let response = app
.clone()
.oneshot(upsert_request(Some(TEST_TOKEN), None)?)
.await?;
assert_eq!(response.status(), StatusCode::CREATED);
let response = app
.clone()
.oneshot(
Request::builder()
.method(Method::POST)
.uri("/api/vectors/search")
.header(header::CONTENT_TYPE, "application/json")
.header(header::AUTHORIZATION, format!("Bearer {TEST_TOKEN}"))
.body(Body::from(r#"{"embedding":[1.0,0.0],"limit":5}"#))?,
)
.await?;
assert_eq!(response.status(), StatusCode::OK);
let bytes = axum::body::to_bytes(response.into_body(), 64 * 1024).await?;
let json: serde_json::Value = serde_json::from_slice(&bytes)?;
assert_eq!(json["matches"][0]["id"], "note-1");
let response = app
.clone()
.oneshot(
Request::builder()
.method(Method::DELETE)
.uri("/api/vectors/note-1")
.header(header::AUTHORIZATION, format!("Bearer {TEST_TOKEN}"))
.body(Body::empty())?,
)
.await?;
assert_eq!(response.status(), StatusCode::NO_CONTENT);
Ok(())
}
#[tokio::test]
async fn browser_mutations_enforce_same_origin() -> Result<()> {
let directory = tempfile::tempdir()?;
fs::write(directory.path().join("index.html"), "home")?;
let app = test_app(directory.path(), true)?;
let response = app
.clone()
.oneshot(upsert_request(
Some(TEST_TOKEN),
Some("https://evil.example"),
)?)
.await?;
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let response = app
.clone()
.oneshot(upsert_request(
Some(TEST_TOKEN),
Some("http://localhost:3100"),
)?)
.await?;
assert_eq!(response.status(), StatusCode::CREATED);
Ok(())
}
#[tokio::test]
async fn origin_host_fallback_is_loopback_only() -> Result<()> {
let directory = tempfile::tempdir()?;
fs::write(directory.path().join("index.html"), "home")?;
let app = test_app(directory.path(), false)?;
// Both Origin and Host are attacker-supplied on public listeners, so
// even a matching pair is rejected without a configured public origin.
let response = app
.clone()
.oneshot(upsert_request(
Some(TEST_TOKEN),
Some("http://localhost:3100"),
)?)
.await?;
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
// Non-browser API clients without an Origin header are unaffected.
let response = app
.clone()
.oneshot(upsert_request(Some(TEST_TOKEN), None)?)
.await?;
assert_eq!(response.status(), StatusCode::CREATED);
Ok(())
}
#[tokio::test]
async fn native_listener_serves_pages_probes_and_bounds_bodies() -> Result<()> {
let directory = tempfile::tempdir()?;
fs::write(directory.path().join("index.html"), "home {{ pathname }}")?;
fs::write(directory.path().join("_404.html"), "missing")?;
let app = test_app(directory.path(), true)?;
let response = app
.clone()
.oneshot(Request::builder().uri("/").body(Body::empty())?)
.await?;
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get(header::CONTENT_SECURITY_POLICY)
.and_then(|value| value.to_str().ok())
.map(|value| value.contains("default-src 'self'")),
Some(true)
);
let bytes = axum::body::to_bytes(response.into_body(), 64 * 1024).await?;
// The pathname renders HTML-escaped, proving auto-escape stays on.
assert_eq!(bytes, "home /");
let response = app
.clone()
.oneshot(Request::builder().uri("/nowhere").body(Body::empty())?)
.await?;
assert_eq!(response.status(), StatusCode::NOT_FOUND);
let bytes = axum::body::to_bytes(response.into_body(), 64 * 1024).await?;
assert_eq!(bytes, "missing");
let response = app
.clone()
.oneshot(Request::builder().uri("/api/ready").body(Body::empty())?)
.await?;
assert_eq!(response.status(), StatusCode::OK);
// Non-GET/HEAD methods on page routes are refused with Allow.
let response = app
.clone()
.oneshot(
Request::builder()
.method(Method::POST)
.uri("/")
.body(Body::empty())?,
)
.await?;
assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(
response
.headers()
.get(header::ALLOW)
.and_then(|value| value.to_str().ok()),
Some("GET, HEAD")
);
let oversized = vec![b'a'; MAX_BODY_BYTES + 1];
let response = app
.clone()
.oneshot(
Request::builder()
.method(Method::PUT)
.uri("/api/vectors/note-1")
.header(header::CONTENT_TYPE, "application/json")
.header(header::AUTHORIZATION, format!("Bearer {TEST_TOKEN}"))
.body(Body::from(oversized))?,
)
.await?;
assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
Ok(())
}
}