inbound: Return HTTP-level authorization errors (#1220)

The initial policy/authorization implementation only operates on
connections. When an connection is not authorized, the connection is
dropped and no information is passed back to the client. This also means
that all authorizations are made at connect-time and policy
changes--especially those that revoke access--are not observed
until a new connection is established. This is at the very least
awkward when trying to test policies.

This change modifies the inbound proxy's behavior for HTTP connections:
when a server configures a port to use (or be detected) as HTTP, the
authorization decision is now deferred until a request is processed. As
each request is handled, the policy's state is checked to determine
whether the connection is still permitted. If the connection is not
authorized, a `403 Forbidden` status code is returned for HTTP
requests (and the grpc-status `PermissionDenied` is used for gRPC
requests). The http error metrics also now reflect an `unauthorized`
reason.

This change, unfortunately, adds some wrinkles to our transport metric
labeling: HTTP connections will no longer have a `saz_name`
annotation--as no authorization policy is associated with these
connections (though they will continue to include the `srv_name` from
when the connection was established).

Author: olix0r

linkerd/app/core/src/classify.rs

@@ -193,6 +193,7 @@ fn grpc_class(headers: &http::HeaderMap) -> Option<Class> {
                 | grpc::Code::DeadlineExceeded
                 | grpc::Code::Internal
                 | grpc::Code::Unavailable
+                | grpc::Code::PermissionDenied
                 | grpc::Code::DataLoss => SuccessOrFailure::Failure,
                 _ => SuccessOrFailure::Success,
             };

linkerd/app/core/src/errors.rs

@@ -1,3 +1,4 @@
+use crate::transport::DeniedUnauthorized;
 use http::{header::HeaderValue, StatusCode};
 use linkerd_errno::Errno;
 use linkerd_error::Error;
@@ -62,6 +63,7 @@ pub enum Reason {
     FailFast,
     GatewayLoop,
     NotFound,
+    Unauthorized,
     Unexpected,
 }
 
@@ -307,7 +309,7 @@ fn set_http_status(
         builder.status(StatusCode::SERVICE_UNAVAILABLE)
     } else if error.is::<tower::timeout::error::Elapsed>() {
         builder.status(StatusCode::SERVICE_UNAVAILABLE)
-    } else if error.is::<IdentityRequired>() {
+    } else if error.is::<IdentityRequired>() || error.is::<DeniedUnauthorized>() {
         builder.status(StatusCode::FORBIDDEN)
     } else if let Some(source) = error.source() {
         set_http_status(builder, source)
@@ -351,6 +353,13 @@ fn set_grpc_status(
             HeaderValue::from_static("proxy dispatch timed out"),
         );
         code
+    } else if error.is::<DeniedUnauthorized>() {
+        let code = Code::PermissionDenied;
+        headers.insert(GRPC_STATUS, code_header(code));
+        if let Ok(msg) = HeaderValue::from_str(&error.to_string()) {
+            headers.insert(GRPC_MESSAGE, msg);
+        }
+        code
     } else if error.is::<IdentityRequired>() {
         let code = Code::FailedPrecondition;
         headers.insert(GRPC_STATUS, code_header(code));
@@ -433,6 +442,8 @@ impl LabelError {
             Reason::FailFast
         } else if err.is::<tower::timeout::error::Elapsed>() {
             Reason::DispatchTimeout
+        } else if err.is::<DeniedUnauthorized>() {
+            Reason::Unauthorized
         } else if err.is::<IdentityRequired>() {
             Reason::IdentityRequired
         } else if let Some(e) = err.downcast_ref::<std::io::Error>() {
@@ -466,6 +477,7 @@ impl FmtLabels for Reason {
                 Reason::GatewayLoop => "gateway loop",
                 Reason::NotFound => "not found",
                 Reason::Io(_) => "i/o",
+                Reason::Unauthorized => "unauthorized",
                 Reason::Unexpected => "unexpected",
             }
         )?;

linkerd/app/inbound/src/accept.rs

@@ -120,7 +120,7 @@ mod tests {
     #[tokio::test(flavor = "current_thread")]
     async fn default_allow() {
         let (io, _) = io::duplex(1);
-        let policies = Store::fixed(
+        let (policies, _tx) = Store::fixed(
             ServerPolicy {
                 protocol: linkerd_server_policy::Protocol::Opaque,
                 authorizations: vec![Authorization {
@@ -144,7 +144,7 @@ mod tests {
 
     #[tokio::test(flavor = "current_thread")]
     async fn default_deny() {
-        let policies = Store::fixed(DefaultPolicy::Deny, None);
+        let (policies, _tx) = Store::fixed(DefaultPolicy::Deny, None);
         let (io, _) = io::duplex(1);
         inbound()
             .with_stack(new_ok())
@@ -158,7 +158,7 @@ mod tests {
 
     #[tokio::test(flavor = "current_thread")]
     async fn direct() {
-        let policies = Store::fixed(DefaultPolicy::Deny, None);
+        let (policies, _tx) = Store::fixed(DefaultPolicy::Deny, None);
         let (io, _) = io::duplex(1);
         inbound()
             .with_stack(new_panic("detect stack must not be built"))