use crate::server::middleware::apply_custom_to_method_router;
use crate::server::{OrdinaryAppRouter, OrdinaryAppServerState, cors};
use arrayvec::ArrayVec;
use async_compression::Level;
use axum::extract::{ConnectInfo, MatchedPath, Path, State};
use axum::http::header::{CONTENT_TYPE, LOCATION};
use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, Uri, header};
use axum::response::IntoResponse;
use axum::routing::{delete, get, patch, post, put};
use axum::{Extension, Router};
use axum_extra::extract::CookieJar;
use base64::{Engine as B64Engine, engine::general_purpose::URL_SAFE_NO_PAD as b64};
use bytes::{BufMut, Bytes, BytesMut};
use flexbuffers::{BuilderOptions, VectorBuilder};
use hashbrown::HashMap;
use hyper::header::{CONTENT_SECURITY_POLICY, IF_NONE_MATCH};
use ordinary_auth::OrdinaryAuth;
use ordinary_config::{
CompressionAlgorithm, DatabaseModelConfig, HttpCsp, HttpMethod, HttpRouteConfig, OrdinaryConfig,
};
use ordinary_function::{Engine, FunctionCallResult, OrdinaryFunction, RenderMode};
use ordinary_server_utils::compression::get_compressed;
use ordinary_server_utils::middleware::{check_if_none_match, get_etag_hash};
use ordinary_server_utils::tcp::Sni;
use ordinary_server_utils::{GMT_FORMAT, REPORTING_ENDPOINTS, get_host_fwd};
use ordinary_storage::{CacheKind, Lookup, OrdinaryStorage};
use ordinary_types::{Kind, flexbuffer_reader_to_json, json_to_flexbuffer_vec};
use serde::Deserialize;
use smallvec::SmallVec;
use std::collections::BTreeMap;
use std::hash::BuildHasher;
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::{Arc, LazyLock, OnceLock};
use std::time::Duration;
use time::UtcDateTime;
use tower::ServiceBuilder;
use tower_http::request_id::RequestId;
use tower_http::timeout::TimeoutLayer;
use tracing::Instrument;
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub(super) async fn setup(
config: &Arc<OrdinaryConfig>,
secure: bool,
auth: &Arc<OrdinaryAuth>,
storage: &Arc<OrdinaryStorage>,
engine: &Engine,
model_map: &mut HashMap<String, DatabaseModelConfig>,
) -> (
HashMap<(Method, String), (Arc<OrdinaryFunction>, Option<HttpRouteConfig>)>,
Option<Arc<OrdinaryFunction>>,
) {
let mut function_route_map = HashMap::new();
let mut registration_function = None;
if let Some(function_configs) = config.functions.clone()
&& !function_configs.is_empty()
{
for function_config in function_configs {
if let Ok(function) = OrdinaryFunction::new(
config.clone(),
secure,
None,
engine.clone(),
function_config.clone(),
auth.clone(),
storage.clone(),
model_map,
)
.await
{
let function = Arc::new(function);
if let Some(auth) = &config.auth
&& let Some(reg) = &auth.registration
&& let Some(reg_fn) = ®.function
&& reg_fn == &function.name
{
registration_function = Some(function.clone());
}
if let Some(http_config) = &config.http
&& let Some(routes) = &http_config.routes
{
for route in routes {
if route.function == function.name {
let method = match &route.method {
HttpMethod::GET => Some(Method::GET),
HttpMethod::QUERY => Some(Method::QUERY),
HttpMethod::PUT => Some(Method::PUT),
HttpMethod::POST => Some(Method::POST),
HttpMethod::PATCH => Some(Method::PATCH),
HttpMethod::DELETE => Some(Method::DELETE),
HttpMethod::Other(other) => {
match Method::from_str(other.as_str()) {
Ok(other) => Some(other),
Err(err) => {
tracing::error!(%err);
None
}
}
}
};
if let Some(method) = method {
function_route_map.insert(
(method, route.path.clone()),
(function.clone(), route.config.clone()),
);
}
}
}
}
function_route_map.insert(
(Method::POST, function.name.clone()),
(function.clone(), None),
);
}
}
}
function_route_map.shrink_to_fit();
(function_route_map, registration_function)
}
pub(crate) fn setup_router(
config: &Arc<OrdinaryConfig>,
state: &Arc<OrdinaryAppServerState>,
api_domain: Option<&str>,
forwarded_by: &str,
forwarded_proto: &str,
) -> Option<OrdinaryAppRouter> {
if let Some(function_configs) = &config.functions
&& !function_configs.is_empty()
{
let mut router = Router::new();
router = router.route("/.ordinary/v1/functions/call/{name}", post(call));
for function_config in function_configs {
if let Some(http_config) = &config.http
&& let Some(routes) = &http_config.routes
{
for route in routes {
let timeout_s = u64::from(
route
.config
.as_ref()
.unwrap_or(
http_config
.default_route_config
.as_ref()
.unwrap_or(&HttpRouteConfig::default()),
)
.timeout
.unwrap_or(function_config.timeout.unwrap_or(10)),
);
let timeout_layer = TimeoutLayer::with_status_code(
StatusCode::REQUEST_TIMEOUT,
Duration::from_secs(timeout_s),
);
if route.function == function_config.name_validated() {
let (path, mut mr) = match route.method {
HttpMethod::GET => (route.path.as_str(), get(render)),
HttpMethod::PATCH => (route.path.as_str(), patch(render)),
HttpMethod::PUT => (route.path.as_str(), put(render)),
HttpMethod::POST => (route.path.as_str(), post(render)),
HttpMethod::DELETE => (route.path.as_str(), delete(render)),
_ => continue,
};
if let Some(names) = &route.middlewares {
mr = apply_custom_to_method_router(
mr,
config,
state,
names,
config.domain.clone(),
forwarded_by.to_string(),
forwarded_proto.to_string(),
api_domain,
);
}
mr = mr.route_layer(ServiceBuilder::new().layer(timeout_layer));
mr = cors::apply_to_route(
config
.http
.as_ref()
.and_then(|http_config| {
http_config
.default_route_config
.as_ref()
.map(|drc| drc.cors.as_ref())
})
.flatten(),
route
.config
.as_ref()
.and_then(|route_config| route_config.cors.as_ref()),
mr,
);
router = router.route(path, mr);
}
}
}
}
return Some(router);
}
None
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call(
State(state): State<Arc<OrdinaryAppServerState>>,
addr: ConnectInfo<SocketAddr>,
matched_path: MatchedPath,
Path(name): Path<String>,
sni: Option<Extension<Sni>>,
rid: Option<Extension<RequestId>>,
headers: HeaderMap,
method: Method,
uri: Uri,
body: Bytes,
) -> impl IntoResponse {
let Some((function, _)) = state.function_route_map.get(&(Method::POST, name.clone())) else {
tracing::error!("no function");
return StatusCode::NOT_FOUND.into_response();
};
let Ok(root) = flexbuffers::Reader::get_root(body.as_ref()) else {
tracing::error!("failed to get root");
return StatusCode::BAD_REQUEST.into_response();
};
let span = tracing::info_span!(
"fn",
nm = %function.name,
kind = %"call",
);
async {
let path = uri.path();
let query = uri.query().map(ToString::to_string);
let Some(host) = get_host_fwd(&headers, &uri, sni.map(|e| e.0).as_ref()) else {
tracing::error!("no host");
return StatusCode::BAD_REQUEST.into_response();
};
let Some(rid) = rid.and_then(|v| v.0.header_value().to_str().map(ToString::to_string).ok())
else {
tracing::error!("no rid");
return StatusCode::BAD_REQUEST.into_response();
};
let mut input_builder = flexbuffers::Builder::new(&BuilderOptions::SHARE_NONE);
let mut input_vec = input_builder.start_vector();
if let Err(err) = function
.input
.copy_to(&root.as_vector().idx(0), &mut input_vec, None)
{
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
input_vec.end_vector();
let claims = if function.fn_config.protected == Some(true) {
let mut claims_builder = flexbuffers::Builder::new(&BuilderOptions::SHARE_NONE);
let mut claims_vec = claims_builder.start_vector();
if let Some(value) = check_authn(&state, &headers, &mut claims_vec, None) {
return value.into_response();
}
claims_vec.end_vector();
Some(Bytes::copy_from_slice(claims_builder.view()))
} else {
None
};
if let Ok(res) = function
.call(
Bytes::copy_from_slice(input_builder.view()),
claims,
&addr.0.ip(),
addr.port(),
&host,
&method,
rid.as_str(),
&headers,
path,
matched_path.as_str(),
vec![("name".to_string(), name)],
query.as_deref(),
)
.await
{
match res {
FunctionCallResult::Result(res) => (StatusCode::OK, res).into_response(),
FunctionCallResult::StatusCode(code) => code.into_response(),
}
} else {
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
.instrument(span)
.await
}
enum Parsed {
Void,
Blob(Bytes),
String(String),
JsonBlob(String),
JsonValue(Option<serde_json::Value>),
}
#[derive(Deserialize)]
struct CspHashes {
script: Vec<String>,
style: Vec<String>,
}
static FOLDHASH: OnceLock<foldhash::fast::FixedState> = OnceLock::new();
#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
pub(crate) async fn render(
State(state): State<Arc<OrdinaryAppServerState>>,
addr: ConnectInfo<SocketAddr>,
jar: CookieJar,
matched_path: MatchedPath,
Path(path_params): Path<Vec<(String, String)>>,
sni: Option<Extension<Sni>>,
request_id: Option<Extension<RequestId>>,
headers: HeaderMap,
method: Method,
uri: Uri,
body: Bytes,
) -> impl IntoResponse {
let path = uri.path();
let query = uri.query().map(ToString::to_string);
let Some((function, http_config)) = state
.function_route_map
.get(&(method.clone(), matched_path.as_str().to_owned()))
else {
tracing::error!("no function");
return StatusCode::NOT_FOUND.into_response();
};
let span = tracing::info_span!(
"fn",
nm = %function.name,
kind = %"render",
);
async {
let Some(host) = get_host_fwd(&headers, &uri, sni.map(|s| s.0).as_ref()) else {
tracing::error!("no host");
return StatusCode::BAD_REQUEST.into_response();
};
let cache_info = if function.readonly && matches!(method, Method::GET | Method::QUERY) && let Some(http_config) = http_config && let Some(http_cache) = &http_config.cache && let Some(stored_cache) = &http_cache.stored {
let cache_span = tracing::info_span!("cache");
let mut cache_key = BytesMut::new();
cache_key.put_slice(host.as_bytes());
cache_key.put_u8(match method {
Method::GET => 0,
Method::QUERY => 1,
_ => unreachable!()
});
cache_key.put_slice(matched_path.as_str().as_bytes());
let mut check_etag = true;
if let Some(key_on) = &stored_cache.key_on {
if let Some(pp) = &key_on.path_params {
for (key, value) in &path_params {
if pp.contains(key) {
cache_key.put_slice(value.as_bytes());
}
}
}
if let Some(query) = &query && let Some(qk) = &key_on.query_keys && let Ok(mut params) = serde_html_form::from_str::<BTreeMap<String, Vec<String>>>(query) {
for query_key in qk {
if let Some(list) = params.get_mut(query_key) {
list.sort();
for value in list {
cache_key.put_slice(value.as_bytes());
}
}
}
}
if key_on.body_hash == Some(true) && !body.is_empty() {
let hasher = FOLDHASH.get_or_init(foldhash::fast::FixedState::default);
let hash = hasher.hash_one(body.as_ref());
cache_key.put_u64(hash);
}
if let Some(etag) = &key_on.etag {
check_etag = *etag;
}
}
let mut checks = SmallVec::<[Bytes; 5]>::new();
if check_etag && let Some(incoming_etag) = headers.get(IF_NONE_MATCH) {
let etag = incoming_etag.as_bytes();
let cache_key_len = cache_key.len();
if etag.len() > 10 && etag.len() < 22 {
cache_key.put_slice(&etag[0..11]);
} else if etag.len() > 22 {
cache_key.put_slice(&etag[0..22]);
}
checks.push(cache_key.clone().into());
cache_key.truncate(cache_key_len);
}
let mut compression_intersection = ArrayVec::<CompressionAlgorithm, 5>::new();
let incoming_compressions = if let Some(accept_encoding) = headers.get(header::ACCEPT_ENCODING) && let Ok(accept_encoding_str) = accept_encoding.to_str() {
accept_encoding_str.split(", ").collect::<ArrayVec<_, 4>>()
} else {
ArrayVec::new()
};
if let Some(compressions) = &stored_cache.internal_compressions {
let mut uncompressed = false;
for compression in compressions {
match compression {
CompressionAlgorithm::All => tracing::error!("'All' should be unreachable"),
CompressionAlgorithm::Uncompressed => uncompressed = true,
CompressionAlgorithm::Brotli => {
if incoming_compressions.contains(&"br") {
compression_intersection.push(compression.to_owned());
}
}
CompressionAlgorithm::Deflate => {
if incoming_compressions.contains(&"deflate") {
compression_intersection.push(compression.to_owned());
}
}
CompressionAlgorithm::Zstd { level: _ } => {
if incoming_compressions.contains(&"zstd") {
compression_intersection.push(compression.to_owned());
}
}
CompressionAlgorithm::Gzip => {
if incoming_compressions.contains(&"gzip") {
compression_intersection.push(compression.to_owned());
}
}
}
}
if uncompressed {
compression_intersection.push(CompressionAlgorithm::Uncompressed);
}
}
let mut content_type_intersection = SmallVec::<[String; 2]>::new();
let incoming_content_types = if let Some(accept) = headers.get(header::ACCEPT) && let Ok(accept_str) = accept.to_str() {
accept_str.split(',').collect::<SmallVec<[_; 4]>>()
} else {
SmallVec::new()
};
if let Some(content_types) = &stored_cache.internal_content_types {
'outer: for content_type in content_types {
for incoming_content_type in &incoming_content_types {
if *incoming_content_type == "*/*" {
content_type_intersection.clone_from(content_types);
break 'outer;
} else if content_type.starts_with(incoming_content_type) {
content_type_intersection.push(content_type.to_owned());
break;
}
}
}
}
for content_type in &content_type_intersection {
let cache_key_len = cache_key.len();
cache_key.put_slice(content_type.as_bytes());
for compression in &compression_intersection {
let cache_key_len = cache_key.len();
cache_key.put_u8(compression.as_u8());
checks.push(cache_key.clone().into());
cache_key.truncate(cache_key_len);
}
cache_key.truncate(cache_key_len);
}
let check = cache_span.in_scope(|| {
state.storage.cache.check(
stored_cache,
CacheKind::Http,
Lookup::new(cache_key.clone().into(), &checks),
)
});
if let Ok(hit) = check && let Some(res) = hit && let Ok(root) = flexbuffers::Reader::get_root(res.as_ref()) {
let root_vec = root.as_vector();
let last_modified = root_vec.idx(0).as_str();
let etag = root_vec.idx(1).as_str();
let mut header_map = HeaderMap::with_capacity(11);
if etag.is_empty() && let Some(etag_header) = headers.get(IF_NONE_MATCH) && let Ok(etag) = etag_header.to_str() {
match process_readonly_response(
true,
true,
&headers,
Some(http_config),
header_map,
None,
etag,
last_modified,
StatusCode::OK,
Bytes::new(),
"",
) {
Ok(res) => return res.into_response(),
Err(err) => {
tracing::error!(%err);
}
}
} else {
let content_type = root_vec.idx(2).as_str();
let compression = CompressionAlgorithm::from_u8(root_vec.idx(3).as_u8(), None);
let res = Bytes::copy_from_slice(root_vec.idx(4).as_blob().0);
if matches!(content_type, "text/html" | "text/html; charset=utf-8") {
let csp = root_vec.idx(5).as_str();
if !csp.is_empty() && let Ok(csp_header) = HeaderValue::from_str(csp) {
header_map.insert(CONTENT_SECURITY_POLICY, csp_header);
header_map.insert(REPORTING_ENDPOINTS, function.reporting_endpoints.clone());
}
}
match process_readonly_response(
false,
true,
&headers,
Some(http_config),
header_map,
Some(&compression),
etag,
last_modified,
StatusCode::OK,
res,
content_type,
) {
Ok(res) => return res.into_response(),
Err(err) => {
tracing::error!(%err);
}
}
}
}
Some((cache_key, compression_intersection.first().map(ToOwned::to_owned)))
} else {
None
};
let Some(rid) =
request_id.and_then(|v| v.header_value().to_str().map(ToString::to_string).ok())
else {
tracing::error!("no rid");
return StatusCode::BAD_REQUEST.into_response();
};
let input: Parsed = match function.input {
Kind::Void => Parsed::Void,
Kind::Blob | Kind::Json | Kind::String => {
if matches!(method, Method::GET | Method::DELETE) {
tracing::error!("method not allowed for blob input type");
return StatusCode::METHOD_NOT_ALLOWED.into_response();
}
match function.input {
Kind::Json => {
let Ok(json) = std::str::from_utf8(body.as_ref()) else {
tracing::error!("body failed to convert to str");
return StatusCode::BAD_REQUEST.into_response();
};
let Some(content_type) = headers.get(CONTENT_TYPE) else {
tracing::error!("content-type header missing from request");
return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response();
};
let Ok(content_type_str) = content_type.to_str() else {
tracing::error!("content-type failed to convert to str");
return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response();
};
if !content_type_str.starts_with("application/json") {
tracing::error!("content type is not application/json");
return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response();
}
Parsed::JsonBlob(json.to_string())
}
Kind::String => {
let Ok(string) = std::str::from_utf8(body.as_ref()) else {
tracing::error!("failed to parse string");
return StatusCode::BAD_REQUEST.into_response();
};
Parsed::String(string.into())
}
_ => Parsed::Blob(body)
}
}
_ => Parsed::JsonValue(if matches!(method, Method::GET | Method::DELETE) {
if let Some(query) = &query {
serde_html_form::from_str(query).ok()
} else if path_params.is_empty() {
tracing::error!("no query or path params supplied");
return StatusCode::BAD_REQUEST.into_response();
} else {
None
}
} else if let Some(content_type) = headers.get(CONTENT_TYPE) {
let Ok(content_type_str) = content_type.to_str() else {
tracing::error!("content-type failed to convert to str");
return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response();
};
let mut body: BytesMut = body.into();
if content_type_str.starts_with("application/json") {
simd_json::from_slice(&mut body[..]).ok()
} else if content_type_str.starts_with("application/x-www-form-urlencoded") {
serde_html_form::from_bytes(body.as_ref()).ok()
} else {
tracing::error!("unsupported content-type");
return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response();
}
} else if let Some(query) = &query {
serde_html_form::from_str(query).ok()
} else if path_params.is_empty() {
tracing::error!("no query or path params and no content-type on headers");
return StatusCode::BAD_REQUEST.into_response();
} else {
None
}),
};
let mut builder = flexbuffers::Builder::new(&BuilderOptions::SHARE_NONE);
let mut input_vec = builder.start_vector();
match input {
Parsed::Void => {
input_vec.push(());
}
Parsed::Blob(bytes) => {
input_vec.push(flexbuffers::Blob(bytes.as_ref()));
}
Parsed::String(string) => {
input_vec.push(string.as_str());
}
Parsed::JsonBlob(json) => {
input_vec.push(json.as_str());
}
Parsed::JsonValue(input) => {
if let Some(mut input) = input {
if let Some(obj) = input.as_object_mut() {
for (key, value) in &path_params {
obj.insert(key.to_owned(), value.to_owned().into());
}
}
if let Err(err) =
json_to_flexbuffer_vec(&function.input, &input, &mut input_vec)
{
tracing::error!(%err);
return StatusCode::BAD_REQUEST.into_response();
}
} else {
match &function.input {
Kind::List { .. } => {
let list = serde_json::Value::Array(path_params.iter().map(|(_, v)| v.to_owned().into()).collect::<Vec<serde_json::Value>>());
if let Err(err) = json_to_flexbuffer_vec(&function.input, &list, &mut input_vec) {
tracing::error!(%err);
return StatusCode::BAD_REQUEST.into_response();
}
}
Kind::Object { .. } => {
let mut obj = serde_json::Map::new();
for (key, value) in &path_params {
obj.insert(key.to_owned(), value.to_owned().into());
}
if let Err(err) = json_to_flexbuffer_vec(&function.input, &serde_json::Value::Object(obj), &mut input_vec) {
tracing::error!(%err);
return StatusCode::BAD_REQUEST.into_response();
}
}
_ => {
if let Some((_, val)) = &path_params.last() && let Err(err) = json_to_flexbuffer_vec(&function.input, &val.to_owned().into(), &mut input_vec) {
tracing::error!(%err);
return StatusCode::BAD_REQUEST.into_response();
}
}
}
}
}
}
input_vec
.end_vector();
let claims = if function.fn_config.protected == Some(true) {
let mut claims_builder = flexbuffers::Builder::new(&BuilderOptions::SHARE_NONE);
let mut claims_vec = claims_builder.start_vector();
if let Some(value) = check_authn(&state, &headers, &mut claims_vec, Some(jar)) {
return value.into_response();
}
claims_vec.end_vector();
Some(Bytes::copy_from_slice(claims_builder.view()))
} else {
None
};
let (res, code, metadata) = match function
.render(
Bytes::copy_from_slice(builder.view()),
RenderMode::Input(claims),
&addr.0.ip(),
addr.port(),
&host,
&method,
rid.as_str(),
&headers,
path,
matched_path.as_str(),
path_params,
query.as_deref(),
)
.await
{
Ok(res) => res,
Err(err) => {
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
};
if code == 0 {
if let Some(accept) = headers.get(header::ACCEPT)
&& let Ok(accept_str) = accept.to_str()
{
let content_type = if let Some(content_type) = metadata.get("content-type") {
if accept_str == "*/*" || accept_str.contains(content_type) {
Some(content_type.as_str())
} else {
None
}
} else if accept_str.contains("application/json") {
Some("application/json")
} else {
None
};
#[allow(clippy::collapsible_if)]
if let Some(content_type) = content_type {
if content_type.starts_with("application/json") {
return match flexbuffers::Reader::get_root(res.as_ref()) {
Ok(root) => match flexbuffer_reader_to_json(&function.output, &root) {
Ok(val) => match simd_json::to_string(&val) {
Ok(serialized) => {
(StatusCode::OK, [(CONTENT_TYPE, content_type)], serialized)
.into_response()
}
Err(err) => {
tracing::error!(%err);
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
},
Err(err) => {
tracing::error!(%err);
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
},
Err(err) => {
tracing::error!(%err);
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
};
}
}
}
(StatusCode::OK, res).into_response()
} else if let Ok(status_code) = StatusCode::from_u16(code) {
match status_code {
StatusCode::MOVED_PERMANENTLY | StatusCode::PERMANENT_REDIRECT | StatusCode::FOUND | StatusCode::SEE_OTHER | StatusCode::TEMPORARY_REDIRECT | StatusCode::CREATED => {
if let Some(location) = metadata.get("location") {
(status_code, [(LOCATION, location)], res).into_response()
} else {
tracing::warn!(code = %status_code, "no location header provided on redirecting status.");
(status_code, res).into_response()
}
}
_ => {
let content_type = if let Some(content_type) = metadata.get("content-type") {
content_type
} else {
tracing::warn!("no content-type provided. default application/octet-stream sent");
&"application/octet-stream".to_string()
};
let mut header_map = HeaderMap::with_capacity(11);
let csp_string = if matches!(content_type.as_str(), "text/html" | "text/html; charset=utf-8") {
let csp_hashes = if let Some(csp_hashes) = metadata.get("csp-hashes") {
match simd_json::from_slice(&mut csp_hashes.as_bytes().to_vec()[..]) {
Ok(ch) => ch,
Err(err) => {
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
}
} else {
CspHashes {
script: vec![],
style: vec![],
}
};
let http_csp = if let Some(http_config) = http_config {
if let Some(http_csp) = &http_config.csp {
http_csp
} else {
&HttpCsp::default()
}
} else {
&HttpCsp::default()
};
let csp_string = http_csp.build_string(
&state.config.http.as_ref().and_then(|http_config| http_config.default_route_config.as_ref().map(|drc| drc.csp.clone())).flatten().unwrap_or_default(),
Some(csp_hashes.style),
Some(csp_hashes.script),
function.secure,
false,
);
let csp_header = match HeaderValue::from_str(&csp_string) {
Ok(ch) => ch,
Err(err) => {
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
};
header_map.insert(CONTENT_SECURITY_POLICY, csp_header);
header_map.insert(REPORTING_ENDPOINTS, function.reporting_endpoints.clone());
Some(csp_string)
} else {
None
};
if function.readonly && matches!(method, Method::GET | Method::QUERY) {
let last_modified = match UtcDateTime::now().format(&GMT_FORMAT) {
Ok(lm) => lm,
Err(err) => {
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
};
let etag = if let Some(config) = &http_config
&& let Some(http_cache) = &config.cache
{
get_etag_hash(res.as_ref(), Some(http_cache))
} else {
get_etag_hash(res.as_ref(), None)
};
let compression_info = if let Some(http_config) = http_config && let Some(http_cache) = &http_config.cache && let Some(stored_cache) = &http_cache.stored {
if let Some((mut cache_key, compression)) = cache_info {
let compressed = if let Some(compression) = compression && let Some(content_types) = &stored_cache.internal_content_types && content_types.contains(content_type) {
Some((get_compressed_res(&res, &compression).await, compression))
} else {
None
};
let state_clone = state.clone();
let stored_cache_clone = stored_cache.clone();
let content_type_clone = content_type.clone();
let last_modified_clone = last_modified.clone();
let etag_clone = etag.clone();
let compressed_clone = compressed.clone();
let cache_span = tracing::info_span!("cache");
tokio::task::spawn_blocking(move || {
let mut should_etag_only_cache = true;
if let Some(key_on) = &stored_cache_clone.key_on && key_on.etag == Some(false) {
should_etag_only_cache = false;
}
if should_etag_only_cache {
let cache_key_len = cache_key.len();
cache_key.put_slice(etag_clone.as_bytes());
let mut builder = flexbuffers::Builder::new(
&BuilderOptions::SHARE_NONE,
);
let mut builder_vec = builder.start_vector();
builder_vec.push(last_modified_clone.as_str());
builder_vec.end_vector();
cache_span.in_scope(|| {
if let Err(err) = state_clone.storage.cache.write(
&stored_cache_clone,
CacheKind::Http,
cache_key.as_ref(),
builder.view(),
None,
true,
) {
tracing::error!(%err, "failed to write to cache");
}
});
cache_key.truncate(cache_key_len);
}
if let Some((compressed, compression)) = &compressed_clone {
cache_key.put_slice(content_type_clone.as_bytes());
cache_key.put_u8(compression.as_u8());
let mut builder = flexbuffers::Builder::new(
&BuilderOptions::SHARE_NONE,
);
let mut builder_vec = builder.start_vector();
builder_vec.push(last_modified_clone.as_str());
builder_vec.push(etag_clone.as_str());
builder_vec.push(content_type_clone.as_str());
builder_vec.push(compression.as_u8());
builder_vec.push(flexbuffers::Blob(compressed.as_ref()));
if let Some(csp_string) = csp_string {
builder_vec.push(csp_string.as_str());
}
builder_vec.end_vector();
cache_span.in_scope(|| {
if let Err(err) = state_clone.storage.cache.write(
&stored_cache_clone,
CacheKind::Http,
cache_key.as_ref(),
builder.view(),
None,
true,
) {
tracing::error!(%err, "failed to write to cache");
}
});
}
});
compressed
} else {
None
}
} else {
None
};
match process_readonly_response(
false,
false,
&headers,
http_config.as_ref(),
header_map,
compression_info.as_ref().map(|ci| &ci.1),
etag.as_str(),
last_modified.as_str(),
status_code,
compression_info.as_ref().map(|ci| ci.0.clone()).unwrap_or(res),
content_type,
) {
Ok(res) => res.into_response(),
Err(err) => {
tracing::error!(%err);
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
} else {
if let Ok(val) = HeaderValue::from_str(content_type) {
header_map.insert(CONTENT_TYPE, val);
}
header_map.insert(header::VARY, VARY_HEADER.clone());
(status_code, header_map, res).into_response()
}
}
}
} else {
tracing::error!(code, "invalid HTTP status");
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
.instrument(span)
.await
}
#[inline]
fn check_authn(
state: &Arc<OrdinaryAppServerState>,
headers: &HeaderMap,
claims_vec: &mut VectorBuilder,
jar: Option<CookieJar>,
) -> Option<StatusCode> {
let b64_token = if let Some(val) = headers.get("authorization")
&& let Ok(str_val) = val.to_str()
&& let Some(b64_token) = str_val.strip_prefix("Bearer ")
{
b64_token.to_owned()
} else if let Some(jar) = jar {
let cookie_name = if state.secure_cookies {
"__Host-ORDINARY-ACCESS-TOKEN"
} else {
"ORDINARY-ACCESS-TOKEN"
};
if let Some(cookie) = jar.get(cookie_name) {
cookie.value().to_owned()
} else {
return Some(StatusCode::UNAUTHORIZED);
}
} else {
return Some(StatusCode::UNAUTHORIZED);
};
if let Ok(token) = b64.decode(&b64_token) {
return match state.auth.verify_access_token(&token) {
Ok((account, claims_root)) => {
claims_vec.push(account);
for field in &state.auth.config.access_token.claims {
if let Err(err) =
field
.kind
.copy_to(&claims_root.idx(field.idx as usize), claims_vec, None)
{
tracing::error!(%err);
return Some(StatusCode::INTERNAL_SERVER_ERROR);
}
}
None
}
Err(err) => {
tracing::error!(%err);
Some(StatusCode::UNAUTHORIZED)
}
};
}
Some(StatusCode::UNAUTHORIZED)
}
static VARY_HEADER: LazyLock<HeaderValue> = LazyLock::new(|| {
HeaderValue::from_str(&format!(
"{}, {}",
header::ACCEPT.as_str(),
header::ACCEPT_ENCODING.as_str()
))
.unwrap_or_else(|_| HeaderValue::from_static(header::ACCEPT_ENCODING.as_str()))
});
#[inline]
#[allow(clippy::too_many_arguments)]
fn process_readonly_response(
force_304: bool,
check_last_modified: bool,
headers: &HeaderMap,
http: Option<&HttpRouteConfig>,
mut header_map: HeaderMap,
compression: Option<&CompressionAlgorithm>,
etag: &str,
last_modified: &str,
status_code: StatusCode,
res: Bytes,
content_type: &str,
) -> anyhow::Result<impl IntoResponse> {
header_map.insert(header::VARY, VARY_HEADER.clone());
if let Ok(etag) = HeaderValue::from_str(etag) {
header_map.insert(header::ETAG, etag);
}
if let Ok(last_modified) = HeaderValue::from_str(last_modified) {
header_map.insert(header::LAST_MODIFIED, last_modified);
}
if let Some(http) = http
&& let Some(http_cache) = &http.cache
{
if let Some(cache_control) = &http_cache.cache_control {
let mut header_val = String::new();
cache_control.header_value(&mut header_val, "")?;
if let Ok(cache_control) = HeaderValue::from_str(header_val.as_str()) {
header_map.insert(header::CACHE_CONTROL, cache_control);
}
}
if let Some(expires_s) = http_cache.expires {
let future = UtcDateTime::now() + time::Duration::seconds(expires_s.cast_signed());
if let Ok(formatted) = future.format(&GMT_FORMAT)
&& let Ok(expires) = HeaderValue::from_str(formatted.as_str())
{
header_map.insert(header::EXPIRES, expires);
}
}
}
if let Some(etag) = check_if_none_match(headers, etag)
&& let Ok(etag_header) = HeaderValue::from_str(etag)
{
header_map.insert(header::ETAG, etag_header);
Ok((StatusCode::NOT_MODIFIED, header_map).into_response())
} else if check_last_modified
&& let Some(if_modified_since) = headers.get(header::IF_MODIFIED_SINCE)
&& let Ok(if_modified_since_str) = if_modified_since.to_str()
&& let Ok(if_modified_since) = UtcDateTime::parse(if_modified_since_str, &GMT_FORMAT)
&& let Ok(last_modified) = UtcDateTime::parse(last_modified, &GMT_FORMAT)
&& if_modified_since >= last_modified
{
Ok((StatusCode::NOT_MODIFIED, header_map).into_response())
} else {
if force_304 {
return Ok((StatusCode::NOT_MODIFIED, header_map).into_response());
}
if let Some(compression) = compression {
match compression {
CompressionAlgorithm::All | CompressionAlgorithm::Uncompressed => (),
_ => {
header_map.insert(
header::CONTENT_ENCODING,
HeaderValue::from_static(compression.as_str()),
);
}
}
}
header_map.insert(CONTENT_TYPE, HeaderValue::from_str(content_type)?);
Ok((status_code, header_map, res).into_response())
}
}
#[inline]
async fn get_compressed_res(res: &Bytes, compression: &CompressionAlgorithm) -> Bytes {
match compression {
CompressionAlgorithm::Zstd { level } => {
get_compressed(
res.as_ref(),
compression.as_str(),
Some(Level::Precise(i32::from(*level))),
)
.await
}
_ => get_compressed(res.as_ref(), compression.as_str(), None).await,
}
}