1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217
/*
* Copyright 2020 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
//! Filters for processing packets.
mod chain;
mod error;
mod factory;
mod read;
mod registry;
mod set;
mod write;
pub mod capture;
pub mod compress;
pub mod concatenate;
pub mod debug;
pub mod drop;
pub mod firewall;
pub mod load_balancer;
pub mod local_rate_limit;
pub mod r#match;
pub mod metrics;
pub mod pass;
pub mod timestamp;
pub mod token_router;
/// Prelude containing all types and traits required to implement [`Filter`] and
/// [`FilterFactory`].
pub mod prelude {
pub use super::{
ConvertProtoConfigError, CreateFilterArgs, CreationError, Filter, FilterError,
FilterInstance, ReadContext, StaticFilter, WriteContext,
};
}
// Core Filter types
#[doc(inline)]
pub use self::{
capture::Capture,
compress::Compress,
concatenate::Concatenate,
debug::Debug,
drop::Drop,
error::{ConvertProtoConfigError, CreationError, FilterError},
factory::{CreateFilterArgs, DynFilterFactory, FilterFactory, FilterInstance},
firewall::Firewall,
load_balancer::LoadBalancer,
local_rate_limit::LocalRateLimit,
pass::Pass,
r#match::Match,
read::ReadContext,
registry::FilterRegistry,
set::{FilterMap, FilterSet},
timestamp::Timestamp,
token_router::{HashedTokenRouter, TokenRouter},
write::WriteContext,
};
pub use self::chain::FilterChain;
/// Statically safe version of [`Filter`], if you're writing a Rust filter, you
/// should implement [`StaticFilter`] in addition to [`Filter`], as
/// [`StaticFilter`] guarantees all of the required properties through the type
/// system, allowing Quilkin take care of the virtual table boilerplate
/// automatically at compile-time.
/// ```
/// use quilkin::filters::prelude::*;
///
/// struct Greet;
///
/// /// Prepends data on each packet
/// #[async_trait::async_trait]
/// impl Filter for Greet {
/// async fn read(&self, ctx: &mut ReadContext) -> Result<(), FilterError> {
/// ctx.contents.prepend_from_slice(b"Hello ");
/// Ok(())
/// }
/// async fn write(&self, ctx: &mut WriteContext) -> Result<(), FilterError> {
/// ctx.contents.prepend_from_slice(b"Goodbye ");
/// Ok(())
/// }
/// }
///
/// impl StaticFilter for Greet {
/// const NAME: &'static str = "greet.v1";
/// type Configuration = ();
/// type BinaryConfiguration = ();
///
/// fn try_from_config(_: Option<Self::Configuration>) -> Result<Self, CreationError> {
/// Ok(Self)
/// }
/// }
/// ```
pub trait StaticFilter: Filter + Sized
// This where clause simply states that `Configuration`'s and
// `BinaryConfiguration`'s `Error` types are compatible with `filters::Error`.
where
CreationError: From<<Self::Configuration as TryFrom<Self::BinaryConfiguration>>::Error>
+ From<<Self::BinaryConfiguration as TryFrom<Self::Configuration>>::Error>,
{
/// The globally unique name of the filter.
const NAME: &'static str;
/// The human-readable configuration of the filter. **Must** be [`serde`]
/// compatible, have a JSON schema, and be convertible to and
/// from [`Self::BinaryConfiguration`].
type Configuration: schemars::JsonSchema
+ serde::Serialize
+ for<'de> serde::Deserialize<'de>
+ TryFrom<Self::BinaryConfiguration>;
/// The binary configuration of the filter. **Must** be [`prost`] compatible,
/// and be convertible to and from [`Self::Configuration`].
type BinaryConfiguration: prost::Message
+ Default
+ TryFrom<Self::Configuration>
+ Send
+ Sync
+ Sized;
/// Instantiates a new [`StaticFilter`] from the given configuration, if any.
/// # Errors
/// If the provided configuration is invalid.
fn try_from_config(config: Option<Self::Configuration>) -> Result<Self, CreationError>;
/// Instantiates a new [`StaticFilter`] from the given configuration, if any.
/// # Panics
/// If the provided configuration is invalid.
fn from_config(config: Option<Self::Configuration>) -> Self {
Self::try_from_config(config).unwrap()
}
/// Creates a new dynamic [`FilterFactory`] virtual table.
fn factory() -> DynFilterFactory
where
Self: 'static,
{
Box::from(std::marker::PhantomData::<fn() -> Self>)
}
/// Convenience method for providing a consistent error message for filters
/// which require a fully initialized [`Self::Configuration`].
fn ensure_config_exists(
config: Option<Self::Configuration>,
) -> Result<Self::Configuration, CreationError> {
config.ok_or(CreationError::MissingConfig(Self::NAME))
}
fn as_filter_config(
config: impl Into<Option<Self::Configuration>>,
) -> Result<crate::config::Filter, CreationError> {
Ok(crate::config::Filter {
name: Self::NAME.into(),
label: None,
config: config
.into()
.map(|config| serde_json::to_value(&config))
.transpose()?,
})
}
}
/// Trait for routing and manipulating packets.
///
/// An implementation of [`Filter`] provides a `read` and a `write` method. Both
/// methods are invoked by the proxy when it consults the filter chain - their
/// arguments contain information about the packet being processed.
/// - `read` is invoked when a packet is received on the local downstream port
/// and is to be sent to an upstream endpoint.
/// - `write` is invoked in the opposite direction when a packet is received
/// from an upstream endpoint and is to be sent to a downstream client.
///
/// **Metrics**
///
/// * `filter_read_duration_seconds` The duration it took for a `filter`'s
/// `read` implementation to execute.
/// * Labels
/// * `filter` The name of the filter being executed.
///
/// * `filter_write_duration_seconds` The duration it took for a `filter`'s
/// `write` implementation to execute.
/// * Labels
/// * `filter` The name of the filter being executed.
#[async_trait::async_trait]
pub trait Filter: Send + Sync {
/// [`Filter::read`] is invoked when the proxy receives data from a
/// downstream connection on the listening port.
///
/// This function should return an `Some` if the packet processing should
/// proceed. If the packet should be rejected, it will return [`None`]
/// instead. By default, the context passes through unchanged.
async fn read(&self, _: &mut ReadContext) -> Result<(), FilterError> {
Ok(())
}
/// [`Filter::write`] is invoked when the proxy is about to send data to a
/// downstream connection via the listening port after receiving it via one
/// of the upstream Endpoints.
///
/// This function should return an `Some` if the packet processing should
/// proceed. If the packet should be rejected, it will return [`None`]
async fn write(&self, _: &mut WriteContext) -> Result<(), FilterError> {
Ok(())
}
}