use camber::RuntimeError;
#[cfg(feature = "ws")]
use camber::http::WsConn;
use camber::http::{
NegotiatedResponseMetadata, Rejection, RejectionContext, RejectionKind, RejectionProtocol,
Request, Response,
};
use serde::Deserialize;
use std::net::IpAddr;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use super::http::HttpResponse;
pub const REDACTED_BODY: &str = "internal server error";
pub const UNAVAILABLE_BODY: &str = "service unavailable";
pub const REJECTION_MESSAGE: &str = "message=request rejected";
pub const COMPLETION_MESSAGE: &str = "message=request completed";
pub const COLLAPSED_STATUS: u16 = 418;
pub const MALFORMED_JSON: &str = "{\"id\":";
pub const MALFORMED_MULTIPART: &str = "--not-the-boundary\r\nnothing here\r\n";
pub fn multipart_content_type(boundary: &str) -> Box<str> {
format!("multipart/form-data; boundary={boundary}").into_boxed_str()
}
pub const DECLARED_MESSAGE: &str = "field id is required";
pub const MAPPER_VERSION: &str = "7";
#[derive(Debug, Deserialize)]
pub struct Ticket {
pub id: Box<str>,
}
pub const UNREPRESENTABLE_HEADER: &str = "X-Broken\nName";
pub const REQUEST_ID_HEADER: &str = "X-Request-Id";
#[derive(Debug)]
pub struct Observed {
pub origin: &'static str,
pub kind: RejectionKind,
pub status: u16,
pub message: Box<str>,
pub method: Box<str>,
pub raw_path: Box<str>,
pub route: Option<Box<str>>,
pub protocol: Option<RejectionProtocol>,
pub content_type: Option<Box<str>>,
pub subprotocol: Option<Box<str>>,
pub allow: Option<Box<str>>,
pub remote: Option<IpAddr>,
pub request_id: Box<str>,
}
pub type Journal = Arc<Mutex<Vec<Observed>>>;
fn default_allow(rejection: &Rejection) -> Option<Box<str>> {
rejection
.headers()
.find(|(name, _)| name.eq_ignore_ascii_case("Allow"))
.map(|(_, value)| Box::from(value))
}
fn observe(
journal: &Journal,
origin: &'static str,
rejection: &Rejection,
context: &RejectionContext,
) {
journal
.lock()
.unwrap_or_else(|error| error.into_inner())
.push(Observed {
origin,
kind: rejection.kind(),
status: rejection.status(),
message: rejection.message().into(),
method: context.method().into(),
raw_path: context.raw_path().into(),
route: context.route().map(Box::from),
protocol: context
.negotiated()
.map(NegotiatedResponseMetadata::protocol),
content_type: context
.negotiated()
.and_then(NegotiatedResponseMetadata::content_type)
.map(Box::from),
subprotocol: context
.negotiated()
.and_then(NegotiatedResponseMetadata::subprotocol)
.map(Box::from),
allow: default_allow(rejection),
remote: context.remote_addr(),
request_id: context.request_id().as_str().into(),
});
}
fn mapper(
journal: &Journal,
origin: &'static str,
status: Option<u16>,
) -> impl Fn(&Rejection, &RejectionContext) -> Result<Response, RuntimeError> + Send + Sync + 'static
{
let journal = Arc::clone(journal);
move |rejection: &Rejection, context: &RejectionContext| {
observe(&journal, origin, rejection, context);
Response::text(
status.unwrap_or_else(|| rejection.status()),
rejection.message(),
)
}
}
pub fn recording_mapper(
journal: &Journal,
origin: &'static str,
) -> impl Fn(&Rejection, &RejectionContext) -> Result<Response, RuntimeError> + Send + Sync + 'static
{
mapper(journal, origin, None)
}
pub fn collapsing_mapper(
journal: &Journal,
origin: &'static str,
status: u16,
) -> impl Fn(&Rejection, &RejectionContext) -> Result<Response, RuntimeError> + Send + Sync + 'static
{
mapper(journal, origin, Some(status))
}
pub fn drain(journal: &Journal) -> Box<[Observed]> {
std::mem::take(&mut *journal.lock().unwrap_or_else(|error| error.into_inner()))
.into_boxed_slice()
}
pub fn only(journal: &Journal, label: &str) -> Observed {
let seen = drain(journal);
assert_eq!(
seen.len(),
1,
"{label}: one refusal invokes one mapper once, not {seen:?}"
);
seen.into_vec().remove(0)
}
pub fn counting_handler(
handled: &Arc<AtomicUsize>,
body: &'static str,
) -> impl Fn(&Request) -> std::future::Ready<Result<Response, RuntimeError>> + Send + Sync + 'static
{
let handled = Arc::clone(handled);
move |_request: &Request| {
handled.fetch_add(1, Ordering::SeqCst);
std::future::ready(Response::text(200, body))
}
}
const UNSENDABLE_BODY: &str = "unsendable";
pub fn unrepresentable_handler(
handled: &Arc<AtomicUsize>,
) -> impl Fn(&Request) -> std::future::Ready<Result<Response, RuntimeError>> + Send + Sync + 'static
{
let counted = counting_handler(handled, UNSENDABLE_BODY);
move |request: &Request| {
let answered = counted(request).into_inner();
std::future::ready(
answered.map(|response| response.with_header(UNREPRESENTABLE_HEADER, "present")),
)
}
}
#[cfg(feature = "ws")]
pub fn counting_ws_handler(
entries: &Arc<AtomicUsize>,
) -> impl Fn(&Request, WsConn) -> Result<(), RuntimeError> + Send + Sync + 'static {
let entries = Arc::clone(entries);
move |_request: &Request, _socket: WsConn| {
entries.fetch_add(1, Ordering::SeqCst);
Ok(())
}
}
pub fn ticket_handler()
-> impl Fn(&Request) -> std::future::Ready<Result<Response, RuntimeError>> + Send + Sync + 'static {
|request: &Request| {
let parsed: Result<Ticket, RuntimeError> = request.json();
std::future::ready(parsed.and_then(|ticket| Response::text(200, &ticket.id)))
}
}
pub fn multipart_count_handler()
-> impl Fn(&Request) -> std::future::Ready<Result<Response, RuntimeError>> + Send + Sync + 'static {
|request: &Request| {
let parsed = request.multipart().map(|reader| reader.parts().len());
std::future::ready(parsed.and_then(|count| Response::text(200, &count.to_string())))
}
}
pub fn naming(
answered: Result<Response, RuntimeError>,
context: &RejectionContext,
) -> Result<Response, RuntimeError> {
answered.map(|response| response.with_header(REQUEST_ID_HEADER, context.request_id().as_str()))
}
pub fn assert_no_private_text(response: &HttpResponse, private: &[&str], label: &str) {
assert!(
!private.is_empty(),
"{label}: no private text was declared, so no leak can be found"
);
let raw = String::from_utf8_lossy(response.raw());
for text in private {
assert!(
!raw.contains(text),
"{label}: the private text {text:?} reached the peer: {raw}"
);
}
}
pub type Trail = Arc<Mutex<Vec<&'static str>>>;
pub fn mark(trail: &Trail, marker: &'static str) {
trail
.lock()
.unwrap_or_else(|error| error.into_inner())
.push(marker);
}
pub fn take(trail: &Trail) -> Box<[&'static str]> {
std::mem::take(&mut *trail.lock().unwrap_or_else(|error| error.into_inner())).into_boxed_slice()
}
pub fn marking<M>(
trail: &Trail,
marker: &'static str,
inner: M,
) -> impl Fn(&Rejection, &RejectionContext) -> Result<Response, RuntimeError> + Send + Sync + 'static
where
M: Fn(&Rejection, &RejectionContext) -> Result<Response, RuntimeError> + Send + Sync + 'static,
{
let trail = Arc::clone(trail);
move |rejection: &Rejection, context: &RejectionContext| {
mark(&trail, marker);
inner(rejection, context)
}
}
pub fn request_id_of(response: &HttpResponse, label: &str) -> Box<str> {
let header = response.header(REQUEST_ID_HEADER);
assert_request_id_shape(header, label).into()
}
pub fn assert_request_id_shape<'a>(id: Option<&'a str>, label: &str) -> &'a str {
let id = id.unwrap_or_else(|| panic!("{label}: nothing carried a request identifier"));
assert_eq!(id.len(), 32, "{label}: identifier length");
assert!(
id.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)),
"{label}: identifier is lowercase hexadecimal: {id:?}"
);
id
}
pub fn assert_fixed_fallback(response: &HttpResponse, label: &str) {
assert_eq!(response.status, 500, "{label}: status");
assert_eq!(
response.header("content-type"),
Some("text/plain"),
"{label}: content type"
);
request_id_of(response, label);
assert_eq!(response.text().as_ref(), REDACTED_BODY, "{label}: body");
}
pub struct Established<'a> {
pub method: &'a str,
pub raw_path: &'a str,
pub route: Option<&'a str>,
pub protocol: Option<RejectionProtocol>,
pub content_type: Option<&'a str>,
}
pub fn assert_established(seen: &Observed, expected: &Established<'_>, label: &str) {
assert_eq!(seen.method.as_ref(), expected.method, "{label}: method");
assert_eq!(
seen.raw_path.as_ref(),
expected.raw_path,
"{label}: raw path"
);
assert_eq!(seen.route.as_deref(), expected.route, "{label}: route");
assert_eq!(seen.protocol, expected.protocol, "{label}: dispatch class");
assert_eq!(
seen.content_type.as_deref(),
expected.content_type,
"{label}: negotiated representation"
);
assert!(seen.remote.is_some(), "{label}: the transport named a peer");
assert_request_id_shape(Some(seen.request_id.as_ref()), label);
}
pub struct Collapsed<'a> {
pub kind: RejectionKind,
pub status: u16,
pub message: &'a str,
}
pub fn assert_classification(seen: &Observed, expected: &Collapsed<'_>, label: &str) {
assert_eq!(seen.kind, expected.kind, "{label}: category");
assert_eq!(seen.status, expected.status, "{label}: default status");
assert_eq!(
seen.message.as_ref(),
expected.message,
"{label}: safe message"
);
}
pub fn assert_collapsed(
journal: &Journal,
response: &HttpResponse,
label: &str,
expected: &Collapsed<'_>,
) -> Observed {
assert_eq!(
response.status, COLLAPSED_STATUS,
"{label}: every category collapses onto one wire status"
);
let seen = only(journal, label);
assert_classification(&seen, expected, label);
seen
}
pub const ROOT_CAUSE: &str = "signing key unreadable at /var/lib/camber/keys";
pub const MIDDLE_CAUSE: &str = "credential refresh failed";
#[derive(Debug)]
struct RootCause;
impl std::fmt::Display for RootCause {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(ROOT_CAUSE)
}
}
impl std::error::Error for RootCause {}
#[derive(Debug)]
struct MiddleCause(RootCause);
impl std::fmt::Display for MiddleCause {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(MIDDLE_CAUSE)
}
}
impl std::error::Error for MiddleCause {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(&self.0)
}
}
pub fn two_level_failure() -> RuntimeError {
RuntimeError::Io(std::io::Error::other(MiddleCause(RootCause)))
}
pub fn one_level_failure() -> RuntimeError {
RuntimeError::Io(std::io::Error::other(RootCause))
}