use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::hash::Hash;
use std::sync::Arc;
use affinidi_rate_limit::{
ClientIpExtractor, InvalidTrustedProxy, IpNet, KeyedRateLimiterState, Refusal,
};
use axum::extract::ConnectInfo;
use axum::http::Request;
use ipnetwork::IpNetwork;
use tokio_util::sync::{CancellationToken, DropGuard};
pub const VSOCK_PARENT_PEER: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 3)), 0);
pub const IPV6_CLIENT_PREFIX: u8 = 64;
#[derive(Debug, Clone)]
pub struct TrustedProxyKeyExtractor {
clients: ClientIpExtractor,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Client {
pub ip: IpAddr,
pub key: IpAddr,
}
impl TrustedProxyKeyExtractor {
pub fn new(trusted_cidrs: Vec<IpNetwork>) -> Result<Self, InvalidTrustedProxy> {
let ranges = trusted_cidrs
.iter()
.map(|cidr| {
IpNet::new(cidr.network(), cidr.prefix())
.expect("an IpNetwork's prefix is valid for its address family")
})
.collect();
let clients = ClientIpExtractor::trusted_proxies(ranges)?
.ipv6_prefix(IPV6_CLIENT_PREFIX)
.expect("64 is a valid IPv6 prefix length");
Ok(Self { clients })
}
pub fn client_of<B>(&self, req: &Request<B>) -> Result<Client, Refusal> {
let peer = req
.extensions()
.get::<ConnectInfo<SocketAddr>>()
.map(|ci| ci.0.ip())
.ok_or(Refusal::NoClientIp)?;
let unreadable = |_| Refusal::InvalidForwardedFor;
let ip = self
.clients
.client_ip(peer, req.headers())
.map_err(unreadable)?;
let key = self
.clients
.client_key(peer, req.headers())
.map_err(unreadable)?;
Ok(Client { ip, key })
}
}
pub fn check_trusted_proxy_cidrs(trusted_cidrs: &[IpNetwork]) -> Result<(), InvalidTrustedProxy> {
TrustedProxyKeyExtractor::new(trusted_cidrs.to_vec()).map(drop)
}
pub fn deserialize_trusted_proxy_cidrs<'de, D>(deserializer: D) -> Result<Vec<IpNetwork>, D::Error>
where
D: serde::Deserializer<'de>,
{
let cidrs = <Vec<IpNetwork> as serde::Deserialize>::deserialize(deserializer)?;
check_trusted_proxy_cidrs(&cidrs)
.map_err(|e| serde::de::Error::custom(format!("trust_xff_cidrs: {e}")))?;
Ok(cidrs)
}
#[derive(Clone)]
pub struct LimiterSweep {
token: CancellationToken,
_stop: Arc<DropGuard>,
}
impl LimiterSweep {
pub fn new() -> Self {
let token = CancellationToken::new();
Self {
_stop: Arc::new(token.clone().drop_guard()),
token,
}
}
pub fn swept<K>(&self, limiter: KeyedRateLimiterState<K>) -> KeyedRateLimiterState<K>
where
K: Hash + Eq + Clone + Send + Sync + 'static,
{
if tokio::runtime::Handle::try_current().is_ok() {
limiter.spawn_gc(self.token.clone());
}
limiter
}
}
impl Default for LimiterSweep {
fn default() -> Self {
Self::new()
}
}
pub async fn insert_default_connect_info_if_missing(
mut request: axum::extract::Request,
next: axum::middleware::Next,
) -> axum::response::Response {
use std::net::Ipv4Addr;
if request
.extensions()
.get::<ConnectInfo<SocketAddr>>()
.is_none()
{
let synthetic = ConnectInfo(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0));
request.extensions_mut().insert(synthetic);
}
next.run(request).await
}
#[cfg(test)]
mod tests {
use super::*;
fn req_with(peer: Option<&str>, xff: Option<&str>) -> Request<()> {
req_with_all(peer, xff.into_iter().collect())
}
fn req_with_all(peer: Option<&str>, xffs: Vec<&str>) -> Request<()> {
let mut builder = Request::builder().uri("/");
for xff in xffs {
builder = builder.header("x-forwarded-for", xff);
}
let mut req = builder.body(()).unwrap();
if let Some(peer) = peer {
req.extensions_mut()
.insert(ConnectInfo(peer.parse::<SocketAddr>().unwrap()));
}
req
}
fn cidrs(nets: &[&str]) -> Vec<IpNetwork> {
nets.iter().map(|n| n.parse().unwrap()).collect()
}
fn ip(s: &str) -> IpAddr {
s.parse().unwrap()
}
fn loopback() -> TrustedProxyKeyExtractor {
TrustedProxyKeyExtractor::new(cidrs(&["127.0.0.1/32"])).unwrap()
}
const PARENT: &str = "0.0.0.3:0";
#[test]
fn vsock_parent_peer_constant_is_the_documented_address() {
assert_eq!(VSOCK_PARENT_PEER, PARENT.parse::<SocketAddr>().unwrap());
}
fn parent_trusted() -> TrustedProxyKeyExtractor {
TrustedProxyKeyExtractor::new(cidrs(&["0.0.0.3/32"])).unwrap()
}
#[test]
fn xff_from_the_vsock_parent_is_honoured_when_declared() {
let req = req_with(Some(PARENT), Some("9.9.9.9, 203.0.113.9"));
assert_eq!(
parent_trusted().client_of(&req).unwrap().ip,
ip("203.0.113.9")
);
}
#[test]
fn the_vsock_parent_without_xff_is_charged_to_itself() {
let req = req_with(Some(PARENT), None);
assert_eq!(parent_trusted().client_of(&req).unwrap().ip, ip("0.0.0.3"));
}
#[test]
fn the_vsock_parent_is_not_trusted_undeclared() {
let none = TrustedProxyKeyExtractor::new(vec![]).unwrap();
let req = req_with(Some(PARENT), Some("203.0.113.9"));
assert_eq!(none.client_of(&req).unwrap().ip, ip("0.0.0.3"));
}
#[test]
fn a_stale_loopback_entry_does_not_trust_the_vsock_parent() {
let req = req_with(Some(PARENT), Some("203.0.113.9"));
assert_eq!(loopback().client_of(&req).unwrap().ip, ip("0.0.0.3"));
}
#[test]
fn declaring_the_parent_does_not_trust_loopback() {
let req = req_with(Some("127.0.0.1:1234"), Some("203.0.113.9"));
assert_eq!(
parent_trusted().client_of(&req).unwrap().ip,
ip("127.0.0.1")
);
}
#[test]
fn xff_from_an_untrusted_tcp_peer_is_ignored_even_naming_the_parent() {
let req = req_with(Some("203.0.113.9:1234"), Some("0.0.0.3"));
assert_eq!(
parent_trusted().client_of(&req).unwrap().ip,
ip("203.0.113.9")
);
}
#[test]
fn untrusted_peer_keys_on_peer_ignoring_forged_xff() {
let req = req_with(Some("203.0.113.9:1234"), Some("9.9.9.9"));
assert_eq!(loopback().client_of(&req).unwrap().ip, ip("203.0.113.9"));
}
#[test]
fn trusted_peer_keys_on_the_client_the_proxy_appended() {
let req = req_with(Some("127.0.0.1:1234"), Some("9.9.9.9, 203.0.113.9"));
assert_eq!(loopback().client_of(&req).unwrap().ip, ip("203.0.113.9"));
}
#[test]
fn distinct_clients_through_trusted_peer_key_differently() {
let a = req_with(Some("127.0.0.1:1"), Some("198.51.100.1"));
let b = req_with(Some("127.0.0.1:2"), Some("198.51.100.2"));
let extractor = loopback();
assert_ne!(
extractor.client_of(&a).unwrap().ip,
extractor.client_of(&b).unwrap().ip
);
}
#[test]
fn empty_cidr_list_never_trusts_xff() {
let extractor = TrustedProxyKeyExtractor::new(Vec::new()).unwrap();
let req = req_with(Some("127.0.0.1:1234"), Some("9.9.9.9"));
assert_eq!(extractor.client_of(&req).unwrap().ip, ip("127.0.0.1"));
}
#[test]
fn trusted_peer_without_xff_falls_back_to_peer() {
let req = req_with(Some("127.0.0.1:1234"), None);
assert_eq!(loopback().client_of(&req).unwrap().ip, ip("127.0.0.1"));
}
#[test]
fn missing_connect_info_is_unable_to_extract() {
let req = req_with(None, Some("9.9.9.9"));
assert_eq!(loopback().client_of(&req), Err(Refusal::NoClientIp));
}
#[test]
fn chained_trusted_proxies_are_skipped_to_reach_the_client() {
let extractor =
TrustedProxyKeyExtractor::new(cidrs(&["127.0.0.1/32", "10.0.0.0/8"])).unwrap();
let req = req_with(
Some("127.0.0.1:1234"),
Some("198.51.100.7, 10.0.0.5, 10.0.0.6"),
);
assert_eq!(extractor.client_of(&req).unwrap().ip, ip("198.51.100.7"));
}
#[test]
fn forged_trusted_looking_prefix_cannot_reach_past_the_real_hop() {
let extractor =
TrustedProxyKeyExtractor::new(cidrs(&["127.0.0.1/32", "10.0.0.0/8"])).unwrap();
let req = req_with(
Some("127.0.0.1:1234"),
Some("10.0.0.1, 10.0.0.2, 203.0.113.9"),
);
assert_eq!(extractor.client_of(&req).unwrap().ip, ip("203.0.113.9"));
}
#[test]
fn multiple_header_lines_concatenate_in_order() {
let extractor =
TrustedProxyKeyExtractor::new(cidrs(&["127.0.0.1/32", "10.0.0.0/8"])).unwrap();
let req = req_with_all(Some("127.0.0.1:1"), vec!["198.51.100.7", "10.0.0.5"]);
assert_eq!(extractor.client_of(&req).unwrap().ip, ip("198.51.100.7"));
}
fn assert_unreadable(result: Result<Client, Refusal>) {
assert_eq!(result, Err(Refusal::InvalidForwardedFor));
}
#[test]
fn malformed_entry_is_refused() {
let req = req_with(Some("127.0.0.1:1234"), Some("not-an-ip"));
assert_unreadable(loopback().client_of(&req));
}
#[test]
fn empty_entries_do_not_end_the_walk() {
for header in ["203.0.113.9,", "9.9.9.9,, 203.0.113.9", " , 203.0.113.9"] {
let req = req_with(Some("127.0.0.1:1234"), Some(header));
assert_eq!(
loopback().client_of(&req).unwrap().ip,
ip("203.0.113.9"),
"{header}"
);
}
}
#[test]
fn chain_trusted_end_to_end_falls_back_to_peer() {
let extractor =
TrustedProxyKeyExtractor::new(cidrs(&["127.0.0.1/32", "10.0.0.0/8"])).unwrap();
let req = req_with(Some("127.0.0.1:1"), Some("10.0.0.5, 10.0.0.6"));
assert_eq!(extractor.client_of(&req).unwrap().ip, ip("127.0.0.1"));
}
#[test]
fn absurdly_long_chain_is_refused() {
let chain = (0..affinidi_rate_limit::MAX_FORWARDED_ENTRIES + 1)
.map(|_| "203.0.113.9")
.collect::<Vec<_>>()
.join(", ");
let req = req_with(Some("127.0.0.1:1"), Some(&chain));
assert_unreadable(loopback().client_of(&req));
}
#[test]
fn port_and_bracket_forms_are_accepted() {
let extractor = TrustedProxyKeyExtractor::new(cidrs(&["127.0.0.1/32"])).unwrap();
for (entry, expected) in [
("203.0.113.9:443", "203.0.113.9"),
("[2001:db8::1]:443", "2001:db8::1"),
("[2001:db8::1]", "2001:db8::1"),
("2001:db8::1", "2001:db8::1"),
] {
let req = req_with(Some("127.0.0.1:1"), Some(entry));
assert_eq!(
extractor.client_of(&req).unwrap().ip,
ip(expected),
"{entry}"
);
}
}
#[test]
fn an_ipv6_client_is_charged_as_its_64() {
let a = loopback()
.client_of(&req_with(Some("[2001:db8:1:2::a]:1"), None))
.unwrap();
let b = loopback()
.client_of(&req_with(Some("[2001:db8:1:2::b]:1"), None))
.unwrap();
assert_eq!(a.ip, ip("2001:db8:1:2::a"));
assert_eq!(a.key, ip("2001:db8:1:2::"));
assert_eq!(a.key, b.key);
let other = loopback()
.client_of(&req_with(Some("[2001:db8:1:3::a]:1"), None))
.unwrap();
assert_ne!(a.key, other.key, "another /64 is another client");
let proxied = loopback()
.client_of(&req_with(Some("127.0.0.1:1"), Some("2001:db8:1:2::c")))
.unwrap();
assert_eq!(proxied.key, a.key);
}
#[test]
fn an_ipv4_client_is_charged_as_itself() {
let client = loopback()
.client_of(&req_with(Some("203.0.113.9:1"), None))
.unwrap();
assert_eq!(client.key, client.ip);
}
#[test]
fn a_catch_all_range_is_refused_where_the_list_is_read() {
for cidr in ["0.0.0.0/0", "::/0"] {
assert!(
check_trusted_proxy_cidrs(&cidrs(&[cidr])).is_err(),
"{cidr}"
);
let err = toml::from_str::<Wrapper>(&format!("cidrs = [\"{cidr}\"]")).unwrap_err();
assert!(err.to_string().contains("covers every address"), "{err}");
}
assert!(check_trusted_proxy_cidrs(&cidrs(&["10.0.0.0/8", "::1/128"])).is_ok());
}
#[derive(Debug, serde::Deserialize)]
#[allow(dead_code)]
struct Wrapper {
#[serde(deserialize_with = "deserialize_trusted_proxy_cidrs")]
cidrs: Vec<IpNetwork>,
}
#[tokio::test(start_paused = true)]
async fn a_swept_limiter_forgets_refilled_buckets() {
let sweep = LimiterSweep::new();
let limiter = sweep.swept(KeyedRateLimiterState::<u32>::per_period(
std::time::Duration::from_millis(1),
1,
));
for key in 0..100 {
limiter.charge(key).unwrap();
}
assert_eq!(limiter.tracked_keys(), 100);
std::thread::sleep(std::time::Duration::from_millis(5));
tokio::time::sleep(affinidi_rate_limit::GC_INTERVAL * 2).await;
assert_eq!(limiter.tracked_keys(), 0);
}
#[test]
fn ipv4_mapped_peer_matches_an_ipv4_cidr() {
let req = req_with(Some("[::ffff:127.0.0.1]:1234"), Some("203.0.113.9"));
assert_eq!(loopback().client_of(&req).unwrap().ip, ip("203.0.113.9"));
}
#[tokio::test]
async fn synthetic_connect_info_inserted_only_when_missing() {
use axum::body::Body;
use axum::routing::get;
use tower::ServiceExt;
async fn handler(ConnectInfo(addr): ConnectInfo<SocketAddr>) -> String {
addr.ip().to_string()
}
let app = axum::Router::new()
.route("/", get(handler))
.layer(axum::middleware::from_fn(
insert_default_connect_info_if_missing,
));
let resp = app
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
.await
.unwrap();
let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"127.0.0.1");
}
}