Set l5d-remote-ip on inbound requests and outbound responses (#200)

## Problem

[Context](linkerd/linkerd2#2310)

## Solution

We now add the `l5d-remote-ip` header to all inbound requests and outbound responses. We also make sure to strip the header before setting so that clients may not set this header.

We were already adding a header in similar style, so in order to deduplicate code, `add_header.rs` holds the generic logic for adding a header to a `http::Request` or `http::Response`. `l5d-client-id` and `l5d-server-id` now also use the same logic in order to add their respective headers.

## Validation

An additional test module was added to `tests/discovery.rs` that tests the following:
- outbound responses strip `l5d-remote-ip`
- inbound requests strip `l5d-remote-ip`
- outbound responses set `l5d-remote-ip` to the endpoint IP
- inbound requests set `l5d-remote-ip` to the source IP

Closes linkerd/linkerd2#2310

Signed-off-by: Kevin Leimkuhler <kevinl@buoyant.io>

Author: kleimkuhler

src/app/inbound.rs

@@ -261,130 +261,50 @@ pub mod rewrite_loopback_addr {
 
 /// Adds `l5d-client-id` headers to http::Requests derived from the
 /// TlsIdentity of a `Source`.
-pub mod client_id {
-    use std::marker::PhantomData;
-
-    use futures::Poll;
-    use http::{self, header::HeaderValue};
-
-    use proxy::server::Source;
-    use svc;
+pub mod set_client_id_on_req {
+    use super::super::L5D_CLIENT_ID;
+    use http::header::HeaderValue;
+
+    use proxy::{
+        http::add_header::{self, request::ReqHeader, Layer},
+        server::Source,
+    };
     use Conditional;
 
-    #[derive(Debug)]
-    pub struct Layer<B>(PhantomData<fn() -> B>);
-
-    #[derive(Debug)]
-    pub struct Stack<M, B> {
-        inner: M,
-        _marker: PhantomData<fn() -> B>,
-    }
-
-    #[derive(Debug)]
-    pub struct Service<S, B> {
-        inner: S,
-        value: HeaderValue,
-        _marker: PhantomData<fn() -> B>,
-    }
-
-    pub fn layer<B>() -> Layer<B> {
-        Layer(PhantomData)
-    }
-
-    impl<B> Clone for Layer<B> {
-        fn clone(&self) -> Self {
-            Layer(PhantomData)
-        }
-    }
-
-    impl<M, B> svc::Layer<Source, Source, M> for Layer<B>
-    where
-        M: svc::Stack<Source>,
-    {
-        type Value = <Stack<M, B> as svc::Stack<Source>>::Value;
-        type Error = <Stack<M, B> as svc::Stack<Source>>::Error;
-        type Stack = Stack<M, B>;
-
-        fn bind(&self, inner: M) -> Self::Stack {
-            Stack {
-                inner,
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    // === impl Stack ===
-
-    impl<M: Clone, B> Clone for Stack<M, B> {
-        fn clone(&self) -> Self {
-            Stack {
-                inner: self.inner.clone(),
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    impl<M, B> svc::Stack<Source> for Stack<M, B>
-    where
-        M: svc::Stack<Source>,
-    {
-        type Value = svc::Either<Service<M::Value, B>, M::Value>;
-        type Error = M::Error;
-
-        fn make(&self, source: &Source) -> Result<Self::Value, Self::Error> {
-            let svc = self.inner.make(source)?;
-
+    pub fn layer() -> Layer<&'static str, Source, ReqHeader> {
+        add_header::request::layer(L5D_CLIENT_ID, |source: &Source| {
             if let Conditional::Some(ref id) = source.tls_peer {
                 match HeaderValue::from_str(id.as_ref()) {
                     Ok(value) => {
                         debug!("l5d-client-id enabled for {:?}", source);
-                        return Ok(svc::Either::A(Service {
-                            inner: svc,
-                            value,
-                            _marker: PhantomData,
-                        }));
+                        return Some(value);
                     }
                     Err(_err) => {
                         warn!("l5d-client-id identity header is invalid: {:?}", source);
                     }
-                }
+                };
             }
 
-            trace!("l5d-client-id not enabled for {:?}", source);
-            Ok(svc::Either::B(svc))
-        }
-    }
-
-    // === impl Service ===
-
-    impl<S: Clone, B> Clone for Service<S, B> {
-        fn clone(&self) -> Self {
-            Service {
-                inner: self.inner.clone(),
-                value: self.value.clone(),
-                _marker: PhantomData,
-            }
-        }
+            None
+        })
     }
+}
 
-    impl<S, B> svc::Service<http::Request<B>> for Service<S, B>
-    where
-        S: svc::Service<http::Request<B>>,
-    {
-        type Response = S::Response;
-        type Error = S::Error;
-        type Future = S::Future;
-
-        fn poll_ready(&mut self) -> Poll<(), Self::Error> {
-            self.inner.poll_ready()
-        }
-
-        fn call(&mut self, mut req: http::Request<B>) -> Self::Future {
-            req.headers_mut()
-                .insert(super::super::L5D_CLIENT_ID, self.value.clone());
-
-            self.inner.call(req)
-        }
+/// Adds `l5d-remote-ip` headers to http::Requests derived from the
+/// `remote` of a `Source`.
+pub mod set_remote_ip_on_req {
+    use super::super::L5D_REMOTE_IP;
+    use bytes::Bytes;
+    use http::header::HeaderValue;
+    use proxy::{
+        http::add_header::{self, request::ReqHeader, Layer},
+        server::Source,
+    };
+
+    pub fn layer() -> Layer<&'static str, Source, ReqHeader> {
+        add_header::request::layer(L5D_REMOTE_IP, |source: &Source| {
+            HeaderValue::from_shared(Bytes::from(source.remote.ip().to_string())).ok()
+        })
     }
 }
 

src/app/main.rs

@@ -302,7 +302,8 @@ where
 
             let outbound = {
                 use super::outbound::{
-                    discovery::Resolve, orig_proto_upgrade, server_id, Endpoint,
+                    add_remote_ip_on_rsp, add_server_id_on_rsp, discovery::Resolve,
+                    orig_proto_upgrade, Endpoint,
                 };
                 use proxy::{
                     canonicalize,
@@ -348,8 +349,10 @@ where
                 let endpoint_stack = client_stack
                     .push(buffer::layer(MAX_IN_FLIGHT))
                     .push(strip_header::response::layer(super::L5D_SERVER_ID))
+                    .push(strip_header::response::layer(super::L5D_REMOTE_IP))
                     .push(settings::router::layer::<Endpoint, _>())
-                    .push(server_id::layer())
+                    .push(add_server_id_on_rsp::layer())
+                    .push(add_remote_ip_on_rsp::layer())
                     .push(orig_proto_upgrade::layer())
                     .push(tap_layer.clone())
                     .push(metrics::layer::<_, classify::Response>(
@@ -490,8 +493,8 @@ where
 
             let inbound = {
                 use super::inbound::{
-                    client_id, orig_proto_downgrade, rewrite_loopback_addr, Endpoint,
-                    RecognizeEndpoint,
+                    orig_proto_downgrade, rewrite_loopback_addr, set_client_id_on_req,
+                    set_remote_ip_on_req, Endpoint, RecognizeEndpoint,
                 };
 
                 let capacity = config.inbound_router_capacity;
@@ -609,9 +612,11 @@ where
                 let source_stack = dst_router
                     .push(orig_proto_downgrade::layer())
                     .push(insert_target::layer())
-                    .push(client_id::layer())
-                    .push(strip_header::response::layer(super::L5D_SERVER_ID))
+                    .push(set_remote_ip_on_req::layer())
+                    .push(strip_header::request::layer(super::L5D_REMOTE_IP))
+                    .push(set_client_id_on_req::layer())
                     .push(strip_header::request::layer(super::L5D_CLIENT_ID))
+                    .push(strip_header::response::layer(super::L5D_SERVER_ID))
                     .push(strip_header::request::layer(super::DST_OVERRIDE_HEADER));
 
                 // As the inbound proxy accepts connections, we don't do any

src/app/mod.rs

@@ -17,6 +17,7 @@ use addr::{self, Addr};
 
 const CANONICAL_DST_HEADER: &'static str = "l5d-dst-canonical";
 pub const DST_OVERRIDE_HEADER: &'static str = "l5d-dst-override";
+const L5D_REMOTE_IP: &'static str = "l5d-remote-ip";
 const L5D_SERVER_ID: &'static str = "l5d-server-id";
 const L5D_CLIENT_ID: &'static str = "l5d-client-id";
 

src/app/outbound.rs

@@ -271,154 +271,44 @@ pub mod orig_proto_upgrade {
 
 /// Adds `l5d-server-id` headers to http::Responses derived from the
 /// TlsIdentity of an `Endpoint`.
-pub mod server_id {
-    use std::marker::PhantomData;
-
-    use futures::{Future, Poll};
-    use http::{self, header::HeaderValue};
-
+pub mod add_server_id_on_rsp {
+    use super::super::L5D_SERVER_ID;
     use super::Endpoint;
-    use proxy::http::ClientUsedTls;
-    use svc;
+    use http::header::HeaderValue;
+    use proxy::http::add_header::{self, response::ResHeader, Layer};
     use Conditional;
 
-    #[derive(Debug)]
-    pub struct Layer<B>(PhantomData<fn() -> B>);
-
-    #[derive(Debug)]
-    pub struct Stack<M, B> {
-        inner: M,
-        _marker: PhantomData<fn() -> B>,
-    }
-
-    #[derive(Debug)]
-    pub struct Service<S, B> {
-        inner: S,
-        value: HeaderValue,
-        _marker: PhantomData<fn() -> B>,
-    }
-
-    pub struct ResponseFuture<F> {
-        inner: F,
-        value: HeaderValue,
-    }
-
-    pub fn layer<B>() -> Layer<B> {
-        Layer(PhantomData)
-    }
-
-    impl<B> Clone for Layer<B> {
-        fn clone(&self) -> Self {
-            Layer(PhantomData)
-        }
-    }
-
-    impl<M, B> svc::Layer<Endpoint, Endpoint, M> for Layer<B>
-    where
-        M: svc::Stack<Endpoint>,
-    {
-        type Value = <Stack<M, B> as svc::Stack<Endpoint>>::Value;
-        type Error = <Stack<M, B> as svc::Stack<Endpoint>>::Error;
-        type Stack = Stack<M, B>;
-
-        fn bind(&self, inner: M) -> Self::Stack {
-            Stack {
-                inner,
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    // === impl Stack ===
-
-    impl<M: Clone, B> Clone for Stack<M, B> {
-        fn clone(&self) -> Self {
-            Stack {
-                inner: self.inner.clone(),
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    impl<M, B> svc::Stack<Endpoint> for Stack<M, B>
-    where
-        M: svc::Stack<Endpoint>,
-    {
-        type Value = svc::Either<Service<M::Value, B>, M::Value>;
-        type Error = M::Error;
-
-        fn make(&self, endpoint: &Endpoint) -> Result<Self::Value, Self::Error> {
-            let svc = self.inner.make(endpoint)?;
-
+    pub fn layer() -> Layer<&'static str, Endpoint, ResHeader> {
+        add_header::response::layer(L5D_SERVER_ID, |endpoint: &Endpoint| {
             if let Conditional::Some(id) = endpoint.connect.tls_server_identity() {
                 match HeaderValue::from_str(id.as_ref()) {
                     Ok(value) => {
                         debug!("l5d-server-id enabled for {:?}", endpoint);
-                        return Ok(svc::Either::A(Service {
-                            inner: svc,
-                            value,
-                            _marker: PhantomData,
-                        }));
+                        return Some(value);
                     }
                     Err(_err) => {
                         warn!("l5d-server-id identity header is invalid: {:?}", endpoint);
                     }
-                }
-            }
-
-            trace!("l5d-server-id not enabled for {:?}", endpoint);
-            Ok(svc::Either::B(svc))
-        }
-    }
-
-    // === impl Service ===
-
-    impl<S: Clone, B> Clone for Service<S, B> {
-        fn clone(&self) -> Self {
-            Service {
-                inner: self.inner.clone(),
-                value: self.value.clone(),
-                _marker: PhantomData,
+                };
             }
-        }
-    }
-
-    impl<S, B, Req> svc::Service<Req> for Service<S, B>
-    where
-        S: svc::Service<Req, Response = http::Response<B>>,
-    {
-        type Response = S::Response;
-        type Error = S::Error;
-        type Future = ResponseFuture<S::Future>;
-
-        fn poll_ready(&mut self) -> Poll<(), Self::Error> {
-            self.inner.poll_ready()
-        }
 
-        fn call(&mut self, req: Req) -> Self::Future {
-            let fut = self.inner.call(req);
-
-            ResponseFuture {
-                inner: fut,
-                value: self.value.clone(),
-            }
-        }
+            None
+        })
     }
+}
 
-    impl<F, B> Future for ResponseFuture<F>
-    where
-        F: Future<Item = http::Response<B>>,
-    {
-        type Item = F::Item;
-        type Error = F::Error;
-
-        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-            let mut res = try_ready!(self.inner.poll());
-            if res.extensions().get::<ClientUsedTls>().is_some() {
-                res.headers_mut()
-                    .insert(super::super::L5D_SERVER_ID, self.value.clone());
-            }
-            Ok(res.into())
-        }
+/// Adds `l5d-remote-ip` headers to http::Responses derived from the
+/// `remote` of a `Source`.
+pub mod add_remote_ip_on_rsp {
+    use super::super::L5D_REMOTE_IP;
+    use super::Endpoint;
+    use bytes::Bytes;
+    use http::header::HeaderValue;
+    use proxy::http::add_header::{self, response::ResHeader, Layer};
+
+    pub fn layer() -> Layer<&'static str, Endpoint, ResHeader> {
+        add_header::response::layer(L5D_REMOTE_IP, |endpoint: &Endpoint| {
+            HeaderValue::from_shared(Bytes::from(endpoint.connect.addr.ip().to_string())).ok()
+        })
     }
 }