handle AuthorizationFailedRejection (#1339)

Update the default handler so it will use a 401 code
rather than 400. Also adds a custom rejection type
that can be used for custom directives to provide more
context into why the authorization failed.

atlas-akka/src/main/scala/com/netflix/atlas/akka/CustomRejection.scala

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2014-2021 Netflix, Inc.
+ *
+ * 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.
+ */
+package com.netflix.atlas.akka
+
+import akka.http.scaladsl.model.StatusCode
+import akka.http.scaladsl.server.Rejection
+
+/**
+  * Rejection type that can be used with the RequestHandler to fully customize the status
+  * code and diagnostic message returned to the user.
+  */
+case class CustomRejection(status: StatusCode, message: String) extends Rejection

atlas-akka/src/main/scala/com/netflix/atlas/akka/RequestHandler.scala

@@ -25,6 +25,7 @@ import akka.http.scaladsl.model.StatusCode
 import akka.http.scaladsl.model.StatusCodes
 import akka.http.scaladsl.model.headers.RawHeader
 import akka.http.scaladsl.server.AuthenticationFailedRejection
+import akka.http.scaladsl.server.AuthorizationFailedRejection
 import akka.http.scaladsl.server.CircuitBreakerOpenRejection
 import akka.http.scaladsl.server.Directives._
 import akka.http.scaladsl.server.MalformedRequestContentRejection
@@ -226,6 +227,10 @@ object RequestHandler extends StrictLogging {
           error(StatusCodes.MethodNotAllowed, s"method not allowed: ${m.name()}")
         case AuthenticationFailedRejection(_, _) =>
           error(StatusCodes.Forbidden, "not authorized")
+        case AuthorizationFailedRejection =>
+          error(StatusCodes.Unauthorized, "not authorized")
+        case CustomRejection(status, message) =>
+          error(status, message)
         case r: Rejection =>
           error(StatusCodes.BadRequest, r.toString)
       }

atlas-akka/src/test/scala/com/netflix/atlas/akka/RequestHandlerSuite.scala

@@ -223,4 +223,10 @@ class RequestHandlerSuite extends AnyFunSuite with ScalatestRouteTest {
       assert(response.status === StatusCodes.ServiceUnavailable)
     }
   }
+
+  test("authorization rejection") {
+    Post("/unauthorized") ~> routes ~> check {
+      assert(response.status === StatusCodes.Unauthorized)
+    }
+  }
 }

atlas-akka/src/test/scala/com/netflix/atlas/akka/TestApi.scala

@@ -92,6 +92,11 @@ class TestApi(val system: ActorSystem) extends WebApi {
           case Failure(e) => complete(StatusCodes.InternalServerError, e.getMessage)
         }
       }
+    } ~
+    path("unauthorized") {
+      authorize(false) {
+        complete(StatusCodes.OK)
+      }
     }
   }
 }