diff options
Diffstat (limited to 'third_party/rust/warp/src')
46 files changed, 11001 insertions, 0 deletions
diff --git a/third_party/rust/warp/src/error.rs b/third_party/rust/warp/src/error.rs new file mode 100644 index 0000000000..64220b633e --- /dev/null +++ b/third_party/rust/warp/src/error.rs @@ -0,0 +1,79 @@ +use std::convert::Infallible; +use std::error::Error as StdError; +use std::fmt; + +type BoxError = Box<dyn std::error::Error + Send + Sync>; + +/// Errors that can happen inside warp. +pub struct Error { + inner: BoxError, +} + +impl Error { + pub(crate) fn new<E: Into<BoxError>>(err: E) -> Error { + Error { inner: err.into() } + } +} + +impl fmt::Debug for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // Skip showing worthless `Error { .. }` wrapper. + fmt::Debug::fmt(&self.inner, f) + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&self.inner, f) + } +} + +impl StdError for Error { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(self.inner.as_ref()) + } +} + +impl From<Infallible> for Error { + fn from(infallible: Infallible) -> Error { + match infallible {} + } +} + +#[test] +fn error_size_of() { + assert_eq!( + ::std::mem::size_of::<Error>(), + ::std::mem::size_of::<usize>() * 2 + ); +} + +#[test] +fn error_source() { + let e = Error::new(std::fmt::Error {}); + assert!(e.source().unwrap().is::<std::fmt::Error>()); +} + +macro_rules! unit_error { + ( + $(#[$docs:meta])* + $pub:vis $typ:ident: $display:literal + ) => ( + $(#[$docs])* + $pub struct $typ { _p: (), } + + impl ::std::fmt::Debug for $typ { + fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result { + f.debug_struct(stringify!($typ)).finish() + } + } + + impl ::std::fmt::Display for $typ { + fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result { + f.write_str($display) + } + } + + impl ::std::error::Error for $typ {} + ) +} diff --git a/third_party/rust/warp/src/filter/and.rs b/third_party/rust/warp/src/filter/and.rs new file mode 100644 index 0000000000..5edd90fed8 --- /dev/null +++ b/third_party/rust/warp/src/filter/and.rs @@ -0,0 +1,97 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::ready; +use pin_project::pin_project; + +use super::{Combine, Filter, FilterBase, Internal, Tuple}; +use crate::generic::CombinedTuples; +use crate::reject::CombineRejection; + +#[derive(Clone, Copy, Debug)] +pub struct And<T, U> { + pub(super) first: T, + pub(super) second: U, +} + +impl<T, U> FilterBase for And<T, U> +where + T: Filter, + T::Extract: Send, + U: Filter + Clone + Send, + <T::Extract as Tuple>::HList: Combine<<U::Extract as Tuple>::HList> + Send, + CombinedTuples<T::Extract, U::Extract>: Send, + U::Error: CombineRejection<T::Error>, +{ + type Extract = CombinedTuples<T::Extract, U::Extract>; + type Error = <U::Error as CombineRejection<T::Error>>::One; + type Future = AndFuture<T, U>; + + fn filter(&self, _: Internal) -> Self::Future { + AndFuture { + state: State::First(self.first.filter(Internal), self.second.clone()), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct AndFuture<T: Filter, U: Filter> { + #[pin] + state: State<T::Future, T::Extract, U>, +} + +#[pin_project(project = StateProj)] +enum State<T, TE, U: Filter> { + First(#[pin] T, U), + Second(Option<TE>, #[pin] U::Future), + Done, +} + +impl<T, U> Future for AndFuture<T, U> +where + T: Filter, + U: Filter, + <T::Extract as Tuple>::HList: Combine<<U::Extract as Tuple>::HList> + Send, + U::Error: CombineRejection<T::Error>, +{ + type Output = Result< + CombinedTuples<T::Extract, U::Extract>, + <U::Error as CombineRejection<T::Error>>::One, + >; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + self.project().state.poll(cx) + } +} + +impl<T, TE, U, E> Future for State<T, TE, U> +where + T: Future<Output = Result<TE, E>>, + U: Filter, + TE: Tuple, + TE::HList: Combine<<U::Extract as Tuple>::HList> + Send, + U::Error: CombineRejection<E>, +{ + type Output = Result<CombinedTuples<TE, U::Extract>, <U::Error as CombineRejection<E>>::One>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + loop { + match self.as_mut().project() { + StateProj::First(first, second) => { + let ex1 = ready!(first.poll(cx))?; + let fut2 = second.filter(Internal); + self.set(State::Second(Some(ex1), fut2)); + } + StateProj::Second(ex1, second) => { + let ex2 = ready!(second.poll(cx))?; + let ex3 = ex1.take().unwrap().combine(ex2); + self.set(State::Done); + return Poll::Ready(Ok(ex3)); + } + StateProj::Done => panic!("polled after complete"), + } + } + } +} diff --git a/third_party/rust/warp/src/filter/and_then.rs b/third_party/rust/warp/src/filter/and_then.rs new file mode 100644 index 0000000000..efed5fe8cf --- /dev/null +++ b/third_party/rust/warp/src/filter/and_then.rs @@ -0,0 +1,110 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Func, Internal}; +use crate::reject::CombineRejection; + +#[derive(Clone, Copy, Debug)] +pub struct AndThen<T, F> { + pub(super) filter: T, + pub(super) callback: F, +} + +impl<T, F> FilterBase for AndThen<T, F> +where + T: Filter, + F: Func<T::Extract> + Clone + Send, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: CombineRejection<T::Error>, +{ + type Extract = (<F::Output as TryFuture>::Ok,); + type Error = <<F::Output as TryFuture>::Error as CombineRejection<T::Error>>::One; + type Future = AndThenFuture<T, F>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + AndThenFuture { + state: State::First(self.filter.filter(Internal), self.callback.clone()), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct AndThenFuture<T, F> +where + T: Filter, + F: Func<T::Extract>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: CombineRejection<T::Error>, +{ + #[pin] + state: State<T::Future, F>, +} + +#[pin_project(project = StateProj)] +enum State<T, F> +where + T: TryFuture, + F: Func<T::Ok>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: CombineRejection<T::Error>, +{ + First(#[pin] T, F), + Second(#[pin] F::Output), + Done, +} + +impl<T, F> Future for AndThenFuture<T, F> +where + T: Filter, + F: Func<T::Extract>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: CombineRejection<T::Error>, +{ + type Output = Result< + (<F::Output as TryFuture>::Ok,), + <<F::Output as TryFuture>::Error as CombineRejection<T::Error>>::One, + >; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + self.project().state.poll(cx) + } +} + +impl<T, F> Future for State<T, F> +where + T: TryFuture, + F: Func<T::Ok>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: CombineRejection<T::Error>, +{ + type Output = Result< + (<F::Output as TryFuture>::Ok,), + <<F::Output as TryFuture>::Error as CombineRejection<T::Error>>::One, + >; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + loop { + match self.as_mut().project() { + StateProj::First(first, second) => { + let ex1 = ready!(first.try_poll(cx))?; + let fut2 = second.call(ex1); + self.set(State::Second(fut2)); + } + StateProj::Second(second) => { + let ex2 = match ready!(second.try_poll(cx)) { + Ok(item) => Ok((item,)), + Err(err) => Err(From::from(err)), + }; + self.set(State::Done); + return Poll::Ready(ex2); + } + StateProj::Done => panic!("polled after complete"), + } + } + } +} diff --git a/third_party/rust/warp/src/filter/boxed.rs b/third_party/rust/warp/src/filter/boxed.rs new file mode 100644 index 0000000000..5dd43cbcab --- /dev/null +++ b/third_party/rust/warp/src/filter/boxed.rs @@ -0,0 +1,100 @@ +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; + +use futures_util::TryFutureExt; + +use super::{Filter, FilterBase, Internal, Tuple}; +use crate::reject::Rejection; + +/// A type representing a boxed [`Filter`](crate::Filter) trait object. +/// +/// The filter inside is a dynamic trait object. The purpose of this type is +/// to ease returning `Filter`s from other functions. +/// +/// To create one, call `Filter::boxed` on any filter. +/// +/// # Examples +/// +/// ``` +/// use warp::{Filter, filters::BoxedFilter, Reply}; +/// +/// pub fn assets_filter() -> BoxedFilter<(impl Reply,)> { +/// warp::path("assets") +/// .and(warp::fs::dir("./assets")) +/// .boxed() +/// } +/// ``` +/// +pub struct BoxedFilter<T: Tuple> { + filter: Arc< + dyn Filter< + Extract = T, + Error = Rejection, + Future = Pin<Box<dyn Future<Output = Result<T, Rejection>> + Send>>, + > + Send + + Sync, + >, +} + +impl<T: Tuple + Send> BoxedFilter<T> { + pub(super) fn new<F>(filter: F) -> BoxedFilter<T> + where + F: Filter<Extract = T> + Send + Sync + 'static, + F::Error: Into<Rejection>, + { + BoxedFilter { + filter: Arc::new(BoxingFilter { + filter: filter.map_err(super::Internal, Into::into), + }), + } + } +} + +impl<T: Tuple> Clone for BoxedFilter<T> { + fn clone(&self) -> BoxedFilter<T> { + BoxedFilter { + filter: self.filter.clone(), + } + } +} + +impl<T: Tuple> fmt::Debug for BoxedFilter<T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BoxedFilter").finish() + } +} + +fn _assert_send() { + fn _assert<T: Send>() {} + _assert::<BoxedFilter<()>>(); +} + +impl<T: Tuple + Send> FilterBase for BoxedFilter<T> { + type Extract = T; + type Error = Rejection; + type Future = Pin<Box<dyn Future<Output = Result<T, Rejection>> + Send>>; + + fn filter(&self, _: Internal) -> Self::Future { + self.filter.filter(Internal) + } +} + +struct BoxingFilter<F> { + filter: F, +} + +impl<F> FilterBase for BoxingFilter<F> +where + F: Filter, + F::Future: Send + 'static, +{ + type Extract = F::Extract; + type Error = F::Error; + type Future = Pin<Box<dyn Future<Output = Result<Self::Extract, Self::Error>> + Send>>; + + fn filter(&self, _: Internal) -> Self::Future { + Box::pin(self.filter.filter(Internal).into_future()) + } +} diff --git a/third_party/rust/warp/src/filter/map.rs b/third_party/rust/warp/src/filter/map.rs new file mode 100644 index 0000000000..ec3173994e --- /dev/null +++ b/third_party/rust/warp/src/filter/map.rs @@ -0,0 +1,59 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Func, Internal}; + +#[derive(Clone, Copy, Debug)] +pub struct Map<T, F> { + pub(super) filter: T, + pub(super) callback: F, +} + +impl<T, F> FilterBase for Map<T, F> +where + T: Filter, + F: Func<T::Extract> + Clone + Send, +{ + type Extract = (F::Output,); + type Error = T::Error; + type Future = MapFuture<T, F>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + MapFuture { + extract: self.filter.filter(Internal), + callback: self.callback.clone(), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct MapFuture<T: Filter, F> { + #[pin] + extract: T::Future, + callback: F, +} + +impl<T, F> Future for MapFuture<T, F> +where + T: Filter, + F: Func<T::Extract>, +{ + type Output = Result<(F::Output,), T::Error>; + + #[inline] + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + let pin = self.project(); + match ready!(pin.extract.try_poll(cx)) { + Ok(ex) => { + let ex = (pin.callback.call(ex),); + Poll::Ready(Ok(ex)) + } + Err(err) => Poll::Ready(Err(err)), + } + } +} diff --git a/third_party/rust/warp/src/filter/map_err.rs b/third_party/rust/warp/src/filter/map_err.rs new file mode 100644 index 0000000000..f4659e6ff6 --- /dev/null +++ b/third_party/rust/warp/src/filter/map_err.rs @@ -0,0 +1,58 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::TryFuture; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Internal}; +use crate::reject::IsReject; + +#[derive(Clone, Copy, Debug)] +pub struct MapErr<T, F> { + pub(super) filter: T, + pub(super) callback: F, +} + +impl<T, F, E> FilterBase for MapErr<T, F> +where + T: Filter, + F: Fn(T::Error) -> E + Clone + Send, + E: IsReject, +{ + type Extract = T::Extract; + type Error = E; + type Future = MapErrFuture<T, F>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + MapErrFuture { + extract: self.filter.filter(Internal), + callback: self.callback.clone(), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct MapErrFuture<T: Filter, F> { + #[pin] + extract: T::Future, + callback: F, +} + +impl<T, F, E> Future for MapErrFuture<T, F> +where + T: Filter, + F: Fn(T::Error) -> E, +{ + type Output = Result<T::Extract, E>; + + #[inline] + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + self.as_mut() + .project() + .extract + .try_poll(cx) + .map_err(|err| (self.callback)(err)) + } +} diff --git a/third_party/rust/warp/src/filter/mod.rs b/third_party/rust/warp/src/filter/mod.rs new file mode 100644 index 0000000000..263650768d --- /dev/null +++ b/third_party/rust/warp/src/filter/mod.rs @@ -0,0 +1,492 @@ +mod and; +mod and_then; +mod boxed; +mod map; +mod map_err; +mod or; +mod or_else; +mod recover; +pub(crate) mod service; +mod then; +mod unify; +mod untuple_one; +mod wrap; + +use std::future::Future; + +use futures_util::{future, TryFuture, TryFutureExt}; + +pub(crate) use crate::generic::{one, Combine, Either, Func, One, Tuple}; +use crate::reject::{CombineRejection, IsReject, Rejection}; +use crate::route::{self, Route}; + +pub(crate) use self::and::And; +use self::and_then::AndThen; +pub use self::boxed::BoxedFilter; +pub(crate) use self::map::Map; +pub(crate) use self::map_err::MapErr; +pub(crate) use self::or::Or; +use self::or_else::OrElse; +use self::recover::Recover; +use self::then::Then; +use self::unify::Unify; +use self::untuple_one::UntupleOne; +pub use self::wrap::wrap_fn; +pub(crate) use self::wrap::{Wrap, WrapSealed}; + +// A crate-private base trait, allowing the actual `filter` method to change +// signatures without it being a breaking change. +pub trait FilterBase { + type Extract: Tuple; // + Send; + type Error: IsReject; + type Future: Future<Output = Result<Self::Extract, Self::Error>> + Send; + + fn filter(&self, internal: Internal) -> Self::Future; + + fn map_err<F, E>(self, _internal: Internal, fun: F) -> MapErr<Self, F> + where + Self: Sized, + F: Fn(Self::Error) -> E + Clone, + E: ::std::fmt::Debug + Send, + { + MapErr { + filter: self, + callback: fun, + } + } +} + +// A crate-private argument to prevent users from calling methods on +// the `FilterBase` trait. +// +// For instance, this innocent user code could otherwise call `filter`: +// +// ``` +// async fn with_filter<F: Filter>(f: F) -> Result<F::Extract, F::Error> { +// f.filter().await +// } +// ``` +#[allow(missing_debug_implementations)] +pub struct Internal; + +/// Composable request filters. +/// +/// A `Filter` can optionally extract some data from a request, combine +/// it with others, mutate it, and return back some value as a reply. The +/// power of `Filter`s come from being able to isolate small subsets, and then +/// chain and reuse them in various parts of your app. +/// +/// # Extracting Tuples +/// +/// You may notice that several of these filters extract some tuple, often +/// times a tuple of just 1 item! Why? +/// +/// If a filter extracts a `(String,)`, that simply means that it +/// extracts a `String`. If you were to `map` the filter, the argument type +/// would be exactly that, just a `String`. +/// +/// What is it? It's just some type magic that allows for automatic combining +/// and flattening of tuples. Without it, combining two filters together with +/// `and`, where one extracted `()`, and another `String`, would mean the +/// `map` would be given a single argument of `((), String,)`, which is just +/// no fun. +pub trait Filter: FilterBase { + /// Composes a new `Filter` that requires both this and the other to filter a request. + /// + /// Additionally, this will join together the extracted values of both + /// filters, so that `map` and `and_then` receive them as separate arguments. + /// + /// If a `Filter` extracts nothing (so, `()`), combining with any other + /// filter will simply discard the `()`. If a `Filter` extracts one or + /// more items, combining will mean it extracts the values of itself + /// combined with the other. + /// + /// # Example + /// + /// ``` + /// use warp::Filter; + /// + /// // Match `/hello/:name`... + /// warp::path("hello") + /// .and(warp::path::param::<String>()); + /// ``` + fn and<F>(self, other: F) -> And<Self, F> + where + Self: Sized, + <Self::Extract as Tuple>::HList: Combine<<F::Extract as Tuple>::HList>, + F: Filter + Clone, + F::Error: CombineRejection<Self::Error>, + { + And { + first: self, + second: other, + } + } + + /// Composes a new `Filter` of either this or the other filter. + /// + /// # Example + /// + /// ``` + /// use std::net::SocketAddr; + /// use warp::Filter; + /// + /// // Match either `/:u32` or `/:socketaddr` + /// warp::path::param::<u32>() + /// .or(warp::path::param::<SocketAddr>()); + /// ``` + fn or<F>(self, other: F) -> Or<Self, F> + where + Self: Filter<Error = Rejection> + Sized, + F: Filter, + F::Error: CombineRejection<Self::Error>, + { + Or { + first: self, + second: other, + } + } + + /// Composes this `Filter` with a function receiving the extracted value. + /// + /// + /// # Example + /// + /// ``` + /// use warp::Filter; + /// + /// // Map `/:id` + /// warp::path::param().map(|id: u64| { + /// format!("Hello #{}", id) + /// }); + /// ``` + /// + /// # `Func` + /// + /// The generic `Func` trait is implemented for any function that receives + /// the same arguments as this `Filter` extracts. In practice, this + /// shouldn't ever bother you, and simply makes things feel more natural. + /// + /// For example, if three `Filter`s were combined together, suppose one + /// extracts nothing (so `()`), and the other two extract two integers, + /// a function that accepts exactly two integer arguments is allowed. + /// Specifically, any `Fn(u32, u32)`. + /// + /// Without `Product` and `Func`, this would be a lot messier. First of + /// all, the `()`s couldn't be discarded, and the tuples would be nested. + /// So, instead, you'd need to pass an `Fn(((), (u32, u32)))`. That's just + /// a single argument. Bleck! + /// + /// Even worse, the tuples would shuffle the types around depending on + /// the exact invocation of `and`s. So, `unit.and(int).and(int)` would + /// result in a different extracted type from `unit.and(int.and(int))`, + /// or from `int.and(unit).and(int)`. If you changed around the order + /// of filters, while still having them be semantically equivalent, you'd + /// need to update all your `map`s as well. + /// + /// `Product`, `HList`, and `Func` do all the heavy work so that none of + /// this is a bother to you. What's more, the types are enforced at + /// compile-time, and tuple flattening is optimized away to nothing by + /// LLVM. + fn map<F>(self, fun: F) -> Map<Self, F> + where + Self: Sized, + F: Func<Self::Extract> + Clone, + { + Map { + filter: self, + callback: fun, + } + } + + /// Composes this `Filter` with an async function receiving + /// the extracted value. + /// + /// The function should return some `Future` type. + /// + /// # Example + /// + /// ``` + /// use warp::Filter; + /// + /// // Map `/:id` + /// warp::path::param().then(|id: u64| async move { + /// format!("Hello #{}", id) + /// }); + /// ``` + fn then<F>(self, fun: F) -> Then<Self, F> + where + Self: Sized, + F: Func<Self::Extract> + Clone, + F::Output: Future + Send, + { + Then { + filter: self, + callback: fun, + } + } + + /// Composes this `Filter` with a fallible async function receiving + /// the extracted value. + /// + /// The function should return some `TryFuture` type. + /// + /// The `Error` type of the return `Future` needs be a `Rejection`, which + /// means most futures will need to have their error mapped into one. + /// + /// Rejections are meant to say "this filter didn't accept the request, + /// maybe another can". So for application-level errors, consider using + /// [`Filter::then`] instead. + /// + /// # Example + /// + /// ``` + /// use warp::Filter; + /// + /// // Validate after `/:id` + /// warp::path::param().and_then(|id: u64| async move { + /// if id != 0 { + /// Ok(format!("Hello #{}", id)) + /// } else { + /// Err(warp::reject::not_found()) + /// } + /// }); + /// ``` + fn and_then<F>(self, fun: F) -> AndThen<Self, F> + where + Self: Sized, + F: Func<Self::Extract> + Clone, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: CombineRejection<Self::Error>, + { + AndThen { + filter: self, + callback: fun, + } + } + + /// Compose this `Filter` with a function receiving an error. + /// + /// The function should return some `TryFuture` type yielding the + /// same item and error types. + fn or_else<F>(self, fun: F) -> OrElse<Self, F> + where + Self: Filter<Error = Rejection> + Sized, + F: Func<Rejection>, + F::Output: TryFuture<Ok = Self::Extract> + Send, + <F::Output as TryFuture>::Error: IsReject, + { + OrElse { + filter: self, + callback: fun, + } + } + + /// Compose this `Filter` with a function receiving an error and + /// returning a *new* type, instead of the *same* type. + /// + /// This is useful for "customizing" rejections into new response types. + /// See also the [rejections example][ex]. + /// + /// [ex]: https://github.com/seanmonstar/warp/blob/master/examples/rejections.rs + fn recover<F>(self, fun: F) -> Recover<Self, F> + where + Self: Filter<Error = Rejection> + Sized, + F: Func<Rejection>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: IsReject, + { + Recover { + filter: self, + callback: fun, + } + } + + /// Unifies the extracted value of `Filter`s composed with `or`. + /// + /// When a `Filter` extracts some `Either<T, T>`, where both sides + /// are the same type, this combinator can be used to grab the + /// inner value, regardless of which side of `Either` it was. This + /// is useful for values that could be extracted from multiple parts + /// of a request, and the exact place isn't important. + /// + /// # Example + /// + /// ```rust + /// use std::net::SocketAddr; + /// use warp::Filter; + /// + /// let client_ip = warp::header("x-real-ip") + /// .or(warp::header("x-forwarded-for")) + /// .unify() + /// .map(|ip: SocketAddr| { + /// // Get the IP from either header, + /// // and unify into the inner type. + /// }); + /// ``` + fn unify<T>(self) -> Unify<Self> + where + Self: Filter<Extract = (Either<T, T>,)> + Sized, + T: Tuple, + { + Unify { filter: self } + } + + /// Convenience method to remove one layer of tupling. + /// + /// This is useful for when things like `map` don't return a new value, + /// but just `()`, since warp will wrap it up into a `((),)`. + /// + /// # Example + /// + /// ``` + /// use warp::Filter; + /// + /// let route = warp::path::param() + /// .map(|num: u64| { + /// println!("just logging: {}", num); + /// // returning "nothing" + /// }) + /// .untuple_one() + /// .map(|| { + /// println!("the ((),) was removed"); + /// warp::reply() + /// }); + /// ``` + /// + /// ``` + /// use warp::Filter; + /// + /// let route = warp::any() + /// .map(|| { + /// // wanting to return a tuple + /// (true, 33) + /// }) + /// .untuple_one() + /// .map(|is_enabled: bool, count: i32| { + /// println!("untupled: ({}, {})", is_enabled, count); + /// }); + /// ``` + fn untuple_one<T>(self) -> UntupleOne<Self> + where + Self: Filter<Extract = (T,)> + Sized, + T: Tuple, + { + UntupleOne { filter: self } + } + + /// Wraps the current filter with some wrapper. + /// + /// The wrapper may do some preparation work before starting this filter, + /// and may do post-processing after the filter completes. + /// + /// # Example + /// + /// ``` + /// use warp::Filter; + /// + /// let route = warp::any() + /// .map(warp::reply); + /// + /// // Wrap the route with a log wrapper. + /// let route = route.with(warp::log("example")); + /// ``` + fn with<W>(self, wrapper: W) -> W::Wrapped + where + Self: Sized, + W: Wrap<Self>, + { + wrapper.wrap(self) + } + + /// Boxes this filter into a trait object, making it easier to name the type. + /// + /// # Example + /// + /// ``` + /// use warp::Filter; + /// + /// fn impl_reply() -> warp::filters::BoxedFilter<(impl warp::Reply,)> { + /// warp::any() + /// .map(warp::reply) + /// .boxed() + /// } + /// + /// fn named_i32() -> warp::filters::BoxedFilter<(i32,)> { + /// warp::path::param::<i32>() + /// .boxed() + /// } + /// + /// fn named_and() -> warp::filters::BoxedFilter<(i32, String)> { + /// warp::path::param::<i32>() + /// .and(warp::header::<String>("host")) + /// .boxed() + /// } + /// ``` + fn boxed(self) -> BoxedFilter<Self::Extract> + where + Self: Sized + Send + Sync + 'static, + Self::Extract: Send, + Self::Error: Into<Rejection>, + { + BoxedFilter::new(self) + } +} + +impl<T: FilterBase> Filter for T {} + +pub trait FilterClone: Filter + Clone {} + +impl<T: Filter + Clone> FilterClone for T {} + +fn _assert_object_safe() { + fn _assert(_f: &dyn Filter<Extract = (), Error = (), Future = future::Ready<()>>) {} +} + +// ===== FilterFn ===== + +pub(crate) fn filter_fn<F, U>(func: F) -> FilterFn<F> +where + F: Fn(&mut Route) -> U, + U: TryFuture, + U::Ok: Tuple, + U::Error: IsReject, +{ + FilterFn { func } +} + +pub(crate) fn filter_fn_one<F, U>( + func: F, +) -> impl Filter<Extract = (U::Ok,), Error = U::Error> + Copy +where + F: Fn(&mut Route) -> U + Copy, + U: TryFuture + Send + 'static, + U::Ok: Send, + U::Error: IsReject, +{ + filter_fn(move |route| func(route).map_ok(|item| (item,))) +} + +#[derive(Copy, Clone)] +#[allow(missing_debug_implementations)] +pub(crate) struct FilterFn<F> { + // TODO: could include a `debug_str: &'static str` to be used in Debug impl + func: F, +} + +impl<F, U> FilterBase for FilterFn<F> +where + F: Fn(&mut Route) -> U, + U: TryFuture + Send + 'static, + U::Ok: Tuple + Send, + U::Error: IsReject, +{ + type Extract = U::Ok; + type Error = U::Error; + type Future = future::IntoFuture<U>; + + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + route::with(|route| (self.func)(route)).into_future() + } +} diff --git a/third_party/rust/warp/src/filter/or.rs b/third_party/rust/warp/src/filter/or.rs new file mode 100644 index 0000000000..774067fb59 --- /dev/null +++ b/third_party/rust/warp/src/filter/or.rs @@ -0,0 +1,110 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Internal}; +use crate::generic::Either; +use crate::reject::CombineRejection; +use crate::route; + +type Combined<E1, E2> = <E1 as CombineRejection<E2>>::Combined; + +#[derive(Clone, Copy, Debug)] +pub struct Or<T, U> { + pub(super) first: T, + pub(super) second: U, +} + +impl<T, U> FilterBase for Or<T, U> +where + T: Filter, + U: Filter + Clone + Send, + U::Error: CombineRejection<T::Error>, +{ + type Extract = (Either<T::Extract, U::Extract>,); + //type Error = <U::Error as CombineRejection<T::Error>>::Combined; + type Error = Combined<U::Error, T::Error>; + type Future = EitherFuture<T, U>; + + fn filter(&self, _: Internal) -> Self::Future { + let idx = route::with(|route| route.matched_path_index()); + EitherFuture { + state: State::First(self.first.filter(Internal), self.second.clone()), + original_path_index: PathIndex(idx), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct EitherFuture<T: Filter, U: Filter> { + #[pin] + state: State<T, U>, + original_path_index: PathIndex, +} + +#[pin_project(project = StateProj)] +enum State<T: Filter, U: Filter> { + First(#[pin] T::Future, U), + Second(Option<T::Error>, #[pin] U::Future), + Done, +} + +#[derive(Copy, Clone)] +struct PathIndex(usize); + +impl PathIndex { + fn reset_path(&self) { + route::with(|route| route.reset_matched_path_index(self.0)); + } +} + +impl<T, U> Future for EitherFuture<T, U> +where + T: Filter, + U: Filter, + U::Error: CombineRejection<T::Error>, +{ + type Output = Result<(Either<T::Extract, U::Extract>,), Combined<U::Error, T::Error>>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + loop { + let pin = self.as_mut().project(); + let (err1, fut2) = match pin.state.project() { + StateProj::First(first, second) => match ready!(first.try_poll(cx)) { + Ok(ex1) => { + return Poll::Ready(Ok((Either::A(ex1),))); + } + Err(e) => { + pin.original_path_index.reset_path(); + (e, second.filter(Internal)) + } + }, + StateProj::Second(err1, second) => { + let ex2 = match ready!(second.try_poll(cx)) { + Ok(ex2) => Ok((Either::B(ex2),)), + Err(e) => { + pin.original_path_index.reset_path(); + let err1 = err1.take().expect("polled after complete"); + Err(e.combine(err1)) + } + }; + self.set(EitherFuture { + state: State::Done, + ..*self + }); + return Poll::Ready(ex2); + } + StateProj::Done => panic!("polled after complete"), + }; + + self.set(EitherFuture { + state: State::Second(Some(err1), fut2), + ..*self + }); + } + } +} diff --git a/third_party/rust/warp/src/filter/or_else.rs b/third_party/rust/warp/src/filter/or_else.rs new file mode 100644 index 0000000000..aaf23243ec --- /dev/null +++ b/third_party/rust/warp/src/filter/or_else.rs @@ -0,0 +1,107 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Func, Internal}; +use crate::reject::IsReject; +use crate::route; + +#[derive(Clone, Copy, Debug)] +pub struct OrElse<T, F> { + pub(super) filter: T, + pub(super) callback: F, +} + +impl<T, F> FilterBase for OrElse<T, F> +where + T: Filter, + F: Func<T::Error> + Clone + Send, + F::Output: TryFuture<Ok = T::Extract> + Send, + <F::Output as TryFuture>::Error: IsReject, +{ + type Extract = <F::Output as TryFuture>::Ok; + type Error = <F::Output as TryFuture>::Error; + type Future = OrElseFuture<T, F>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + let idx = route::with(|route| route.matched_path_index()); + OrElseFuture { + state: State::First(self.filter.filter(Internal), self.callback.clone()), + original_path_index: PathIndex(idx), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct OrElseFuture<T: Filter, F> +where + T: Filter, + F: Func<T::Error>, + F::Output: TryFuture<Ok = T::Extract> + Send, +{ + #[pin] + state: State<T, F>, + original_path_index: PathIndex, +} + +#[pin_project(project = StateProj)] +enum State<T, F> +where + T: Filter, + F: Func<T::Error>, + F::Output: TryFuture<Ok = T::Extract> + Send, +{ + First(#[pin] T::Future, F), + Second(#[pin] F::Output), + Done, +} + +#[derive(Copy, Clone)] +struct PathIndex(usize); + +impl PathIndex { + fn reset_path(&self) { + route::with(|route| route.reset_matched_path_index(self.0)); + } +} + +impl<T, F> Future for OrElseFuture<T, F> +where + T: Filter, + F: Func<T::Error>, + F::Output: TryFuture<Ok = T::Extract> + Send, +{ + type Output = Result<<F::Output as TryFuture>::Ok, <F::Output as TryFuture>::Error>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + loop { + let pin = self.as_mut().project(); + let (err, second) = match pin.state.project() { + StateProj::First(first, second) => match ready!(first.try_poll(cx)) { + Ok(ex) => return Poll::Ready(Ok(ex)), + Err(err) => (err, second), + }, + StateProj::Second(second) => { + let ex2 = ready!(second.try_poll(cx)); + self.set(OrElseFuture { + state: State::Done, + ..*self + }); + return Poll::Ready(ex2); + } + StateProj::Done => panic!("polled after complete"), + }; + + pin.original_path_index.reset_path(); + let fut2 = second.call(err); + self.set(OrElseFuture { + state: State::Second(fut2), + ..*self + }); + } + } +} diff --git a/third_party/rust/warp/src/filter/recover.rs b/third_party/rust/warp/src/filter/recover.rs new file mode 100644 index 0000000000..100e9398c2 --- /dev/null +++ b/third_party/rust/warp/src/filter/recover.rs @@ -0,0 +1,117 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Func, Internal}; +use crate::generic::Either; +use crate::reject::IsReject; +use crate::route; + +#[derive(Clone, Copy, Debug)] +pub struct Recover<T, F> { + pub(super) filter: T, + pub(super) callback: F, +} + +impl<T, F> FilterBase for Recover<T, F> +where + T: Filter, + F: Func<T::Error> + Clone + Send, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: IsReject, +{ + type Extract = (Either<T::Extract, (<F::Output as TryFuture>::Ok,)>,); + type Error = <F::Output as TryFuture>::Error; + type Future = RecoverFuture<T, F>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + let idx = route::with(|route| route.matched_path_index()); + RecoverFuture { + state: State::First(self.filter.filter(Internal), self.callback.clone()), + original_path_index: PathIndex(idx), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct RecoverFuture<T: Filter, F> +where + T: Filter, + F: Func<T::Error>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: IsReject, +{ + #[pin] + state: State<T, F>, + original_path_index: PathIndex, +} + +#[pin_project(project = StateProj)] +enum State<T, F> +where + T: Filter, + F: Func<T::Error>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: IsReject, +{ + First(#[pin] T::Future, F), + Second(#[pin] F::Output), + Done, +} + +#[derive(Copy, Clone)] +struct PathIndex(usize); + +impl PathIndex { + fn reset_path(&self) { + route::with(|route| route.reset_matched_path_index(self.0)); + } +} + +impl<T, F> Future for RecoverFuture<T, F> +where + T: Filter, + F: Func<T::Error>, + F::Output: TryFuture + Send, + <F::Output as TryFuture>::Error: IsReject, +{ + type Output = Result< + (Either<T::Extract, (<F::Output as TryFuture>::Ok,)>,), + <F::Output as TryFuture>::Error, + >; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + loop { + let pin = self.as_mut().project(); + let (err, second) = match pin.state.project() { + StateProj::First(first, second) => match ready!(first.try_poll(cx)) { + Ok(ex) => return Poll::Ready(Ok((Either::A(ex),))), + Err(err) => (err, second), + }, + StateProj::Second(second) => { + let ex2 = match ready!(second.try_poll(cx)) { + Ok(ex2) => Ok((Either::B((ex2,)),)), + Err(e) => Err(e), + }; + self.set(RecoverFuture { + state: State::Done, + ..*self + }); + return Poll::Ready(ex2); + } + StateProj::Done => panic!("polled after complete"), + }; + + pin.original_path_index.reset_path(); + let fut2 = second.call(err); + self.set(RecoverFuture { + state: State::Second(fut2), + ..*self + }); + } + } +} diff --git a/third_party/rust/warp/src/filter/service.rs b/third_party/rust/warp/src/filter/service.rs new file mode 100644 index 0000000000..3de12a02ed --- /dev/null +++ b/third_party/rust/warp/src/filter/service.rs @@ -0,0 +1,137 @@ +use std::convert::Infallible; +use std::future::Future; +use std::net::SocketAddr; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::future::TryFuture; +use hyper::service::Service; +use pin_project::pin_project; + +use crate::reject::IsReject; +use crate::reply::{Reply, Response}; +use crate::route::{self, Route}; +use crate::{Filter, Request}; + +/// Convert a `Filter` into a `Service`. +/// +/// Filters are normally what APIs are built on in warp. However, it can be +/// useful to convert a `Filter` into a [`Service`][Service], such as if +/// further customizing a `hyper::Service`, or if wanting to make use of +/// the greater [Tower][tower] set of middleware. +/// +/// # Example +/// +/// Running a `warp::Filter` on a regular `hyper::Server`: +/// +/// ``` +/// # async fn run() -> Result<(), Box<dyn std::error::Error>> { +/// use std::convert::Infallible; +/// use warp::Filter; +/// +/// // Our Filter... +/// let route = warp::any().map(|| "Hello From Warp!"); +/// +/// // Convert it into a `Service`... +/// let svc = warp::service(route); +/// +/// // Typical hyper setup... +/// let make_svc = hyper::service::make_service_fn(move |_| async move { +/// Ok::<_, Infallible>(svc) +/// }); +/// +/// hyper::Server::bind(&([127, 0, 0, 1], 3030).into()) +/// .serve(make_svc) +/// .await?; +/// # Ok(()) +/// # } +/// ``` +/// +/// [Service]: https://docs.rs/hyper/0.13.*/hyper/service/trait.Service.html +/// [tower]: https://docs.rs/tower +pub fn service<F>(filter: F) -> FilteredService<F> +where + F: Filter, + <F::Future as TryFuture>::Ok: Reply, + <F::Future as TryFuture>::Error: IsReject, +{ + FilteredService { filter } +} + +#[derive(Copy, Clone, Debug)] +pub struct FilteredService<F> { + filter: F, +} + +impl<F> FilteredService<F> +where + F: Filter, + <F::Future as TryFuture>::Ok: Reply, + <F::Future as TryFuture>::Error: IsReject, +{ + #[inline] + pub(crate) fn call_with_addr( + &self, + req: Request, + remote_addr: Option<SocketAddr>, + ) -> FilteredFuture<F::Future> { + debug_assert!(!route::is_set(), "nested route::set calls"); + + let route = Route::new(req, remote_addr); + let fut = route::set(&route, || self.filter.filter(super::Internal)); + FilteredFuture { future: fut, route } + } +} + +impl<F> Service<Request> for FilteredService<F> +where + F: Filter, + <F::Future as TryFuture>::Ok: Reply, + <F::Future as TryFuture>::Error: IsReject, +{ + type Response = Response; + type Error = Infallible; + type Future = FilteredFuture<F::Future>; + + fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { + Poll::Ready(Ok(())) + } + + #[inline] + fn call(&mut self, req: Request) -> Self::Future { + self.call_with_addr(req, None) + } +} + +#[pin_project] +#[derive(Debug)] +pub struct FilteredFuture<F> { + #[pin] + future: F, + route: ::std::cell::RefCell<Route>, +} + +impl<F> Future for FilteredFuture<F> +where + F: TryFuture, + F::Ok: Reply, + F::Error: IsReject, +{ + type Output = Result<Response, Infallible>; + + #[inline] + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + debug_assert!(!route::is_set(), "nested route::set calls"); + + let pin = self.project(); + let fut = pin.future; + match route::set(pin.route, || fut.try_poll(cx)) { + Poll::Ready(Ok(ok)) => Poll::Ready(Ok(ok.into_response())), + Poll::Pending => Poll::Pending, + Poll::Ready(Err(err)) => { + tracing::debug!("rejected: {:?}", err); + Poll::Ready(Ok(err.into_response())) + } + } + } +} diff --git a/third_party/rust/warp/src/filter/then.rs b/third_party/rust/warp/src/filter/then.rs new file mode 100644 index 0000000000..543a22669a --- /dev/null +++ b/third_party/rust/warp/src/filter/then.rs @@ -0,0 +1,95 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Func, Internal}; + +#[derive(Clone, Copy, Debug)] +pub struct Then<T, F> { + pub(super) filter: T, + pub(super) callback: F, +} + +impl<T, F> FilterBase for Then<T, F> +where + T: Filter, + F: Func<T::Extract> + Clone + Send, + F::Output: Future + Send, +{ + type Extract = (<F::Output as Future>::Output,); + type Error = T::Error; + type Future = ThenFuture<T, F>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + ThenFuture { + state: State::First(self.filter.filter(Internal), self.callback.clone()), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct ThenFuture<T, F> +where + T: Filter, + F: Func<T::Extract>, + F::Output: Future + Send, +{ + #[pin] + state: State<T::Future, F>, +} + +#[pin_project(project = StateProj)] +enum State<T, F> +where + T: TryFuture, + F: Func<T::Ok>, + F::Output: Future + Send, +{ + First(#[pin] T, F), + Second(#[pin] F::Output), + Done, +} + +impl<T, F> Future for ThenFuture<T, F> +where + T: Filter, + F: Func<T::Extract>, + F::Output: Future + Send, +{ + type Output = Result<(<F::Output as Future>::Output,), T::Error>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + self.project().state.poll(cx) + } +} + +impl<T, F> Future for State<T, F> +where + T: TryFuture, + F: Func<T::Ok>, + F::Output: Future + Send, +{ + type Output = Result<(<F::Output as Future>::Output,), T::Error>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + loop { + match self.as_mut().project() { + StateProj::First(first, second) => { + let ex1 = ready!(first.try_poll(cx))?; + let fut2 = second.call(ex1); + self.set(State::Second(fut2)); + } + StateProj::Second(second) => { + let ex2 = (ready!(second.poll(cx)),); + self.set(State::Done); + return Poll::Ready(Ok(ex2)); + } + StateProj::Done => panic!("polled after complete"), + } + } + } +} diff --git a/third_party/rust/warp/src/filter/unify.rs b/third_party/rust/warp/src/filter/unify.rs new file mode 100644 index 0000000000..0cf670daa2 --- /dev/null +++ b/third_party/rust/warp/src/filter/unify.rs @@ -0,0 +1,50 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Either, Filter, FilterBase, Internal, Tuple}; + +#[derive(Clone, Copy, Debug)] +pub struct Unify<F> { + pub(super) filter: F, +} + +impl<F, T> FilterBase for Unify<F> +where + F: Filter<Extract = (Either<T, T>,)>, + T: Tuple, +{ + type Extract = T; + type Error = F::Error; + type Future = UnifyFuture<F::Future>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + UnifyFuture { + inner: self.filter.filter(Internal), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct UnifyFuture<F> { + #[pin] + inner: F, +} + +impl<F, T> Future for UnifyFuture<F> +where + F: TryFuture<Ok = (Either<T, T>,)>, +{ + type Output = Result<T, F::Error>; + + #[inline] + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + Poll::Ready(match ready!(self.project().inner.try_poll(cx))? { + (Either::A(x),) | (Either::B(x),) => Ok(x), + }) + } +} diff --git a/third_party/rust/warp/src/filter/untuple_one.rs b/third_party/rust/warp/src/filter/untuple_one.rs new file mode 100644 index 0000000000..0fb0de6748 --- /dev/null +++ b/third_party/rust/warp/src/filter/untuple_one.rs @@ -0,0 +1,52 @@ +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures_util::{ready, TryFuture}; +use pin_project::pin_project; + +use super::{Filter, FilterBase, Internal, Tuple}; + +#[derive(Clone, Copy, Debug)] +pub struct UntupleOne<F> { + pub(super) filter: F, +} + +impl<F, T> FilterBase for UntupleOne<F> +where + F: Filter<Extract = (T,)>, + T: Tuple, +{ + type Extract = T; + type Error = F::Error; + type Future = UntupleOneFuture<F>; + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + UntupleOneFuture { + extract: self.filter.filter(Internal), + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +pub struct UntupleOneFuture<F: Filter> { + #[pin] + extract: F::Future, +} + +impl<F, T> Future for UntupleOneFuture<F> +where + F: Filter<Extract = (T,)>, + T: Tuple, +{ + type Output = Result<T, F::Error>; + + #[inline] + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + match ready!(self.project().extract.try_poll(cx)) { + Ok((t,)) => Poll::Ready(Ok(t)), + Err(err) => Poll::Ready(Err(err)), + } + } +} diff --git a/third_party/rust/warp/src/filter/wrap.rs b/third_party/rust/warp/src/filter/wrap.rs new file mode 100644 index 0000000000..faceb705f3 --- /dev/null +++ b/third_party/rust/warp/src/filter/wrap.rs @@ -0,0 +1,67 @@ +use super::Filter; + +pub trait WrapSealed<F: Filter> { + type Wrapped: Filter; + + fn wrap(&self, filter: F) -> Self::Wrapped; +} + +impl<'a, T, F> WrapSealed<F> for &'a T +where + T: WrapSealed<F>, + F: Filter, +{ + type Wrapped = T::Wrapped; + fn wrap(&self, filter: F) -> Self::Wrapped { + (*self).wrap(filter) + } +} + +pub trait Wrap<F: Filter>: WrapSealed<F> {} + +impl<T, F> Wrap<F> for T +where + T: WrapSealed<F>, + F: Filter, +{ +} + +/// Combines received filter with pre and after filters +/// +/// # Example +/// +/// ``` +/// use crate::warp::Filter; +/// +/// let route = warp::any() +/// .map(|| "hello world") +/// .with(warp::wrap_fn(|filter| filter)); +/// ``` +/// +/// You can find the full example in the [usage example](https://github.com/seanmonstar/warp/blob/master/examples/wrapping.rs). +pub fn wrap_fn<F, T, U>(func: F) -> WrapFn<F> +where + F: Fn(T) -> U, + T: Filter, + U: Filter, +{ + WrapFn { func } +} + +#[derive(Debug)] +pub struct WrapFn<F> { + func: F, +} + +impl<F, T, U> WrapSealed<T> for WrapFn<F> +where + F: Fn(T) -> U, + T: Filter, + U: Filter, +{ + type Wrapped = U; + + fn wrap(&self, filter: T) -> Self::Wrapped { + (self.func)(filter) + } +} diff --git a/third_party/rust/warp/src/filters/addr.rs b/third_party/rust/warp/src/filters/addr.rs new file mode 100644 index 0000000000..3d630705a1 --- /dev/null +++ b/third_party/rust/warp/src/filters/addr.rs @@ -0,0 +1,26 @@ +//! Socket Address filters. + +use std::convert::Infallible; +use std::net::SocketAddr; + +use crate::filter::{filter_fn_one, Filter}; + +/// Creates a `Filter` to get the remote address of the connection. +/// +/// If the underlying transport doesn't use socket addresses, this will yield +/// `None`. +/// +/// # Example +/// +/// ``` +/// use std::net::SocketAddr; +/// use warp::Filter; +/// +/// let route = warp::addr::remote() +/// .map(|addr: Option<SocketAddr>| { +/// println!("remote address = {:?}", addr); +/// }); +/// ``` +pub fn remote() -> impl Filter<Extract = (Option<SocketAddr>,), Error = Infallible> + Copy { + filter_fn_one(|route| futures_util::future::ok(route.remote_addr())) +} diff --git a/third_party/rust/warp/src/filters/any.rs b/third_party/rust/warp/src/filters/any.rs new file mode 100644 index 0000000000..7328b8a64a --- /dev/null +++ b/third_party/rust/warp/src/filters/any.rs @@ -0,0 +1,76 @@ +//! A filter that matches any route. +use std::convert::Infallible; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use crate::filter::{Filter, FilterBase, Internal}; + +/// A [`Filter`](crate::Filter) that matches any route. +/// +/// This can be a useful building block to build new filters from, +/// since [`Filter`] is otherwise a sealed trait. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::any() +/// .map(|| { +/// "I always return this string!" +/// }); +/// ``` +/// +/// This could allow creating a single `impl Filter` returning a specific +/// reply, that can then be used as the end of several different filter +/// chains. +/// +/// Another use case is turning some clone-able resource into a `Filter`, +/// thus allowing to easily `and` it together with others. +/// +/// ``` +/// use std::sync::Arc; +/// use warp::Filter; +/// +/// let state = Arc::new(vec![33, 41]); +/// let with_state = warp::any().map(move || state.clone()); +/// +/// // Now we could `and` with any other filter: +/// +/// let route = warp::path::param() +/// .and(with_state) +/// .map(|param_id: u32, db: Arc<Vec<u32>>| { +/// db.contains(¶m_id) +/// }); +/// ``` +pub fn any() -> impl Filter<Extract = (), Error = Infallible> + Copy { + Any +} + +#[derive(Copy, Clone)] +#[allow(missing_debug_implementations)] +struct Any; + +impl FilterBase for Any { + type Extract = (); + type Error = Infallible; + type Future = AnyFut; + + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + AnyFut + } +} + +#[allow(missing_debug_implementations)] +struct AnyFut; + +impl Future for AnyFut { + type Output = Result<(), Infallible>; + + #[inline] + fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> { + Poll::Ready(Ok(())) + } +} diff --git a/third_party/rust/warp/src/filters/body.rs b/third_party/rust/warp/src/filters/body.rs new file mode 100644 index 0000000000..85dabbfeae --- /dev/null +++ b/third_party/rust/warp/src/filters/body.rs @@ -0,0 +1,347 @@ +//! Body filters +//! +//! Filters that extract a body for a route. + +use std::error::Error as StdError; +use std::fmt; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use bytes::{Buf, Bytes}; +use futures_util::{future, ready, Stream, TryFutureExt}; +use headers::ContentLength; +use http::header::CONTENT_TYPE; +use hyper::Body; +use mime; +use serde::de::DeserializeOwned; +use serde_json; +use serde_urlencoded; + +use crate::filter::{filter_fn, filter_fn_one, Filter, FilterBase}; +use crate::reject::{self, Rejection}; + +type BoxError = Box<dyn StdError + Send + Sync>; + +// Extracts the `Body` Stream from the route. +// +// Does not consume any of it. +pub(crate) fn body() -> impl Filter<Extract = (Body,), Error = Rejection> + Copy { + filter_fn_one(|route| { + future::ready(route.take_body().ok_or_else(|| { + tracing::error!("request body already taken in previous filter"); + reject::known(BodyConsumedMultipleTimes { _p: () }) + })) + }) +} + +/// Require a `content-length` header to have a value no greater than some limit. +/// +/// Rejects if `content-length` header is missing, is invalid, or has a number +/// larger than the limit provided. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // Limit the upload to 4kb... +/// let upload = warp::body::content_length_limit(4096) +/// .and(warp::body::aggregate()); +/// ``` +pub fn content_length_limit(limit: u64) -> impl Filter<Extract = (), Error = Rejection> + Copy { + crate::filters::header::header2() + .map_err(crate::filter::Internal, |_| { + tracing::debug!("content-length missing"); + reject::length_required() + }) + .and_then(move |ContentLength(length)| { + if length <= limit { + future::ok(()) + } else { + tracing::debug!("content-length: {} is over limit {}", length, limit); + future::err(reject::payload_too_large()) + } + }) + .untuple_one() +} + +/// Create a `Filter` that extracts the request body as a `futures::Stream`. +/// +/// If other filters have already extracted the body, this filter will reject +/// with a `500 Internal Server Error`. +/// +/// For example usage, please take a look at [examples/stream.rs](https://github.com/seanmonstar/warp/blob/master/examples/stream.rs). +/// +/// # Warning +/// +/// This does not have a default size limit, it would be wise to use one to +/// prevent a overly large request from using too much memory. +pub fn stream( +) -> impl Filter<Extract = (impl Stream<Item = Result<impl Buf, crate::Error>>,), Error = Rejection> + Copy +{ + body().map(|body: Body| BodyStream { body }) +} + +/// Returns a `Filter` that matches any request and extracts a `Future` of a +/// concatenated body. +/// +/// The contents of the body will be flattened into a single contiguous +/// `Bytes`, which may require memory copies. If you don't require a +/// contiguous buffer, using `aggregate` can be give better performance. +/// +/// # Warning +/// +/// This does not have a default size limit, it would be wise to use one to +/// prevent a overly large request from using too much memory. +/// +/// # Example +/// +/// ``` +/// use warp::{Buf, Filter}; +/// +/// let route = warp::body::content_length_limit(1024 * 32) +/// .and(warp::body::bytes()) +/// .map(|bytes: bytes::Bytes| { +/// println!("bytes = {:?}", bytes); +/// }); +/// ``` +pub fn bytes() -> impl Filter<Extract = (Bytes,), Error = Rejection> + Copy { + body().and_then(|body: hyper::Body| { + hyper::body::to_bytes(body).map_err(|err| { + tracing::debug!("to_bytes error: {}", err); + reject::known(BodyReadError(err)) + }) + }) +} + +/// Returns a `Filter` that matches any request and extracts a `Future` of an +/// aggregated body. +/// +/// The `Buf` may contain multiple, non-contiguous buffers. This can be more +/// performant (by reducing copies) when receiving large bodies. +/// +/// # Warning +/// +/// This does not have a default size limit, it would be wise to use one to +/// prevent a overly large request from using too much memory. +/// +/// # Example +/// +/// ``` +/// use warp::{Buf, Filter}; +/// +/// fn full_body(mut body: impl Buf) { +/// // It could have several non-contiguous slices of memory... +/// while body.has_remaining() { +/// println!("slice = {:?}", body.chunk()); +/// let cnt = body.chunk().len(); +/// body.advance(cnt); +/// } +/// } +/// +/// let route = warp::body::content_length_limit(1024 * 32) +/// .and(warp::body::aggregate()) +/// .map(full_body); +/// ``` +pub fn aggregate() -> impl Filter<Extract = (impl Buf,), Error = Rejection> + Copy { + body().and_then(|body: ::hyper::Body| { + hyper::body::aggregate(body).map_err(|err| { + tracing::debug!("aggregate error: {}", err); + reject::known(BodyReadError(err)) + }) + }) +} + +/// Returns a `Filter` that matches any request and extracts a `Future` of a +/// JSON-decoded body. +/// +/// # Warning +/// +/// This does not have a default size limit, it would be wise to use one to +/// prevent a overly large request from using too much memory. +/// +/// # Example +/// +/// ``` +/// use std::collections::HashMap; +/// use warp::Filter; +/// +/// let route = warp::body::content_length_limit(1024 * 32) +/// .and(warp::body::json()) +/// .map(|simple_map: HashMap<String, String>| { +/// "Got a JSON body!" +/// }); +/// ``` +pub fn json<T: DeserializeOwned + Send>() -> impl Filter<Extract = (T,), Error = Rejection> + Copy { + is_content_type::<Json>() + .and(bytes()) + .and_then(|buf| async move { + Json::decode(buf).map_err(|err| { + tracing::debug!("request json body error: {}", err); + reject::known(BodyDeserializeError { cause: err }) + }) + }) +} + +/// Returns a `Filter` that matches any request and extracts a +/// `Future` of a form encoded body. +/// +/// # Note +/// +/// This filter is for the simpler `application/x-www-form-urlencoded` format, +/// not `multipart/form-data`. +/// +/// # Warning +/// +/// This does not have a default size limit, it would be wise to use one to +/// prevent a overly large request from using too much memory. +/// +/// +/// ``` +/// use std::collections::HashMap; +/// use warp::Filter; +/// +/// let route = warp::body::content_length_limit(1024 * 32) +/// .and(warp::body::form()) +/// .map(|simple_map: HashMap<String, String>| { +/// "Got a urlencoded body!" +/// }); +/// ``` +pub fn form<T: DeserializeOwned + Send>() -> impl Filter<Extract = (T,), Error = Rejection> + Copy { + is_content_type::<Form>() + .and(aggregate()) + .and_then(|buf| async move { + Form::decode(buf).map_err(|err| { + tracing::debug!("request form body error: {}", err); + reject::known(BodyDeserializeError { cause: err }) + }) + }) +} + +// ===== Decoders ===== + +trait Decode { + const MIME: (mime::Name<'static>, mime::Name<'static>); + const WITH_NO_CONTENT_TYPE: bool; + + fn decode<B: Buf, T: DeserializeOwned>(buf: B) -> Result<T, BoxError>; +} + +struct Json; + +impl Decode for Json { + const MIME: (mime::Name<'static>, mime::Name<'static>) = (mime::APPLICATION, mime::JSON); + const WITH_NO_CONTENT_TYPE: bool = true; + + fn decode<B: Buf, T: DeserializeOwned>(mut buf: B) -> Result<T, BoxError> { + serde_json::from_slice(&buf.copy_to_bytes(buf.remaining())).map_err(Into::into) + } +} + +struct Form; + +impl Decode for Form { + const MIME: (mime::Name<'static>, mime::Name<'static>) = + (mime::APPLICATION, mime::WWW_FORM_URLENCODED); + const WITH_NO_CONTENT_TYPE: bool = true; + + fn decode<B: Buf, T: DeserializeOwned>(buf: B) -> Result<T, BoxError> { + serde_urlencoded::from_reader(buf.reader()).map_err(Into::into) + } +} + +// Require the `content-type` header to be this type (or, if there's no `content-type` +// header at all, optimistically hope it's the right type). +fn is_content_type<D: Decode>() -> impl Filter<Extract = (), Error = Rejection> + Copy { + filter_fn(move |route| { + let (type_, subtype) = D::MIME; + if let Some(value) = route.headers().get(CONTENT_TYPE) { + tracing::trace!("is_content_type {}/{}? {:?}", type_, subtype, value); + let ct = value + .to_str() + .ok() + .and_then(|s| s.parse::<mime::Mime>().ok()); + if let Some(ct) = ct { + if ct.type_() == type_ && ct.subtype() == subtype { + future::ok(()) + } else { + tracing::debug!( + "content-type {:?} doesn't match {}/{}", + value, + type_, + subtype + ); + future::err(reject::unsupported_media_type()) + } + } else { + tracing::debug!("content-type {:?} couldn't be parsed", value); + future::err(reject::unsupported_media_type()) + } + } else if D::WITH_NO_CONTENT_TYPE { + // Optimistically assume its correct! + tracing::trace!("no content-type header, assuming {}/{}", type_, subtype); + future::ok(()) + } else { + tracing::debug!("no content-type found"); + future::err(reject::unsupported_media_type()) + } + }) +} + +// ===== BodyStream ===== + +struct BodyStream { + body: Body, +} + +impl Stream for BodyStream { + type Item = Result<Bytes, crate::Error>; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + let opt_item = ready!(Pin::new(&mut self.get_mut().body).poll_next(cx)); + + match opt_item { + None => Poll::Ready(None), + Some(item) => { + let stream_buf = item.map_err(crate::Error::new); + + Poll::Ready(Some(stream_buf)) + } + } + } +} + +// ===== Rejections ===== + +/// An error used in rejections when deserializing a request body fails. +#[derive(Debug)] +pub struct BodyDeserializeError { + cause: BoxError, +} + +impl fmt::Display for BodyDeserializeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Request body deserialize error: {}", self.cause) + } +} + +impl StdError for BodyDeserializeError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(self.cause.as_ref()) + } +} + +#[derive(Debug)] +pub(crate) struct BodyReadError(::hyper::Error); + +impl fmt::Display for BodyReadError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Request body read error: {}", self.0) + } +} + +impl StdError for BodyReadError {} + +unit_error! { + pub(crate) BodyConsumedMultipleTimes: "Request body consumed multiple times" +} diff --git a/third_party/rust/warp/src/filters/compression.rs b/third_party/rust/warp/src/filters/compression.rs new file mode 100644 index 0000000000..244e768356 --- /dev/null +++ b/third_party/rust/warp/src/filters/compression.rs @@ -0,0 +1,292 @@ +//! Compression Filters +//! +//! Filters that compress the body of a response. + +#[cfg(feature = "compression-brotli")] +use async_compression::tokio::bufread::BrotliEncoder; + +#[cfg(feature = "compression-gzip")] +use async_compression::tokio::bufread::{DeflateEncoder, GzipEncoder}; + +use http::header::HeaderValue; +use hyper::{ + header::{CONTENT_ENCODING, CONTENT_LENGTH}, + Body, +}; +use tokio_util::io::{ReaderStream, StreamReader}; + +use crate::filter::{Filter, WrapSealed}; +use crate::reject::IsReject; +use crate::reply::{Reply, Response}; + +use self::internal::{CompressionProps, WithCompression}; + +enum CompressionAlgo { + #[cfg(feature = "compression-brotli")] + BR, + #[cfg(feature = "compression-gzip")] + DEFLATE, + #[cfg(feature = "compression-gzip")] + GZIP, +} + +impl From<CompressionAlgo> for HeaderValue { + #[inline] + fn from(algo: CompressionAlgo) -> Self { + HeaderValue::from_static(match algo { + #[cfg(feature = "compression-brotli")] + CompressionAlgo::BR => "br", + #[cfg(feature = "compression-gzip")] + CompressionAlgo::DEFLATE => "deflate", + #[cfg(feature = "compression-gzip")] + CompressionAlgo::GZIP => "gzip", + }) + } +} + +/// Compression +#[derive(Clone, Copy, Debug)] +pub struct Compression<F> { + func: F, +} + +// TODO: The implementation of `gzip()`, `deflate()`, and `brotli()` could be replaced with +// generics or a macro + +/// Create a wrapping filter that compresses the Body of a [`Response`](crate::reply::Response) +/// using gzip, adding `content-encoding: gzip` to the Response's [`HeaderMap`](hyper::HeaderMap) +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::get() +/// .and(warp::path::end()) +/// .and(warp::fs::file("./README.md")) +/// .with(warp::compression::gzip()); +/// ``` +#[cfg(feature = "compression-gzip")] +pub fn gzip() -> Compression<impl Fn(CompressionProps) -> Response + Copy> { + let func = move |mut props: CompressionProps| { + let body = Body::wrap_stream(ReaderStream::new(GzipEncoder::new(StreamReader::new( + props.body, + )))); + props + .head + .headers + .append(CONTENT_ENCODING, CompressionAlgo::GZIP.into()); + props.head.headers.remove(CONTENT_LENGTH); + Response::from_parts(props.head, body) + }; + Compression { func } +} + +/// Create a wrapping filter that compresses the Body of a [`Response`](crate::reply::Response) +/// using deflate, adding `content-encoding: deflate` to the Response's [`HeaderMap`](hyper::HeaderMap) +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::get() +/// .and(warp::path::end()) +/// .and(warp::fs::file("./README.md")) +/// .with(warp::compression::deflate()); +/// ``` +#[cfg(feature = "compression-gzip")] +pub fn deflate() -> Compression<impl Fn(CompressionProps) -> Response + Copy> { + let func = move |mut props: CompressionProps| { + let body = Body::wrap_stream(ReaderStream::new(DeflateEncoder::new(StreamReader::new( + props.body, + )))); + props + .head + .headers + .append(CONTENT_ENCODING, CompressionAlgo::DEFLATE.into()); + props.head.headers.remove(CONTENT_LENGTH); + Response::from_parts(props.head, body) + }; + Compression { func } +} + +/// Create a wrapping filter that compresses the Body of a [`Response`](crate::reply::Response) +/// using brotli, adding `content-encoding: br` to the Response's [`HeaderMap`](hyper::HeaderMap) +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::get() +/// .and(warp::path::end()) +/// .and(warp::fs::file("./README.md")) +/// .with(warp::compression::brotli()); +/// ``` +#[cfg(feature = "compression-brotli")] +pub fn brotli() -> Compression<impl Fn(CompressionProps) -> Response + Copy> { + let func = move |mut props: CompressionProps| { + let body = Body::wrap_stream(ReaderStream::new(BrotliEncoder::new(StreamReader::new( + props.body, + )))); + props + .head + .headers + .append(CONTENT_ENCODING, CompressionAlgo::BR.into()); + props.head.headers.remove(CONTENT_LENGTH); + Response::from_parts(props.head, body) + }; + Compression { func } +} + +impl<FN, F> WrapSealed<F> for Compression<FN> +where + FN: Fn(CompressionProps) -> Response + Clone + Send, + F: Filter + Clone + Send, + F::Extract: Reply, + F::Error: IsReject, +{ + type Wrapped = WithCompression<FN, F>; + + fn wrap(&self, filter: F) -> Self::Wrapped { + WithCompression { + filter, + compress: self.clone(), + } + } +} + +mod internal { + use std::future::Future; + use std::pin::Pin; + use std::task::{Context, Poll}; + + use bytes::Bytes; + use futures_util::{ready, Stream, TryFuture}; + use hyper::Body; + use pin_project::pin_project; + + use crate::filter::{Filter, FilterBase, Internal}; + use crate::reject::IsReject; + use crate::reply::{Reply, Response}; + + use super::Compression; + + /// A wrapper around any type that implements [`Stream`](futures::Stream) to be + /// compatible with async_compression's Stream based encoders + #[pin_project] + #[derive(Debug)] + pub struct CompressableBody<S, E> + where + E: std::error::Error, + S: Stream<Item = Result<Bytes, E>>, + { + #[pin] + body: S, + } + + impl<S, E> Stream for CompressableBody<S, E> + where + E: std::error::Error, + S: Stream<Item = Result<Bytes, E>>, + { + type Item = std::io::Result<Bytes>; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + use std::io::{Error, ErrorKind}; + + let pin = self.project(); + S::poll_next(pin.body, cx).map_err(|_| Error::from(ErrorKind::InvalidData)) + } + } + + impl From<Body> for CompressableBody<Body, hyper::Error> { + fn from(body: Body) -> Self { + CompressableBody { body } + } + } + + /// Compression Props + #[derive(Debug)] + pub struct CompressionProps { + pub(super) body: CompressableBody<Body, hyper::Error>, + pub(super) head: http::response::Parts, + } + + impl From<http::Response<Body>> for CompressionProps { + fn from(resp: http::Response<Body>) -> Self { + let (head, body) = resp.into_parts(); + CompressionProps { + body: body.into(), + head, + } + } + } + + #[allow(missing_debug_implementations)] + pub struct Compressed(pub(super) Response); + + impl Reply for Compressed { + #[inline] + fn into_response(self) -> Response { + self.0 + } + } + + #[allow(missing_debug_implementations)] + #[derive(Clone, Copy)] + pub struct WithCompression<FN, F> { + pub(super) compress: Compression<FN>, + pub(super) filter: F, + } + + impl<FN, F> FilterBase for WithCompression<FN, F> + where + FN: Fn(CompressionProps) -> Response + Clone + Send, + F: Filter + Clone + Send, + F::Extract: Reply, + F::Error: IsReject, + { + type Extract = (Compressed,); + type Error = F::Error; + type Future = WithCompressionFuture<FN, F::Future>; + + fn filter(&self, _: Internal) -> Self::Future { + WithCompressionFuture { + compress: self.compress.clone(), + future: self.filter.filter(Internal), + } + } + } + + #[allow(missing_debug_implementations)] + #[pin_project] + pub struct WithCompressionFuture<FN, F> { + compress: Compression<FN>, + #[pin] + future: F, + } + + impl<FN, F> Future for WithCompressionFuture<FN, F> + where + FN: Fn(CompressionProps) -> Response, + F: TryFuture, + F::Ok: Reply, + F::Error: IsReject, + { + type Output = Result<(Compressed,), F::Error>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + let pin = self.as_mut().project(); + let result = ready!(pin.future.try_poll(cx)); + match result { + Ok(reply) => { + let resp = (self.compress.func)(reply.into_response().into()); + Poll::Ready(Ok((Compressed(resp),))) + } + Err(reject) => Poll::Ready(Err(reject)), + } + } + } +} diff --git a/third_party/rust/warp/src/filters/cookie.rs b/third_party/rust/warp/src/filters/cookie.rs new file mode 100644 index 0000000000..53695e85c8 --- /dev/null +++ b/third_party/rust/warp/src/filters/cookie.rs @@ -0,0 +1,46 @@ +//! Cookie Filters + +use futures_util::future; +use headers::Cookie; + +use super::header; +use crate::filter::{Filter, One}; +use crate::reject::Rejection; +use std::convert::Infallible; +use std::str::FromStr; + +/// Creates a `Filter` that requires a cookie by name. +/// +/// If found, extracts the value of the cookie, otherwise rejects. +pub fn cookie<T>(name: &'static str) -> impl Filter<Extract = One<T>, Error = Rejection> + Copy +where + T: FromStr + Send + 'static, +{ + header::header2().and_then(move |cookie: Cookie| { + let cookie = cookie + .get(name) + .ok_or_else(|| crate::reject::missing_cookie(name)) + .and_then(|s| T::from_str(s).map_err(|_| crate::reject::missing_cookie(name))); + future::ready(cookie) + }) +} + +/// Creates a `Filter` that looks for an optional cookie by name. +/// +/// If found, extracts the value of the cookie, otherwise continues +/// the request, extracting `None`. +pub fn optional<T>( + name: &'static str, +) -> impl Filter<Extract = One<Option<T>>, Error = Infallible> + Copy +where + T: FromStr + Send + 'static, +{ + header::optional2().map(move |opt: Option<Cookie>| { + let cookie = opt.and_then(|cookie| cookie.get(name).map(|x| T::from_str(x))); + match cookie { + Some(Ok(t)) => Some(t), + Some(Err(_)) => None, + None => None, + } + }) +} diff --git a/third_party/rust/warp/src/filters/cors.rs b/third_party/rust/warp/src/filters/cors.rs new file mode 100644 index 0000000000..39ea240e32 --- /dev/null +++ b/third_party/rust/warp/src/filters/cors.rs @@ -0,0 +1,626 @@ +//! CORS Filters + +use std::collections::HashSet; +use std::convert::TryFrom; +use std::error::Error as StdError; +use std::fmt; +use std::sync::Arc; + +use headers::{ + AccessControlAllowHeaders, AccessControlAllowMethods, AccessControlExposeHeaders, HeaderMapExt, +}; +use http::{ + self, + header::{self, HeaderName, HeaderValue}, +}; + +use crate::filter::{Filter, WrapSealed}; +use crate::reject::{CombineRejection, Rejection}; +use crate::reply::Reply; + +use self::internal::{CorsFilter, IntoOrigin, Seconds}; + +/// Create a wrapping [`Filter`](crate::Filter) that exposes [CORS][] behavior for a wrapped +/// filter. +/// +/// [CORS]: https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let cors = warp::cors() +/// .allow_origin("https://hyper.rs") +/// .allow_methods(vec!["GET", "POST", "DELETE"]); +/// +/// let route = warp::any() +/// .map(warp::reply) +/// .with(cors); +/// ``` +/// If you want to allow any route: +/// ``` +/// use warp::Filter; +/// let cors = warp::cors() +/// .allow_any_origin(); +/// ``` +/// You can find more usage examples [here](https://github.com/seanmonstar/warp/blob/7fa54eaecd0fe12687137372791ff22fc7995766/tests/cors.rs). +pub fn cors() -> Builder { + Builder { + credentials: false, + allowed_headers: HashSet::new(), + exposed_headers: HashSet::new(), + max_age: None, + methods: HashSet::new(), + origins: None, + } +} + +/// A wrapping [`Filter`](crate::Filter) constructed via `warp::cors()`. +#[derive(Clone, Debug)] +pub struct Cors { + config: Arc<Configured>, +} + +/// A constructed via `warp::cors()`. +#[derive(Clone, Debug)] +pub struct Builder { + credentials: bool, + allowed_headers: HashSet<HeaderName>, + exposed_headers: HashSet<HeaderName>, + max_age: Option<u64>, + methods: HashSet<http::Method>, + origins: Option<HashSet<HeaderValue>>, +} + +impl Builder { + /// Sets whether to add the `Access-Control-Allow-Credentials` header. + pub fn allow_credentials(mut self, allow: bool) -> Self { + self.credentials = allow; + self + } + + /// Adds a method to the existing list of allowed request methods. + /// + /// # Panics + /// + /// Panics if the provided argument is not a valid `http::Method`. + pub fn allow_method<M>(mut self, method: M) -> Self + where + http::Method: TryFrom<M>, + { + let method = match TryFrom::try_from(method) { + Ok(m) => m, + Err(_) => panic!("illegal Method"), + }; + self.methods.insert(method); + self + } + + /// Adds multiple methods to the existing list of allowed request methods. + /// + /// # Panics + /// + /// Panics if the provided argument is not a valid `http::Method`. + pub fn allow_methods<I>(mut self, methods: I) -> Self + where + I: IntoIterator, + http::Method: TryFrom<I::Item>, + { + let iter = methods.into_iter().map(|m| match TryFrom::try_from(m) { + Ok(m) => m, + Err(_) => panic!("illegal Method"), + }); + self.methods.extend(iter); + self + } + + /// Adds a header to the list of allowed request headers. + /// + /// **Note**: These should match the values the browser sends via `Access-Control-Request-Headers`, e.g. `content-type`. + /// + /// # Panics + /// + /// Panics if the provided argument is not a valid `http::header::HeaderName`. + pub fn allow_header<H>(mut self, header: H) -> Self + where + HeaderName: TryFrom<H>, + { + let header = match TryFrom::try_from(header) { + Ok(m) => m, + Err(_) => panic!("illegal Header"), + }; + self.allowed_headers.insert(header); + self + } + + /// Adds multiple headers to the list of allowed request headers. + /// + /// **Note**: These should match the values the browser sends via `Access-Control-Request-Headers`, e.g.`content-type`. + /// + /// # Panics + /// + /// Panics if any of the headers are not a valid `http::header::HeaderName`. + pub fn allow_headers<I>(mut self, headers: I) -> Self + where + I: IntoIterator, + HeaderName: TryFrom<I::Item>, + { + let iter = headers.into_iter().map(|h| match TryFrom::try_from(h) { + Ok(h) => h, + Err(_) => panic!("illegal Header"), + }); + self.allowed_headers.extend(iter); + self + } + + /// Adds a header to the list of exposed headers. + /// + /// # Panics + /// + /// Panics if the provided argument is not a valid `http::header::HeaderName`. + pub fn expose_header<H>(mut self, header: H) -> Self + where + HeaderName: TryFrom<H>, + { + let header = match TryFrom::try_from(header) { + Ok(m) => m, + Err(_) => panic!("illegal Header"), + }; + self.exposed_headers.insert(header); + self + } + + /// Adds multiple headers to the list of exposed headers. + /// + /// # Panics + /// + /// Panics if any of the headers are not a valid `http::header::HeaderName`. + pub fn expose_headers<I>(mut self, headers: I) -> Self + where + I: IntoIterator, + HeaderName: TryFrom<I::Item>, + { + let iter = headers.into_iter().map(|h| match TryFrom::try_from(h) { + Ok(h) => h, + Err(_) => panic!("illegal Header"), + }); + self.exposed_headers.extend(iter); + self + } + + /// Sets that *any* `Origin` header is allowed. + /// + /// # Warning + /// + /// This can allow websites you didn't intend to access this resource, + /// it is usually better to set an explicit list. + pub fn allow_any_origin(mut self) -> Self { + self.origins = None; + self + } + + /// Add an origin to the existing list of allowed `Origin`s. + /// + /// # Panics + /// + /// Panics if the provided argument is not a valid `Origin`. + pub fn allow_origin(self, origin: impl IntoOrigin) -> Self { + self.allow_origins(Some(origin)) + } + + /// Add multiple origins to the existing list of allowed `Origin`s. + /// + /// # Panics + /// + /// Panics if the provided argument is not a valid `Origin`. + pub fn allow_origins<I>(mut self, origins: I) -> Self + where + I: IntoIterator, + I::Item: IntoOrigin, + { + let iter = origins + .into_iter() + .map(IntoOrigin::into_origin) + .map(|origin| { + origin + .to_string() + .parse() + .expect("Origin is always a valid HeaderValue") + }); + + self.origins.get_or_insert_with(HashSet::new).extend(iter); + + self + } + + /// Sets the `Access-Control-Max-Age` header. + /// + /// # Example + /// + /// + /// ``` + /// use std::time::Duration; + /// use warp::Filter; + /// + /// let cors = warp::cors() + /// .max_age(30) // 30u32 seconds + /// .max_age(Duration::from_secs(30)); // or a Duration + /// ``` + pub fn max_age(mut self, seconds: impl Seconds) -> Self { + self.max_age = Some(seconds.seconds()); + self + } + + /// Builds the `Cors` wrapper from the configured settings. + /// + /// This step isn't *required*, as the `Builder` itself can be passed + /// to `Filter::with`. This just allows constructing once, thus not needing + /// to pay the cost of "building" every time. + pub fn build(self) -> Cors { + let expose_headers_header = if self.exposed_headers.is_empty() { + None + } else { + Some(self.exposed_headers.iter().cloned().collect()) + }; + let allowed_headers_header = self.allowed_headers.iter().cloned().collect(); + let methods_header = self.methods.iter().cloned().collect(); + + let config = Arc::new(Configured { + cors: self, + allowed_headers_header, + expose_headers_header, + methods_header, + }); + + Cors { config } + } +} + +impl<F> WrapSealed<F> for Builder +where + F: Filter + Clone + Send + Sync + 'static, + F::Extract: Reply, + F::Error: CombineRejection<Rejection>, + <F::Error as CombineRejection<Rejection>>::One: CombineRejection<Rejection>, +{ + type Wrapped = CorsFilter<F>; + + fn wrap(&self, inner: F) -> Self::Wrapped { + let Cors { config } = self.clone().build(); + + CorsFilter { config, inner } + } +} + +impl<F> WrapSealed<F> for Cors +where + F: Filter + Clone + Send + Sync + 'static, + F::Extract: Reply, + F::Error: CombineRejection<Rejection>, + <F::Error as CombineRejection<Rejection>>::One: CombineRejection<Rejection>, +{ + type Wrapped = CorsFilter<F>; + + fn wrap(&self, inner: F) -> Self::Wrapped { + let config = self.config.clone(); + + CorsFilter { config, inner } + } +} + +/// An error used to reject requests that are forbidden by a `cors` filter. +pub struct CorsForbidden { + kind: Forbidden, +} + +#[derive(Debug)] +enum Forbidden { + OriginNotAllowed, + MethodNotAllowed, + HeaderNotAllowed, +} + +impl fmt::Debug for CorsForbidden { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_tuple("CorsForbidden").field(&self.kind).finish() + } +} + +impl fmt::Display for CorsForbidden { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let detail = match self.kind { + Forbidden::OriginNotAllowed => "origin not allowed", + Forbidden::MethodNotAllowed => "request-method not allowed", + Forbidden::HeaderNotAllowed => "header not allowed", + }; + write!(f, "CORS request forbidden: {}", detail) + } +} + +impl StdError for CorsForbidden {} + +#[derive(Clone, Debug)] +struct Configured { + cors: Builder, + allowed_headers_header: AccessControlAllowHeaders, + expose_headers_header: Option<AccessControlExposeHeaders>, + methods_header: AccessControlAllowMethods, +} + +enum Validated { + Preflight(HeaderValue), + Simple(HeaderValue), + NotCors, +} + +impl Configured { + fn check_request( + &self, + method: &http::Method, + headers: &http::HeaderMap, + ) -> Result<Validated, Forbidden> { + match (headers.get(header::ORIGIN), method) { + (Some(origin), &http::Method::OPTIONS) => { + // OPTIONS requests are preflight CORS requests... + + if !self.is_origin_allowed(origin) { + return Err(Forbidden::OriginNotAllowed); + } + + if let Some(req_method) = headers.get(header::ACCESS_CONTROL_REQUEST_METHOD) { + if !self.is_method_allowed(req_method) { + return Err(Forbidden::MethodNotAllowed); + } + } else { + tracing::trace!( + "preflight request missing access-control-request-method header" + ); + return Err(Forbidden::MethodNotAllowed); + } + + if let Some(req_headers) = headers.get(header::ACCESS_CONTROL_REQUEST_HEADERS) { + let headers = req_headers + .to_str() + .map_err(|_| Forbidden::HeaderNotAllowed)?; + for header in headers.split(',') { + if !self.is_header_allowed(header.trim()) { + return Err(Forbidden::HeaderNotAllowed); + } + } + } + + Ok(Validated::Preflight(origin.clone())) + } + (Some(origin), _) => { + // Any other method, simply check for a valid origin... + + tracing::trace!("origin header: {:?}", origin); + if self.is_origin_allowed(origin) { + Ok(Validated::Simple(origin.clone())) + } else { + Err(Forbidden::OriginNotAllowed) + } + } + (None, _) => { + // No `ORIGIN` header means this isn't CORS! + Ok(Validated::NotCors) + } + } + } + + fn is_method_allowed(&self, header: &HeaderValue) -> bool { + http::Method::from_bytes(header.as_bytes()) + .map(|method| self.cors.methods.contains(&method)) + .unwrap_or(false) + } + + fn is_header_allowed(&self, header: &str) -> bool { + HeaderName::from_bytes(header.as_bytes()) + .map(|header| self.cors.allowed_headers.contains(&header)) + .unwrap_or(false) + } + + fn is_origin_allowed(&self, origin: &HeaderValue) -> bool { + if let Some(ref allowed) = self.cors.origins { + allowed.contains(origin) + } else { + true + } + } + + fn append_preflight_headers(&self, headers: &mut http::HeaderMap) { + self.append_common_headers(headers); + + headers.typed_insert(self.allowed_headers_header.clone()); + headers.typed_insert(self.methods_header.clone()); + + if let Some(max_age) = self.cors.max_age { + headers.insert(header::ACCESS_CONTROL_MAX_AGE, max_age.into()); + } + } + + fn append_common_headers(&self, headers: &mut http::HeaderMap) { + if self.cors.credentials { + headers.insert( + header::ACCESS_CONTROL_ALLOW_CREDENTIALS, + HeaderValue::from_static("true"), + ); + } + if let Some(expose_headers_header) = &self.expose_headers_header { + headers.typed_insert(expose_headers_header.clone()) + } + } +} + +mod internal { + use std::future::Future; + use std::pin::Pin; + use std::sync::Arc; + use std::task::{Context, Poll}; + + use futures_util::{future, ready, TryFuture}; + use headers::Origin; + use http::header; + use pin_project::pin_project; + + use super::{Configured, CorsForbidden, Validated}; + use crate::filter::{Filter, FilterBase, Internal, One}; + use crate::generic::Either; + use crate::reject::{CombineRejection, Rejection}; + use crate::route; + + #[derive(Clone, Debug)] + pub struct CorsFilter<F> { + pub(super) config: Arc<Configured>, + pub(super) inner: F, + } + + impl<F> FilterBase for CorsFilter<F> + where + F: Filter, + F::Extract: Send, + F::Future: Future, + F::Error: CombineRejection<Rejection>, + { + type Extract = + One<Either<One<Preflight>, One<Either<One<Wrapped<F::Extract>>, F::Extract>>>>; + type Error = <F::Error as CombineRejection<Rejection>>::One; + type Future = future::Either< + future::Ready<Result<Self::Extract, Self::Error>>, + WrappedFuture<F::Future>, + >; + + fn filter(&self, _: Internal) -> Self::Future { + let validated = + route::with(|route| self.config.check_request(route.method(), route.headers())); + + match validated { + Ok(Validated::Preflight(origin)) => { + let preflight = Preflight { + config: self.config.clone(), + origin, + }; + future::Either::Left(future::ok((Either::A((preflight,)),))) + } + Ok(Validated::Simple(origin)) => future::Either::Right(WrappedFuture { + inner: self.inner.filter(Internal), + wrapped: Some((self.config.clone(), origin)), + }), + Ok(Validated::NotCors) => future::Either::Right(WrappedFuture { + inner: self.inner.filter(Internal), + wrapped: None, + }), + Err(err) => { + let rejection = crate::reject::known(CorsForbidden { kind: err }); + future::Either::Left(future::err(rejection.into())) + } + } + } + } + + #[derive(Debug)] + pub struct Preflight { + config: Arc<Configured>, + origin: header::HeaderValue, + } + + impl crate::reply::Reply for Preflight { + fn into_response(self) -> crate::reply::Response { + let mut res = crate::reply::Response::default(); + self.config.append_preflight_headers(res.headers_mut()); + res.headers_mut() + .insert(header::ACCESS_CONTROL_ALLOW_ORIGIN, self.origin); + res + } + } + + #[derive(Debug)] + pub struct Wrapped<R> { + config: Arc<Configured>, + inner: R, + origin: header::HeaderValue, + } + + impl<R> crate::reply::Reply for Wrapped<R> + where + R: crate::reply::Reply, + { + fn into_response(self) -> crate::reply::Response { + let mut res = self.inner.into_response(); + self.config.append_common_headers(res.headers_mut()); + res.headers_mut() + .insert(header::ACCESS_CONTROL_ALLOW_ORIGIN, self.origin); + res + } + } + + #[pin_project] + #[derive(Debug)] + pub struct WrappedFuture<F> { + #[pin] + inner: F, + wrapped: Option<(Arc<Configured>, header::HeaderValue)>, + } + + impl<F> Future for WrappedFuture<F> + where + F: TryFuture, + F::Error: CombineRejection<Rejection>, + { + type Output = Result< + One<Either<One<Preflight>, One<Either<One<Wrapped<F::Ok>>, F::Ok>>>>, + <F::Error as CombineRejection<Rejection>>::One, + >; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + let pin = self.project(); + match ready!(pin.inner.try_poll(cx)) { + Ok(inner) => { + let item = if let Some((config, origin)) = pin.wrapped.take() { + (Either::A((Wrapped { + config, + inner, + origin, + },)),) + } else { + (Either::B(inner),) + }; + let item = (Either::B(item),); + Poll::Ready(Ok(item)) + } + Err(err) => Poll::Ready(Err(err.into())), + } + } + } + + pub trait Seconds { + fn seconds(self) -> u64; + } + + impl Seconds for u32 { + fn seconds(self) -> u64 { + self.into() + } + } + + impl Seconds for ::std::time::Duration { + fn seconds(self) -> u64 { + self.as_secs() + } + } + + pub trait IntoOrigin { + fn into_origin(self) -> Origin; + } + + impl<'a> IntoOrigin for &'a str { + fn into_origin(self) -> Origin { + let mut parts = self.splitn(2, "://"); + let scheme = parts.next().expect("missing scheme"); + let rest = parts.next().expect("missing scheme"); + + Origin::try_from_parts(scheme, rest, None).expect("invalid Origin") + } + } +} diff --git a/third_party/rust/warp/src/filters/ext.rs b/third_party/rust/warp/src/filters/ext.rs new file mode 100644 index 0000000000..985bbfb61c --- /dev/null +++ b/third_party/rust/warp/src/filters/ext.rs @@ -0,0 +1,36 @@ +//! Request Extensions + +use std::convert::Infallible; + +use futures_util::future; + +use crate::filter::{filter_fn_one, Filter}; +use crate::reject::{self, Rejection}; + +/// Get a previously set extension of the current route. +/// +/// If the extension doesn't exist, this rejects with a `MissingExtension`. +pub fn get<T: Clone + Send + Sync + 'static>( +) -> impl Filter<Extract = (T,), Error = Rejection> + Copy { + filter_fn_one(|route| { + let route = route + .extensions() + .get::<T>() + .cloned() + .ok_or_else(|| reject::known(MissingExtension { _p: () })); + future::ready(route) + }) +} + +/// Get a previously set extension of the current route. +/// +/// If the extension doesn't exist, it yields `None`. +pub fn optional<T: Clone + Send + Sync + 'static>( +) -> impl Filter<Extract = (Option<T>,), Error = Infallible> + Copy { + filter_fn_one(|route| future::ok(route.extensions().get::<T>().cloned())) +} + +unit_error! { + /// An error used to reject if `get` cannot find the extension. + pub MissingExtension: "Missing request extension" +} diff --git a/third_party/rust/warp/src/filters/fs.rs b/third_party/rust/warp/src/filters/fs.rs new file mode 100644 index 0000000000..fdfa709687 --- /dev/null +++ b/third_party/rust/warp/src/filters/fs.rs @@ -0,0 +1,539 @@ +//! File System Filters + +use std::cmp; +use std::convert::Infallible; +use std::fs::Metadata; +use std::future::Future; +use std::io; +use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::sync::Arc; +use std::task::Poll; + +use bytes::{Bytes, BytesMut}; +use futures_util::future::Either; +use futures_util::{future, ready, stream, FutureExt, Stream, StreamExt, TryFutureExt}; +use headers::{ + AcceptRanges, ContentLength, ContentRange, ContentType, HeaderMapExt, IfModifiedSince, IfRange, + IfUnmodifiedSince, LastModified, Range, +}; +use http::StatusCode; +use hyper::Body; +use mime_guess; +use percent_encoding::percent_decode_str; +use tokio::fs::File as TkFile; +use tokio::io::AsyncSeekExt; +use tokio_util::io::poll_read_buf; + +use crate::filter::{Filter, FilterClone, One}; +use crate::reject::{self, Rejection}; +use crate::reply::{Reply, Response}; + +/// Creates a `Filter` that serves a File at the `path`. +/// +/// Does not filter out based on any information of the request. Always serves +/// the file at the exact `path` provided. Thus, this can be used to serve a +/// single file with `GET`s, but could also be used in combination with other +/// filters, such as after validating in `POST` request, wanting to return a +/// specific file as the body. +/// +/// For serving a directory, see [dir]. +/// +/// # Example +/// +/// ``` +/// // Always serves this file from the file system. +/// let route = warp::fs::file("/www/static/app.js"); +/// ``` +pub fn file(path: impl Into<PathBuf>) -> impl FilterClone<Extract = One<File>, Error = Rejection> { + let path = Arc::new(path.into()); + crate::any() + .map(move || { + tracing::trace!("file: {:?}", path); + ArcPath(path.clone()) + }) + .and(conditionals()) + .and_then(file_reply) +} + +/// Creates a `Filter` that serves a directory at the base `path` joined +/// by the request path. +/// +/// This can be used to serve "static files" from a directory. By far the most +/// common pattern of serving static files is for `GET` requests, so this +/// filter automatically includes a `GET` check. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // Matches requests that start with `/static`, +/// // and then uses the rest of that path to lookup +/// // and serve a file from `/www/static`. +/// let route = warp::path("static") +/// .and(warp::fs::dir("/www/static")); +/// +/// // For example: +/// // - `GET /static/app.js` would serve the file `/www/static/app.js` +/// // - `GET /static/css/app.css` would serve the file `/www/static/css/app.css` +/// ``` +pub fn dir(path: impl Into<PathBuf>) -> impl FilterClone<Extract = One<File>, Error = Rejection> { + let base = Arc::new(path.into()); + crate::get() + .or(crate::head()) + .unify() + .and(path_from_tail(base)) + .and(conditionals()) + .and_then(file_reply) +} + +fn path_from_tail( + base: Arc<PathBuf>, +) -> impl FilterClone<Extract = One<ArcPath>, Error = Rejection> { + crate::path::tail().and_then(move |tail: crate::path::Tail| { + future::ready(sanitize_path(base.as_ref(), tail.as_str())).and_then(|mut buf| async { + let is_dir = tokio::fs::metadata(buf.clone()) + .await + .map(|m| m.is_dir()) + .unwrap_or(false); + + if is_dir { + tracing::debug!("dir: appending index.html to directory path"); + buf.push("index.html"); + } + tracing::trace!("dir: {:?}", buf); + Ok(ArcPath(Arc::new(buf))) + }) + }) +} + +fn sanitize_path(base: impl AsRef<Path>, tail: &str) -> Result<PathBuf, Rejection> { + let mut buf = PathBuf::from(base.as_ref()); + let p = match percent_decode_str(tail).decode_utf8() { + Ok(p) => p, + Err(err) => { + tracing::debug!("dir: failed to decode route={:?}: {:?}", tail, err); + return Err(reject::not_found()); + } + }; + tracing::trace!("dir? base={:?}, route={:?}", base.as_ref(), p); + for seg in p.split('/') { + if seg.starts_with("..") { + tracing::warn!("dir: rejecting segment starting with '..'"); + return Err(reject::not_found()); + } else if seg.contains('\\') { + tracing::warn!("dir: rejecting segment containing backslash (\\)"); + return Err(reject::not_found()); + } else if cfg!(windows) && seg.contains(':') { + tracing::warn!("dir: rejecting segment containing colon (:)"); + return Err(reject::not_found()); + } else { + buf.push(seg); + } + } + Ok(buf) +} + +#[derive(Debug)] +struct Conditionals { + if_modified_since: Option<IfModifiedSince>, + if_unmodified_since: Option<IfUnmodifiedSince>, + if_range: Option<IfRange>, + range: Option<Range>, +} + +enum Cond { + NoBody(Response), + WithBody(Option<Range>), +} + +impl Conditionals { + fn check(self, last_modified: Option<LastModified>) -> Cond { + if let Some(since) = self.if_unmodified_since { + let precondition = last_modified + .map(|time| since.precondition_passes(time.into())) + .unwrap_or(false); + + tracing::trace!( + "if-unmodified-since? {:?} vs {:?} = {}", + since, + last_modified, + precondition + ); + if !precondition { + let mut res = Response::new(Body::empty()); + *res.status_mut() = StatusCode::PRECONDITION_FAILED; + return Cond::NoBody(res); + } + } + + if let Some(since) = self.if_modified_since { + tracing::trace!( + "if-modified-since? header = {:?}, file = {:?}", + since, + last_modified + ); + let unmodified = last_modified + .map(|time| !since.is_modified(time.into())) + // no last_modified means its always modified + .unwrap_or(false); + if unmodified { + let mut res = Response::new(Body::empty()); + *res.status_mut() = StatusCode::NOT_MODIFIED; + return Cond::NoBody(res); + } + } + + if let Some(if_range) = self.if_range { + tracing::trace!("if-range? {:?} vs {:?}", if_range, last_modified); + let can_range = !if_range.is_modified(None, last_modified.as_ref()); + + if !can_range { + return Cond::WithBody(None); + } + } + + Cond::WithBody(self.range) + } +} + +fn conditionals() -> impl Filter<Extract = One<Conditionals>, Error = Infallible> + Copy { + crate::header::optional2() + .and(crate::header::optional2()) + .and(crate::header::optional2()) + .and(crate::header::optional2()) + .map( + |if_modified_since, if_unmodified_since, if_range, range| Conditionals { + if_modified_since, + if_unmodified_since, + if_range, + range, + }, + ) +} + +/// A file response. +#[derive(Debug)] +pub struct File { + resp: Response, + path: ArcPath, +} + +impl File { + /// Extract the `&Path` of the file this `Response` delivers. + /// + /// # Example + /// + /// The example below changes the Content-Type response header for every file called `video.mp4`. + /// + /// ``` + /// use warp::{Filter, reply::Reply}; + /// + /// let route = warp::path("static") + /// .and(warp::fs::dir("/www/static")) + /// .map(|reply: warp::filters::fs::File| { + /// if reply.path().ends_with("video.mp4") { + /// warp::reply::with_header(reply, "Content-Type", "video/mp4").into_response() + /// } else { + /// reply.into_response() + /// } + /// }); + /// ``` + pub fn path(&self) -> &Path { + self.path.as_ref() + } +} + +// Silly wrapper since Arc<PathBuf> doesn't implement AsRef<Path> ;_; +#[derive(Clone, Debug)] +struct ArcPath(Arc<PathBuf>); + +impl AsRef<Path> for ArcPath { + fn as_ref(&self) -> &Path { + (*self.0).as_ref() + } +} + +impl Reply for File { + fn into_response(self) -> Response { + self.resp + } +} + +fn file_reply( + path: ArcPath, + conditionals: Conditionals, +) -> impl Future<Output = Result<File, Rejection>> + Send { + TkFile::open(path.clone()).then(move |res| match res { + Ok(f) => Either::Left(file_conditional(f, path, conditionals)), + Err(err) => { + let rej = match err.kind() { + io::ErrorKind::NotFound => { + tracing::debug!("file not found: {:?}", path.as_ref().display()); + reject::not_found() + } + io::ErrorKind::PermissionDenied => { + tracing::warn!("file permission denied: {:?}", path.as_ref().display()); + reject::known(FilePermissionError { _p: () }) + } + _ => { + tracing::error!( + "file open error (path={:?}): {} ", + path.as_ref().display(), + err + ); + reject::known(FileOpenError { _p: () }) + } + }; + Either::Right(future::err(rej)) + } + }) +} + +async fn file_metadata(f: TkFile) -> Result<(TkFile, Metadata), Rejection> { + match f.metadata().await { + Ok(meta) => Ok((f, meta)), + Err(err) => { + tracing::debug!("file metadata error: {}", err); + Err(reject::not_found()) + } + } +} + +fn file_conditional( + f: TkFile, + path: ArcPath, + conditionals: Conditionals, +) -> impl Future<Output = Result<File, Rejection>> + Send { + file_metadata(f).map_ok(move |(file, meta)| { + let mut len = meta.len(); + let modified = meta.modified().ok().map(LastModified::from); + + let resp = match conditionals.check(modified) { + Cond::NoBody(resp) => resp, + Cond::WithBody(range) => { + bytes_range(range, len) + .map(|(start, end)| { + let sub_len = end - start; + let buf_size = optimal_buf_size(&meta); + let stream = file_stream(file, buf_size, (start, end)); + let body = Body::wrap_stream(stream); + + let mut resp = Response::new(body); + + if sub_len != len { + *resp.status_mut() = StatusCode::PARTIAL_CONTENT; + resp.headers_mut().typed_insert( + ContentRange::bytes(start..end, len).expect("valid ContentRange"), + ); + + len = sub_len; + } + + let mime = mime_guess::from_path(path.as_ref()).first_or_octet_stream(); + + resp.headers_mut().typed_insert(ContentLength(len)); + resp.headers_mut().typed_insert(ContentType::from(mime)); + resp.headers_mut().typed_insert(AcceptRanges::bytes()); + + if let Some(last_modified) = modified { + resp.headers_mut().typed_insert(last_modified); + } + + resp + }) + .unwrap_or_else(|BadRange| { + // bad byte range + let mut resp = Response::new(Body::empty()); + *resp.status_mut() = StatusCode::RANGE_NOT_SATISFIABLE; + resp.headers_mut() + .typed_insert(ContentRange::unsatisfied_bytes(len)); + resp + }) + } + }; + + File { resp, path } + }) +} + +struct BadRange; + +fn bytes_range(range: Option<Range>, max_len: u64) -> Result<(u64, u64), BadRange> { + use std::ops::Bound; + + let range = if let Some(range) = range { + range + } else { + return Ok((0, max_len)); + }; + + let ret = range + .iter() + .map(|(start, end)| { + let start = match start { + Bound::Unbounded => 0, + Bound::Included(s) => s, + Bound::Excluded(s) => s + 1, + }; + + let end = match end { + Bound::Unbounded => max_len, + Bound::Included(s) => { + // For the special case where s == the file size + if s == max_len { + s + } else { + s + 1 + } + } + Bound::Excluded(s) => s, + }; + + if start < end && end <= max_len { + Ok((start, end)) + } else { + tracing::trace!("unsatisfiable byte range: {}-{}/{}", start, end, max_len); + Err(BadRange) + } + }) + .next() + .unwrap_or(Ok((0, max_len))); + ret +} + +fn file_stream( + mut file: TkFile, + buf_size: usize, + (start, end): (u64, u64), +) -> impl Stream<Item = Result<Bytes, io::Error>> + Send { + use std::io::SeekFrom; + + let seek = async move { + if start != 0 { + file.seek(SeekFrom::Start(start)).await?; + } + Ok(file) + }; + + seek.into_stream() + .map(move |result| { + let mut buf = BytesMut::new(); + let mut len = end - start; + let mut f = match result { + Ok(f) => f, + Err(f) => return Either::Left(stream::once(future::err(f))), + }; + + Either::Right(stream::poll_fn(move |cx| { + if len == 0 { + return Poll::Ready(None); + } + reserve_at_least(&mut buf, buf_size); + + let n = match ready!(poll_read_buf(Pin::new(&mut f), cx, &mut buf)) { + Ok(n) => n as u64, + Err(err) => { + tracing::debug!("file read error: {}", err); + return Poll::Ready(Some(Err(err))); + } + }; + + if n == 0 { + tracing::debug!("file read found EOF before expected length"); + return Poll::Ready(None); + } + + let mut chunk = buf.split().freeze(); + if n > len { + chunk = chunk.split_to(len as usize); + len = 0; + } else { + len -= n; + } + + Poll::Ready(Some(Ok(chunk))) + })) + }) + .flatten() +} + +fn reserve_at_least(buf: &mut BytesMut, cap: usize) { + if buf.capacity() - buf.len() < cap { + buf.reserve(cap); + } +} + +const DEFAULT_READ_BUF_SIZE: usize = 8_192; + +fn optimal_buf_size(metadata: &Metadata) -> usize { + let block_size = get_block_size(metadata); + + // If file length is smaller than block size, don't waste space + // reserving a bigger-than-needed buffer. + cmp::min(block_size as u64, metadata.len()) as usize +} + +#[cfg(unix)] +fn get_block_size(metadata: &Metadata) -> usize { + use std::os::unix::fs::MetadataExt; + //TODO: blksize() returns u64, should handle bad cast... + //(really, a block size bigger than 4gb?) + + // Use device blocksize unless it's really small. + cmp::max(metadata.blksize() as usize, DEFAULT_READ_BUF_SIZE) +} + +#[cfg(not(unix))] +fn get_block_size(_metadata: &Metadata) -> usize { + DEFAULT_READ_BUF_SIZE +} + +// ===== Rejections ===== + +unit_error! { + pub(crate) FileOpenError: "file open error" +} + +unit_error! { + pub(crate) FilePermissionError: "file perimission error" +} + +#[cfg(test)] +mod tests { + use super::sanitize_path; + use bytes::BytesMut; + + #[test] + fn test_sanitize_path() { + let base = "/var/www"; + + fn p(s: &str) -> &::std::path::Path { + s.as_ref() + } + + assert_eq!( + sanitize_path(base, "/foo.html").unwrap(), + p("/var/www/foo.html") + ); + + // bad paths + sanitize_path(base, "/../foo.html").expect_err("dot dot"); + + sanitize_path(base, "/C:\\/foo.html").expect_err("C:\\"); + } + + #[test] + fn test_reserve_at_least() { + let mut buf = BytesMut::new(); + let cap = 8_192; + + assert_eq!(buf.len(), 0); + assert_eq!(buf.capacity(), 0); + + super::reserve_at_least(&mut buf, cap); + assert_eq!(buf.len(), 0); + assert_eq!(buf.capacity(), cap); + } +} diff --git a/third_party/rust/warp/src/filters/header.rs b/third_party/rust/warp/src/filters/header.rs new file mode 100644 index 0000000000..0c535a38b5 --- /dev/null +++ b/third_party/rust/warp/src/filters/header.rs @@ -0,0 +1,230 @@ +//! Header Filters +//! +//! These filters are used to interact with the Request HTTP headers. Some +//! of them, like `exact` and `exact_ignore_case`, are just predicates, +//! they don't extract any values. The `header` filter allows parsing +//! a type from any header. +use std::convert::Infallible; +use std::str::FromStr; + +use futures_util::future; +use headers::{Header, HeaderMapExt}; +use http::header::HeaderValue; +use http::HeaderMap; + +use crate::filter::{filter_fn, filter_fn_one, Filter, One}; +use crate::reject::{self, Rejection}; + +/// Create a `Filter` that tries to parse the specified header. +/// +/// This `Filter` will look for a header with supplied name, and try to +/// parse to a `T`, otherwise rejects the request. +/// +/// # Example +/// +/// ``` +/// use std::net::SocketAddr; +/// +/// // Parse `content-length: 100` as a `u64` +/// let content_length = warp::header::<u64>("content-length"); +/// +/// // Parse `host: 127.0.0.1:8080` as a `SocketAddr +/// let local_host = warp::header::<SocketAddr>("host"); +/// +/// // Parse `foo: bar` into a `String` +/// let foo = warp::header::<String>("foo"); +/// ``` +pub fn header<T: FromStr + Send + 'static>( + name: &'static str, +) -> impl Filter<Extract = One<T>, Error = Rejection> + Copy { + filter_fn_one(move |route| { + tracing::trace!("header({:?})", name); + let route = route + .headers() + .get(name) + .ok_or_else(|| reject::missing_header(name)) + .and_then(|value| value.to_str().map_err(|_| reject::invalid_header(name))) + .and_then(|s| T::from_str(s).map_err(|_| reject::invalid_header(name))); + future::ready(route) + }) +} + +pub(crate) fn header2<T: Header + Send + 'static>( +) -> impl Filter<Extract = One<T>, Error = Rejection> + Copy { + filter_fn_one(move |route| { + tracing::trace!("header2({:?})", T::name()); + let route = route + .headers() + .typed_get() + .ok_or_else(|| reject::invalid_header(T::name().as_str())); + future::ready(route) + }) +} + +/// Create a `Filter` that tries to parse the specified header, if it exists. +/// +/// If the header does not exist, it yields `None`. Otherwise, it will try to +/// parse as a `T`, and if it fails, a invalid header rejection is return. If +/// successful, the filter yields `Some(T)`. +/// +/// # Example +/// +/// ``` +/// // Grab the `authorization` header if it exists. +/// let opt_auth = warp::header::optional::<String>("authorization"); +/// ``` +pub fn optional<T>( + name: &'static str, +) -> impl Filter<Extract = One<Option<T>>, Error = Rejection> + Copy +where + T: FromStr + Send + 'static, +{ + filter_fn_one(move |route| { + tracing::trace!("optional({:?})", name); + let result = route.headers().get(name).map(|value| { + value + .to_str() + .map_err(|_| reject::invalid_header(name))? + .parse::<T>() + .map_err(|_| reject::invalid_header(name)) + }); + + match result { + Some(Ok(t)) => future::ok(Some(t)), + Some(Err(e)) => future::err(e), + None => future::ok(None), + } + }) +} + +pub(crate) fn optional2<T>() -> impl Filter<Extract = One<Option<T>>, Error = Infallible> + Copy +where + T: Header + Send + 'static, +{ + filter_fn_one(move |route| future::ready(Ok(route.headers().typed_get()))) +} + +/* TODO +pub fn exact2<T>(header: T) -> impl FilterClone<Extract=(), Error=Rejection> +where + T: Header + PartialEq + Clone + Send, +{ + filter_fn(move |route| { + tracing::trace!("exact2({:?})", T::NAME); + route.headers() + .typed_get::<T>() + .and_then(|val| if val == header { + Some(()) + } else { + None + }) + .ok_or_else(|| reject::bad_request()) + }) +} +*/ + +/// Create a `Filter` that requires a header to match the value exactly. +/// +/// This `Filter` will look for a header with supplied name and the exact +/// value, otherwise rejects the request. +/// +/// # Example +/// +/// ``` +/// // Require `dnt: 1` header to be set. +/// let must_dnt = warp::header::exact("dnt", "1"); +/// ``` +pub fn exact( + name: &'static str, + value: &'static str, +) -> impl Filter<Extract = (), Error = Rejection> + Copy { + filter_fn(move |route| { + tracing::trace!("exact?({:?}, {:?})", name, value); + let route = route + .headers() + .get(name) + .ok_or_else(|| reject::missing_header(name)) + .and_then(|val| { + if val == value { + Ok(()) + } else { + Err(reject::invalid_header(name)) + } + }); + future::ready(route) + }) +} + +/// Create a `Filter` that requires a header to match the value exactly. +/// +/// This `Filter` will look for a header with supplied name and the exact +/// value, ignoring ASCII case, otherwise rejects the request. +/// +/// # Example +/// +/// ``` +/// // Require `connection: keep-alive` header to be set. +/// let keep_alive = warp::header::exact_ignore_case("connection", "keep-alive"); +/// ``` +pub fn exact_ignore_case( + name: &'static str, + value: &'static str, +) -> impl Filter<Extract = (), Error = Rejection> + Copy { + filter_fn(move |route| { + tracing::trace!("exact_ignore_case({:?}, {:?})", name, value); + let route = route + .headers() + .get(name) + .ok_or_else(|| reject::missing_header(name)) + .and_then(|val| { + if val.as_bytes().eq_ignore_ascii_case(value.as_bytes()) { + Ok(()) + } else { + Err(reject::invalid_header(name)) + } + }); + future::ready(route) + }) +} + +/// Create a `Filter` that gets a `HeaderValue` for the name. +/// +/// # Example +/// +/// ``` +/// use warp::{Filter, http::header::HeaderValue}; +/// +/// let filter = warp::header::value("x-token") +/// .map(|value: HeaderValue| { +/// format!("header value bytes: {:?}", value) +/// }); +/// ``` +pub fn value( + name: &'static str, +) -> impl Filter<Extract = One<HeaderValue>, Error = Rejection> + Copy { + filter_fn_one(move |route| { + tracing::trace!("value({:?})", name); + let route = route + .headers() + .get(name) + .cloned() + .ok_or_else(|| reject::missing_header(name)); + future::ready(route) + }) +} + +/// Create a `Filter` that returns a clone of the request's `HeaderMap`. +/// +/// # Example +/// +/// ``` +/// use warp::{Filter, http::HeaderMap}; +/// +/// let headers = warp::header::headers_cloned() +/// .map(|headers: HeaderMap| { +/// format!("header count: {}", headers.len()) +/// }); +/// ``` +pub fn headers_cloned() -> impl Filter<Extract = One<HeaderMap>, Error = Infallible> + Copy { + filter_fn_one(|route| future::ok(route.headers().clone())) +} diff --git a/third_party/rust/warp/src/filters/host.rs b/third_party/rust/warp/src/filters/host.rs new file mode 100644 index 0000000000..9aae039c19 --- /dev/null +++ b/third_party/rust/warp/src/filters/host.rs @@ -0,0 +1,96 @@ +//! Host ("authority") filter +//! +use crate::filter::{filter_fn_one, Filter, One}; +use crate::reject::{self, Rejection}; +use futures_util::future; +pub use http::uri::Authority; +use std::str::FromStr; + +/// Creates a `Filter` that requires a specific authority (target server's +/// host and port) in the request. +/// +/// Authority is specified either in the `Host` header or in the target URI. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let multihost = +/// warp::host::exact("foo.com").map(|| "you've reached foo.com") +/// .or(warp::host::exact("bar.com").map(|| "you've reached bar.com")); +/// ``` +pub fn exact(expected: &str) -> impl Filter<Extract = (), Error = Rejection> + Clone { + let expected = Authority::from_str(expected).expect("invalid host/authority"); + optional() + .and_then(move |option: Option<Authority>| match option { + Some(authority) if authority == expected => future::ok(()), + _ => future::err(reject::not_found()), + }) + .untuple_one() +} + +/// Creates a `Filter` that looks for an authority (target server's host +/// and port) in the request. +/// +/// Authority is specified either in the `Host` header or in the target URI. +/// +/// If found, extracts the `Authority`, otherwise continues the request, +/// extracting `None`. +/// +/// Rejects with `400 Bad Request` if the `Host` header is malformed or if there +/// is a mismatch between the `Host` header and the target URI. +/// +/// # Example +/// +/// ``` +/// use warp::{Filter, host::Authority}; +/// +/// let host = warp::host::optional() +/// .map(|authority: Option<Authority>| { +/// if let Some(a) = authority { +/// format!("{} is currently not at home", a.host()) +/// } else { +/// "please state who you're trying to reach".to_owned() +/// } +/// }); +/// ``` +pub fn optional() -> impl Filter<Extract = One<Option<Authority>>, Error = Rejection> + Copy { + filter_fn_one(move |route| { + // The authority can be sent by clients in various ways: + // + // 1) in the "target URI" + // a) serialized in the start line (HTTP/1.1 proxy requests) + // b) serialized in `:authority` pseudo-header (HTTP/2 generated - "SHOULD") + // 2) in the `Host` header (HTTP/1.1 origin requests, HTTP/2 converted) + // + // Hyper transparently handles 1a/1b, but not 2, so we must look at both. + + let from_uri = route.uri().authority(); + + let name = "host"; + let from_header = route.headers() + .get(name) + .map(|value| + // Header present, parse it + value.to_str().map_err(|_| reject::invalid_header(name)) + .and_then(|value| Authority::from_str(value).map_err(|_| reject::invalid_header(name))) + ); + + future::ready(match (from_uri, from_header) { + // no authority in the request (HTTP/1.0 or non-conforming) + (None, None) => Ok(None), + + // authority specified in either or both matching + (Some(a), None) => Ok(Some(a.clone())), + (None, Some(Ok(a))) => Ok(Some(a)), + (Some(a), Some(Ok(b))) if *a == b => Ok(Some(b)), + + // mismatch + (Some(_), Some(Ok(_))) => Err(reject::invalid_header(name)), + + // parse error + (_, Some(Err(r))) => Err(r), + }) + }) +} diff --git a/third_party/rust/warp/src/filters/log.rs b/third_party/rust/warp/src/filters/log.rs new file mode 100644 index 0000000000..3790fd8a8b --- /dev/null +++ b/third_party/rust/warp/src/filters/log.rs @@ -0,0 +1,280 @@ +//! Logger Filters + +use std::fmt; +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +use http::{self, header, StatusCode}; + +use crate::filter::{Filter, WrapSealed}; +use crate::reject::IsReject; +use crate::reply::Reply; +use crate::route::Route; + +use self::internal::WithLog; + +/// Create a wrapping [`Filter`](crate::Filter) with the specified `name` as the `target`. +/// +/// This uses the default access logging format, and log records produced +/// will have their `target` set to `name`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // If using something like `pretty_env_logger`, +/// // view logs by setting `RUST_LOG=example::api`. +/// let log = warp::log("example::api"); +/// let route = warp::any() +/// .map(warp::reply) +/// .with(log); +/// ``` +pub fn log(name: &'static str) -> Log<impl Fn(Info<'_>) + Copy> { + let func = move |info: Info<'_>| { + // TODO? + // - response content length? + log::info!( + target: name, + "{} \"{} {} {:?}\" {} \"{}\" \"{}\" {:?}", + OptFmt(info.route.remote_addr()), + info.method(), + info.path(), + info.route.version(), + info.status().as_u16(), + OptFmt(info.referer()), + OptFmt(info.user_agent()), + info.elapsed(), + ); + }; + Log { func } +} + +/// Create a wrapping [`Filter`](crate::Filter) that receives `warp::log::Info`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let log = warp::log::custom(|info| { +/// // Use a log macro, or slog, or println, or whatever! +/// eprintln!( +/// "{} {} {}", +/// info.method(), +/// info.path(), +/// info.status(), +/// ); +/// }); +/// let route = warp::any() +/// .map(warp::reply) +/// .with(log); +/// ``` +pub fn custom<F>(func: F) -> Log<F> +where + F: Fn(Info<'_>), +{ + Log { func } +} + +/// Decorates a [`Filter`] to log requests and responses. +#[derive(Clone, Copy, Debug)] +pub struct Log<F> { + func: F, +} + +/// Information about the request/response that can be used to prepare log lines. +#[allow(missing_debug_implementations)] +pub struct Info<'a> { + route: &'a Route, + start: Instant, + status: StatusCode, +} + +impl<FN, F> WrapSealed<F> for Log<FN> +where + FN: Fn(Info<'_>) + Clone + Send, + F: Filter + Clone + Send, + F::Extract: Reply, + F::Error: IsReject, +{ + type Wrapped = WithLog<FN, F>; + + fn wrap(&self, filter: F) -> Self::Wrapped { + WithLog { + filter, + log: self.clone(), + } + } +} + +impl<'a> Info<'a> { + /// View the remote `SocketAddr` of the request. + pub fn remote_addr(&self) -> Option<SocketAddr> { + self.route.remote_addr() + } + + /// View the `http::Method` of the request. + pub fn method(&self) -> &http::Method { + self.route.method() + } + + /// View the URI path of the request. + pub fn path(&self) -> &str { + self.route.full_path() + } + + /// View the `http::Version` of the request. + pub fn version(&self) -> http::Version { + self.route.version() + } + + /// View the `http::StatusCode` of the response. + pub fn status(&self) -> http::StatusCode { + self.status + } + + /// View the referer of the request. + pub fn referer(&self) -> Option<&str> { + self.route + .headers() + .get(header::REFERER) + .and_then(|v| v.to_str().ok()) + } + + /// View the user agent of the request. + pub fn user_agent(&self) -> Option<&str> { + self.route + .headers() + .get(header::USER_AGENT) + .and_then(|v| v.to_str().ok()) + } + + /// View the `Duration` that elapsed for the request. + pub fn elapsed(&self) -> Duration { + tokio::time::Instant::now().into_std() - self.start + } + + /// View the host of the request + pub fn host(&self) -> Option<&str> { + self.route + .headers() + .get(header::HOST) + .and_then(|v| v.to_str().ok()) + } + + /// Access the full headers of the request + pub fn request_headers(&self) -> &http::HeaderMap { + self.route.headers() + } +} + +struct OptFmt<T>(Option<T>); + +impl<T: fmt::Display> fmt::Display for OptFmt<T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + if let Some(ref t) = self.0 { + fmt::Display::fmt(t, f) + } else { + f.write_str("-") + } + } +} + +mod internal { + use std::future::Future; + use std::pin::Pin; + use std::task::{Context, Poll}; + use std::time::Instant; + + use futures_util::{ready, TryFuture}; + use pin_project::pin_project; + + use super::{Info, Log}; + use crate::filter::{Filter, FilterBase, Internal}; + use crate::reject::IsReject; + use crate::reply::{Reply, Response}; + use crate::route; + + #[allow(missing_debug_implementations)] + pub struct Logged(pub(super) Response); + + impl Reply for Logged { + #[inline] + fn into_response(self) -> Response { + self.0 + } + } + + #[allow(missing_debug_implementations)] + #[derive(Clone, Copy)] + pub struct WithLog<FN, F> { + pub(super) filter: F, + pub(super) log: Log<FN>, + } + + impl<FN, F> FilterBase for WithLog<FN, F> + where + FN: Fn(Info<'_>) + Clone + Send, + F: Filter + Clone + Send, + F::Extract: Reply, + F::Error: IsReject, + { + type Extract = (Logged,); + type Error = F::Error; + type Future = WithLogFuture<FN, F::Future>; + + fn filter(&self, _: Internal) -> Self::Future { + let started = tokio::time::Instant::now().into_std(); + WithLogFuture { + log: self.log.clone(), + future: self.filter.filter(Internal), + started, + } + } + } + + #[allow(missing_debug_implementations)] + #[pin_project] + pub struct WithLogFuture<FN, F> { + log: Log<FN>, + #[pin] + future: F, + started: Instant, + } + + impl<FN, F> Future for WithLogFuture<FN, F> + where + FN: Fn(Info<'_>), + F: TryFuture, + F::Ok: Reply, + F::Error: IsReject, + { + type Output = Result<(Logged,), F::Error>; + + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { + let pin = self.as_mut().project(); + let (result, status) = match ready!(pin.future.try_poll(cx)) { + Ok(reply) => { + let resp = reply.into_response(); + let status = resp.status(); + (Poll::Ready(Ok((Logged(resp),))), status) + } + Err(reject) => { + let status = reject.status(); + (Poll::Ready(Err(reject)), status) + } + }; + + route::with(|route| { + (self.log.func)(Info { + route, + start: self.started, + status, + }); + }); + + result + } + } +} diff --git a/third_party/rust/warp/src/filters/method.rs b/third_party/rust/warp/src/filters/method.rs new file mode 100644 index 0000000000..c4d7462720 --- /dev/null +++ b/third_party/rust/warp/src/filters/method.rs @@ -0,0 +1,150 @@ +//! HTTP Method filters. +//! +//! The filters deal with the HTTP Method part of a request. Several here will +//! match the request `Method`, and if not matched, will reject the request +//! with a `405 Method Not Allowed`. +//! +//! There is also [`warp::method()`](method), which never rejects +//! a request, and just extracts the method to be used in your filter chains. +use futures_util::future; +use http::Method; + +use crate::filter::{filter_fn, filter_fn_one, Filter, One}; +use crate::reject::Rejection; +use std::convert::Infallible; + +/// Create a `Filter` that requires the request method to be `GET`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let get_only = warp::get().map(warp::reply); +/// ``` +pub fn get() -> impl Filter<Extract = (), Error = Rejection> + Copy { + method_is(|| &Method::GET) +} + +/// Create a `Filter` that requires the request method to be `POST`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let post_only = warp::post().map(warp::reply); +/// ``` +pub fn post() -> impl Filter<Extract = (), Error = Rejection> + Copy { + method_is(|| &Method::POST) +} + +/// Create a `Filter` that requires the request method to be `PUT`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let put_only = warp::put().map(warp::reply); +/// ``` +pub fn put() -> impl Filter<Extract = (), Error = Rejection> + Copy { + method_is(|| &Method::PUT) +} + +/// Create a `Filter` that requires the request method to be `DELETE`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let delete_only = warp::delete().map(warp::reply); +/// ``` +pub fn delete() -> impl Filter<Extract = (), Error = Rejection> + Copy { + method_is(|| &Method::DELETE) +} + +/// Create a `Filter` that requires the request method to be `HEAD`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let head_only = warp::head().map(warp::reply); +/// ``` +pub fn head() -> impl Filter<Extract = (), Error = Rejection> + Copy { + method_is(|| &Method::HEAD) +} + +/// Create a `Filter` that requires the request method to be `OPTIONS`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let options_only = warp::options().map(warp::reply); +/// ``` +pub fn options() -> impl Filter<Extract = (), Error = Rejection> + Copy { + method_is(|| &Method::OPTIONS) +} + +/// Create a `Filter` that requires the request method to be `PATCH`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let patch_only = warp::patch().map(warp::reply); +/// ``` +pub fn patch() -> impl Filter<Extract = (), Error = Rejection> + Copy { + method_is(|| &Method::PATCH) +} + +/// Extract the `Method` from the request. +/// +/// This never rejects a request. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::method() +/// .map(|method| { +/// format!("You sent a {} request!", method) +/// }); +/// ``` +pub fn method() -> impl Filter<Extract = One<Method>, Error = Infallible> + Copy { + filter_fn_one(|route| future::ok::<_, Infallible>(route.method().clone())) +} + +// NOTE: This takes a static function instead of `&'static Method` directly +// so that the `impl Filter` can be zero-sized. Moving it around should be +// cheaper than holding a single static pointer (which would make it 1 word). +fn method_is<F>(func: F) -> impl Filter<Extract = (), Error = Rejection> + Copy +where + F: Fn() -> &'static Method + Copy, +{ + filter_fn(move |route| { + let method = func(); + tracing::trace!("method::{:?}?: {:?}", method, route.method()); + if route.method() == method { + future::ok(()) + } else { + future::err(crate::reject::method_not_allowed()) + } + }) +} + +#[cfg(test)] +mod tests { + #[test] + fn method_size_of() { + // See comment on `method_is` function. + assert_eq!(std::mem::size_of_val(&super::get()), 0,); + } +} diff --git a/third_party/rust/warp/src/filters/mod.rs b/third_party/rust/warp/src/filters/mod.rs new file mode 100644 index 0000000000..bd1c48718c --- /dev/null +++ b/third_party/rust/warp/src/filters/mod.rs @@ -0,0 +1,29 @@ +//! Built-in Filters +//! +//! This module mostly serves as documentation to group together the list of +//! built-in filters. Most of these are available at more convenient paths. + +pub mod addr; +pub mod any; +pub mod body; +#[cfg(any(feature = "compression-brotli", feature = "compression-gzip"))] +pub mod compression; +pub mod cookie; +pub mod cors; +pub mod ext; +pub mod fs; +pub mod header; +pub mod host; +pub mod log; +pub mod method; +#[cfg(feature = "multipart")] +pub mod multipart; +pub mod path; +pub mod query; +pub mod reply; +pub mod sse; +pub mod trace; +#[cfg(feature = "websocket")] +pub mod ws; + +pub use crate::filter::BoxedFilter; diff --git a/third_party/rust/warp/src/filters/multipart.rs b/third_party/rust/warp/src/filters/multipart.rs new file mode 100644 index 0000000000..103f257607 --- /dev/null +++ b/third_party/rust/warp/src/filters/multipart.rs @@ -0,0 +1,228 @@ +//! Multipart body filters +//! +//! [`Filter`](crate::Filter)s that extract a multipart body for a route. + +use std::error::Error as StdError; +use std::fmt::{Display, Formatter}; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::{fmt, io}; + +use bytes::{Buf, Bytes}; +use futures_util::{future, Stream}; +use headers::ContentType; +use hyper::Body; +use mime::Mime; +use multer::{Field as PartInner, Multipart as FormDataInner}; + +use crate::filter::{Filter, FilterBase, Internal}; +use crate::reject::{self, Rejection}; + +// If not otherwise configured, default to 2MB. +const DEFAULT_FORM_DATA_MAX_LENGTH: u64 = 1024 * 1024 * 2; + +/// A [`Filter`](crate::Filter) to extract a `multipart/form-data` body from a request. +/// +/// Create with the `warp::multipart::form()` function. +#[derive(Debug, Clone)] +pub struct FormOptions { + max_length: Option<u64>, +} + +/// A `Stream` of multipart/form-data `Part`s. +/// +/// Extracted with a `warp::multipart::form` filter. +pub struct FormData { + inner: FormDataInner<'static>, +} + +/// A single "part" of a multipart/form-data body. +/// +/// Yielded from the `FormData` stream. +pub struct Part { + part: PartInner<'static>, +} + +/// Create a [`Filter`](crate::Filter) to extract a `multipart/form-data` body from a request. +/// +/// The extracted `FormData` type is a `Stream` of `Part`s, and each `Part` +/// in turn is a `Stream` of bytes. +pub fn form() -> FormOptions { + FormOptions { + max_length: Some(DEFAULT_FORM_DATA_MAX_LENGTH), + } +} + +// ===== impl Form ===== + +impl FormOptions { + /// Set the maximum byte length allowed for this body. + /// + /// `max_length(None)` means that maximum byte length is not checked. + /// Defaults to 2MB. + pub fn max_length(mut self, max: impl Into<Option<u64>>) -> Self { + self.max_length = max.into(); + self + } +} + +type FormFut = Pin<Box<dyn Future<Output = Result<(FormData,), Rejection>> + Send>>; + +impl FilterBase for FormOptions { + type Extract = (FormData,); + type Error = Rejection; + type Future = FormFut; + + fn filter(&self, _: Internal) -> Self::Future { + let boundary = super::header::header2::<ContentType>().and_then(|ct| { + let mime = Mime::from(ct); + let mime = mime + .get_param("boundary") + .map(|v| v.to_string()) + .ok_or_else(|| reject::invalid_header("content-type")); + future::ready(mime) + }); + + let filt = boundary + .and(super::body::body()) + .map(|boundary: String, body| { + let body = BodyIoError(body); + FormData { + inner: FormDataInner::new(body, &boundary), + } + }); + + if let Some(max_length) = self.max_length { + Box::pin( + super::body::content_length_limit(max_length) + .and(filt) + .filter(Internal), + ) + } else { + Box::pin(filt.filter(Internal)) + } + } +} + +// ===== impl FormData ===== + +impl fmt::Debug for FormData { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("FormData").finish() + } +} + +impl Stream for FormData { + type Item = Result<Part, crate::Error>; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + match self.inner.poll_next_field(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Ok(Some(part))) => { + if part.name().is_some() { + Poll::Ready(Some(Ok(Part { part }))) + } else { + Poll::Ready(Some(Err(crate::Error::new(MultipartFieldMissingName)))) + } + } + Poll::Ready(Ok(None)) => Poll::Ready(None), + Poll::Ready(Err(err)) => Poll::Ready(Some(Err(crate::Error::new(err)))), + } + } +} + +// ===== impl Part ===== + +impl Part { + /// Get the name of this part. + pub fn name(&self) -> &str { + self.part.name().expect("checked for name previously") + } + + /// Get the filename of this part, if present. + pub fn filename(&self) -> Option<&str> { + self.part.file_name() + } + + /// Get the content-type of this part, if present. + pub fn content_type(&self) -> Option<&str> { + let content_type = self.part.content_type(); + content_type.map(|t| t.as_ref()) + } + + /// Asynchronously get some of the data for this `Part`. + pub async fn data(&mut self) -> Option<Result<impl Buf, crate::Error>> { + future::poll_fn(|cx| self.poll_next(cx)).await + } + + /// Convert this `Part` into a `Stream` of `Buf`s. + pub fn stream(self) -> impl Stream<Item = Result<impl Buf, crate::Error>> { + PartStream(self) + } + + fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, crate::Error>>> { + match Pin::new(&mut self.part).poll_next(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(Ok(bytes))), + Poll::Ready(None) => Poll::Ready(None), + Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(crate::Error::new(err)))), + } + } +} + +impl fmt::Debug for Part { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut builder = f.debug_struct("Part"); + builder.field("name", &self.part.name()); + + if let Some(ref filename) = self.part.file_name() { + builder.field("filename", filename); + } + + if let Some(ref mime) = self.part.content_type() { + builder.field("content_type", mime); + } + + builder.finish() + } +} + +struct PartStream(Part); + +impl Stream for PartStream { + type Item = Result<Bytes, crate::Error>; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + self.0.poll_next(cx) + } +} + +struct BodyIoError(Body); + +impl Stream for BodyIoError { + type Item = io::Result<Bytes>; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + match Pin::new(&mut self.0).poll_next(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(Ok(bytes))), + Poll::Ready(None) => Poll::Ready(None), + Poll::Ready(Some(Err(err))) => { + Poll::Ready(Some(Err(io::Error::new(io::ErrorKind::Other, err)))) + } + } + } +} + +/// An error used when a multipart field is missing a name. +#[derive(Debug)] +struct MultipartFieldMissingName; + +impl Display for MultipartFieldMissingName { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!(f, "Multipart field is missing a name") + } +} + +impl StdError for MultipartFieldMissingName {} diff --git a/third_party/rust/warp/src/filters/path.rs b/third_party/rust/warp/src/filters/path.rs new file mode 100644 index 0000000000..29676a7294 --- /dev/null +++ b/third_party/rust/warp/src/filters/path.rs @@ -0,0 +1,652 @@ +//! Path Filters +//! +//! The [`Filter`](crate::Filter)s here work on the "path" of requests. +//! +//! - [`path`](./fn.path.html) matches a specific segment, like `/foo`. +//! - [`param`](./fn.param.html) tries to parse a segment into a type, like `/:u16`. +//! - [`end`](./fn.end.html) matches when the path end is found. +//! - [`path!`](../../macro.path.html) eases combining multiple `path` and `param` filters. +//! +//! # Routing +//! +//! Routing in warp is simple yet powerful. +//! +//! First up, matching a single segment: +//! +//! ``` +//! use warp::Filter; +//! +//! // GET /hi +//! let hi = warp::path("hi").map(|| { +//! "Hello, World!" +//! }); +//! ``` +//! +//! How about multiple segments? It's easiest with the `path!` macro: +//! +//! ``` +//! # use warp::Filter; +//! // GET /hello/from/warp +//! let hello_from_warp = warp::path!("hello" / "from" / "warp").map(|| { +//! "Hello from warp!" +//! }); +//! ``` +//! +//! Neat! But how do I handle **parameters** in paths? +//! +//! ``` +//! # use warp::Filter; +//! // GET /sum/:u32/:u32 +//! let sum = warp::path!("sum" / u32 / u32).map(|a, b| { +//! format!("{} + {} = {}", a, b, a + b) +//! }); +//! ``` +//! +//! In fact, any type that implements `FromStr` can be used, in any order: +//! +//! ``` +//! # use warp::Filter; +//! // GET /:u16/times/:u16 +//! let times = warp::path!(u16 / "times" / u16).map(|a, b| { +//! format!("{} times {} = {}", a, b, a * b) +//! }); +//! ``` +//! +//! Oh shoot, those math routes should be **mounted** at a different path, +//! is that possible? Yep! +//! +//! ``` +//! # use warp::Filter; +//! # let sum = warp::any().map(warp::reply); +//! # let times = sum.clone(); +//! // GET /math/sum/:u32/:u32 +//! // GET /math/:u16/times/:u16 +//! let math = warp::path("math"); +//! let math_sum = math.and(sum); +//! let math_times = math.and(times); +//! ``` +//! +//! What! `and`? What's that do? +//! +//! It combines the filters in a sort of "this and then that" order. In fact, +//! it's exactly what the `path!` macro has been doing internally. +//! +//! ``` +//! # use warp::Filter; +//! // GET /bye/:string +//! let bye = warp::path("bye") +//! .and(warp::path::param()) +//! .map(|name: String| { +//! format!("Good bye, {}!", name) +//! }); +//! ``` +//! +//! Ah, so, can filters do things besides `and`? +//! +//! Why, yes they can! They can also `or`! As you might expect, `or` creates a +//! "this or else that" chain of filters. If the first doesn't succeed, then +//! it tries the other. +//! +//! So, those `math` routes could have been **mounted** all as one, with `or`. +//! +//! +//! ``` +//! # use warp::Filter; +//! # let sum = warp::path("sum"); +//! # let times = warp::path("times"); +//! // GET /math/sum/:u32/:u32 +//! // GET /math/:u16/times/:u16 +//! let math = warp::path("math") +//! .and(sum.or(times)); +//! ``` +//! +//! It turns out, using `or` is how you combine everything together into a +//! single API. +//! +//! ``` +//! # use warp::Filter; +//! # let hi = warp::path("hi"); +//! # let hello_from_warp = hi.clone(); +//! # let bye = hi.clone(); +//! # let math = hi.clone(); +//! // GET /hi +//! // GET /hello/from/warp +//! // GET /bye/:string +//! // GET /math/sum/:u32/:u32 +//! // GET /math/:u16/times/:u16 +//! let routes = hi +//! .or(hello_from_warp) +//! .or(bye) +//! .or(math); +//! ``` +//! +//! Note that you will generally want path filters to come **before** other filters +//! like `body` or `headers`. If a different type of filter comes first, a request +//! with an invalid body for route `/right-path-wrong-body` may try matching against `/wrong-path` +//! and return the error from `/wrong-path` instead of the correct body-related error. + +use std::convert::Infallible; +use std::fmt; +use std::str::FromStr; + +use futures_util::future; +use http::uri::PathAndQuery; + +use self::internal::Opaque; +use crate::filter::{filter_fn, one, Filter, FilterBase, Internal, One, Tuple}; +use crate::reject::{self, Rejection}; +use crate::route::{self, Route}; + +/// Create an exact match path segment [`Filter`](crate::Filter). +/// +/// This will try to match exactly to the current request path segment. +/// +/// # Note +/// +/// - [`end()`](./fn.end.html) should be used to match the end of a path to avoid having +/// filters for shorter paths like `/math` unintentionally match a longer +/// path such as `/math/sum` +/// - Path-related filters should generally come **before** other types of filters, such +/// as those checking headers or body types. Including those other filters before +/// the path checks may result in strange errors being returned because a given request +/// does not match the parameters for a completely separate route. +/// +/// # Panics +/// +/// Exact path filters cannot be empty, or contain slashes. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // Matches '/hello' +/// let hello = warp::path("hello") +/// .map(|| "Hello, World!"); +/// ``` +pub fn path<P>(p: P) -> Exact<Opaque<P>> +where + P: AsRef<str>, +{ + let s = p.as_ref(); + assert!(!s.is_empty(), "exact path segments should not be empty"); + assert!( + !s.contains('/'), + "exact path segments should not contain a slash: {:?}", + s + ); + + Exact(Opaque(p)) + /* + segment(move |seg| { + tracing::trace!("{:?}?: {:?}", p, seg); + if seg == p { + Ok(()) + } else { + Err(reject::not_found()) + } + }) + */ +} + +/// A [`Filter`](crate::Filter) matching an exact path segment. +/// +/// Constructed from `path()` or `path!()`. +#[allow(missing_debug_implementations)] +#[derive(Clone, Copy)] +pub struct Exact<P>(P); + +impl<P> FilterBase for Exact<P> +where + P: AsRef<str>, +{ + type Extract = (); + type Error = Rejection; + type Future = future::Ready<Result<Self::Extract, Self::Error>>; + + #[inline] + fn filter(&self, _: Internal) -> Self::Future { + route::with(|route| { + let p = self.0.as_ref(); + future::ready(with_segment(route, |seg| { + tracing::trace!("{:?}?: {:?}", p, seg); + + if seg == p { + Ok(()) + } else { + Err(reject::not_found()) + } + })) + }) + } +} + +/// Matches the end of a route. +/// +/// Note that _not_ including `end()` may result in shorter paths like +/// `/math` unintentionally matching `/math/sum`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // Matches '/' +/// let hello = warp::path::end() +/// .map(|| "Hello, World!"); +/// ``` +pub fn end() -> impl Filter<Extract = (), Error = Rejection> + Copy { + filter_fn(move |route| { + if route.path().is_empty() { + future::ok(()) + } else { + future::err(reject::not_found()) + } + }) +} + +/// Extract a parameter from a path segment. +/// +/// This will try to parse a value from the current request path +/// segment, and if successful, the value is returned as the `Filter`'s +/// "extracted" value. +/// +/// If the value could not be parsed, rejects with a `404 Not Found`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::path::param() +/// .map(|id: u32| { +/// format!("You asked for /{}", id) +/// }); +/// ``` +pub fn param<T: FromStr + Send + 'static>( +) -> impl Filter<Extract = One<T>, Error = Rejection> + Copy { + filter_segment(|seg| { + tracing::trace!("param?: {:?}", seg); + if seg.is_empty() { + return Err(reject::not_found()); + } + T::from_str(seg).map(one).map_err(|_| reject::not_found()) + }) +} + +/// Extract the unmatched tail of the path. +/// +/// This will return a `Tail`, which allows access to the rest of the path +/// that previous filters have not already matched. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::path("foo") +/// .and(warp::path::tail()) +/// .map(|tail| { +/// // GET /foo/bar/baz would return "bar/baz". +/// format!("The tail after foo is {:?}", tail) +/// }); +/// ``` +pub fn tail() -> impl Filter<Extract = One<Tail>, Error = Infallible> + Copy { + filter_fn(move |route| { + let path = path_and_query(route); + let idx = route.matched_path_index(); + + // Giving the user the full tail means we assume the full path + // has been matched now. + let end = path.path().len() - idx; + route.set_unmatched_path(end); + + future::ok(one(Tail { + path, + start_index: idx, + })) + }) +} + +/// Represents the tail part of a request path, returned by the [`tail()`] filter. +pub struct Tail { + path: PathAndQuery, + start_index: usize, +} + +impl Tail { + /// Get the `&str` representation of the remaining path. + pub fn as_str(&self) -> &str { + &self.path.path()[self.start_index..] + } +} + +impl fmt::Debug for Tail { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(self.as_str(), f) + } +} + +/// Peek at the unmatched tail of the path, without affecting the matched path. +/// +/// This will return a `Peek`, which allows access to the rest of the path +/// that previous filters have not already matched. This differs from `tail` +/// in that `peek` will **not** set the entire path as matched. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::path("foo") +/// .and(warp::path::peek()) +/// .map(|peek| { +/// // GET /foo/bar/baz would return "bar/baz". +/// format!("The path after foo is {:?}", peek) +/// }); +/// ``` +pub fn peek() -> impl Filter<Extract = One<Peek>, Error = Infallible> + Copy { + filter_fn(move |route| { + let path = path_and_query(route); + let idx = route.matched_path_index(); + + future::ok(one(Peek { + path, + start_index: idx, + })) + }) +} + +/// Represents the tail part of a request path, returned by the [`peek()`] filter. +pub struct Peek { + path: PathAndQuery, + start_index: usize, +} + +impl Peek { + /// Get the `&str` representation of the remaining path. + pub fn as_str(&self) -> &str { + &self.path.path()[self.start_index..] + } + + /// Get an iterator over the segments of the peeked path. + pub fn segments(&self) -> impl Iterator<Item = &str> { + self.as_str().split('/').filter(|seg| !seg.is_empty()) + } +} + +impl fmt::Debug for Peek { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(self.as_str(), f) + } +} + +/// Returns the full request path, irrespective of other filters. +/// +/// This will return a `FullPath`, which can be stringified to return the +/// full path of the request. +/// +/// This is more useful in generic pre/post-processing filters, and should +/// probably not be used for request matching/routing. +/// +/// # Example +/// +/// ``` +/// use warp::{Filter, path::FullPath}; +/// use std::{collections::HashMap, sync::{Arc, Mutex}}; +/// +/// let counts = Arc::new(Mutex::new(HashMap::new())); +/// let access_counter = warp::path::full() +/// .map(move |path: FullPath| { +/// let mut counts = counts.lock().unwrap(); +/// +/// *counts.entry(path.as_str().to_string()) +/// .and_modify(|c| *c += 1) +/// .or_insert(0) +/// }); +/// +/// let route = warp::path("foo") +/// .and(warp::path("bar")) +/// .and(access_counter) +/// .map(|count| { +/// format!("This is the {}th visit to this URL!", count) +/// }); +/// ``` +pub fn full() -> impl Filter<Extract = One<FullPath>, Error = Infallible> + Copy { + filter_fn(move |route| future::ok(one(FullPath(path_and_query(route))))) +} + +/// Represents the full request path, returned by the [`full()`] filter. +pub struct FullPath(PathAndQuery); + +impl FullPath { + /// Get the `&str` representation of the request path. + pub fn as_str(&self) -> &str { + self.0.path() + } +} + +impl fmt::Debug for FullPath { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(self.as_str(), f) + } +} + +fn filter_segment<F, U>(func: F) -> impl Filter<Extract = U, Error = Rejection> + Copy +where + F: Fn(&str) -> Result<U, Rejection> + Copy, + U: Tuple + Send + 'static, +{ + filter_fn(move |route| future::ready(with_segment(route, func))) +} + +fn with_segment<F, U>(route: &mut Route, func: F) -> Result<U, Rejection> +where + F: Fn(&str) -> Result<U, Rejection>, +{ + let seg = segment(route); + let ret = func(seg); + if ret.is_ok() { + let idx = seg.len(); + route.set_unmatched_path(idx); + } + ret +} + +fn segment(route: &Route) -> &str { + route + .path() + .splitn(2, '/') + .next() + .expect("split always has at least 1") +} + +fn path_and_query(route: &Route) -> PathAndQuery { + route + .uri() + .path_and_query() + .cloned() + .unwrap_or_else(|| PathAndQuery::from_static("")) +} + +/// Convenient way to chain multiple path filters together. +/// +/// Any number of either type identifiers or string expressions can be passed, +/// each separated by a forward slash (`/`). Strings will be used to match +/// path segments exactly, and type identifiers are used just like +/// [`param`](crate::path::param) filters. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // Match `/sum/:a/:b` +/// let route = warp::path!("sum" / u32 / u32) +/// .map(|a, b| { +/// format!("{} + {} = {}", a, b, a + b) +/// }); +/// ``` +/// +/// The equivalent filter chain without using the `path!` macro looks this: +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::path("sum") +/// .and(warp::path::param::<u32>()) +/// .and(warp::path::param::<u32>()) +/// .and(warp::path::end()) +/// .map(|a, b| { +/// format!("{} + {} = {}", a, b, a + b) +/// }); +/// ``` +/// +/// # Path Prefixes +/// +/// The `path!` macro automatically assumes the path should include an `end()` +/// filter. To build up a path filter *prefix*, such that the `end()` isn't +/// included, use the `/ ..` syntax. +/// +/// +/// ``` +/// use warp::Filter; +/// +/// let prefix = warp::path!("math" / "sum" / ..); +/// +/// let sum = warp::path!(u32 / u32) +/// .map(|a, b| { +/// format!("{} + {} = {}", a, b, a + b) +/// }); +/// +/// let help = warp::path::end() +/// .map(|| "This API returns the sum of two u32's"); +/// +/// let api = prefix.and(sum.or(help)); +/// ``` +#[macro_export] +macro_rules! path { + ($($pieces:tt)*) => ({ + $crate::__internal_path!(@start $($pieces)*) + }); +} + +#[doc(hidden)] +#[macro_export] +// not public API +macro_rules! __internal_path { + (@start) => ( + $crate::path::end() + ); + (@start ..) => ({ + compile_error!("'..' cannot be the only segment") + }); + (@start $first:tt $(/ $tail:tt)*) => ({ + $crate::__internal_path!(@munch $crate::any(); [$first] [$(/ $tail)*]) + }); + + (@munch $sum:expr; [$cur:tt] [/ $next:tt $(/ $tail:tt)*]) => ({ + $crate::__internal_path!(@munch $crate::Filter::and($sum, $crate::__internal_path!(@segment $cur)); [$next] [$(/ $tail)*]) + }); + (@munch $sum:expr; [$cur:tt] []) => ({ + $crate::__internal_path!(@last $sum; $cur) + }); + + (@last $sum:expr; ..) => ( + $sum + ); + (@last $sum:expr; $end:tt) => ( + $crate::Filter::and( + $crate::Filter::and($sum, $crate::__internal_path!(@segment $end)), + $crate::path::end() + ) + ); + + (@segment ..) => ( + compile_error!("'..' must be the last segment") + ); + (@segment $param:ty) => ( + $crate::path::param::<$param>() + ); + // Constructs a unique ZST so the &'static str pointer doesn't need to + // be carried around. + (@segment $s:literal) => ({ + #[derive(Clone, Copy)] + struct __StaticPath; + impl ::std::convert::AsRef<str> for __StaticPath { + fn as_ref(&self) -> &str { + static S: &str = $s; + S + } + } + $crate::path(__StaticPath) + }); +} + +// path! compile fail tests + +/// ```compile_fail +/// warp::path!("foo" / .. / "bar"); +/// ``` +/// +/// ```compile_fail +/// warp::path!(.. / "bar"); +/// ``` +/// +/// ```compile_fail +/// warp::path!("foo" ..); +/// ``` +/// +/// ```compile_fail +/// warp::path!("foo" / .. /); +/// ``` +/// +/// ```compile_fail +/// warp::path!(..); +/// ``` +fn _path_macro_compile_fail() {} + +mod internal { + // Used to prevent users from naming this type. + // + // For instance, `Exact<Opaque<String>>` means a user cannot depend + // on it being `Exact<String>`. + #[allow(missing_debug_implementations)] + #[derive(Clone, Copy)] + pub struct Opaque<T>(pub(super) T); + + impl<T: AsRef<str>> AsRef<str> for Opaque<T> { + #[inline] + fn as_ref(&self) -> &str { + self.0.as_ref() + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_path_exact_size() { + use std::mem::{size_of, size_of_val}; + + assert_eq!( + size_of_val(&path("hello")), + size_of::<&str>(), + "exact(&str) is size of &str" + ); + + assert_eq!( + size_of_val(&path(String::from("world"))), + size_of::<String>(), + "exact(String) is size of String" + ); + + assert_eq!( + size_of_val(&path!("zst")), + size_of::<()>(), + "path!(&str) is ZST" + ); + } +} diff --git a/third_party/rust/warp/src/filters/query.rs b/third_party/rust/warp/src/filters/query.rs new file mode 100644 index 0000000000..7aee569484 --- /dev/null +++ b/third_party/rust/warp/src/filters/query.rs @@ -0,0 +1,91 @@ +//! Query Filters + +use futures_util::future; +use serde::de::DeserializeOwned; +use serde_urlencoded; + +use crate::filter::{filter_fn_one, Filter, One}; +use crate::reject::{self, Rejection}; + +/// Creates a `Filter` that decodes query parameters to the type `T`. +/// +/// If cannot decode into a `T`, the request is rejected with a `400 Bad Request`. +/// +/// # Example +/// +/// ``` +/// use std::collections::HashMap; +/// use warp::{ +/// http::Response, +/// Filter, +/// }; +/// +/// let route = warp::any() +/// .and(warp::query::<HashMap<String, String>>()) +/// .map(|map: HashMap<String, String>| { +/// let mut response: Vec<String> = Vec::new(); +/// for (key, value) in map.into_iter() { +/// response.push(format!("{}={}", key, value)) +/// } +/// Response::builder().body(response.join(";")) +/// }); +/// ``` +/// +/// You can define your custom query object and deserialize with [Serde][Serde]. Ensure to include +/// the crate in your dependencies before usage. +/// +/// ``` +/// use serde_derive::{Deserialize, Serialize}; +/// use std::collections::HashMap; +/// use warp::{ +/// http::Response, +/// Filter, +/// }; +/// +/// #[derive(Serialize, Deserialize)] +/// struct FooQuery { +/// foo: Option<String>, +/// bar: u8, +/// } +/// +/// let route = warp::any() +/// .and(warp::query::<FooQuery>()) +/// .map(|q: FooQuery| { +/// if let Some(foo) = q.foo { +/// Response::builder().body(format!("foo={}", foo)) +/// } else { +/// Response::builder().body(format!("bar={}", q.bar)) +/// } +/// }); +/// ``` +/// +/// For more examples, please take a look at [examples/query_string.rs](https://github.com/seanmonstar/warp/blob/master/examples/query_string.rs). +/// +/// [Serde]: https://docs.rs/serde +pub fn query<T: DeserializeOwned + Send + 'static>( +) -> impl Filter<Extract = One<T>, Error = Rejection> + Copy { + filter_fn_one(|route| { + let query_string = route.query().unwrap_or_else(|| { + tracing::debug!("route was called without a query string, defaulting to empty"); + "" + }); + + let query_encoded = serde_urlencoded::from_str(query_string).map_err(|e| { + tracing::debug!("failed to decode query string '{}': {:?}", query_string, e); + reject::invalid_query() + }); + future::ready(query_encoded) + }) +} + +/// Creates a `Filter` that returns the raw query string as type String. +pub fn raw() -> impl Filter<Extract = One<String>, Error = Rejection> + Copy { + filter_fn_one(|route| { + let route = route + .query() + .map(|q| q.to_owned()) + .map(Ok) + .unwrap_or_else(|| Err(reject::invalid_query())); + future::ready(route) + }) +} diff --git a/third_party/rust/warp/src/filters/reply.rs b/third_party/rust/warp/src/filters/reply.rs new file mode 100644 index 0000000000..42b661d3e0 --- /dev/null +++ b/third_party/rust/warp/src/filters/reply.rs @@ -0,0 +1,257 @@ +//! Reply Filters +//! +//! These "filters" behave a little differently than the rest. Instead of +//! being used directly on requests, these filters "wrap" other filters. +//! +//! +//! ## Wrapping a `Filter` (`with`) +//! +//! ``` +//! use warp::Filter; +//! +//! let with_server = warp::reply::with::header("server", "warp"); +//! +//! let route = warp::any() +//! .map(warp::reply) +//! .with(with_server); +//! ``` +//! +//! Wrapping allows adding in conditional logic *before* the request enters +//! the inner filter (though the `with::header` wrapper does not). + +use std::convert::TryFrom; +use std::sync::Arc; + +use http::header::{HeaderMap, HeaderName, HeaderValue}; + +use self::sealed::{WithDefaultHeader_, WithHeader_, WithHeaders_}; +use crate::filter::{Filter, Map, WrapSealed}; +use crate::reply::Reply; + +/// Wrap a [`Filter`] that adds a header to the reply. +/// +/// # Note +/// +/// This **only** adds a header if the underlying filter is successful, and +/// returns a [`Reply`] If the underlying filter was rejected, the +/// header is not added. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // Always set `foo: bar` header. +/// let route = warp::any() +/// .map(warp::reply) +/// .with(warp::reply::with::header("foo", "bar")); +/// ``` +pub fn header<K, V>(name: K, value: V) -> WithHeader +where + HeaderName: TryFrom<K>, + <HeaderName as TryFrom<K>>::Error: Into<http::Error>, + HeaderValue: TryFrom<V>, + <HeaderValue as TryFrom<V>>::Error: Into<http::Error>, +{ + let (name, value) = assert_name_and_value(name, value); + WithHeader { name, value } +} + +/// Wrap a [`Filter`] that adds multiple headers to the reply. +/// +/// # Note +/// +/// This **only** adds a header if the underlying filter is successful, and +/// returns a [`Reply`] If the underlying filter was rejected, the +/// header is not added. +/// +/// # Example +/// +/// ``` +/// use warp::http::header::{HeaderMap, HeaderValue}; +/// use warp::Filter; +/// +/// let mut headers = HeaderMap::new(); +/// headers.insert("server", HeaderValue::from_static("wee/0")); +/// headers.insert("foo", HeaderValue::from_static("bar")); +/// +/// // Always set `server: wee/0` and `foo: bar` headers. +/// let route = warp::any() +/// .map(warp::reply) +/// .with(warp::reply::with::headers(headers)); +/// ``` +pub fn headers(headers: HeaderMap) -> WithHeaders { + WithHeaders { + headers: Arc::new(headers), + } +} + +// pub fn headers? + +/// Wrap a [`Filter`] that adds a header to the reply, if they +/// aren't already set. +/// +/// # Note +/// +/// This **only** adds a header if the underlying filter is successful, and +/// returns a [`Reply`] If the underlying filter was rejected, the +/// header is not added. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // Set `server: warp` if not already set. +/// let route = warp::any() +/// .map(warp::reply) +/// .with(warp::reply::with::default_header("server", "warp")); +/// ``` +pub fn default_header<K, V>(name: K, value: V) -> WithDefaultHeader +where + HeaderName: TryFrom<K>, + <HeaderName as TryFrom<K>>::Error: Into<http::Error>, + HeaderValue: TryFrom<V>, + <HeaderValue as TryFrom<V>>::Error: Into<http::Error>, +{ + let (name, value) = assert_name_and_value(name, value); + WithDefaultHeader { name, value } +} + +/// Wrap a `Filter` to always set a header. +#[derive(Clone, Debug)] +pub struct WithHeader { + name: HeaderName, + value: HeaderValue, +} + +impl<F, R> WrapSealed<F> for WithHeader +where + F: Filter<Extract = (R,)>, + R: Reply, +{ + type Wrapped = Map<F, WithHeader_>; + + fn wrap(&self, filter: F) -> Self::Wrapped { + let with = WithHeader_ { with: self.clone() }; + filter.map(with) + } +} + +/// Wrap a `Filter` to always set multiple headers. +#[derive(Clone, Debug)] +pub struct WithHeaders { + headers: Arc<HeaderMap>, +} + +impl<F, R> WrapSealed<F> for WithHeaders +where + F: Filter<Extract = (R,)>, + R: Reply, +{ + type Wrapped = Map<F, WithHeaders_>; + + fn wrap(&self, filter: F) -> Self::Wrapped { + let with = WithHeaders_ { with: self.clone() }; + filter.map(with) + } +} + +/// Wrap a `Filter` to set a header if it is not already set. +#[derive(Clone, Debug)] +pub struct WithDefaultHeader { + name: HeaderName, + value: HeaderValue, +} + +impl<F, R> WrapSealed<F> for WithDefaultHeader +where + F: Filter<Extract = (R,)>, + R: Reply, +{ + type Wrapped = Map<F, WithDefaultHeader_>; + + fn wrap(&self, filter: F) -> Self::Wrapped { + let with = WithDefaultHeader_ { with: self.clone() }; + filter.map(with) + } +} + +fn assert_name_and_value<K, V>(name: K, value: V) -> (HeaderName, HeaderValue) +where + HeaderName: TryFrom<K>, + <HeaderName as TryFrom<K>>::Error: Into<http::Error>, + HeaderValue: TryFrom<V>, + <HeaderValue as TryFrom<V>>::Error: Into<http::Error>, +{ + let name = <HeaderName as TryFrom<K>>::try_from(name) + .map_err(Into::into) + .unwrap_or_else(|_| panic!("invalid header name")); + + let value = <HeaderValue as TryFrom<V>>::try_from(value) + .map_err(Into::into) + .unwrap_or_else(|_| panic!("invalid header value")); + + (name, value) +} + +mod sealed { + use super::{WithDefaultHeader, WithHeader, WithHeaders}; + use crate::generic::{Func, One}; + use crate::reply::{Reply, Reply_}; + + #[derive(Clone)] + #[allow(missing_debug_implementations)] + pub struct WithHeader_ { + pub(super) with: WithHeader, + } + + impl<R: Reply> Func<One<R>> for WithHeader_ { + type Output = Reply_; + + fn call(&self, args: One<R>) -> Self::Output { + let mut resp = args.0.into_response(); + // Use "insert" to replace any set header... + resp.headers_mut() + .insert(&self.with.name, self.with.value.clone()); + Reply_(resp) + } + } + + #[derive(Clone)] + #[allow(missing_debug_implementations)] + pub struct WithHeaders_ { + pub(super) with: WithHeaders, + } + + impl<R: Reply> Func<One<R>> for WithHeaders_ { + type Output = Reply_; + + fn call(&self, args: One<R>) -> Self::Output { + let mut resp = args.0.into_response(); + for (name, value) in &*self.with.headers { + resp.headers_mut().insert(name, value.clone()); + } + Reply_(resp) + } + } + + #[derive(Clone)] + #[allow(missing_debug_implementations)] + pub struct WithDefaultHeader_ { + pub(super) with: WithDefaultHeader, + } + + impl<R: Reply> Func<One<R>> for WithDefaultHeader_ { + type Output = Reply_; + + fn call(&self, args: One<R>) -> Self::Output { + let mut resp = args.0.into_response(); + resp.headers_mut() + .entry(&self.with.name) + .or_insert_with(|| self.with.value.clone()); + + Reply_(resp) + } + } +} diff --git a/third_party/rust/warp/src/filters/sse.rs b/third_party/rust/warp/src/filters/sse.rs new file mode 100644 index 0000000000..413456aab2 --- /dev/null +++ b/third_party/rust/warp/src/filters/sse.rs @@ -0,0 +1,513 @@ +//! Server-Sent Events (SSE) +//! +//! # Example +//! +//! ``` +//! +//! use std::time::Duration; +//! use std::convert::Infallible; +//! use warp::{Filter, sse::Event}; +//! use futures_util::{stream::iter, Stream}; +//! +//! fn sse_events() -> impl Stream<Item = Result<Event, Infallible>> { +//! iter(vec![ +//! Ok(Event::default().data("unnamed event")), +//! Ok( +//! Event::default().event("chat") +//! .data("chat message") +//! ), +//! Ok( +//! Event::default().id(13.to_string()) +//! .event("chat") +//! .data("other chat message\nwith next line") +//! .retry(Duration::from_millis(5000)) +//! ) +//! ]) +//! } +//! +//! let app = warp::path("push-notifications") +//! .and(warp::get()) +//! .map(|| { +//! warp::sse::reply(warp::sse::keep_alive().stream(sse_events())) +//! }); +//! ``` +//! +//! Each field already is event which can be sent to client. +//! The events with multiple fields can be created by combining fields using tuples. +//! +//! See also the [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource) API, +//! which specifies the expected behavior of Server Sent Events. +//! + +#![allow(rustdoc::invalid_html_tags)] + +use serde::Serialize; +use std::borrow::Cow; +use std::error::Error as StdError; +use std::fmt::{self, Write}; +use std::future::Future; +use std::pin::Pin; +use std::str::FromStr; +use std::task::{Context, Poll}; +use std::time::Duration; + +use futures_util::{future, Stream, TryStream, TryStreamExt}; +use http::header::{HeaderValue, CACHE_CONTROL, CONTENT_TYPE}; +use hyper::Body; +use pin_project::pin_project; +use serde_json::{self, Error}; +use tokio::time::{self, Sleep}; + +use self::sealed::SseError; +use super::header; +use crate::filter::One; +use crate::reply::Response; +use crate::{Filter, Rejection, Reply}; + +// Server-sent event data type +#[derive(Debug)] +enum DataType { + Text(String), + Json(String), +} + +/// Server-sent event +#[derive(Default, Debug)] +pub struct Event { + id: Option<String>, + data: Option<DataType>, + event: Option<String>, + comment: Option<String>, + retry: Option<Duration>, +} + +impl Event { + /// Set Server-sent event data + /// data field(s) ("data:<content>") + pub fn data<T: Into<String>>(mut self, data: T) -> Event { + self.data = Some(DataType::Text(data.into())); + self + } + + /// Set Server-sent event data + /// data field(s) ("data:<content>") + pub fn json_data<T: Serialize>(mut self, data: T) -> Result<Event, Error> { + self.data = Some(DataType::Json(serde_json::to_string(&data)?)); + Ok(self) + } + + /// Set Server-sent event comment + /// Comment field (":<comment-text>") + pub fn comment<T: Into<String>>(mut self, comment: T) -> Event { + self.comment = Some(comment.into()); + self + } + + /// Set Server-sent event event + /// Event name field ("event:<event-name>") + pub fn event<T: Into<String>>(mut self, event: T) -> Event { + self.event = Some(event.into()); + self + } + + /// Set Server-sent event retry + /// Retry timeout field ("retry:<timeout>") + pub fn retry(mut self, duration: Duration) -> Event { + self.retry = Some(duration); + self + } + + /// Set Server-sent event id + /// Identifier field ("id:<identifier>") + pub fn id<T: Into<String>>(mut self, id: T) -> Event { + self.id = Some(id.into()); + self + } +} + +impl fmt::Display for Event { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + if let Some(ref comment) = &self.comment { + ":".fmt(f)?; + comment.fmt(f)?; + f.write_char('\n')?; + } + + if let Some(ref event) = &self.event { + "event:".fmt(f)?; + event.fmt(f)?; + f.write_char('\n')?; + } + + match self.data { + Some(DataType::Text(ref data)) => { + for line in data.split('\n') { + "data:".fmt(f)?; + line.fmt(f)?; + f.write_char('\n')?; + } + } + Some(DataType::Json(ref data)) => { + "data:".fmt(f)?; + data.fmt(f)?; + f.write_char('\n')?; + } + None => {} + } + + if let Some(ref id) = &self.id { + "id:".fmt(f)?; + id.fmt(f)?; + f.write_char('\n')?; + } + + if let Some(ref duration) = &self.retry { + "retry:".fmt(f)?; + + let secs = duration.as_secs(); + let millis = duration.subsec_millis(); + + if secs > 0 { + // format seconds + secs.fmt(f)?; + + // pad milliseconds + if millis < 10 { + f.write_str("00")?; + } else if millis < 100 { + f.write_char('0')?; + } + } + + // format milliseconds + millis.fmt(f)?; + + f.write_char('\n')?; + } + + f.write_char('\n')?; + Ok(()) + } +} + +/// Gets the optional last event id from request. +/// Typically this identifier represented as number or string. +/// +/// ``` +/// let app = warp::sse::last_event_id::<u32>(); +/// +/// // The identifier is present +/// async { +/// assert_eq!( +/// warp::test::request() +/// .header("Last-Event-ID", "12") +/// .filter(&app) +/// .await +/// .unwrap(), +/// Some(12) +/// ); +/// +/// // The identifier is missing +/// assert_eq!( +/// warp::test::request() +/// .filter(&app) +/// .await +/// .unwrap(), +/// None +/// ); +/// +/// // The identifier is not a valid +/// assert!( +/// warp::test::request() +/// .header("Last-Event-ID", "abc") +/// .filter(&app) +/// .await +/// .is_err(), +/// ); +///}; +/// ``` +pub fn last_event_id<T>() -> impl Filter<Extract = One<Option<T>>, Error = Rejection> + Copy +where + T: FromStr + Send + Sync + 'static, +{ + header::optional("last-event-id") +} + +/// Server-sent events reply +/// +/// This function converts stream of server events into a `Reply` with: +/// +/// - Status of `200 OK` +/// - Header `content-type: text/event-stream` +/// - Header `cache-control: no-cache`. +/// +/// # Example +/// +/// ``` +/// +/// use std::time::Duration; +/// use futures_util::Stream; +/// use futures_util::stream::iter; +/// use std::convert::Infallible; +/// use warp::{Filter, sse::Event}; +/// use serde_derive::Serialize; +/// +/// #[derive(Serialize)] +/// struct Msg { +/// from: u32, +/// text: String, +/// } +/// +/// fn event_stream() -> impl Stream<Item = Result<Event, Infallible>> { +/// iter(vec![ +/// // Unnamed event with data only +/// Ok(Event::default().data("payload")), +/// // Named event with ID and retry timeout +/// Ok( +/// Event::default().data("other message\nwith next line") +/// .event("chat") +/// .id(1.to_string()) +/// .retry(Duration::from_millis(15000)) +/// ), +/// // Event with JSON data +/// Ok( +/// Event::default().id(2.to_string()) +/// .json_data(Msg { +/// from: 2, +/// text: "hello".into(), +/// }).unwrap(), +/// ) +/// ]) +/// } +/// +/// async { +/// let app = warp::path("sse").and(warp::get()).map(|| { +/// warp::sse::reply(event_stream()) +/// }); +/// +/// let res = warp::test::request() +/// .method("GET") +/// .header("Connection", "Keep-Alive") +/// .path("/sse") +/// .reply(&app) +/// .await +/// .into_body(); +/// +/// assert_eq!( +/// res, +/// r#"data:payload +/// +/// event:chat +/// data:other message +/// data:with next line +/// id:1 +/// retry:15000 +/// +/// data:{"from":2,"text":"hello"} +/// id:2 +/// +/// "# +/// ); +/// }; +/// ``` +pub fn reply<S>(event_stream: S) -> impl Reply +where + S: TryStream<Ok = Event> + Send + 'static, + S::Error: StdError + Send + Sync + 'static, +{ + SseReply { event_stream } +} + +#[allow(missing_debug_implementations)] +struct SseReply<S> { + event_stream: S, +} + +impl<S> Reply for SseReply<S> +where + S: TryStream<Ok = Event> + Send + 'static, + S::Error: StdError + Send + Sync + 'static, +{ + #[inline] + fn into_response(self) -> Response { + let body_stream = self + .event_stream + .map_err(|error| { + // FIXME: error logging + log::error!("sse stream error: {}", error); + SseError + }) + .into_stream() + .and_then(|event| future::ready(Ok(event.to_string()))); + + let mut res = Response::new(Body::wrap_stream(body_stream)); + // Set appropriate content type + res.headers_mut() + .insert(CONTENT_TYPE, HeaderValue::from_static("text/event-stream")); + // Disable response body caching + res.headers_mut() + .insert(CACHE_CONTROL, HeaderValue::from_static("no-cache")); + res + } +} + +/// Configure the interval between keep-alive messages, the content +/// of each message, and the associated stream. +#[derive(Debug)] +pub struct KeepAlive { + comment_text: Cow<'static, str>, + max_interval: Duration, +} + +impl KeepAlive { + /// Customize the interval between keep-alive messages. + /// + /// Default is 15 seconds. + pub fn interval(mut self, time: Duration) -> Self { + self.max_interval = time; + self + } + + /// Customize the text of the keep-alive message. + /// + /// Default is an empty comment. + pub fn text(mut self, text: impl Into<Cow<'static, str>>) -> Self { + self.comment_text = text.into(); + self + } + + /// Wrap an event stream with keep-alive functionality. + /// + /// See [`keep_alive`] for more. + pub fn stream<S>( + self, + event_stream: S, + ) -> impl TryStream<Ok = Event, Error = impl StdError + Send + Sync + 'static> + Send + 'static + where + S: TryStream<Ok = Event> + Send + 'static, + S::Error: StdError + Send + Sync + 'static, + { + let alive_timer = time::sleep(self.max_interval); + SseKeepAlive { + event_stream, + comment_text: self.comment_text, + max_interval: self.max_interval, + alive_timer, + } + } +} + +#[allow(missing_debug_implementations)] +#[pin_project] +struct SseKeepAlive<S> { + #[pin] + event_stream: S, + comment_text: Cow<'static, str>, + max_interval: Duration, + #[pin] + alive_timer: Sleep, +} + +/// Keeps event source connection alive when no events sent over a some time. +/// +/// Some proxy servers may drop HTTP connection after a some timeout of inactivity. +/// This function helps to prevent such behavior by sending comment events every +/// `keep_interval` of inactivity. +/// +/// By default the comment is `:` (an empty comment) and the time interval between +/// events is 15 seconds. Both may be customized using the builder pattern +/// as shown below. +/// +/// ``` +/// use std::time::Duration; +/// use std::convert::Infallible; +/// use futures_util::StreamExt; +/// use tokio::time::interval; +/// use tokio_stream::wrappers::IntervalStream; +/// use warp::{Filter, Stream, sse::Event}; +/// +/// // create server-sent event +/// fn sse_counter(counter: u64) -> Result<Event, Infallible> { +/// Ok(Event::default().data(counter.to_string())) +/// } +/// +/// fn main() { +/// let routes = warp::path("ticks") +/// .and(warp::get()) +/// .map(|| { +/// let mut counter: u64 = 0; +/// let interval = interval(Duration::from_secs(15)); +/// let stream = IntervalStream::new(interval); +/// let event_stream = stream.map(move |_| { +/// counter += 1; +/// sse_counter(counter) +/// }); +/// // reply using server-sent events +/// let stream = warp::sse::keep_alive() +/// .interval(Duration::from_secs(5)) +/// .text("thump".to_string()) +/// .stream(event_stream); +/// warp::sse::reply(stream) +/// }); +/// } +/// ``` +/// +/// See [notes](https://www.w3.org/TR/2009/WD-eventsource-20090421/#notes). +pub fn keep_alive() -> KeepAlive { + KeepAlive { + comment_text: Cow::Borrowed(""), + max_interval: Duration::from_secs(15), + } +} + +impl<S> Stream for SseKeepAlive<S> +where + S: TryStream<Ok = Event> + Send + 'static, + S::Error: StdError + Send + Sync + 'static, +{ + type Item = Result<Event, SseError>; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + let mut pin = self.project(); + match pin.event_stream.try_poll_next(cx) { + Poll::Pending => match Pin::new(&mut pin.alive_timer).poll(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(_) => { + // restart timer + pin.alive_timer + .reset(tokio::time::Instant::now() + *pin.max_interval); + let comment_str = pin.comment_text.clone(); + let event = Event::default().comment(comment_str); + Poll::Ready(Some(Ok(event))) + } + }, + Poll::Ready(Some(Ok(event))) => { + // restart timer + pin.alive_timer + .reset(tokio::time::Instant::now() + *pin.max_interval); + Poll::Ready(Some(Ok(event))) + } + Poll::Ready(None) => Poll::Ready(None), + Poll::Ready(Some(Err(error))) => { + log::error!("sse::keep error: {}", error); + Poll::Ready(Some(Err(SseError))) + } + } + } +} + +mod sealed { + use super::*; + + /// SSE error type + #[derive(Debug)] + pub struct SseError; + + impl fmt::Display for SseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "sse error") + } + } + + impl StdError for SseError {} +} diff --git a/third_party/rust/warp/src/filters/trace.rs b/third_party/rust/warp/src/filters/trace.rs new file mode 100644 index 0000000000..60686b430f --- /dev/null +++ b/third_party/rust/warp/src/filters/trace.rs @@ -0,0 +1,309 @@ +//! [`tracing`] filters. +//! +//! [`tracing`] is a framework for instrumenting Rust programs to +//! collect scoped, structured, and async-aware diagnostics. This module +//! provides a set of filters for instrumenting Warp applications with `tracing` +//! spans. [`Spans`] can be used to associate individual events with a request, +//! and track contexts through the application. +//! +//! [`tracing`]: https://crates.io/crates/tracing +//! [`Spans`]: https://docs.rs/tracing/latest/tracing/#spans +use tracing::Span; + +use std::net::SocketAddr; + +use http::{self, header}; + +use crate::filter::{Filter, WrapSealed}; +use crate::reject::IsReject; +use crate::reply::Reply; +use crate::route::Route; + +use self::internal::WithTrace; + +/// Create a wrapping filter that instruments every request with a `tracing` +/// [`Span`] at the [`INFO`] level, containing a summary of the request. +/// Additionally, if the [`DEBUG`] level is enabled, the span will contain an +/// event recording the request's headers. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::any() +/// .map(warp::reply) +/// .with(warp::trace::request()); +/// ``` +/// +/// [`Span`]: https://docs.rs/tracing/latest/tracing/#spans +/// [`INFO`]: https://docs.rs/tracing/0.1.16/tracing/struct.Level.html#associatedconstant.INFO +/// [`DEBUG`]: https://docs.rs/tracing/0.1.16/tracing/struct.Level.html#associatedconstant.DEBUG +pub fn request() -> Trace<impl Fn(Info<'_>) -> Span + Clone> { + use tracing::field::{display, Empty}; + trace(|info: Info<'_>| { + let span = tracing::info_span!( + "request", + remote.addr = Empty, + method = %info.method(), + path = %info.path(), + version = ?info.route.version(), + referer = Empty, + ); + + // Record optional fields. + if let Some(remote_addr) = info.remote_addr() { + span.record("remote.addr", &display(remote_addr)); + } + + if let Some(referer) = info.referer() { + span.record("referer", &display(referer)); + } + + tracing::debug!(parent: &span, "received request"); + + span + }) +} + +/// Create a wrapping filter that instruments every request with a custom +/// `tracing` [`Span`] provided by a function. +/// +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::any() +/// .map(warp::reply) +/// .with(warp::trace(|info| { +/// // Create a span using tracing macros +/// tracing::info_span!( +/// "request", +/// method = %info.method(), +/// path = %info.path(), +/// ) +/// })); +/// ``` +/// +/// [`Span`]: https://docs.rs/tracing/latest/tracing/#spans +pub fn trace<F>(func: F) -> Trace<F> +where + F: Fn(Info<'_>) -> Span + Clone, +{ + Trace { func } +} + +/// Create a wrapping filter that instruments every request with a `tracing` +/// [`Span`] at the [`DEBUG`] level representing a named context. +/// +/// This can be used to instrument multiple routes with their own sub-spans in a +/// per-request trace. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let hello = warp::path("hello") +/// .map(warp::reply) +/// .with(warp::trace::named("hello")); +/// +/// let goodbye = warp::path("goodbye") +/// .map(warp::reply) +/// .with(warp::trace::named("goodbye")); +/// +/// let routes = hello.or(goodbye); +/// ``` +/// +/// [`Span`]: https://docs.rs/tracing/latest/tracing/#spans +/// [`DEBUG`]: https://docs.rs/tracing/0.1.16/tracing/struct.Level.html#associatedconstant.DEBUG +pub fn named(name: &'static str) -> Trace<impl Fn(Info<'_>) -> Span + Copy> { + trace(move |_| tracing::debug_span!("context", "{}", name,)) +} + +/// Decorates a [`Filter`] to create a [`tracing`] [span] for +/// requests and responses. +/// +/// [`tracing`]: https://crates.io/crates/tracing +/// [span]: https://docs.rs/tracing/latest/tracing/#spans +#[derive(Clone, Copy, Debug)] +pub struct Trace<F> { + func: F, +} + +/// Information about the request/response that can be used to prepare log lines. +#[allow(missing_debug_implementations)] +pub struct Info<'a> { + route: &'a Route, +} + +impl<FN, F> WrapSealed<F> for Trace<FN> +where + FN: Fn(Info<'_>) -> Span + Clone + Send, + F: Filter + Clone + Send, + F::Extract: Reply, + F::Error: IsReject, +{ + type Wrapped = WithTrace<FN, F>; + + fn wrap(&self, filter: F) -> Self::Wrapped { + WithTrace { + filter, + trace: self.clone(), + } + } +} + +impl<'a> Info<'a> { + /// View the remote `SocketAddr` of the request. + pub fn remote_addr(&self) -> Option<SocketAddr> { + self.route.remote_addr() + } + + /// View the `http::Method` of the request. + pub fn method(&self) -> &http::Method { + self.route.method() + } + + /// View the URI path of the request. + pub fn path(&self) -> &str { + self.route.full_path() + } + + /// View the `http::Version` of the request. + pub fn version(&self) -> http::Version { + self.route.version() + } + + /// View the referer of the request. + pub fn referer(&self) -> Option<&str> { + self.route + .headers() + .get(header::REFERER) + .and_then(|v| v.to_str().ok()) + } + + /// View the user agent of the request. + pub fn user_agent(&self) -> Option<&str> { + self.route + .headers() + .get(header::USER_AGENT) + .and_then(|v| v.to_str().ok()) + } + + /// View the host of the request + pub fn host(&self) -> Option<&str> { + self.route + .headers() + .get(header::HOST) + .and_then(|v| v.to_str().ok()) + } + + /// View the request headers. + pub fn request_headers(&self) -> &http::HeaderMap { + self.route.headers() + } +} + +mod internal { + use futures_util::{future::Inspect, future::MapOk, FutureExt, TryFutureExt}; + + use super::{Info, Trace}; + use crate::filter::{Filter, FilterBase, Internal}; + use crate::reject::IsReject; + use crate::reply::Reply; + use crate::reply::Response; + use crate::route; + + #[allow(missing_debug_implementations)] + pub struct Traced(pub(super) Response); + + impl Reply for Traced { + #[inline] + fn into_response(self) -> Response { + self.0 + } + } + + #[allow(missing_debug_implementations)] + #[derive(Clone, Copy)] + pub struct WithTrace<FN, F> { + pub(super) filter: F, + pub(super) trace: Trace<FN>, + } + + use tracing::instrument::{Instrument, Instrumented}; + use tracing::Span; + + fn finished_logger<E: IsReject>(reply: &Result<(Traced,), E>) { + let (status, error) = match reply { + Ok((Traced(resp),)) => (resp.status(), None), + Err(error) => (error.status(), Some(error)), + }; + + if status.is_success() { + tracing::info!( + target: "warp::filters::trace", + status = status.as_u16(), + "finished processing with success" + ); + } else if status.is_server_error() { + tracing::error!( + target: "warp::filters::trace", + status = status.as_u16(), + error = ?error, + "unable to process request (internal error)" + ); + } else if status.is_client_error() { + tracing::warn!( + target: "warp::filters::trace", + status = status.as_u16(), + error = ?error, + "unable to serve request (client error)" + ); + } else { + // Either informational or redirect + tracing::info!( + target: "warp::filters::trace", + status = status.as_u16(), + error = ?error, + "finished processing with status" + ); + } + } + + fn convert_reply<R: Reply>(reply: R) -> (Traced,) { + (Traced(reply.into_response()),) + } + + impl<FN, F> FilterBase for WithTrace<FN, F> + where + FN: Fn(Info<'_>) -> Span + Clone + Send, + F: Filter + Clone + Send, + F::Extract: Reply, + F::Error: IsReject, + { + type Extract = (Traced,); + type Error = F::Error; + type Future = Instrumented< + Inspect< + MapOk<F::Future, fn(F::Extract) -> Self::Extract>, + fn(&Result<Self::Extract, F::Error>), + >, + >; + + fn filter(&self, _: Internal) -> Self::Future { + let span = route::with(|route| (self.trace.func)(Info { route })); + let _entered = span.enter(); + + tracing::info!(target: "warp::filters::trace", "processing request"); + self.filter + .filter(Internal) + .map_ok(convert_reply as fn(F::Extract) -> Self::Extract) + .inspect(finished_logger as fn(&Result<Self::Extract, F::Error>)) + .instrument(span.clone()) + } + } +} diff --git a/third_party/rust/warp/src/filters/ws.rs b/third_party/rust/warp/src/filters/ws.rs new file mode 100644 index 0000000000..81a6273cbd --- /dev/null +++ b/third_party/rust/warp/src/filters/ws.rs @@ -0,0 +1,420 @@ +//! Websockets Filters + +use std::borrow::Cow; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use super::header; +use crate::filter::{filter_fn_one, Filter, One}; +use crate::reject::Rejection; +use crate::reply::{Reply, Response}; +use futures_util::{future, ready, FutureExt, Sink, Stream, TryFutureExt}; +use headers::{Connection, HeaderMapExt, SecWebsocketAccept, SecWebsocketKey, Upgrade}; +use http; +use hyper::upgrade::OnUpgrade; +use tokio_tungstenite::{ + tungstenite::protocol::{self, WebSocketConfig}, + WebSocketStream, +}; + +/// Creates a Websocket Filter. +/// +/// The yielded `Ws` is used to finish the websocket upgrade. +/// +/// # Note +/// +/// This filter combines multiple filters internally, so you don't need them: +/// +/// - Method must be `GET` +/// - Header `connection` must be `upgrade` +/// - Header `upgrade` must be `websocket` +/// - Header `sec-websocket-version` must be `13` +/// - Header `sec-websocket-key` must be set. +/// +/// If the filters are met, yields a `Ws`. Calling `Ws::on_upgrade` will +/// return a reply with: +/// +/// - Status of `101 Switching Protocols` +/// - Header `connection: upgrade` +/// - Header `upgrade: websocket` +/// - Header `sec-websocket-accept` with the hash value of the received key. +pub fn ws() -> impl Filter<Extract = One<Ws>, Error = Rejection> + Copy { + let connection_has_upgrade = header::header2() + .and_then(|conn: ::headers::Connection| { + if conn.contains("upgrade") { + future::ok(()) + } else { + future::err(crate::reject::known(MissingConnectionUpgrade)) + } + }) + .untuple_one(); + + crate::get() + .and(connection_has_upgrade) + .and(header::exact_ignore_case("upgrade", "websocket")) + .and(header::exact("sec-websocket-version", "13")) + //.and(header::exact2(Upgrade::websocket())) + //.and(header::exact2(SecWebsocketVersion::V13)) + .and(header::header2::<SecWebsocketKey>()) + .and(on_upgrade()) + .map( + move |key: SecWebsocketKey, on_upgrade: Option<OnUpgrade>| Ws { + config: None, + key, + on_upgrade, + }, + ) +} + +/// Extracted by the [`ws`] filter, and used to finish an upgrade. +pub struct Ws { + config: Option<WebSocketConfig>, + key: SecWebsocketKey, + on_upgrade: Option<OnUpgrade>, +} + +impl Ws { + /// Finish the upgrade, passing a function to handle the `WebSocket`. + /// + /// The passed function must return a `Future`. + pub fn on_upgrade<F, U>(self, func: F) -> impl Reply + where + F: FnOnce(WebSocket) -> U + Send + 'static, + U: Future<Output = ()> + Send + 'static, + { + WsReply { + ws: self, + on_upgrade: func, + } + } + + // config + + /// Does nothing. + /// + /// # Deprecated + /// + /// Use `max_write_buffer_size()` instead. + #[deprecated = "use max_write_buffer_size instead"] + pub fn max_send_queue(self, _max: usize) -> Self { + self + } + + /// The max size of the write buffer, in bytes. + pub fn max_write_buffer_size(mut self, max: usize) -> Self { + self.config + .get_or_insert_with(WebSocketConfig::default) + .max_write_buffer_size = max; + self + } + + /// Set the maximum message size (defaults to 64 megabytes) + pub fn max_message_size(mut self, max: usize) -> Self { + self.config + .get_or_insert_with(WebSocketConfig::default) + .max_message_size = Some(max); + self + } + + /// Set the maximum frame size (defaults to 16 megabytes) + pub fn max_frame_size(mut self, max: usize) -> Self { + self.config + .get_or_insert_with(WebSocketConfig::default) + .max_frame_size = Some(max); + self + } +} + +impl fmt::Debug for Ws { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Ws").finish() + } +} + +#[allow(missing_debug_implementations)] +struct WsReply<F> { + ws: Ws, + on_upgrade: F, +} + +impl<F, U> Reply for WsReply<F> +where + F: FnOnce(WebSocket) -> U + Send + 'static, + U: Future<Output = ()> + Send + 'static, +{ + fn into_response(self) -> Response { + if let Some(on_upgrade) = self.ws.on_upgrade { + let on_upgrade_cb = self.on_upgrade; + let config = self.ws.config; + let fut = on_upgrade + .and_then(move |upgraded| { + tracing::trace!("websocket upgrade complete"); + WebSocket::from_raw_socket(upgraded, protocol::Role::Server, config).map(Ok) + }) + .and_then(move |socket| on_upgrade_cb(socket).map(Ok)) + .map(|result| { + if let Err(err) = result { + tracing::debug!("ws upgrade error: {}", err); + } + }); + ::tokio::task::spawn(fut); + } else { + tracing::debug!("ws couldn't be upgraded since no upgrade state was present"); + } + + let mut res = http::Response::default(); + + *res.status_mut() = http::StatusCode::SWITCHING_PROTOCOLS; + + res.headers_mut().typed_insert(Connection::upgrade()); + res.headers_mut().typed_insert(Upgrade::websocket()); + res.headers_mut() + .typed_insert(SecWebsocketAccept::from(self.ws.key)); + + res + } +} + +// Extracts OnUpgrade state from the route. +fn on_upgrade() -> impl Filter<Extract = (Option<OnUpgrade>,), Error = Rejection> + Copy { + filter_fn_one(|route| future::ready(Ok(route.extensions_mut().remove::<OnUpgrade>()))) +} + +/// A websocket `Stream` and `Sink`, provided to `ws` filters. +/// +/// Ping messages sent from the client will be handled internally by replying with a Pong message. +/// Close messages need to be handled explicitly: usually by closing the `Sink` end of the +/// `WebSocket`. +/// +/// **Note!** +/// Due to rust futures nature, pings won't be handled until read part of `WebSocket` is polled + +pub struct WebSocket { + inner: WebSocketStream<hyper::upgrade::Upgraded>, +} + +impl WebSocket { + pub(crate) async fn from_raw_socket( + upgraded: hyper::upgrade::Upgraded, + role: protocol::Role, + config: Option<protocol::WebSocketConfig>, + ) -> Self { + WebSocketStream::from_raw_socket(upgraded, role, config) + .map(|inner| WebSocket { inner }) + .await + } + + /// Gracefully close this websocket. + pub async fn close(mut self) -> Result<(), crate::Error> { + future::poll_fn(|cx| Pin::new(&mut self).poll_close(cx)).await + } +} + +impl Stream for WebSocket { + type Item = Result<Message, crate::Error>; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + match ready!(Pin::new(&mut self.inner).poll_next(cx)) { + Some(Ok(item)) => Poll::Ready(Some(Ok(Message { inner: item }))), + Some(Err(e)) => { + tracing::debug!("websocket poll error: {}", e); + Poll::Ready(Some(Err(crate::Error::new(e)))) + } + None => { + tracing::trace!("websocket closed"); + Poll::Ready(None) + } + } + } +} + +impl Sink<Message> for WebSocket { + type Error = crate::Error; + + fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { + match ready!(Pin::new(&mut self.inner).poll_ready(cx)) { + Ok(()) => Poll::Ready(Ok(())), + Err(e) => Poll::Ready(Err(crate::Error::new(e))), + } + } + + fn start_send(mut self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error> { + match Pin::new(&mut self.inner).start_send(item.inner) { + Ok(()) => Ok(()), + Err(e) => { + tracing::debug!("websocket start_send error: {}", e); + Err(crate::Error::new(e)) + } + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { + match ready!(Pin::new(&mut self.inner).poll_flush(cx)) { + Ok(()) => Poll::Ready(Ok(())), + Err(e) => Poll::Ready(Err(crate::Error::new(e))), + } + } + + fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { + match ready!(Pin::new(&mut self.inner).poll_close(cx)) { + Ok(()) => Poll::Ready(Ok(())), + Err(err) => { + tracing::debug!("websocket close error: {}", err); + Poll::Ready(Err(crate::Error::new(err))) + } + } + } +} + +impl fmt::Debug for WebSocket { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("WebSocket").finish() + } +} + +/// A WebSocket message. +/// +/// This will likely become a `non-exhaustive` enum in the future, once that +/// language feature has stabilized. +#[derive(Eq, PartialEq, Clone)] +pub struct Message { + inner: protocol::Message, +} + +impl Message { + /// Construct a new Text `Message`. + pub fn text<S: Into<String>>(s: S) -> Message { + Message { + inner: protocol::Message::text(s), + } + } + + /// Construct a new Binary `Message`. + pub fn binary<V: Into<Vec<u8>>>(v: V) -> Message { + Message { + inner: protocol::Message::binary(v), + } + } + + /// Construct a new Ping `Message`. + pub fn ping<V: Into<Vec<u8>>>(v: V) -> Message { + Message { + inner: protocol::Message::Ping(v.into()), + } + } + + /// Construct a new Pong `Message`. + /// + /// Note that one rarely needs to manually construct a Pong message because the underlying tungstenite socket + /// automatically responds to the Ping messages it receives. Manual construction might still be useful in some cases + /// like in tests or to send unidirectional heartbeats. + pub fn pong<V: Into<Vec<u8>>>(v: V) -> Message { + Message { + inner: protocol::Message::Pong(v.into()), + } + } + + /// Construct the default Close `Message`. + pub fn close() -> Message { + Message { + inner: protocol::Message::Close(None), + } + } + + /// Construct a Close `Message` with a code and reason. + pub fn close_with(code: impl Into<u16>, reason: impl Into<Cow<'static, str>>) -> Message { + Message { + inner: protocol::Message::Close(Some(protocol::frame::CloseFrame { + code: protocol::frame::coding::CloseCode::from(code.into()), + reason: reason.into(), + })), + } + } + + /// Returns true if this message is a Text message. + pub fn is_text(&self) -> bool { + self.inner.is_text() + } + + /// Returns true if this message is a Binary message. + pub fn is_binary(&self) -> bool { + self.inner.is_binary() + } + + /// Returns true if this message a is a Close message. + pub fn is_close(&self) -> bool { + self.inner.is_close() + } + + /// Returns true if this message is a Ping message. + pub fn is_ping(&self) -> bool { + self.inner.is_ping() + } + + /// Returns true if this message is a Pong message. + pub fn is_pong(&self) -> bool { + self.inner.is_pong() + } + + /// Try to get the close frame (close code and reason) + pub fn close_frame(&self) -> Option<(u16, &str)> { + if let protocol::Message::Close(Some(ref close_frame)) = self.inner { + Some((close_frame.code.into(), close_frame.reason.as_ref())) + } else { + None + } + } + + /// Try to get a reference to the string text, if this is a Text message. + pub fn to_str(&self) -> Result<&str, ()> { + match self.inner { + protocol::Message::Text(ref s) => Ok(s), + _ => Err(()), + } + } + + /// Return the bytes of this message, if the message can contain data. + pub fn as_bytes(&self) -> &[u8] { + match self.inner { + protocol::Message::Text(ref s) => s.as_bytes(), + protocol::Message::Binary(ref v) => v, + protocol::Message::Ping(ref v) => v, + protocol::Message::Pong(ref v) => v, + protocol::Message::Close(_) => &[], + protocol::Message::Frame(ref frame) => frame.payload(), + } + } + + /// Destructure this message into binary data. + pub fn into_bytes(self) -> Vec<u8> { + self.inner.into_data() + } +} + +impl fmt::Debug for Message { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(&self.inner, f) + } +} + +impl From<Message> for Vec<u8> { + fn from(m: Message) -> Self { + m.into_bytes() + } +} + +// ===== Rejections ===== + +/// Connection header did not include 'upgrade' +#[derive(Debug)] +pub struct MissingConnectionUpgrade; + +impl fmt::Display for MissingConnectionUpgrade { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Connection header did not include 'upgrade'") + } +} + +impl ::std::error::Error for MissingConnectionUpgrade {} diff --git a/third_party/rust/warp/src/generic.rs b/third_party/rust/warp/src/generic.rs new file mode 100644 index 0000000000..5350d22156 --- /dev/null +++ b/third_party/rust/warp/src/generic.rs @@ -0,0 +1,250 @@ +#[derive(Debug)] +pub struct Product<H, T: HList>(pub(crate) H, pub(crate) T); + +pub type One<T> = (T,); + +#[inline] +pub(crate) fn one<T>(val: T) -> One<T> { + (val,) +} + +#[derive(Debug)] +pub enum Either<T, U> { + A(T), + B(U), +} + +// Converts Product (and ()) into tuples. +pub trait HList: Sized { + type Tuple: Tuple<HList = Self>; + + fn flatten(self) -> Self::Tuple; +} + +// Typeclass that tuples can be converted into a Product (or unit ()). +pub trait Tuple: Sized { + type HList: HList<Tuple = Self>; + + fn hlist(self) -> Self::HList; + + #[inline] + fn combine<T>(self, other: T) -> CombinedTuples<Self, T> + where + Self: Sized, + T: Tuple, + Self::HList: Combine<T::HList>, + { + self.hlist().combine(other.hlist()).flatten() + } +} + +pub type CombinedTuples<T, U> = + <<<T as Tuple>::HList as Combine<<U as Tuple>::HList>>::Output as HList>::Tuple; + +// Combines Product together. +pub trait Combine<T: HList> { + type Output: HList; + + fn combine(self, other: T) -> Self::Output; +} + +pub trait Func<Args> { + type Output; + + fn call(&self, args: Args) -> Self::Output; +} + +// ===== impl Combine ===== + +impl<T: HList> Combine<T> for () { + type Output = T; + #[inline] + fn combine(self, other: T) -> Self::Output { + other + } +} + +impl<H, T: HList, U: HList> Combine<U> for Product<H, T> +where + T: Combine<U>, + Product<H, <T as Combine<U>>::Output>: HList, +{ + type Output = Product<H, <T as Combine<U>>::Output>; + + #[inline] + fn combine(self, other: U) -> Self::Output { + Product(self.0, self.1.combine(other)) + } +} + +impl HList for () { + type Tuple = (); + #[inline] + fn flatten(self) -> Self::Tuple {} +} + +impl Tuple for () { + type HList = (); + + #[inline] + fn hlist(self) -> Self::HList {} +} + +impl<F, R> Func<()> for F +where + F: Fn() -> R, +{ + type Output = R; + + #[inline] + fn call(&self, _args: ()) -> Self::Output { + (*self)() + } +} + +impl<F, R> Func<crate::Rejection> for F +where + F: Fn(crate::Rejection) -> R, +{ + type Output = R; + + #[inline] + fn call(&self, arg: crate::Rejection) -> Self::Output { + (*self)(arg) + } +} + +macro_rules! product { + ($H:expr) => { Product($H, ()) }; + ($H:expr, $($T:expr),*) => { Product($H, product!($($T),*)) }; +} + +macro_rules! Product { + ($H:ty) => { Product<$H, ()> }; + ($H:ty, $($T:ty),*) => { Product<$H, Product!($($T),*)> }; +} + +macro_rules! product_pat { + ($H:pat) => { Product($H, ()) }; + ($H:pat, $($T:pat),*) => { Product($H, product_pat!($($T),*)) }; +} + +macro_rules! generics { + ($type:ident) => { + impl<$type> HList for Product!($type) { + type Tuple = ($type,); + + #[inline] + fn flatten(self) -> Self::Tuple { + (self.0,) + } + } + + impl<$type> Tuple for ($type,) { + type HList = Product!($type); + #[inline] + fn hlist(self) -> Self::HList { + product!(self.0) + } + } + + impl<F, R, $type> Func<Product!($type)> for F + where + F: Fn($type) -> R, + { + type Output = R; + + #[inline] + fn call(&self, args: Product!($type)) -> Self::Output { + (*self)(args.0) + } + + } + + impl<F, R, $type> Func<($type,)> for F + where + F: Fn($type) -> R, + { + type Output = R; + + #[inline] + fn call(&self, args: ($type,)) -> Self::Output { + (*self)(args.0) + } + } + + }; + + ($type1:ident, $( $type:ident ),*) => { + generics!($( $type ),*); + + impl<$type1, $( $type ),*> HList for Product!($type1, $($type),*) { + type Tuple = ($type1, $( $type ),*); + + #[inline] + fn flatten(self) -> Self::Tuple { + #[allow(non_snake_case)] + let product_pat!($type1, $( $type ),*) = self; + ($type1, $( $type ),*) + } + } + + impl<$type1, $( $type ),*> Tuple for ($type1, $($type),*) { + type HList = Product!($type1, $( $type ),*); + + #[inline] + fn hlist(self) -> Self::HList { + #[allow(non_snake_case)] + let ($type1, $( $type ),*) = self; + product!($type1, $( $type ),*) + } + } + + impl<F, R, $type1, $( $type ),*> Func<Product!($type1, $($type),*)> for F + where + F: Fn($type1, $( $type ),*) -> R, + { + type Output = R; + + #[inline] + fn call(&self, args: Product!($type1, $($type),*)) -> Self::Output { + #[allow(non_snake_case)] + let product_pat!($type1, $( $type ),*) = args; + (*self)($type1, $( $type ),*) + } + } + + impl<F, R, $type1, $( $type ),*> Func<($type1, $($type),*)> for F + where + F: Fn($type1, $( $type ),*) -> R, + { + type Output = R; + + #[inline] + fn call(&self, args: ($type1, $($type),*)) -> Self::Output { + #[allow(non_snake_case)] + let ($type1, $( $type ),*) = args; + (*self)($type1, $( $type ),*) + } + } + }; +} + +generics! { + T1, + T2, + T3, + T4, + T5, + T6, + T7, + T8, + T9, + T10, + T11, + T12, + T13, + T14, + T15, + T16 +} diff --git a/third_party/rust/warp/src/lib.rs b/third_party/rust/warp/src/lib.rs new file mode 100644 index 0000000000..35ed0dcd27 --- /dev/null +++ b/third_party/rust/warp/src/lib.rs @@ -0,0 +1,179 @@ +#![doc(html_root_url = "https://docs.rs/warp/0.3.6")] +#![deny(missing_docs)] +#![deny(missing_debug_implementations)] +#![deny(rust_2018_idioms)] +#![cfg_attr(test, deny(warnings))] + +//! # warp +//! +//! warp is a super-easy, composable, web server framework for warp speeds. +//! +//! Thanks to its [`Filter`][Filter] system, warp provides these out of the box: +//! +//! - Path routing and parameter extraction +//! - Header requirements and extraction +//! - Query string deserialization +//! - JSON and Form bodies +//! - Multipart form data +//! - Static Files and Directories +//! - Websockets +//! - Access logging +//! - Etc +//! +//! Since it builds on top of [hyper](https://hyper.rs), you automatically get: +//! +//! - HTTP/1 +//! - HTTP/2 +//! - Asynchronous +//! - One of the fastest HTTP implementations +//! - Tested and **correct** +//! +//! ## Filters +//! +//! The main concept in warp is the [`Filter`][Filter], which allows composition +//! to describe various endpoints in your web service. Besides this powerful +//! trait, warp comes with several built in [filters](filters/index.html), which +//! can be combined for your specific needs. +//! +//! As a small example, consider an endpoint that has path and header requirements: +//! +//! ``` +//! use warp::Filter; +//! +//! let hi = warp::path("hello") +//! .and(warp::path::param()) +//! .and(warp::header("user-agent")) +//! .map(|param: String, agent: String| { +//! format!("Hello {}, whose agent is {}", param, agent) +//! }); +//! ``` +//! +//! This example composes several [`Filter`s][Filter] together using `and`: +//! +//! - A path prefix of "hello" +//! - A path parameter of a `String` +//! - The `user-agent` header parsed as a `String` +//! +//! These specific filters will [`reject`][reject] requests that don't match +//! their requirements. +//! +//! This ends up matching requests like: +//! +//! ```notrust +//! GET /hello/sean HTTP/1.1 +//! Host: hyper.rs +//! User-Agent: reqwest/v0.8.6 +//! +//! ``` +//! And it returns a response similar to this: +//! +//! ```notrust +//! HTTP/1.1 200 OK +//! Content-Length: 41 +//! Date: ... +//! +//! Hello sean, whose agent is reqwest/v0.8.6 +//! ``` +//! +//! Take a look at the full list of [`filters`](filters/index.html) to see what +//! you can build. +//! +//! ## Testing +//! +//! Testing your web services easily is extremely important, and warp provides +//! a [`test`](mod@self::test) module to help send mocked requests through your service. +//! +//! [Filter]: trait.Filter.html +//! [reject]: reject/index.html + +#[macro_use] +mod error; +mod filter; +pub mod filters; +mod generic; +pub mod redirect; +pub mod reject; +pub mod reply; +mod route; +mod server; +mod service; +pub mod test; +#[cfg(feature = "tls")] +mod tls; +mod transport; + +pub use self::error::Error; +pub use self::filter::Filter; +// This otherwise shows a big dump of re-exports in the doc homepage, +// with zero context, so just hide it from the docs. Doc examples +// on each can show that a convenient import exists. +#[cfg(feature = "compression")] +#[doc(hidden)] +pub use self::filters::compression; +#[cfg(feature = "multipart")] +#[doc(hidden)] +pub use self::filters::multipart; +#[cfg(feature = "websocket")] +#[doc(hidden)] +pub use self::filters::ws; +#[doc(hidden)] +pub use self::filters::{ + addr, + // any() function + any::any, + body, + cookie, + // cookie() function + cookie::cookie, + cors, + // cors() function + cors::cors, + ext, + fs, + header, + // header() function + header::header, + host, + log, + // log() function + log::log, + method::{delete, get, head, method, options, patch, post, put}, + path, + // path() function and macro + path::path, + query, + // query() function + query::query, + sse, + trace, + // trace() function + trace::trace, +}; +// ws() function +pub use self::filter::wrap_fn; +#[cfg(feature = "websocket")] +#[doc(hidden)] +pub use self::filters::ws::ws; +#[doc(hidden)] +pub use self::redirect::redirect; +#[doc(hidden)] +#[allow(deprecated)] +pub use self::reject::{reject, Rejection}; +#[doc(hidden)] +pub use self::reply::{reply, Reply}; +#[cfg(feature = "tls")] +pub use self::server::TlsServer; +pub use self::server::{serve, Server}; +pub use self::service::service; +#[doc(hidden)] +pub use http; +#[doc(hidden)] +pub use hyper; + +#[doc(hidden)] +pub use bytes::Buf; +#[doc(hidden)] +pub use futures_util::{Future, Sink, Stream}; +#[doc(hidden)] + +pub(crate) type Request = http::Request<hyper::Body>; diff --git a/third_party/rust/warp/src/redirect.rs b/third_party/rust/warp/src/redirect.rs new file mode 100644 index 0000000000..373ab546ff --- /dev/null +++ b/third_party/rust/warp/src/redirect.rs @@ -0,0 +1,147 @@ +//! Redirect requests to a new location. +//! +//! The types in this module are helpers that implement [`Reply`], and easy +//! to use in order to setup redirects. + +use http::{header, StatusCode}; + +pub use self::sealed::AsLocation; +use crate::reply::{self, Reply}; + +/// HTTP 301 Moved Permanently +/// Description: The requested resource has been permanently moved to a new URL. +/// Usage: It is used when a URL has permanently moved to a new location. Search engines will update their index to the new URL. Browsers and clients will automatically cache this redirect, so subsequent requests for the old URL will automatically go to the new URL without making a request to the old URL. +/// Common Use Case: Changing domain names, restructuring website URLs. +/// +/// # Example +/// +/// ``` +/// use warp::{http::Uri, Filter}; +/// +/// let route = warp::path("v1") +/// .map(|| { +/// warp::redirect(Uri::from_static("/v2")) +/// }); +/// ``` +pub fn redirect(uri: impl AsLocation) -> impl Reply { + reply::with_header( + StatusCode::MOVED_PERMANENTLY, + header::LOCATION, + uri.header_value(), + ) +} + +/// HTTP 302 Found (or Temporary Redirect) +/// Description: The requested resource can be found at a different URL temporarily. +/// Usage: Historically, this status code was used for temporary redirects. However, its meaning was often misunderstood, and different clients treated it differently. As a result, it is recommended to use 307 (or 303) for temporary redirects instead. +/// Common Use Case: Rarely used directly due to ambiguity; replaced by 307 or 303. +/// +/// # Example +/// +/// ``` +/// use warp::{http::Uri, Filter}; +/// +/// let route = warp::path("v1") +/// .map(|| { +/// warp::redirect::found(Uri::from_static("/v2")) +/// }); +/// ``` +pub fn found(uri: impl AsLocation) -> impl Reply { + reply::with_header(StatusCode::FOUND, header::LOCATION, uri.header_value()) +} + +/// HTTP 303 See Other +/// Description: The response to the request can be found at a different URL, and the client should retrieve it using the GET method. +/// Usage: It is typically used to redirect the client to another URL using a GET request after processing a POST request. It ensures that the client doesn't repeat the POST request if they refresh the page. +/// Common Use Case: After form submissions or any non-idempotent request. +/// +/// The HTTP method of the request to the new location will always be `GET`. +/// +/// # Example +/// +/// ``` +/// use warp::{http::Uri, Filter}; +/// +/// let route = warp::path("v1") +/// .map(|| { +/// warp::redirect::see_other(Uri::from_static("/v2")) +/// }); +/// ``` +pub fn see_other(uri: impl AsLocation) -> impl Reply { + reply::with_header(StatusCode::SEE_OTHER, header::LOCATION, uri.header_value()) +} + +/// HTTP 307 Temporary Redirect: +/// Description: The requested resource can be found at a different URL temporarily. +/// Usage: Similar to 302, but explicitly defined as a temporary redirect. The main difference between 307 and 302 is that 307 preserves the method of the original request when redirecting. If the original request was a POST, the subsequent request to the new URL will also be a POST. +/// Common Use Case: Temporary redirects that should preserve the original request method. +/// +/// This is similar to [`see_other`](fn@see_other) but the HTTP method and the body of the request +/// to the new location will be the same as the method and body of the current request. +/// +/// # Example +/// +/// ``` +/// use warp::{http::Uri, Filter}; +/// +/// let route = warp::path("v1") +/// .map(|| { +/// warp::redirect::temporary(Uri::from_static("/v2")) +/// }); +/// ``` +pub fn temporary(uri: impl AsLocation) -> impl Reply { + reply::with_header( + StatusCode::TEMPORARY_REDIRECT, + header::LOCATION, + uri.header_value(), + ) +} + +/// HTTP 308 Permanent Redirect +/// Description: The requested resource has been permanently moved to a new URL, and future requests should use the new URL. +/// Usage: Similar to 301, but like 307, it preserves the original request method when redirecting. It indicates that the redirection is permanent, and browsers and clients will cache this redirect like they do for 301. +// Common Use Case: Permanently moving resources to a new URL while maintaining the original request method. +/// +/// This is similar to [`redirect`](fn@redirect) but the HTTP method of the request to the new +/// location will be the same as the method of the current request. +/// +/// # Example +/// +/// ``` +/// use warp::{http::Uri, Filter}; +/// +/// let route = warp::path("v1") +/// .map(|| { +/// warp::redirect::permanent(Uri::from_static("/v2")) +/// }); +/// ``` +pub fn permanent(uri: impl AsLocation) -> impl Reply { + reply::with_header( + StatusCode::PERMANENT_REDIRECT, + header::LOCATION, + uri.header_value(), + ) +} + +mod sealed { + use bytes::Bytes; + use http::{header::HeaderValue, Uri}; + + /// Trait for redirect locations. Currently only a `Uri` can be used in + /// redirect. + /// This sealed trait exists to allow adding possibly new impls so other + /// arguments could be accepted, like maybe just `warp::redirect("/v2")`. + pub trait AsLocation: Sealed {} + pub trait Sealed { + fn header_value(self) -> HeaderValue; + } + + impl AsLocation for Uri {} + + impl Sealed for Uri { + fn header_value(self) -> HeaderValue { + let bytes = Bytes::from(self.to_string()); + HeaderValue::from_maybe_shared(bytes).expect("Uri is a valid HeaderValue") + } + } +} diff --git a/third_party/rust/warp/src/reject.rs b/third_party/rust/warp/src/reject.rs new file mode 100644 index 0000000000..23a4440f80 --- /dev/null +++ b/third_party/rust/warp/src/reject.rs @@ -0,0 +1,872 @@ +//! Rejections +//! +//! Part of the power of the [`Filter`](../trait.Filter.html) system is being able to +//! reject a request from a filter chain. This allows for filters to be +//! combined with `or`, so that if one side of the chain finds that a request +//! doesn't fulfill its requirements, the other side can try to process +//! the request. +//! +//! Many of the built-in [`filters`](../filters) will automatically reject +//! the request with an appropriate rejection. However, you can also build +//! new custom [`Filter`](../trait.Filter.html)s and still want other routes to be +//! matchable in the case a predicate doesn't hold. +//! +//! As a request is processed by a Filter chain, the rejections are accumulated into +//! a list contained by the [`Rejection`](struct.Rejection.html) type. Rejections from +//! filters can be handled using [`Filter::recover`](../trait.Filter.html#method.recover). +//! This is a convenient way to map rejections into a [`Reply`](../reply/trait.Reply.html). +//! +//! For a more complete example see the +//! [Rejection Example](https://github.com/seanmonstar/warp/blob/master/examples/rejections.rs) +//! from the repository. +//! +//! # Example +//! +//! ``` +//! use warp::{reply, Reply, Filter, reject, Rejection, http::StatusCode}; +//! +//! #[derive(Debug)] +//! struct InvalidParameter; +//! +//! impl reject::Reject for InvalidParameter {} +//! +//! // Custom rejection handler that maps rejections into responses. +//! async fn handle_rejection(err: Rejection) -> Result<impl Reply, std::convert::Infallible> { +//! if err.is_not_found() { +//! Ok(reply::with_status("NOT_FOUND", StatusCode::NOT_FOUND)) +//! } else if let Some(e) = err.find::<InvalidParameter>() { +//! Ok(reply::with_status("BAD_REQUEST", StatusCode::BAD_REQUEST)) +//! } else { +//! eprintln!("unhandled rejection: {:?}", err); +//! Ok(reply::with_status("INTERNAL_SERVER_ERROR", StatusCode::INTERNAL_SERVER_ERROR)) +//! } +//! } +//! +//! +//! // Filter on `/:id`, but reject with InvalidParameter if the `id` is `0`. +//! // Recover from this rejection using a custom rejection handler. +//! let route = warp::path::param() +//! .and_then(|id: u32| async move { +//! if id == 0 { +//! Err(warp::reject::custom(InvalidParameter)) +//! } else { +//! Ok("id is valid") +//! } +//! }) +//! .recover(handle_rejection); +//! ``` + +use std::any::Any; +use std::convert::Infallible; +use std::error::Error as StdError; +use std::fmt; + +use http::{ + self, + header::{HeaderValue, CONTENT_TYPE}, + StatusCode, +}; +use hyper::Body; + +pub(crate) use self::sealed::{CombineRejection, IsReject}; + +/// Rejects a request with `404 Not Found`. +#[inline] +pub fn reject() -> Rejection { + not_found() +} + +/// Rejects a request with `404 Not Found`. +#[inline] +pub fn not_found() -> Rejection { + Rejection { + reason: Reason::NotFound, + } +} + +// 400 Bad Request +#[inline] +pub(crate) fn invalid_query() -> Rejection { + known(InvalidQuery { _p: () }) +} + +// 400 Bad Request +#[inline] +pub(crate) fn missing_header(name: &'static str) -> Rejection { + known(MissingHeader { name }) +} + +// 400 Bad Request +#[inline] +pub(crate) fn invalid_header(name: &'static str) -> Rejection { + known(InvalidHeader { name }) +} + +// 400 Bad Request +#[inline] +pub(crate) fn missing_cookie(name: &'static str) -> Rejection { + known(MissingCookie { name }) +} + +// 405 Method Not Allowed +#[inline] +pub(crate) fn method_not_allowed() -> Rejection { + known(MethodNotAllowed { _p: () }) +} + +// 411 Length Required +#[inline] +pub(crate) fn length_required() -> Rejection { + known(LengthRequired { _p: () }) +} + +// 413 Payload Too Large +#[inline] +pub(crate) fn payload_too_large() -> Rejection { + known(PayloadTooLarge { _p: () }) +} + +// 415 Unsupported Media Type +// +// Used by the body filters if the request payload content-type doesn't match +// what can be deserialized. +#[inline] +pub(crate) fn unsupported_media_type() -> Rejection { + known(UnsupportedMediaType { _p: () }) +} + +/// Rejects a request with a custom cause. +/// +/// A [`recover`][] filter should convert this `Rejection` into a `Reply`, +/// or else this will be returned as a `500 Internal Server Error`. +/// +/// [`recover`]: ../trait.Filter.html#method.recover +pub fn custom<T: Reject>(err: T) -> Rejection { + Rejection::custom(Box::new(err)) +} + +/// Protect against re-rejecting a rejection. +/// +/// ```compile_fail +/// fn with(r: warp::Rejection) { +/// let _wat = warp::reject::custom(r); +/// } +/// ``` +fn __reject_custom_compilefail() {} + +/// A marker trait to ensure proper types are used for custom rejections. +/// +/// Can be converted into Rejection. +/// +/// # Example +/// +/// ``` +/// use warp::{Filter, reject::Reject}; +/// +/// #[derive(Debug)] +/// struct RateLimited; +/// +/// impl Reject for RateLimited {} +/// +/// let route = warp::any().and_then(|| async { +/// Err::<(), _>(warp::reject::custom(RateLimited)) +/// }); +/// ``` +// Require `Sized` for now to prevent passing a `Box<dyn Reject>`, since we +// would be double-boxing it, and the downcasting wouldn't work as expected. +pub trait Reject: fmt::Debug + Sized + Send + Sync + 'static {} + +trait Cause: fmt::Debug + Send + Sync + 'static { + fn as_any(&self) -> &dyn Any; +} + +impl<T> Cause for T +where + T: fmt::Debug + Send + Sync + 'static, +{ + fn as_any(&self) -> &dyn Any { + self + } +} + +impl dyn Cause { + fn downcast_ref<T: Any>(&self) -> Option<&T> { + self.as_any().downcast_ref::<T>() + } +} + +pub(crate) fn known<T: Into<Known>>(err: T) -> Rejection { + Rejection::known(err.into()) +} + +/// Rejection of a request by a [`Filter`](crate::Filter). +/// +/// See the [`reject`](module@crate::reject) documentation for more. +pub struct Rejection { + reason: Reason, +} + +enum Reason { + NotFound, + Other(Box<Rejections>), +} + +enum Rejections { + Known(Known), + Custom(Box<dyn Cause>), + Combined(Box<Rejections>, Box<Rejections>), +} + +macro_rules! enum_known { + ($($(#[$attr:meta])* $var:ident($ty:path),)+) => ( + pub(crate) enum Known { + $( + $(#[$attr])* + $var($ty), + )+ + } + + impl Known { + fn inner_as_any(&self) -> &dyn Any { + match *self { + $( + $(#[$attr])* + Known::$var(ref t) => t, + )+ + } + } + } + + impl fmt::Debug for Known { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + $( + $(#[$attr])* + Known::$var(ref t) => t.fmt(f), + )+ + } + } + } + + impl fmt::Display for Known { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + $( + $(#[$attr])* + Known::$var(ref t) => t.fmt(f), + )+ + } + } + } + + $( + #[doc(hidden)] + $(#[$attr])* + impl From<$ty> for Known { + fn from(ty: $ty) -> Known { + Known::$var(ty) + } + } + )+ + ); +} + +enum_known! { + MethodNotAllowed(MethodNotAllowed), + InvalidHeader(InvalidHeader), + MissingHeader(MissingHeader), + MissingCookie(MissingCookie), + InvalidQuery(InvalidQuery), + LengthRequired(LengthRequired), + PayloadTooLarge(PayloadTooLarge), + UnsupportedMediaType(UnsupportedMediaType), + FileOpenError(crate::fs::FileOpenError), + FilePermissionError(crate::fs::FilePermissionError), + BodyReadError(crate::body::BodyReadError), + BodyDeserializeError(crate::body::BodyDeserializeError), + CorsForbidden(crate::cors::CorsForbidden), + #[cfg(feature = "websocket")] + MissingConnectionUpgrade(crate::ws::MissingConnectionUpgrade), + MissingExtension(crate::ext::MissingExtension), + BodyConsumedMultipleTimes(crate::body::BodyConsumedMultipleTimes), +} + +impl Rejection { + fn known(known: Known) -> Self { + Rejection { + reason: Reason::Other(Box::new(Rejections::Known(known))), + } + } + + fn custom(other: Box<dyn Cause>) -> Self { + Rejection { + reason: Reason::Other(Box::new(Rejections::Custom(other))), + } + } + + /// Searches this `Rejection` for a specific cause. + /// + /// A `Rejection` will accumulate causes over a `Filter` chain. This method + /// can search through them and return the first cause of this type. + /// + /// # Example + /// + /// ``` + /// #[derive(Debug)] + /// struct Nope; + /// + /// impl warp::reject::Reject for Nope {} + /// + /// let reject = warp::reject::custom(Nope); + /// + /// if let Some(nope) = reject.find::<Nope>() { + /// println!("found it: {:?}", nope); + /// } + /// ``` + pub fn find<T: 'static>(&self) -> Option<&T> { + if let Reason::Other(ref rejections) = self.reason { + return rejections.find(); + } + None + } + + /// Returns true if this Rejection was made via `warp::reject::not_found`. + /// + /// # Example + /// + /// ``` + /// let rejection = warp::reject(); + /// + /// assert!(rejection.is_not_found()); + /// ``` + pub fn is_not_found(&self) -> bool { + matches!(self.reason, Reason::NotFound) + } +} + +impl<T: Reject> From<T> for Rejection { + #[inline] + fn from(err: T) -> Rejection { + custom(err) + } +} + +impl From<Infallible> for Rejection { + #[inline] + fn from(infallible: Infallible) -> Rejection { + match infallible {} + } +} + +impl IsReject for Infallible { + fn status(&self) -> StatusCode { + match *self {} + } + + fn into_response(&self) -> crate::reply::Response { + match *self {} + } +} + +impl IsReject for Rejection { + fn status(&self) -> StatusCode { + match self.reason { + Reason::NotFound => StatusCode::NOT_FOUND, + Reason::Other(ref other) => other.status(), + } + } + + fn into_response(&self) -> crate::reply::Response { + match self.reason { + Reason::NotFound => { + let mut res = http::Response::default(); + *res.status_mut() = StatusCode::NOT_FOUND; + res + } + Reason::Other(ref other) => other.into_response(), + } + } +} + +impl fmt::Debug for Rejection { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_tuple("Rejection").field(&self.reason).finish() + } +} + +impl fmt::Debug for Reason { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match *self { + Reason::NotFound => f.write_str("NotFound"), + Reason::Other(ref other) => match **other { + Rejections::Known(ref e) => fmt::Debug::fmt(e, f), + Rejections::Custom(ref e) => fmt::Debug::fmt(e, f), + Rejections::Combined(ref a, ref b) => { + let mut list = f.debug_list(); + a.debug_list(&mut list); + b.debug_list(&mut list); + list.finish() + } + }, + } + } +} + +// ===== Rejections ===== + +impl Rejections { + fn status(&self) -> StatusCode { + match *self { + Rejections::Known(ref k) => match *k { + Known::MethodNotAllowed(_) => StatusCode::METHOD_NOT_ALLOWED, + Known::InvalidHeader(_) + | Known::MissingHeader(_) + | Known::MissingCookie(_) + | Known::InvalidQuery(_) + | Known::BodyReadError(_) + | Known::BodyDeserializeError(_) => StatusCode::BAD_REQUEST, + #[cfg(feature = "websocket")] + Known::MissingConnectionUpgrade(_) => StatusCode::BAD_REQUEST, + Known::LengthRequired(_) => StatusCode::LENGTH_REQUIRED, + Known::PayloadTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE, + Known::UnsupportedMediaType(_) => StatusCode::UNSUPPORTED_MEDIA_TYPE, + Known::FilePermissionError(_) | Known::CorsForbidden(_) => StatusCode::FORBIDDEN, + Known::FileOpenError(_) + | Known::MissingExtension(_) + | Known::BodyConsumedMultipleTimes(_) => StatusCode::INTERNAL_SERVER_ERROR, + }, + Rejections::Custom(..) => StatusCode::INTERNAL_SERVER_ERROR, + Rejections::Combined(..) => self.preferred().status(), + } + } + + fn into_response(&self) -> crate::reply::Response { + match *self { + Rejections::Known(ref e) => { + let mut res = http::Response::new(Body::from(e.to_string())); + *res.status_mut() = self.status(); + res.headers_mut().insert( + CONTENT_TYPE, + HeaderValue::from_static("text/plain; charset=utf-8"), + ); + res + } + Rejections::Custom(ref e) => { + tracing::error!( + "unhandled custom rejection, returning 500 response: {:?}", + e + ); + let body = format!("Unhandled rejection: {:?}", e); + let mut res = http::Response::new(Body::from(body)); + *res.status_mut() = self.status(); + res.headers_mut().insert( + CONTENT_TYPE, + HeaderValue::from_static("text/plain; charset=utf-8"), + ); + res + } + Rejections::Combined(..) => self.preferred().into_response(), + } + } + + fn find<T: 'static>(&self) -> Option<&T> { + match *self { + Rejections::Known(ref e) => e.inner_as_any().downcast_ref(), + Rejections::Custom(ref e) => e.downcast_ref(), + Rejections::Combined(ref a, ref b) => a.find().or_else(|| b.find()), + } + } + + fn debug_list(&self, f: &mut fmt::DebugList<'_, '_>) { + match *self { + Rejections::Known(ref e) => { + f.entry(e); + } + Rejections::Custom(ref e) => { + f.entry(e); + } + Rejections::Combined(ref a, ref b) => { + a.debug_list(f); + b.debug_list(f); + } + } + } + + fn preferred(&self) -> &Rejections { + match self { + Rejections::Known(_) | Rejections::Custom(_) => self, + Rejections::Combined(a, b) => { + let a = a.preferred(); + let b = b.preferred(); + // Now both a and b are known or custom, so it is safe + // to get status + // Compare status codes, with this priority: + // - NOT_FOUND is lowest + // - METHOD_NOT_ALLOWED is second + // - if one status code is greater than the other + // - otherwise, prefer A... + match (a.status(), b.status()) { + (_, StatusCode::NOT_FOUND) => a, + (StatusCode::NOT_FOUND, _) => b, + (_, StatusCode::METHOD_NOT_ALLOWED) => a, + (StatusCode::METHOD_NOT_ALLOWED, _) => b, + (sa, sb) if sa < sb => b, + _ => a, + } + } + } + } +} + +unit_error! { + /// Invalid query + pub InvalidQuery: "Invalid query string" +} + +unit_error! { + /// HTTP method not allowed + pub MethodNotAllowed: "HTTP method not allowed" +} + +unit_error! { + /// A content-length header is required + pub LengthRequired: "A content-length header is required" +} + +unit_error! { + /// The request payload is too large + pub PayloadTooLarge: "The request payload is too large" +} + +unit_error! { + /// The request's content-type is not supported + pub UnsupportedMediaType: "The request's content-type is not supported" +} + +/// Missing request header +#[derive(Debug)] +pub struct MissingHeader { + name: &'static str, +} + +impl MissingHeader { + /// Retrieve the name of the header that was missing + pub fn name(&self) -> &str { + self.name + } +} + +impl fmt::Display for MissingHeader { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Missing request header {:?}", self.name) + } +} + +impl StdError for MissingHeader {} + +/// Invalid request header +#[derive(Debug)] +pub struct InvalidHeader { + name: &'static str, +} + +impl InvalidHeader { + /// Retrieve the name of the header that was invalid + pub fn name(&self) -> &str { + self.name + } +} + +impl fmt::Display for InvalidHeader { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Invalid request header {:?}", self.name) + } +} + +impl StdError for InvalidHeader {} + +/// Missing cookie +#[derive(Debug)] +pub struct MissingCookie { + name: &'static str, +} + +impl MissingCookie { + /// Retrieve the name of the cookie that was missing + pub fn name(&self) -> &str { + self.name + } +} + +impl fmt::Display for MissingCookie { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Missing request cookie {:?}", self.name) + } +} + +impl StdError for MissingCookie {} + +mod sealed { + use super::{Reason, Rejection, Rejections}; + use http::StatusCode; + use std::convert::Infallible; + use std::fmt; + + // This sealed trait exists to allow Filters to return either `Rejection` + // or `!`. There are no other types that make sense, and so it is sealed. + pub trait IsReject: fmt::Debug + Send + Sync { + fn status(&self) -> StatusCode; + fn into_response(&self) -> crate::reply::Response; + } + + fn _assert_object_safe() { + fn _assert(_: &dyn IsReject) {} + } + + // This weird trait is to allow optimizations of propagating when a + // rejection can *never* happen (currently with the `Never` type, + // eventually to be replaced with `!`). + // + // Using this trait means the `Never` gets propagated to chained filters, + // allowing LLVM to eliminate more code paths. Without it, such as just + // requiring that `Rejection::from(Never)` were used in those filters, + // would mean that links later in the chain may assume a rejection *could* + // happen, and no longer eliminate those branches. + pub trait CombineRejection<E>: Send + Sized { + /// The type that should be returned when only 1 of the two + /// "rejections" occurs. + /// + /// # For example: + /// + /// `warp::any().and(warp::path("foo"))` has the following steps: + /// + /// 1. Since this is `and`, only **one** of the rejections will occur, + /// and as soon as it does, it will be returned. + /// 2. `warp::any()` rejects with `Never`. So, it will never return `Never`. + /// 3. `warp::path()` rejects with `Rejection`. It may return `Rejection`. + /// + /// Thus, if the above filter rejects, it will definitely be `Rejection`. + type One: IsReject + From<Self> + From<E> + Into<Rejection>; + + /// The type that should be returned when both rejections occur, + /// and need to be combined. + type Combined: IsReject; + + fn combine(self, other: E) -> Self::Combined; + } + + impl CombineRejection<Rejection> for Rejection { + type One = Rejection; + type Combined = Rejection; + + fn combine(self, other: Rejection) -> Self::Combined { + let reason = match (self.reason, other.reason) { + (Reason::Other(left), Reason::Other(right)) => { + Reason::Other(Box::new(Rejections::Combined(left, right))) + } + (Reason::Other(other), Reason::NotFound) + | (Reason::NotFound, Reason::Other(other)) => { + // ignore the NotFound + Reason::Other(other) + } + (Reason::NotFound, Reason::NotFound) => Reason::NotFound, + }; + + Rejection { reason } + } + } + + impl CombineRejection<Infallible> for Rejection { + type One = Rejection; + type Combined = Infallible; + + fn combine(self, other: Infallible) -> Self::Combined { + match other {} + } + } + + impl CombineRejection<Rejection> for Infallible { + type One = Rejection; + type Combined = Infallible; + + fn combine(self, _: Rejection) -> Self::Combined { + match self {} + } + } + + impl CombineRejection<Infallible> for Infallible { + type One = Infallible; + type Combined = Infallible; + + fn combine(self, _: Infallible) -> Self::Combined { + match self {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use http::StatusCode; + + #[derive(Debug, PartialEq)] + struct Left; + + #[derive(Debug, PartialEq)] + struct Right; + + impl Reject for Left {} + impl Reject for Right {} + + #[test] + fn rejection_status() { + assert_eq!(not_found().status(), StatusCode::NOT_FOUND); + assert_eq!( + method_not_allowed().status(), + StatusCode::METHOD_NOT_ALLOWED + ); + assert_eq!(length_required().status(), StatusCode::LENGTH_REQUIRED); + assert_eq!(payload_too_large().status(), StatusCode::PAYLOAD_TOO_LARGE); + assert_eq!( + unsupported_media_type().status(), + StatusCode::UNSUPPORTED_MEDIA_TYPE + ); + assert_eq!(custom(Left).status(), StatusCode::INTERNAL_SERVER_ERROR); + } + + #[tokio::test] + async fn combine_rejection_causes_with_some_left_and_none_right() { + let left = custom(Left); + let right = not_found(); + let reject = left.combine(right); + let resp = reject.into_response(); + + assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + response_body_string(resp).await, + "Unhandled rejection: Left" + ) + } + + #[tokio::test] + async fn combine_rejection_causes_with_none_left_and_some_right() { + let left = not_found(); + let right = custom(Right); + let reject = left.combine(right); + let resp = reject.into_response(); + + assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + response_body_string(resp).await, + "Unhandled rejection: Right" + ) + } + + #[tokio::test] + async fn unhandled_customs() { + let reject = not_found().combine(custom(Right)); + + let resp = reject.into_response(); + assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + response_body_string(resp).await, + "Unhandled rejection: Right" + ); + + // There's no real way to determine which is worse, since both are a 500, + // so pick the first one. + let reject = custom(Left).combine(custom(Right)); + + let resp = reject.into_response(); + assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + response_body_string(resp).await, + "Unhandled rejection: Left" + ); + + // With many rejections, custom still is top priority. + let reject = not_found() + .combine(not_found()) + .combine(not_found()) + .combine(custom(Right)) + .combine(not_found()); + + let resp = reject.into_response(); + assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + response_body_string(resp).await, + "Unhandled rejection: Right" + ); + } + + async fn response_body_string(resp: crate::reply::Response) -> String { + let (_, body) = resp.into_parts(); + let body_bytes = hyper::body::to_bytes(body).await.expect("failed concat"); + String::from_utf8_lossy(&body_bytes).to_string() + } + + #[test] + fn find_cause() { + let rej = custom(Left); + + assert_eq!(rej.find::<Left>(), Some(&Left)); + + let rej = rej.combine(method_not_allowed()); + + assert_eq!(rej.find::<Left>(), Some(&Left)); + assert!(rej.find::<MethodNotAllowed>().is_some(), "MethodNotAllowed"); + } + + #[test] + fn size_of_rejection() { + assert_eq!( + ::std::mem::size_of::<Rejection>(), + ::std::mem::size_of::<usize>(), + ); + } + + #[derive(Debug)] + struct X(u32); + impl Reject for X {} + + fn combine_n<F, R>(n: u32, new_reject: F) -> Rejection + where + F: Fn(u32) -> R, + R: Reject, + { + let mut rej = not_found(); + + for i in 0..n { + rej = rej.combine(custom(new_reject(i))); + } + + rej + } + + #[test] + fn test_debug() { + let rej = combine_n(3, X); + + let s = format!("{:?}", rej); + assert_eq!(s, "Rejection([X(0), X(1), X(2)])"); + } + + #[test] + fn convert_big_rejections_into_response() { + let mut rejections = Rejections::Custom(Box::new(std::io::Error::from_raw_os_error(100))); + for _ in 0..50 { + rejections = Rejections::Combined( + Box::new(Rejections::Known(Known::MethodNotAllowed( + MethodNotAllowed { _p: () }, + ))), + Box::new(rejections), + ); + } + let reason = Reason::Other(Box::new(rejections)); + let rejection = Rejection { reason }; + assert_eq!( + StatusCode::INTERNAL_SERVER_ERROR, + rejection.into_response().status() + ); + } +} diff --git a/third_party/rust/warp/src/reply.rs b/third_party/rust/warp/src/reply.rs new file mode 100644 index 0000000000..79e6dfd0e3 --- /dev/null +++ b/third_party/rust/warp/src/reply.rs @@ -0,0 +1,590 @@ +//! Reply to requests. +//! +//! A [`Reply`](./trait.Reply.html) is a type that can be converted into an HTTP +//! response to be sent to the client. These are typically the successful +//! counterpart to a [rejection](../reject). +//! +//! The functions in this module are helpers for quickly creating a reply. +//! Besides them, you can return a type that implements [`Reply`](./trait.Reply.html). This +//! could be any of the following: +//! +//! - [`http::Response<impl Into<hyper::Body>>`](https://docs.rs/http) +//! - `String` +//! - `&'static str` +//! - `http::StatusCode` +//! +//! # Example +//! +//! ``` +//! use warp::{Filter, http::Response}; +//! +//! // Returns an empty `200 OK` response. +//! let empty_200 = warp::any().map(warp::reply); +//! +//! // Returns a `200 OK` response with custom header and body. +//! let custom = warp::any().map(|| { +//! Response::builder() +//! .header("my-custom-header", "some-value") +//! .body("and a custom body") +//! }); +//! +//! // GET requests return the empty 200, POST return the custom. +//! let routes = warp::get().and(empty_200) +//! .or(warp::post().and(custom)); +//! ``` + +use std::borrow::Cow; +use std::convert::TryFrom; +use std::error::Error as StdError; +use std::fmt; + +use crate::generic::{Either, One}; +use http::header::{HeaderName, HeaderValue, CONTENT_TYPE}; +use http::StatusCode; +use hyper::Body; +use serde::Serialize; +use serde_json; + +// This re-export just looks weird in docs... +pub(crate) use self::sealed::Reply_; +use self::sealed::{BoxedReply, Internal}; +#[doc(hidden)] +pub use crate::filters::reply as with; + +/// Response type into which types implementing the `Reply` trait are convertable. +pub type Response = ::http::Response<Body>; + +/// Returns an empty `Reply` with status code `200 OK`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // GET /just-ok returns an empty `200 OK`. +/// let route = warp::path("just-ok") +/// .map(|| { +/// println!("got a /just-ok request!"); +/// warp::reply() +/// }); +/// ``` +#[inline] +pub fn reply() -> impl Reply { + StatusCode::OK +} + +/// Convert the value into a `Reply` with the value encoded as JSON. +/// +/// The passed value must implement [`Serialize`][ser]. Many +/// collections do, and custom domain types can have `Serialize` derived. +/// +/// [ser]: https://serde.rs +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// // GET /ids returns a `200 OK` with a JSON array of ids: +/// // `[1, 3, 7, 13]` +/// let route = warp::path("ids") +/// .map(|| { +/// let our_ids = vec![1, 3, 7, 13]; +/// warp::reply::json(&our_ids) +/// }); +/// ``` +/// +/// # Note +/// +/// If a type fails to be serialized into JSON, the error is logged at the +/// `error` level, and the returned `impl Reply` will be an empty +/// `500 Internal Server Error` response. +pub fn json<T>(val: &T) -> Json +where + T: Serialize, +{ + Json { + inner: serde_json::to_vec(val).map_err(|err| { + tracing::error!("reply::json error: {}", err); + }), + } +} + +/// A JSON formatted reply. +#[allow(missing_debug_implementations)] +pub struct Json { + inner: Result<Vec<u8>, ()>, +} + +impl Reply for Json { + #[inline] + fn into_response(self) -> Response { + match self.inner { + Ok(body) => { + let mut res = Response::new(body.into()); + res.headers_mut() + .insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + res + } + Err(()) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), + } + } +} + +#[derive(Debug)] +pub(crate) struct ReplyJsonError; + +impl fmt::Display for ReplyJsonError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("warp::reply::json() failed") + } +} + +impl StdError for ReplyJsonError {} + +/// Reply with a body and `content-type` set to `text/html; charset=utf-8`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let body = r#" +/// <html> +/// <head> +/// <title>HTML with warp!</title> +/// </head> +/// <body> +/// <h1>warp + HTML = ♥</h1> +/// </body> +/// </html> +/// "#; +/// +/// let route = warp::any() +/// .map(move || { +/// warp::reply::html(body) +/// }); +/// ``` +pub fn html<T>(body: T) -> Html<T> +where + Body: From<T>, + T: Send, +{ + Html { body } +} + +/// An HTML reply. +#[allow(missing_debug_implementations)] +pub struct Html<T> { + body: T, +} + +impl<T> Reply for Html<T> +where + Body: From<T>, + T: Send, +{ + #[inline] + fn into_response(self) -> Response { + let mut res = Response::new(Body::from(self.body)); + res.headers_mut().insert( + CONTENT_TYPE, + HeaderValue::from_static("text/html; charset=utf-8"), + ); + res + } +} + +/// Types that can be converted into a `Response`. +/// +/// This trait is implemented for the following: +/// +/// - `http::StatusCode` +/// - `http::Response<impl Into<hyper::Body>>` +/// - `String` +/// - `&'static str` +/// +/// # Example +/// +/// ```rust +/// use warp::{Filter, http::Response}; +/// +/// struct Message { +/// msg: String +/// } +/// +/// impl warp::Reply for Message { +/// fn into_response(self) -> warp::reply::Response { +/// Response::new(format!("message: {}", self.msg).into()) +/// } +/// } +/// +/// fn handler() -> Message { +/// Message { msg: "Hello".to_string() } +/// } +/// +/// let route = warp::any().map(handler); +/// ``` +pub trait Reply: BoxedReply + Send { + /// Converts the given value into a [`Response`]. + /// + /// [`Response`]: type.Response.html + fn into_response(self) -> Response; + + /* + TODO: Currently unsure about having trait methods here, as it + requires returning an exact type, which I'd rather not commit to. + Additionally, it doesn't work great with `Box<Reply>`. + + A possible alternative is to have wrappers, like + + - `WithStatus<R: Reply>(StatusCode, R)` + + + /// Change the status code of this `Reply`. + fn with_status(self, status: StatusCode) -> Reply_ + where + Self: Sized, + { + let mut res = self.into_response(); + *res.status_mut() = status; + Reply_(res) + } + + /// Add a header to this `Reply`. + /// + /// # Example + /// + /// ```rust + /// use warp::Reply; + /// + /// let reply = warp::reply() + /// .with_header("x-foo", "bar"); + /// ``` + fn with_header<K, V>(self, name: K, value: V) -> Reply_ + where + Self: Sized, + HeaderName: TryFrom<K>, + HeaderValue: TryFrom<V>, + { + match <HeaderName as TryFrom<K>>::try_from(name) { + Ok(name) => match <HeaderValue as TryFrom<V>>::try_from(value) { + Ok(value) => { + let mut res = self.into_response(); + res.headers_mut().append(name, value); + Reply_(res) + }, + Err(err) => { + tracing::error!("with_header value error: {}", err.into()); + Reply_(::reject::server_error() + .into_response()) + } + }, + Err(err) => { + tracing::error!("with_header name error: {}", err.into()); + Reply_(::reject::server_error() + .into_response()) + } + } + } + */ +} + +impl<T: Reply + ?Sized> Reply for Box<T> { + fn into_response(self) -> Response { + self.boxed_into_response(Internal) + } +} + +fn _assert_object_safe() { + fn _assert(_: &dyn Reply) {} +} + +/// Wrap an `impl Reply` to change its `StatusCode`. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::any() +/// .map(warp::reply) +/// .map(|reply| { +/// warp::reply::with_status(reply, warp::http::StatusCode::CREATED) +/// }); +/// ``` +pub fn with_status<T: Reply>(reply: T, status: StatusCode) -> WithStatus<T> { + WithStatus { reply, status } +} + +/// Wrap an `impl Reply` to change its `StatusCode`. +/// +/// Returned by `warp::reply::with_status`. +#[derive(Debug)] +pub struct WithStatus<T> { + reply: T, + status: StatusCode, +} + +impl<T: Reply> Reply for WithStatus<T> { + fn into_response(self) -> Response { + let mut res = self.reply.into_response(); + *res.status_mut() = self.status; + res + } +} + +/// Wrap an `impl Reply` to add a header when rendering. +/// +/// # Example +/// +/// ``` +/// use warp::Filter; +/// +/// let route = warp::any() +/// .map(warp::reply) +/// .map(|reply| { +/// warp::reply::with_header(reply, "server", "warp") +/// }); +/// ``` +pub fn with_header<T: Reply, K, V>(reply: T, name: K, value: V) -> WithHeader<T> +where + HeaderName: TryFrom<K>, + <HeaderName as TryFrom<K>>::Error: Into<http::Error>, + HeaderValue: TryFrom<V>, + <HeaderValue as TryFrom<V>>::Error: Into<http::Error>, +{ + let header = match <HeaderName as TryFrom<K>>::try_from(name) { + Ok(name) => match <HeaderValue as TryFrom<V>>::try_from(value) { + Ok(value) => Some((name, value)), + Err(err) => { + let err = err.into(); + tracing::error!("with_header value error: {}", err); + None + } + }, + Err(err) => { + let err = err.into(); + tracing::error!("with_header name error: {}", err); + None + } + }; + + WithHeader { header, reply } +} + +/// Wraps an `impl Reply` and adds a header when rendering. +/// +/// Returned by `warp::reply::with_header`. +#[derive(Debug)] +pub struct WithHeader<T> { + header: Option<(HeaderName, HeaderValue)>, + reply: T, +} + +impl<T: Reply> Reply for WithHeader<T> { + fn into_response(self) -> Response { + let mut res = self.reply.into_response(); + if let Some((name, value)) = self.header { + res.headers_mut().insert(name, value); + } + res + } +} + +impl<T: Send> Reply for ::http::Response<T> +where + Body: From<T>, +{ + #[inline] + fn into_response(self) -> Response { + self.map(Body::from) + } +} + +impl Reply for ::http::StatusCode { + #[inline] + fn into_response(self) -> Response { + let mut res = Response::default(); + *res.status_mut() = self; + res + } +} + +impl Reply for ::http::Error { + #[inline] + fn into_response(self) -> Response { + tracing::error!("reply error: {:?}", self); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + } +} + +impl<T, E> Reply for Result<T, E> +where + T: Reply, + E: Reply, +{ + #[inline] + fn into_response(self) -> Response { + match self { + Ok(t) => t.into_response(), + Err(e) => e.into_response(), + } + } +} + +fn text_plain<T: Into<Body>>(body: T) -> Response { + let mut response = ::http::Response::new(body.into()); + response.headers_mut().insert( + CONTENT_TYPE, + HeaderValue::from_static("text/plain; charset=utf-8"), + ); + response +} + +impl Reply for String { + #[inline] + fn into_response(self) -> Response { + text_plain(self) + } +} + +impl Reply for Vec<u8> { + #[inline] + fn into_response(self) -> Response { + ::http::Response::builder() + .header( + CONTENT_TYPE, + HeaderValue::from_static("application/octet-stream"), + ) + .body(Body::from(self)) + .unwrap() + } +} + +impl Reply for &'static str { + #[inline] + fn into_response(self) -> Response { + text_plain(self) + } +} + +impl Reply for Cow<'static, str> { + #[inline] + fn into_response(self) -> Response { + match self { + Cow::Borrowed(s) => s.into_response(), + Cow::Owned(s) => s.into_response(), + } + } +} + +impl Reply for &'static [u8] { + #[inline] + fn into_response(self) -> Response { + ::http::Response::builder() + .header( + CONTENT_TYPE, + HeaderValue::from_static("application/octet-stream"), + ) + .body(Body::from(self)) + .unwrap() + } +} + +impl<T, U> Reply for Either<T, U> +where + T: Reply, + U: Reply, +{ + #[inline] + fn into_response(self) -> Response { + match self { + Either::A(a) => a.into_response(), + Either::B(b) => b.into_response(), + } + } +} + +impl<T> Reply for One<T> +where + T: Reply, +{ + #[inline] + fn into_response(self) -> Response { + self.0.into_response() + } +} + +impl Reply for std::convert::Infallible { + #[inline(always)] + fn into_response(self) -> Response { + match self {} + } +} + +mod sealed { + use super::{Reply, Response}; + + // An opaque type to return `impl Reply` from trait methods. + #[allow(missing_debug_implementations)] + pub struct Reply_(pub(crate) Response); + + impl Reply for Reply_ { + #[inline] + fn into_response(self) -> Response { + self.0 + } + } + + #[allow(missing_debug_implementations)] + pub struct Internal; + + // Implemented for all types that implement `Reply`. + // + // A user doesn't need to worry about this, it's just trait + // hackery to get `Box<dyn Reply>` working. + pub trait BoxedReply { + fn boxed_into_response(self: Box<Self>, internal: Internal) -> Response; + } + + impl<T: Reply> BoxedReply for T { + fn boxed_into_response(self: Box<Self>, _: Internal) -> Response { + (*self).into_response() + } + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + + #[test] + fn json_serde_error() { + // a HashMap<Vec, _> cannot be serialized to JSON + let mut map = HashMap::new(); + map.insert(vec![1, 2], 45); + + let res = json(&map).into_response(); + assert_eq!(res.status(), 500); + } + + #[test] + fn response_builder_error() { + let res = ::http::Response::builder() + .status(1337) + .body("woops") + .into_response(); + + assert_eq!(res.status(), 500); + } + + #[test] + fn boxed_reply() { + let r: Box<dyn Reply> = Box::new(reply()); + let resp = r.into_response(); + assert_eq!(resp.status(), 200); + } +} diff --git a/third_party/rust/warp/src/route.rs b/third_party/rust/warp/src/route.rs new file mode 100644 index 0000000000..afbac4d8ba --- /dev/null +++ b/third_party/rust/warp/src/route.rs @@ -0,0 +1,140 @@ +use scoped_tls::scoped_thread_local; +use std::cell::RefCell; +use std::mem; +use std::net::SocketAddr; + +use hyper::Body; + +use crate::Request; + +scoped_thread_local!(static ROUTE: RefCell<Route>); + +pub(crate) fn set<F, U>(r: &RefCell<Route>, func: F) -> U +where + F: FnOnce() -> U, +{ + ROUTE.set(r, func) +} + +pub(crate) fn is_set() -> bool { + ROUTE.is_set() +} + +pub(crate) fn with<F, R>(func: F) -> R +where + F: FnOnce(&mut Route) -> R, +{ + ROUTE.with(move |route| func(&mut *route.borrow_mut())) +} + +#[derive(Debug)] +pub(crate) struct Route { + body: BodyState, + remote_addr: Option<SocketAddr>, + req: Request, + segments_index: usize, +} + +#[derive(Debug)] +enum BodyState { + Ready, + Taken, +} + +impl Route { + pub(crate) fn new(req: Request, remote_addr: Option<SocketAddr>) -> RefCell<Route> { + let segments_index = if req.uri().path().starts_with('/') { + // Skip the beginning slash. + 1 + } else { + 0 + }; + + RefCell::new(Route { + body: BodyState::Ready, + remote_addr, + req, + segments_index, + }) + } + + pub(crate) fn method(&self) -> &http::Method { + self.req.method() + } + + pub(crate) fn headers(&self) -> &http::HeaderMap { + self.req.headers() + } + + pub(crate) fn version(&self) -> http::Version { + self.req.version() + } + + pub(crate) fn extensions(&self) -> &http::Extensions { + self.req.extensions() + } + + #[cfg(feature = "websocket")] + pub(crate) fn extensions_mut(&mut self) -> &mut http::Extensions { + self.req.extensions_mut() + } + + pub(crate) fn uri(&self) -> &http::Uri { + self.req.uri() + } + + pub(crate) fn path(&self) -> &str { + &self.req.uri().path()[self.segments_index..] + } + + pub(crate) fn full_path(&self) -> &str { + self.req.uri().path() + } + + pub(crate) fn set_unmatched_path(&mut self, index: usize) { + let index = self.segments_index + index; + let path = self.req.uri().path(); + if path.is_empty() { + // malformed path + return; + } else if path.len() == index { + self.segments_index = index; + } else { + debug_assert_eq!(path.as_bytes()[index], b'/'); + self.segments_index = index + 1; + } + } + + pub(crate) fn query(&self) -> Option<&str> { + self.req.uri().query() + } + + pub(crate) fn matched_path_index(&self) -> usize { + self.segments_index + } + + pub(crate) fn reset_matched_path_index(&mut self, index: usize) { + debug_assert!( + index <= self.segments_index, + "reset_match_path_index should not be bigger: current={}, arg={}", + self.segments_index, + index, + ); + self.segments_index = index; + } + + pub(crate) fn remote_addr(&self) -> Option<SocketAddr> { + self.remote_addr + } + + pub(crate) fn take_body(&mut self) -> Option<Body> { + match self.body { + BodyState::Ready => { + let body = mem::replace(self.req.body_mut(), Body::empty()); + self.body = BodyState::Taken; + Some(body) + } + BodyState::Taken => None, + } + } +} diff --git a/third_party/rust/warp/src/server.rs b/third_party/rust/warp/src/server.rs new file mode 100644 index 0000000000..929d96eb34 --- /dev/null +++ b/third_party/rust/warp/src/server.rs @@ -0,0 +1,618 @@ +#[cfg(feature = "tls")] +use crate::tls::TlsConfigBuilder; +use std::convert::Infallible; +use std::error::Error as StdError; +use std::future::Future; +use std::net::SocketAddr; +#[cfg(feature = "tls")] +use std::path::Path; + +use futures_util::{future, FutureExt, TryFuture, TryStream, TryStreamExt}; +use hyper::server::conn::AddrIncoming; +use hyper::service::{make_service_fn, service_fn}; +use hyper::Server as HyperServer; +use tokio::io::{AsyncRead, AsyncWrite}; +use tracing::Instrument; + +use crate::filter::Filter; +use crate::reject::IsReject; +use crate::reply::Reply; +use crate::transport::Transport; + +/// Create a `Server` with the provided `Filter`. +pub fn serve<F>(filter: F) -> Server<F> +where + F: Filter + Clone + Send + Sync + 'static, + F::Extract: Reply, + F::Error: IsReject, +{ + Server { + pipeline: false, + filter, + } +} + +/// A Warp Server ready to filter requests. +#[derive(Debug)] +pub struct Server<F> { + pipeline: bool, + filter: F, +} + +/// A Warp Server ready to filter requests over TLS. +/// +/// *This type requires the `"tls"` feature.* +#[cfg(feature = "tls")] +pub struct TlsServer<F> { + server: Server<F>, + tls: TlsConfigBuilder, +} + +// Getting all various generic bounds to make this a re-usable method is +// very complicated, so instead this is just a macro. +macro_rules! into_service { + ($into:expr) => {{ + let inner = crate::service($into); + make_service_fn(move |transport| { + let inner = inner.clone(); + let remote_addr = Transport::remote_addr(transport); + future::ok::<_, Infallible>(service_fn(move |req| { + inner.call_with_addr(req, remote_addr) + })) + }) + }}; +} + +macro_rules! addr_incoming { + ($addr:expr) => {{ + let mut incoming = AddrIncoming::bind($addr)?; + incoming.set_nodelay(true); + let addr = incoming.local_addr(); + (addr, incoming) + }}; +} + +macro_rules! bind_inner { + ($this:ident, $addr:expr) => {{ + let service = into_service!($this.filter); + let (addr, incoming) = addr_incoming!($addr); + let srv = HyperServer::builder(incoming) + .http1_pipeline_flush($this.pipeline) + .serve(service); + Ok::<_, hyper::Error>((addr, srv)) + }}; + + (tls: $this:ident, $addr:expr) => {{ + let service = into_service!($this.server.filter); + let (addr, incoming) = addr_incoming!($addr); + let tls = $this.tls.build()?; + let srv = HyperServer::builder(crate::tls::TlsAcceptor::new(tls, incoming)) + .http1_pipeline_flush($this.server.pipeline) + .serve(service); + Ok::<_, Box<dyn std::error::Error + Send + Sync>>((addr, srv)) + }}; +} + +macro_rules! bind { + ($this:ident, $addr:expr) => {{ + let addr = $addr.into(); + (|addr| bind_inner!($this, addr))(&addr).unwrap_or_else(|e| { + panic!("error binding to {}: {}", addr, e); + }) + }}; + + (tls: $this:ident, $addr:expr) => {{ + let addr = $addr.into(); + (|addr| bind_inner!(tls: $this, addr))(&addr).unwrap_or_else(|e| { + panic!("error binding to {}: {}", addr, e); + }) + }}; +} + +macro_rules! try_bind { + ($this:ident, $addr:expr) => {{ + (|addr| bind_inner!($this, addr))($addr) + }}; + + (tls: $this:ident, $addr:expr) => {{ + (|addr| bind_inner!(tls: $this, addr))($addr) + }}; +} + +// ===== impl Server ===== + +impl<F> Server<F> +where + F: Filter + Clone + Send + Sync + 'static, + <F::Future as TryFuture>::Ok: Reply, + <F::Future as TryFuture>::Error: IsReject, +{ + /// Run this `Server` forever on the current thread. + /// + /// # Panics + /// + /// Panics if we are unable to bind to the provided address. + pub async fn run(self, addr: impl Into<SocketAddr>) { + let (addr, fut) = self.bind_ephemeral(addr); + let span = tracing::info_span!("Server::run", ?addr); + tracing::info!(parent: &span, "listening on http://{}", addr); + + fut.instrument(span).await; + } + + /// Run this `Server` forever on the current thread with a specific stream + /// of incoming connections. + /// + /// This can be used for Unix Domain Sockets, or TLS, etc. + pub async fn run_incoming<I>(self, incoming: I) + where + I: TryStream + Send, + I::Ok: AsyncRead + AsyncWrite + Send + 'static + Unpin, + I::Error: Into<Box<dyn StdError + Send + Sync>>, + { + self.run_incoming2(incoming.map_ok(crate::transport::LiftIo).into_stream()) + .instrument(tracing::info_span!("Server::run_incoming")) + .await; + } + + async fn run_incoming2<I>(self, incoming: I) + where + I: TryStream + Send, + I::Ok: Transport + Send + 'static + Unpin, + I::Error: Into<Box<dyn StdError + Send + Sync>>, + { + let fut = self.serve_incoming2(incoming); + + tracing::info!("listening with custom incoming"); + + fut.await; + } + + /// Bind to a socket address, returning a `Future` that can be + /// executed on the current runtime. + /// + /// # Panics + /// + /// Panics if we are unable to bind to the provided address. + pub fn bind(self, addr: impl Into<SocketAddr> + 'static) -> impl Future<Output = ()> + 'static { + let (_, fut) = self.bind_ephemeral(addr); + fut + } + + /// Bind to a socket address, returning a `Future` that can be + /// executed on any runtime. + /// + /// In case we are unable to bind to the specified address, resolves to an + /// error and logs the reason. + pub async fn try_bind(self, addr: impl Into<SocketAddr>) { + let addr = addr.into(); + let srv = match try_bind!(self, &addr) { + Ok((_, srv)) => srv, + Err(err) => { + tracing::error!("error binding to {}: {}", addr, err); + return; + } + }; + + srv.map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }) + .await; + } + + /// Bind to a possibly ephemeral socket address. + /// + /// Returns the bound address and a `Future` that can be executed on + /// the current runtime. + /// + /// # Panics + /// + /// Panics if we are unable to bind to the provided address. + pub fn bind_ephemeral( + self, + addr: impl Into<SocketAddr>, + ) -> (SocketAddr, impl Future<Output = ()> + 'static) { + let (addr, srv) = bind!(self, addr); + let srv = srv.map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }); + + (addr, srv) + } + + /// Tried to bind a possibly ephemeral socket address. + /// + /// Returns a `Result` which fails in case we are unable to bind with the + /// underlying error. + /// + /// Returns the bound address and a `Future` that can be executed on + /// the current runtime. + pub fn try_bind_ephemeral( + self, + addr: impl Into<SocketAddr>, + ) -> Result<(SocketAddr, impl Future<Output = ()> + 'static), crate::Error> { + let addr = addr.into(); + let (addr, srv) = try_bind!(self, &addr).map_err(crate::Error::new)?; + let srv = srv.map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }); + + Ok((addr, srv)) + } + + /// Create a server with graceful shutdown signal. + /// + /// When the signal completes, the server will start the graceful shutdown + /// process. + /// + /// Returns the bound address and a `Future` that can be executed on + /// the current runtime. + /// + /// # Example + /// + /// ```no_run + /// use warp::Filter; + /// use futures_util::future::TryFutureExt; + /// use tokio::sync::oneshot; + /// + /// # fn main() { + /// let routes = warp::any() + /// .map(|| "Hello, World!"); + /// + /// let (tx, rx) = oneshot::channel(); + /// + /// let (addr, server) = warp::serve(routes) + /// .bind_with_graceful_shutdown(([127, 0, 0, 1], 3030), async { + /// rx.await.ok(); + /// }); + /// + /// // Spawn the server into a runtime + /// tokio::task::spawn(server); + /// + /// // Later, start the shutdown... + /// let _ = tx.send(()); + /// # } + /// ``` + /// + /// # Panics + /// + /// Panics if we are unable to bind to the provided address. + pub fn bind_with_graceful_shutdown( + self, + addr: impl Into<SocketAddr> + 'static, + signal: impl Future<Output = ()> + Send + 'static, + ) -> (SocketAddr, impl Future<Output = ()> + 'static) { + let (addr, srv) = bind!(self, addr); + let fut = srv.with_graceful_shutdown(signal).map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }); + (addr, fut) + } + + /// Create a server with graceful shutdown signal. + /// + /// When the signal completes, the server will start the graceful shutdown + /// process. + pub fn try_bind_with_graceful_shutdown( + self, + addr: impl Into<SocketAddr> + 'static, + signal: impl Future<Output = ()> + Send + 'static, + ) -> Result<(SocketAddr, impl Future<Output = ()> + 'static), crate::Error> { + let addr = addr.into(); + let (addr, srv) = try_bind!(self, &addr).map_err(crate::Error::new)?; + let srv = srv.with_graceful_shutdown(signal).map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }); + + Ok((addr, srv)) + } + + /// Setup this `Server` with a specific stream of incoming connections. + /// + /// This can be used for Unix Domain Sockets, or TLS, etc. + /// + /// Returns a `Future` that can be executed on the current runtime. + pub fn serve_incoming<I>(self, incoming: I) -> impl Future<Output = ()> + where + I: TryStream + Send, + I::Ok: AsyncRead + AsyncWrite + Send + 'static + Unpin, + I::Error: Into<Box<dyn StdError + Send + Sync>>, + { + let incoming = incoming.map_ok(crate::transport::LiftIo); + self.serve_incoming2(incoming) + .instrument(tracing::info_span!("Server::serve_incoming")) + } + + /// Setup this `Server` with a specific stream of incoming connections and a + /// signal to initiate graceful shutdown. + /// + /// This can be used for Unix Domain Sockets, or TLS, etc. + /// + /// When the signal completes, the server will start the graceful shutdown + /// process. + /// + /// Returns a `Future` that can be executed on the current runtime. + pub fn serve_incoming_with_graceful_shutdown<I>( + self, + incoming: I, + signal: impl Future<Output = ()> + Send + 'static, + ) -> impl Future<Output = ()> + where + I: TryStream + Send, + I::Ok: AsyncRead + AsyncWrite + Send + 'static + Unpin, + I::Error: Into<Box<dyn StdError + Send + Sync>>, + { + let incoming = incoming.map_ok(crate::transport::LiftIo); + let service = into_service!(self.filter); + let pipeline = self.pipeline; + + async move { + let srv = + HyperServer::builder(hyper::server::accept::from_stream(incoming.into_stream())) + .http1_pipeline_flush(pipeline) + .serve(service) + .with_graceful_shutdown(signal) + .await; + + if let Err(err) = srv { + tracing::error!("server error: {}", err); + } + } + .instrument(tracing::info_span!( + "Server::serve_incoming_with_graceful_shutdown" + )) + } + + async fn serve_incoming2<I>(self, incoming: I) + where + I: TryStream + Send, + I::Ok: Transport + Send + 'static + Unpin, + I::Error: Into<Box<dyn StdError + Send + Sync>>, + { + let service = into_service!(self.filter); + + let srv = HyperServer::builder(hyper::server::accept::from_stream(incoming.into_stream())) + .http1_pipeline_flush(self.pipeline) + .serve(service) + .await; + + if let Err(err) = srv { + tracing::error!("server error: {}", err); + } + } + + // Generally shouldn't be used, as it can slow down non-pipelined responses. + // + // It's only real use is to make silly pipeline benchmarks look better. + #[doc(hidden)] + pub fn unstable_pipeline(mut self) -> Self { + self.pipeline = true; + self + } + + /// Configure a server to use TLS. + /// + /// *This function requires the `"tls"` feature.* + #[cfg(feature = "tls")] + pub fn tls(self) -> TlsServer<F> { + TlsServer { + server: self, + tls: TlsConfigBuilder::new(), + } + } +} + +// // ===== impl TlsServer ===== + +#[cfg(feature = "tls")] +impl<F> TlsServer<F> +where + F: Filter + Clone + Send + Sync + 'static, + <F::Future as TryFuture>::Ok: Reply, + <F::Future as TryFuture>::Error: IsReject, +{ + // TLS config methods + + /// Specify the file path to read the private key. + /// + /// *This function requires the `"tls"` feature.* + pub fn key_path(self, path: impl AsRef<Path>) -> Self { + self.with_tls(|tls| tls.key_path(path)) + } + + /// Specify the file path to read the certificate. + /// + /// *This function requires the `"tls"` feature.* + pub fn cert_path(self, path: impl AsRef<Path>) -> Self { + self.with_tls(|tls| tls.cert_path(path)) + } + + /// Specify the file path to read the trust anchor for optional client authentication. + /// + /// Anonymous and authenticated clients will be accepted. If no trust anchor is provided by any + /// of the `client_auth_` methods, then client authentication is disabled by default. + /// + /// *This function requires the `"tls"` feature.* + pub fn client_auth_optional_path(self, path: impl AsRef<Path>) -> Self { + self.with_tls(|tls| tls.client_auth_optional_path(path)) + } + + /// Specify the file path to read the trust anchor for required client authentication. + /// + /// Only authenticated clients will be accepted. If no trust anchor is provided by any of the + /// `client_auth_` methods, then client authentication is disabled by default. + /// + /// *This function requires the `"tls"` feature.* + pub fn client_auth_required_path(self, path: impl AsRef<Path>) -> Self { + self.with_tls(|tls| tls.client_auth_required_path(path)) + } + + /// Specify the in-memory contents of the private key. + /// + /// *This function requires the `"tls"` feature.* + pub fn key(self, key: impl AsRef<[u8]>) -> Self { + self.with_tls(|tls| tls.key(key.as_ref())) + } + + /// Specify the in-memory contents of the certificate. + /// + /// *This function requires the `"tls"` feature.* + pub fn cert(self, cert: impl AsRef<[u8]>) -> Self { + self.with_tls(|tls| tls.cert(cert.as_ref())) + } + + /// Specify the in-memory contents of the trust anchor for optional client authentication. + /// + /// Anonymous and authenticated clients will be accepted. If no trust anchor is provided by any + /// of the `client_auth_` methods, then client authentication is disabled by default. + /// + /// *This function requires the `"tls"` feature.* + pub fn client_auth_optional(self, trust_anchor: impl AsRef<[u8]>) -> Self { + self.with_tls(|tls| tls.client_auth_optional(trust_anchor.as_ref())) + } + + /// Specify the in-memory contents of the trust anchor for required client authentication. + /// + /// Only authenticated clients will be accepted. If no trust anchor is provided by any of the + /// `client_auth_` methods, then client authentication is disabled by default. + /// + /// *This function requires the `"tls"` feature.* + pub fn client_auth_required(self, trust_anchor: impl AsRef<[u8]>) -> Self { + self.with_tls(|tls| tls.client_auth_required(trust_anchor.as_ref())) + } + + /// Specify the DER-encoded OCSP response. + /// + /// *This function requires the `"tls"` feature.* + pub fn ocsp_resp(self, resp: impl AsRef<[u8]>) -> Self { + self.with_tls(|tls| tls.ocsp_resp(resp.as_ref())) + } + + fn with_tls<Func>(self, func: Func) -> Self + where + Func: FnOnce(TlsConfigBuilder) -> TlsConfigBuilder, + { + let TlsServer { server, tls } = self; + let tls = func(tls); + TlsServer { server, tls } + } + + // Server run methods + + /// Run this `TlsServer` forever on the current thread. + /// + /// *This function requires the `"tls"` feature.* + pub async fn run(self, addr: impl Into<SocketAddr>) { + let (addr, fut) = self.bind_ephemeral(addr); + let span = tracing::info_span!("TlsServer::run", %addr); + tracing::info!(parent: &span, "listening on https://{}", addr); + + fut.instrument(span).await; + } + + /// Bind to a socket address, returning a `Future` that can be + /// executed on a runtime. + /// + /// *This function requires the `"tls"` feature.* + /// + /// # Panics + /// + /// Panics if we are unable to bind to the provided address. + pub async fn bind(self, addr: impl Into<SocketAddr>) { + let (_, fut) = self.bind_ephemeral(addr); + fut.await; + } + + /// Bind to a possibly ephemeral socket address. + /// + /// Returns the bound address and a `Future` that can be executed on + /// the current runtime. + /// + /// *This function requires the `"tls"` feature.* + /// + /// # Panics + /// + /// Panics if we are unable to bind to the provided address. + pub fn bind_ephemeral( + self, + addr: impl Into<SocketAddr>, + ) -> (SocketAddr, impl Future<Output = ()> + 'static) { + let (addr, srv) = bind!(tls: self, addr); + let srv = srv.map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }); + + (addr, srv) + } + + /// Create a server with graceful shutdown signal. + /// + /// When the signal completes, the server will start the graceful shutdown + /// process. + /// + /// *This function requires the `"tls"` feature.* + /// + /// # Panics + /// + /// Panics if we are unable to bind to the provided address. + pub fn bind_with_graceful_shutdown( + self, + addr: impl Into<SocketAddr> + 'static, + signal: impl Future<Output = ()> + Send + 'static, + ) -> (SocketAddr, impl Future<Output = ()> + 'static) { + let (addr, srv) = bind!(tls: self, addr); + + let fut = srv.with_graceful_shutdown(signal).map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }); + (addr, fut) + } + + /// Create a server with graceful shutdown signal. + /// + /// When the signal completes, the server will start the graceful shutdown + /// process. + /// + /// *This function requires the `"tls"` feature.* + pub fn try_bind_with_graceful_shutdown( + self, + addr: impl Into<SocketAddr> + 'static, + signal: impl Future<Output = ()> + Send + 'static, + ) -> Result<(SocketAddr, impl Future<Output = ()> + 'static), crate::Error> { + let addr = addr.into(); + let (addr, srv) = try_bind!(tls: self, &addr).map_err(crate::Error::new)?; + let srv = srv.with_graceful_shutdown(signal).map(|result| { + if let Err(err) = result { + tracing::error!("server error: {}", err) + } + }); + + Ok((addr, srv)) + } +} + +#[cfg(feature = "tls")] +impl<F> ::std::fmt::Debug for TlsServer<F> +where + F: ::std::fmt::Debug, +{ + fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result { + f.debug_struct("TlsServer") + .field("server", &self.server) + .finish() + } +} diff --git a/third_party/rust/warp/src/service.rs b/third_party/rust/warp/src/service.rs new file mode 100644 index 0000000000..4f93809c4e --- /dev/null +++ b/third_party/rust/warp/src/service.rs @@ -0,0 +1,3 @@ +//! Convert `Filter`s into `Service`s + +pub use crate::filter::service::service; diff --git a/third_party/rust/warp/src/test.rs b/third_party/rust/warp/src/test.rs new file mode 100644 index 0000000000..ca2710faeb --- /dev/null +++ b/third_party/rust/warp/src/test.rs @@ -0,0 +1,764 @@ +//! Test utilities to test your filters. +//! +//! [`Filter`](../trait.Filter.html)s can be easily tested without starting up an HTTP +//! server, by making use of the [`RequestBuilder`](./struct.RequestBuilder.html) in this +//! module. +//! +//! # Testing Filters +//! +//! It's easy to test filters, especially if smaller filters are used to build +//! up your full set. Consider these example filters: +//! +//! ``` +//! use warp::Filter; +//! +//! fn sum() -> impl Filter<Extract = (u32,), Error = warp::Rejection> + Copy { +//! warp::path::param() +//! .and(warp::path::param()) +//! .map(|x: u32, y: u32| { +//! x + y +//! }) +//! } +//! +//! fn math() -> impl Filter<Extract = (String,), Error = warp::Rejection> + Copy { +//! warp::post() +//! .and(sum()) +//! .map(|z: u32| { +//! format!("Sum = {}", z) +//! }) +//! } +//! ``` +//! +//! We can test some requests against the `sum` filter like this: +//! +//! ``` +//! # use warp::Filter; +//! #[tokio::test] +//! async fn test_sum() { +//! # let sum = || warp::any().map(|| 3); +//! let filter = sum(); +//! +//! // Execute `sum` and get the `Extract` back. +//! let value = warp::test::request() +//! .path("/1/2") +//! .filter(&filter) +//! .await +//! .unwrap(); +//! assert_eq!(value, 3); +//! +//! // Or simply test if a request matches (doesn't reject). +//! assert!( +//! warp::test::request() +//! .path("/1/-5") +//! .matches(&filter) +//! .await +//! ); +//! } +//! ``` +//! +//! If the filter returns something that implements `Reply`, and thus can be +//! turned into a response sent back to the client, we can test what exact +//! response is returned. The `math` filter uses the `sum` filter, but returns +//! a `String` that can be turned into a response. +//! +//! ``` +//! # use warp::Filter; +//! #[test] +//! fn test_math() { +//! # let math = || warp::any().map(warp::reply); +//! let filter = math(); +//! +//! let res = warp::test::request() +//! .path("/1/2") +//! .reply(&filter); +//! assert_eq!(res.status(), 405, "GET is not allowed"); +//! +//! let res = warp::test::request() +//! .method("POST") +//! .path("/1/2") +//! .reply(&filter); +//! assert_eq!(res.status(), 200); +//! assert_eq!(res.body(), "Sum is 3"); +//! } +//! ``` +use std::convert::TryFrom; +use std::error::Error as StdError; +use std::fmt; +use std::future::Future; +use std::net::SocketAddr; +#[cfg(feature = "websocket")] +use std::pin::Pin; +#[cfg(feature = "websocket")] +use std::task::Context; +#[cfg(feature = "websocket")] +use std::task::{self, Poll}; + +use bytes::Bytes; +#[cfg(feature = "websocket")] +use futures_channel::mpsc; +#[cfg(feature = "websocket")] +use futures_util::StreamExt; +use futures_util::{future, FutureExt, TryFutureExt}; +use http::{ + header::{HeaderName, HeaderValue}, + Response, +}; +use serde::Serialize; +use serde_json; +#[cfg(feature = "websocket")] +use tokio::sync::oneshot; + +use crate::filter::Filter; +#[cfg(feature = "websocket")] +use crate::filters::ws::Message; +use crate::reject::IsReject; +use crate::reply::Reply; +use crate::route::{self, Route}; +use crate::Request; +#[cfg(feature = "websocket")] +use crate::{Sink, Stream}; + +use self::inner::OneOrTuple; + +/// Starts a new test `RequestBuilder`. +pub fn request() -> RequestBuilder { + RequestBuilder { + remote_addr: None, + req: Request::default(), + } +} + +/// Starts a new test `WsBuilder`. +#[cfg(feature = "websocket")] +pub fn ws() -> WsBuilder { + WsBuilder { req: request() } +} + +/// A request builder for testing filters. +/// +/// See [module documentation](crate::test) for an overview. +#[must_use = "RequestBuilder does nothing on its own"] +#[derive(Debug)] +pub struct RequestBuilder { + remote_addr: Option<SocketAddr>, + req: Request, +} + +/// A Websocket builder for testing filters. +/// +/// See [module documentation](crate::test) for an overview. +#[cfg(feature = "websocket")] +#[must_use = "WsBuilder does nothing on its own"] +#[derive(Debug)] +pub struct WsBuilder { + req: RequestBuilder, +} + +/// A test client for Websocket filters. +#[cfg(feature = "websocket")] +pub struct WsClient { + tx: mpsc::UnboundedSender<crate::ws::Message>, + rx: mpsc::UnboundedReceiver<Result<crate::ws::Message, crate::error::Error>>, +} + +/// An error from Websocket filter tests. +#[derive(Debug)] +pub struct WsError { + cause: Box<dyn StdError + Send + Sync>, +} + +impl RequestBuilder { + /// Sets the method of this builder. + /// + /// The default if not set is `GET`. + /// + /// # Example + /// + /// ``` + /// let req = warp::test::request() + /// .method("POST"); + /// ``` + /// + /// # Panic + /// + /// This panics if the passed string is not able to be parsed as a valid + /// `Method`. + pub fn method(mut self, method: &str) -> Self { + *self.req.method_mut() = method.parse().expect("valid method"); + self + } + + /// Sets the request path of this builder. + /// + /// The default is not set is `/`. + /// + /// # Example + /// + /// ``` + /// let req = warp::test::request() + /// .path("/todos/33"); + /// ``` + /// + /// # Panic + /// + /// This panics if the passed string is not able to be parsed as a valid + /// `Uri`. + pub fn path(mut self, p: &str) -> Self { + let uri = p.parse().expect("test request path invalid"); + *self.req.uri_mut() = uri; + self + } + + /// Set a header for this request. + /// + /// # Example + /// + /// ``` + /// let req = warp::test::request() + /// .header("accept", "application/json"); + /// ``` + /// + /// # Panic + /// + /// This panics if the passed strings are not able to be parsed as a valid + /// `HeaderName` and `HeaderValue`. + pub fn header<K, V>(mut self, key: K, value: V) -> Self + where + HeaderName: TryFrom<K>, + HeaderValue: TryFrom<V>, + { + let name: HeaderName = TryFrom::try_from(key) + .map_err(|_| ()) + .expect("invalid header name"); + let value = TryFrom::try_from(value) + .map_err(|_| ()) + .expect("invalid header value"); + self.req.headers_mut().insert(name, value); + self + } + + /// Set the remote address of this request + /// + /// Default is no remote address. + /// + /// # Example + /// ``` + /// use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + /// + /// let req = warp::test::request() + /// .remote_addr(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080)); + /// ``` + pub fn remote_addr(mut self, addr: SocketAddr) -> Self { + self.remote_addr = Some(addr); + self + } + + /// Add a type to the request's `http::Extensions`. + pub fn extension<T>(mut self, ext: T) -> Self + where + T: Send + Sync + 'static, + { + self.req.extensions_mut().insert(ext); + self + } + + /// Set the bytes of this request body. + /// + /// Default is an empty body. + /// + /// # Example + /// + /// ``` + /// let req = warp::test::request() + /// .body("foo=bar&baz=quux"); + /// ``` + pub fn body(mut self, body: impl AsRef<[u8]>) -> Self { + let body = body.as_ref().to_vec(); + let len = body.len(); + *self.req.body_mut() = body.into(); + self.header("content-length", len.to_string()) + } + + /// Set the bytes of this request body by serializing a value into JSON. + /// + /// # Example + /// + /// ``` + /// let req = warp::test::request() + /// .json(&true); + /// ``` + pub fn json(mut self, val: &impl Serialize) -> Self { + let vec = serde_json::to_vec(val).expect("json() must serialize to JSON"); + let len = vec.len(); + *self.req.body_mut() = vec.into(); + self.header("content-length", len.to_string()) + .header("content-type", "application/json") + } + + /// Tries to apply the `Filter` on this request. + /// + /// # Example + /// + /// ```no_run + /// async { + /// let param = warp::path::param::<u32>(); + /// + /// let ex = warp::test::request() + /// .path("/41") + /// .filter(¶m) + /// .await + /// .unwrap(); + /// + /// assert_eq!(ex, 41); + /// + /// assert!( + /// warp::test::request() + /// .path("/foo") + /// .filter(¶m) + /// .await + /// .is_err() + /// ); + ///}; + /// ``` + pub async fn filter<F>(self, f: &F) -> Result<<F::Extract as OneOrTuple>::Output, F::Error> + where + F: Filter, + F::Future: Send + 'static, + F::Extract: OneOrTuple + Send + 'static, + F::Error: Send + 'static, + { + self.apply_filter(f).await.map(|ex| ex.one_or_tuple()) + } + + /// Returns whether the `Filter` matches this request, or rejects it. + /// + /// # Example + /// + /// ```no_run + /// async { + /// let get = warp::get(); + /// let post = warp::post(); + /// + /// assert!( + /// warp::test::request() + /// .method("GET") + /// .matches(&get) + /// .await + /// ); + /// + /// assert!( + /// !warp::test::request() + /// .method("GET") + /// .matches(&post) + /// .await + /// ); + ///}; + /// ``` + pub async fn matches<F>(self, f: &F) -> bool + where + F: Filter, + F::Future: Send + 'static, + F::Extract: Send + 'static, + F::Error: Send + 'static, + { + self.apply_filter(f).await.is_ok() + } + + /// Returns `Response` provided by applying the `Filter`. + /// + /// This requires that the supplied `Filter` return a [`Reply`]. + pub async fn reply<F>(self, f: &F) -> Response<Bytes> + where + F: Filter + 'static, + F::Extract: Reply + Send, + F::Error: IsReject + Send, + { + // TODO: de-duplicate this and apply_filter() + assert!(!route::is_set(), "nested test filter calls"); + + let route = Route::new(self.req, self.remote_addr); + let mut fut = Box::pin( + route::set(&route, move || f.filter(crate::filter::Internal)).then(|result| { + let res = match result { + Ok(rep) => rep.into_response(), + Err(rej) => { + tracing::debug!("rejected: {:?}", rej); + rej.into_response() + } + }; + let (parts, body) = res.into_parts(); + hyper::body::to_bytes(body).map_ok(|chunk| Response::from_parts(parts, chunk)) + }), + ); + + let fut = future::poll_fn(move |cx| route::set(&route, || fut.as_mut().poll(cx))); + + fut.await.expect("reply shouldn't fail") + } + + fn apply_filter<F>(self, f: &F) -> impl Future<Output = Result<F::Extract, F::Error>> + where + F: Filter, + F::Future: Send + 'static, + F::Extract: Send + 'static, + F::Error: Send + 'static, + { + assert!(!route::is_set(), "nested test filter calls"); + + let route = Route::new(self.req, self.remote_addr); + let mut fut = Box::pin(route::set(&route, move || { + f.filter(crate::filter::Internal) + })); + future::poll_fn(move |cx| route::set(&route, || fut.as_mut().poll(cx))) + } +} + +#[cfg(feature = "websocket")] +impl WsBuilder { + /// Sets the request path of this builder. + /// + /// The default is not set is `/`. + /// + /// # Example + /// + /// ``` + /// let req = warp::test::ws() + /// .path("/chat"); + /// ``` + /// + /// # Panic + /// + /// This panics if the passed string is not able to be parsed as a valid + /// `Uri`. + pub fn path(self, p: &str) -> Self { + WsBuilder { + req: self.req.path(p), + } + } + + /// Set a header for this request. + /// + /// # Example + /// + /// ``` + /// let req = warp::test::ws() + /// .header("foo", "bar"); + /// ``` + /// + /// # Panic + /// + /// This panics if the passed strings are not able to be parsed as a valid + /// `HeaderName` and `HeaderValue`. + pub fn header<K, V>(self, key: K, value: V) -> Self + where + HeaderName: TryFrom<K>, + HeaderValue: TryFrom<V>, + { + WsBuilder { + req: self.req.header(key, value), + } + } + + /// Execute this Websocket request against the provided filter. + /// + /// If the handshake succeeds, returns a `WsClient`. + /// + /// # Example + /// + /// ```no_run + /// use futures_util::future; + /// use warp::Filter; + /// #[tokio::main] + /// # async fn main() { + /// + /// // Some route that accepts websockets (but drops them immediately). + /// let route = warp::ws() + /// .map(|ws: warp::ws::Ws| { + /// ws.on_upgrade(|_| future::ready(())) + /// }); + /// + /// let client = warp::test::ws() + /// .handshake(route) + /// .await + /// .expect("handshake"); + /// # } + /// ``` + pub async fn handshake<F>(self, f: F) -> Result<WsClient, WsError> + where + F: Filter + Clone + Send + Sync + 'static, + F::Extract: Reply + Send, + F::Error: IsReject + Send, + { + let (upgraded_tx, upgraded_rx) = oneshot::channel(); + let (wr_tx, wr_rx) = mpsc::unbounded(); + let (rd_tx, rd_rx) = mpsc::unbounded(); + + tokio::spawn(async move { + use tokio_tungstenite::tungstenite::protocol; + + let (addr, srv) = crate::serve(f).bind_ephemeral(([127, 0, 0, 1], 0)); + + let mut req = self + .req + .header("connection", "upgrade") + .header("upgrade", "websocket") + .header("sec-websocket-version", "13") + .header("sec-websocket-key", "dGhlIHNhbXBsZSBub25jZQ==") + .req; + + let query_string = match req.uri().query() { + Some(q) => format!("?{}", q), + None => String::from(""), + }; + + let uri = format!("http://{}{}{}", addr, req.uri().path(), query_string) + .parse() + .expect("addr + path is valid URI"); + + *req.uri_mut() = uri; + + // let mut rt = current_thread::Runtime::new().unwrap(); + tokio::spawn(srv); + + let upgrade = ::hyper::Client::builder() + .build(AddrConnect(addr)) + .request(req) + .and_then(hyper::upgrade::on); + + let upgraded = match upgrade.await { + Ok(up) => { + let _ = upgraded_tx.send(Ok(())); + up + } + Err(err) => { + let _ = upgraded_tx.send(Err(err)); + return; + } + }; + let ws = crate::ws::WebSocket::from_raw_socket( + upgraded, + protocol::Role::Client, + Default::default(), + ) + .await; + + let (tx, rx) = ws.split(); + let write = wr_rx.map(Ok).forward(tx).map(|_| ()); + + let read = rx + .take_while(|result| match result { + Err(_) => future::ready(false), + Ok(m) => future::ready(!m.is_close()), + }) + .for_each(move |item| { + rd_tx.unbounded_send(item).expect("ws receive error"); + future::ready(()) + }); + + future::join(write, read).await; + }); + + match upgraded_rx.await { + Ok(Ok(())) => Ok(WsClient { + tx: wr_tx, + rx: rd_rx, + }), + Ok(Err(err)) => Err(WsError::new(err)), + Err(_canceled) => panic!("websocket handshake thread panicked"), + } + } +} + +#[cfg(feature = "websocket")] +impl WsClient { + /// Send a "text" websocket message to the server. + pub async fn send_text(&mut self, text: impl Into<String>) { + self.send(crate::ws::Message::text(text)).await; + } + + /// Send a websocket message to the server. + pub async fn send(&mut self, msg: crate::ws::Message) { + self.tx.unbounded_send(msg).unwrap(); + } + + /// Receive a websocket message from the server. + pub async fn recv(&mut self) -> Result<crate::filters::ws::Message, WsError> { + self.rx + .next() + .await + .map(|result| result.map_err(WsError::new)) + .unwrap_or_else(|| { + // websocket is closed + Err(WsError::new("closed")) + }) + } + + /// Assert the server has closed the connection. + pub async fn recv_closed(&mut self) -> Result<(), WsError> { + self.rx + .next() + .await + .map(|result| match result { + Ok(msg) => Err(WsError::new(format!("received message: {:?}", msg))), + Err(err) => Err(WsError::new(err)), + }) + .unwrap_or_else(|| { + // closed successfully + Ok(()) + }) + } + + fn pinned_tx(self: Pin<&mut Self>) -> Pin<&mut mpsc::UnboundedSender<crate::ws::Message>> { + let this = Pin::into_inner(self); + Pin::new(&mut this.tx) + } +} + +#[cfg(feature = "websocket")] +impl fmt::Debug for WsClient { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("WsClient").finish() + } +} + +#[cfg(feature = "websocket")] +impl Sink<crate::ws::Message> for WsClient { + type Error = WsError; + + fn poll_ready( + self: Pin<&mut Self>, + context: &mut Context<'_>, + ) -> Poll<Result<(), Self::Error>> { + self.pinned_tx().poll_ready(context).map_err(WsError::new) + } + + fn start_send(self: Pin<&mut Self>, message: Message) -> Result<(), Self::Error> { + self.pinned_tx().start_send(message).map_err(WsError::new) + } + + fn poll_flush( + self: Pin<&mut Self>, + context: &mut Context<'_>, + ) -> Poll<Result<(), Self::Error>> { + self.pinned_tx().poll_flush(context).map_err(WsError::new) + } + + fn poll_close( + self: Pin<&mut Self>, + context: &mut Context<'_>, + ) -> Poll<Result<(), Self::Error>> { + self.pinned_tx().poll_close(context).map_err(WsError::new) + } +} + +#[cfg(feature = "websocket")] +impl Stream for WsClient { + type Item = Result<crate::ws::Message, WsError>; + + fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> { + let this = Pin::into_inner(self); + let rx = Pin::new(&mut this.rx); + match rx.poll_next(context) { + Poll::Ready(Some(result)) => Poll::Ready(Some(result.map_err(WsError::new))), + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } + } +} + +// ===== impl WsError ===== + +#[cfg(feature = "websocket")] +impl WsError { + fn new<E: Into<Box<dyn StdError + Send + Sync>>>(cause: E) -> Self { + WsError { + cause: cause.into(), + } + } +} + +impl fmt::Display for WsError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "websocket error: {}", self.cause) + } +} + +impl StdError for WsError { + fn description(&self) -> &str { + "websocket error" + } +} + +// ===== impl AddrConnect ===== + +#[cfg(feature = "websocket")] +#[derive(Clone)] +struct AddrConnect(SocketAddr); + +#[cfg(feature = "websocket")] +impl tower_service::Service<::http::Uri> for AddrConnect { + type Response = ::tokio::net::TcpStream; + type Error = ::std::io::Error; + type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>; + + fn poll_ready(&mut self, _cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _: ::http::Uri) -> Self::Future { + Box::pin(tokio::net::TcpStream::connect(self.0)) + } +} + +mod inner { + pub trait OneOrTuple { + type Output; + + fn one_or_tuple(self) -> Self::Output; + } + + impl OneOrTuple for () { + type Output = (); + fn one_or_tuple(self) -> Self::Output {} + } + + macro_rules! one_or_tuple { + ($type1:ident) => { + impl<$type1> OneOrTuple for ($type1,) { + type Output = $type1; + fn one_or_tuple(self) -> Self::Output { + self.0 + } + } + }; + ($type1:ident, $( $type:ident ),*) => { + one_or_tuple!($( $type ),*); + + impl<$type1, $($type),*> OneOrTuple for ($type1, $($type),*) { + type Output = Self; + fn one_or_tuple(self) -> Self::Output { + self + } + } + } + } + + one_or_tuple! { + T1, + T2, + T3, + T4, + T5, + T6, + T7, + T8, + T9, + T10, + T11, + T12, + T13, + T14, + T15, + T16 + } +} diff --git a/third_party/rust/warp/src/tls.rs b/third_party/rust/warp/src/tls.rs new file mode 100644 index 0000000000..96b6ed74e4 --- /dev/null +++ b/third_party/rust/warp/src/tls.rs @@ -0,0 +1,412 @@ +use std::fmt; +use std::fs::File; +use std::future::Future; +use std::io::{self, BufReader, Cursor, Read}; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +use futures_util::ready; +use hyper::server::accept::Accept; +use hyper::server::conn::{AddrIncoming, AddrStream}; + +use crate::transport::Transport; +use tokio_rustls::rustls::{ + server::{AllowAnyAnonymousOrAuthenticatedClient, AllowAnyAuthenticatedClient, NoClientAuth}, + Certificate, Error as TlsError, PrivateKey, RootCertStore, ServerConfig, +}; + +/// Represents errors that can occur building the TlsConfig +#[derive(Debug)] +pub(crate) enum TlsConfigError { + Io(io::Error), + /// An Error parsing the Certificate + CertParseError, + /// An Error parsing a Pkcs8 key + Pkcs8ParseError, + /// An Error parsing a Rsa key + RsaParseError, + /// An error from an empty key + EmptyKey, + /// An error from an invalid key + InvalidKey(TlsError), +} + +impl fmt::Display for TlsConfigError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TlsConfigError::Io(err) => err.fmt(f), + TlsConfigError::CertParseError => write!(f, "certificate parse error"), + TlsConfigError::Pkcs8ParseError => write!(f, "pkcs8 parse error"), + TlsConfigError::RsaParseError => write!(f, "rsa parse error"), + TlsConfigError::EmptyKey => write!(f, "key contains no private key"), + TlsConfigError::InvalidKey(err) => write!(f, "key contains an invalid key, {}", err), + } + } +} + +impl std::error::Error for TlsConfigError {} + +/// Tls client authentication configuration. +pub(crate) enum TlsClientAuth { + /// No client auth. + Off, + /// Allow any anonymous or authenticated client. + Optional(Box<dyn Read + Send + Sync>), + /// Allow any authenticated client. + Required(Box<dyn Read + Send + Sync>), +} + +/// Builder to set the configuration for the Tls server. +pub(crate) struct TlsConfigBuilder { + cert: Box<dyn Read + Send + Sync>, + key: Box<dyn Read + Send + Sync>, + client_auth: TlsClientAuth, + ocsp_resp: Vec<u8>, +} + +impl fmt::Debug for TlsConfigBuilder { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TlsConfigBuilder").finish() + } +} + +impl TlsConfigBuilder { + /// Create a new TlsConfigBuilder + pub(crate) fn new() -> TlsConfigBuilder { + TlsConfigBuilder { + key: Box::new(io::empty()), + cert: Box::new(io::empty()), + client_auth: TlsClientAuth::Off, + ocsp_resp: Vec::new(), + } + } + + /// sets the Tls key via File Path, returns `TlsConfigError::IoError` if the file cannot be open + pub(crate) fn key_path(mut self, path: impl AsRef<Path>) -> Self { + self.key = Box::new(LazyFile { + path: path.as_ref().into(), + file: None, + }); + self + } + + /// sets the Tls key via bytes slice + pub(crate) fn key(mut self, key: &[u8]) -> Self { + self.key = Box::new(Cursor::new(Vec::from(key))); + self + } + + /// Specify the file path for the TLS certificate to use. + pub(crate) fn cert_path(mut self, path: impl AsRef<Path>) -> Self { + self.cert = Box::new(LazyFile { + path: path.as_ref().into(), + file: None, + }); + self + } + + /// sets the Tls certificate via bytes slice + pub(crate) fn cert(mut self, cert: &[u8]) -> Self { + self.cert = Box::new(Cursor::new(Vec::from(cert))); + self + } + + /// Sets the trust anchor for optional Tls client authentication via file path. + /// + /// Anonymous and authenticated clients will be accepted. If no trust anchor is provided by any + /// of the `client_auth_` methods, then client authentication is disabled by default. + pub(crate) fn client_auth_optional_path(mut self, path: impl AsRef<Path>) -> Self { + let file = Box::new(LazyFile { + path: path.as_ref().into(), + file: None, + }); + self.client_auth = TlsClientAuth::Optional(file); + self + } + + /// Sets the trust anchor for optional Tls client authentication via bytes slice. + /// + /// Anonymous and authenticated clients will be accepted. If no trust anchor is provided by any + /// of the `client_auth_` methods, then client authentication is disabled by default. + pub(crate) fn client_auth_optional(mut self, trust_anchor: &[u8]) -> Self { + let cursor = Box::new(Cursor::new(Vec::from(trust_anchor))); + self.client_auth = TlsClientAuth::Optional(cursor); + self + } + + /// Sets the trust anchor for required Tls client authentication via file path. + /// + /// Only authenticated clients will be accepted. If no trust anchor is provided by any of the + /// `client_auth_` methods, then client authentication is disabled by default. + pub(crate) fn client_auth_required_path(mut self, path: impl AsRef<Path>) -> Self { + let file = Box::new(LazyFile { + path: path.as_ref().into(), + file: None, + }); + self.client_auth = TlsClientAuth::Required(file); + self + } + + /// Sets the trust anchor for required Tls client authentication via bytes slice. + /// + /// Only authenticated clients will be accepted. If no trust anchor is provided by any of the + /// `client_auth_` methods, then client authentication is disabled by default. + pub(crate) fn client_auth_required(mut self, trust_anchor: &[u8]) -> Self { + let cursor = Box::new(Cursor::new(Vec::from(trust_anchor))); + self.client_auth = TlsClientAuth::Required(cursor); + self + } + + /// sets the DER-encoded OCSP response + pub(crate) fn ocsp_resp(mut self, ocsp_resp: &[u8]) -> Self { + self.ocsp_resp = Vec::from(ocsp_resp); + self + } + + pub(crate) fn build(mut self) -> Result<ServerConfig, TlsConfigError> { + let mut cert_rdr = BufReader::new(self.cert); + let cert = rustls_pemfile::certs(&mut cert_rdr) + .map_err(|_e| TlsConfigError::CertParseError)? + .into_iter() + .map(Certificate) + .collect(); + + let key = { + // convert it to Vec<u8> to allow reading it again if key is RSA + let mut key_vec = Vec::new(); + self.key + .read_to_end(&mut key_vec) + .map_err(TlsConfigError::Io)?; + + if key_vec.is_empty() { + return Err(TlsConfigError::EmptyKey); + } + + let mut pkcs8 = rustls_pemfile::pkcs8_private_keys(&mut key_vec.as_slice()) + .map_err(|_e| TlsConfigError::Pkcs8ParseError)?; + + if !pkcs8.is_empty() { + PrivateKey(pkcs8.remove(0)) + } else { + let mut rsa = rustls_pemfile::rsa_private_keys(&mut key_vec.as_slice()) + .map_err(|_e| TlsConfigError::RsaParseError)?; + + if !rsa.is_empty() { + PrivateKey(rsa.remove(0)) + } else { + return Err(TlsConfigError::EmptyKey); + } + } + }; + + fn read_trust_anchor( + trust_anchor: Box<dyn Read + Send + Sync>, + ) -> Result<RootCertStore, TlsConfigError> { + let trust_anchors = { + let mut reader = BufReader::new(trust_anchor); + rustls_pemfile::certs(&mut reader).map_err(TlsConfigError::Io)? + }; + + let mut store = RootCertStore::empty(); + let (added, _skipped) = store.add_parsable_certificates(&trust_anchors); + if added == 0 { + return Err(TlsConfigError::CertParseError); + } + + Ok(store) + } + + let client_auth = match self.client_auth { + TlsClientAuth::Off => NoClientAuth::boxed(), + TlsClientAuth::Optional(trust_anchor) => { + AllowAnyAnonymousOrAuthenticatedClient::new(read_trust_anchor(trust_anchor)?) + .boxed() + } + TlsClientAuth::Required(trust_anchor) => { + AllowAnyAuthenticatedClient::new(read_trust_anchor(trust_anchor)?).boxed() + } + }; + + let mut config = ServerConfig::builder() + .with_safe_defaults() + .with_client_cert_verifier(client_auth) + .with_single_cert_with_ocsp_and_sct(cert, key, self.ocsp_resp, Vec::new()) + .map_err(TlsConfigError::InvalidKey)?; + config.alpn_protocols = vec!["h2".into(), "http/1.1".into()]; + Ok(config) + } +} + +struct LazyFile { + path: PathBuf, + file: Option<File>, +} + +impl LazyFile { + fn lazy_read(&mut self, buf: &mut [u8]) -> io::Result<usize> { + if self.file.is_none() { + self.file = Some(File::open(&self.path)?); + } + + self.file.as_mut().unwrap().read(buf) + } +} + +impl Read for LazyFile { + fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { + self.lazy_read(buf).map_err(|err| { + let kind = err.kind(); + io::Error::new( + kind, + format!("error reading file ({:?}): {}", self.path.display(), err), + ) + }) + } +} + +impl Transport for TlsStream { + fn remote_addr(&self) -> Option<SocketAddr> { + Some(self.remote_addr) + } +} + +enum State { + Handshaking(tokio_rustls::Accept<AddrStream>), + Streaming(tokio_rustls::server::TlsStream<AddrStream>), +} + +// tokio_rustls::server::TlsStream doesn't expose constructor methods, +// so we have to TlsAcceptor::accept and handshake to have access to it +// TlsStream implements AsyncRead/AsyncWrite handshaking tokio_rustls::Accept first +pub(crate) struct TlsStream { + state: State, + remote_addr: SocketAddr, +} + +impl TlsStream { + fn new(stream: AddrStream, config: Arc<ServerConfig>) -> TlsStream { + let remote_addr = stream.remote_addr(); + let accept = tokio_rustls::TlsAcceptor::from(config).accept(stream); + TlsStream { + state: State::Handshaking(accept), + remote_addr, + } + } +} + +impl AsyncRead for TlsStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll<io::Result<()>> { + let pin = self.get_mut(); + match pin.state { + State::Handshaking(ref mut accept) => match ready!(Pin::new(accept).poll(cx)) { + Ok(mut stream) => { + let result = Pin::new(&mut stream).poll_read(cx, buf); + pin.state = State::Streaming(stream); + result + } + Err(err) => Poll::Ready(Err(err)), + }, + State::Streaming(ref mut stream) => Pin::new(stream).poll_read(cx, buf), + } + } +} + +impl AsyncWrite for TlsStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll<io::Result<usize>> { + let pin = self.get_mut(); + match pin.state { + State::Handshaking(ref mut accept) => match ready!(Pin::new(accept).poll(cx)) { + Ok(mut stream) => { + let result = Pin::new(&mut stream).poll_write(cx, buf); + pin.state = State::Streaming(stream); + result + } + Err(err) => Poll::Ready(Err(err)), + }, + State::Streaming(ref mut stream) => Pin::new(stream).poll_write(cx, buf), + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> { + match self.state { + State::Handshaking(_) => Poll::Ready(Ok(())), + State::Streaming(ref mut stream) => Pin::new(stream).poll_flush(cx), + } + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> { + match self.state { + State::Handshaking(_) => Poll::Ready(Ok(())), + State::Streaming(ref mut stream) => Pin::new(stream).poll_shutdown(cx), + } + } +} + +pub(crate) struct TlsAcceptor { + config: Arc<ServerConfig>, + incoming: AddrIncoming, +} + +impl TlsAcceptor { + pub(crate) fn new(config: ServerConfig, incoming: AddrIncoming) -> TlsAcceptor { + TlsAcceptor { + config: Arc::new(config), + incoming, + } + } +} + +impl Accept for TlsAcceptor { + type Conn = TlsStream; + type Error = io::Error; + + fn poll_accept( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll<Option<Result<Self::Conn, Self::Error>>> { + let pin = self.get_mut(); + match ready!(Pin::new(&mut pin.incoming).poll_accept(cx)) { + Some(Ok(sock)) => Poll::Ready(Some(Ok(TlsStream::new(sock, pin.config.clone())))), + Some(Err(e)) => Poll::Ready(Some(Err(e))), + None => Poll::Ready(None), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn file_cert_key() { + TlsConfigBuilder::new() + .key_path("examples/tls/key.rsa") + .cert_path("examples/tls/cert.pem") + .build() + .unwrap(); + } + + #[test] + fn bytes_cert_key() { + let key = include_str!("../examples/tls/key.rsa"); + let cert = include_str!("../examples/tls/cert.pem"); + + TlsConfigBuilder::new() + .key(key.as_bytes()) + .cert(cert.as_bytes()) + .build() + .unwrap(); + } +} diff --git a/third_party/rust/warp/src/transport.rs b/third_party/rust/warp/src/transport.rs new file mode 100644 index 0000000000..be553e706e --- /dev/null +++ b/third_party/rust/warp/src/transport.rs @@ -0,0 +1,53 @@ +use std::io; +use std::net::SocketAddr; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use hyper::server::conn::AddrStream; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +pub trait Transport: AsyncRead + AsyncWrite { + fn remote_addr(&self) -> Option<SocketAddr>; +} + +impl Transport for AddrStream { + fn remote_addr(&self) -> Option<SocketAddr> { + Some(self.remote_addr()) + } +} + +pub(crate) struct LiftIo<T>(pub(crate) T); + +impl<T: AsyncRead + Unpin> AsyncRead for LiftIo<T> { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll<io::Result<()>> { + Pin::new(&mut self.get_mut().0).poll_read(cx, buf) + } +} + +impl<T: AsyncWrite + Unpin> AsyncWrite for LiftIo<T> { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll<io::Result<usize>> { + Pin::new(&mut self.get_mut().0).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> { + Pin::new(&mut self.get_mut().0).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> { + Pin::new(&mut self.get_mut().0).poll_shutdown(cx) + } +} + +impl<T: AsyncRead + AsyncWrite + Unpin> Transport for LiftIo<T> { + fn remote_addr(&self) -> Option<SocketAddr> { + None + } +} |