use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use axum::body::Body;
use axum::http::{Method, Request, Response, StatusCode, header};
use bytes::Bytes;
use futures_util::future::BoxFuture;
use http_body_util::BodyExt;
use tower::{Layer, Service};
use crate::Cache;
pub fn cache_page(ttl: Duration) -> CachePageLayer {
CachePageLayer { ttl, cache: None }
}
#[derive(Clone)]
pub struct CachePageLayer {
ttl: Duration,
cache: Option<Arc<Cache>>,
}
impl CachePageLayer {
pub fn with_cache(mut self, cache: Cache) -> Self {
self.cache = Some(Arc::new(cache));
self
}
}
impl<S> Layer<S> for CachePageLayer {
type Service = CachePageService<S>;
fn layer(&self, inner: S) -> Self::Service {
CachePageService {
inner,
ttl: self.ttl,
cache: self.cache.clone(),
}
}
}
#[derive(Clone)]
pub struct CachePageService<S> {
inner: S,
ttl: Duration,
cache: Option<Arc<Cache>>,
}
impl<S> Service<Request<Body>> for CachePageService<S>
where
S: Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
S::Future: Send + 'static,
S::Error: Send + 'static,
{
type Response = Response<Body>;
type Error = S::Error;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, req: Request<Body>) -> Self::Future {
let mut inner = self.inner.clone();
let ttl = self.ttl;
let explicit_cache = self.cache.clone();
Box::pin(async move {
let method = req.method().clone();
if method != Method::GET && method != Method::HEAD {
return inner.call(req).await;
}
if request_is_personalised(&req) {
return inner.call(req).await;
}
let host = req
.headers()
.get(header::HOST)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_owned();
let uri = req.uri().to_string();
let cache_key = format!("cache:page:{}:{}:{}", method, host, uri);
let cache: Option<&Cache> = if let Some(ref c) = explicit_cache {
Some(c.as_ref())
} else {
crate::ambient()
};
if let Some(cache) = cache {
if let Some(stored) = cache.get_bytes_raw(&cache_key).await {
if let Ok(resp) = deserialise_cached_response(stored) {
return Ok(resp);
}
}
}
let resp = inner.call(req).await?;
let status = resp.status();
if status != StatusCode::OK {
return Ok(resp);
}
let should_skip = response_bypasses_cache(&resp);
let (parts, body) = resp.into_parts();
let body_bytes = match body.collect().await {
Ok(collected) => collected.to_bytes(),
Err(e) => {
tracing::error!(
error = %e,
"cache_page: failed to collect upstream response body; returning 502"
);
let mut resp = Response::new(Body::from("Bad Gateway"));
*resp.status_mut() = StatusCode::BAD_GATEWAY;
return Ok(resp);
}
};
if !should_skip {
if let Some(cache) = explicit_cache.as_deref().or_else(|| crate::ambient()) {
let serialised = serialise_cached_response(&parts, &body_bytes);
cache.set_bytes_raw(&cache_key, serialised, Some(ttl)).await;
}
}
let resp = Response::from_parts(parts, Body::from(body_bytes));
Ok(resp)
})
}
}
fn request_is_personalised<B>(req: &Request<B>) -> bool {
request_has_session_cookie(req) || req.headers().contains_key(header::AUTHORIZATION)
}
fn request_has_session_cookie<B>(req: &Request<B>) -> bool {
req.headers()
.get(header::COOKIE)
.and_then(|v| v.to_str().ok())
.map(|cookie_str| {
cookie_str
.split(';')
.any(|pair| pair.trim().starts_with("umbral_session="))
})
.unwrap_or(false)
}
fn response_bypasses_cache<B>(resp: &Response<B>) -> bool {
let headers = resp.headers();
if let Some(cc) = headers.get(header::CACHE_CONTROL) {
if cc.to_str().unwrap_or("").split(',').any(|d| {
let d = d.trim();
d.eq_ignore_ascii_case("no-store")
|| d.eq_ignore_ascii_case("private")
|| d.eq_ignore_ascii_case("no-cache")
}) {
return true;
}
}
if headers.contains_key(header::SET_COOKIE) {
return true;
}
false
}
fn serialise_cached_response(parts: &http::response::Parts, body: &Bytes) -> Vec<u8> {
let mut out: Vec<u8> = Vec::new();
let header_count = parts.headers.len() as u32;
out.extend_from_slice(&header_count.to_le_bytes());
for (name, value) in &parts.headers {
let name_bytes = name.as_str().as_bytes();
let value_bytes = value.as_bytes();
out.extend_from_slice(&(name_bytes.len() as u32).to_le_bytes());
out.extend_from_slice(name_bytes);
out.extend_from_slice(&(value_bytes.len() as u32).to_le_bytes());
out.extend_from_slice(value_bytes);
}
out.extend_from_slice(body);
out
}
fn deserialise_cached_response(data: Vec<u8>) -> Result<Response<Body>, ()> {
if data.len() < 4 {
return Err(());
}
let mut pos = 0;
let header_count = u32::from_le_bytes(data[pos..pos + 4].try_into().map_err(|_| ())?) as usize;
pos += 4;
let mut builder = Response::builder().status(StatusCode::OK);
for _ in 0..header_count {
if pos + 4 > data.len() {
return Err(());
}
let name_len = u32::from_le_bytes(data[pos..pos + 4].try_into().map_err(|_| ())?) as usize;
pos += 4;
if pos + name_len > data.len() {
return Err(());
}
let name = std::str::from_utf8(&data[pos..pos + name_len]).map_err(|_| ())?;
pos += name_len;
if pos + 4 > data.len() {
return Err(());
}
let val_len = u32::from_le_bytes(data[pos..pos + 4].try_into().map_err(|_| ())?) as usize;
pos += 4;
if pos + val_len > data.len() {
return Err(());
}
let value = &data[pos..pos + val_len];
pos += val_len;
builder = builder.header(name, value);
}
let body_bytes = Bytes::copy_from_slice(&data[pos..]);
builder.body(Body::from(body_bytes)).map_err(|_| ())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn authorization_header_makes_request_personalised() {
let req = Request::builder()
.header(header::AUTHORIZATION, "Bearer abc.def.ghi")
.body(Body::empty())
.unwrap();
assert!(
request_is_personalised(&req),
"an Authorization-bearing request must bypass the shared page cache"
);
}
#[test]
fn private_and_no_cache_responses_bypass_cache() {
for directive in ["private", "no-cache", "no-store"] {
let resp = Response::builder()
.header(header::CACHE_CONTROL, directive)
.body(Body::empty())
.unwrap();
assert!(
response_bypasses_cache(&resp),
"`Cache-Control: {directive}` must not be stored in the shared cache"
);
}
}
#[test]
fn plain_get_is_cacheable() {
let req = Request::builder().body(Body::empty()).unwrap();
assert!(!request_is_personalised(&req));
let resp = Response::builder().body(Body::empty()).unwrap();
assert!(!response_bypasses_cache(&resp));
}
}