Log all authorization errors (digital-asset#6857)

* Log all authorization errors CHANGELOG_BEGIN - [Ledger API Server] The ledger API server now prints detailed log messages whenever a request was rejected due to a failed authorization. CHANGELOG_END
kgeonw · Jul 28, 2020 · 46b87c3 · 46b87c3
1 parent 36a4b8a
commit 46b87c3
Show file tree

Hide file tree

Showing 4 changed files with 182 additions and 55 deletions.
diff --git a/.../ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/AuthorizationError.scala b/.../ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/AuthorizationError.scala
@@ -0,0 +1,56 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.api.auth
+
+import java.time.Instant
+
+sealed abstract class AuthorizationError {
+  def reason: String
+}
+
+object AuthorizationError {
+
+  final case class Expired(authorizedUntil: Instant, currentTime: Instant)
+      extends AuthorizationError {
+    override val reason =
+      s"Claims were valid until $authorizedUntil, current time is $currentTime."
+  }
+
+  case object ExpiredOnStream extends AuthorizationError {
+    override val reason = "Claims have expired after the result stream has started."
+  }
+
+  final case class InvalidLedger(authorized: String, actual: String) extends AuthorizationError {
+    override val reason =
+      s"Claims are only valid for ledgerId $authorized, actual ledgerId is $actual."
+  }
+
+  final case class InvalidParticipant(authorized: String, actual: String)
+      extends AuthorizationError {
+    override val reason =
+      s"Claims are only valid for participantId $authorized, actual participantId is $actual."
+  }
+
+  final case class InvalidApplication(authorized: String, actual: String)
+      extends AuthorizationError {
+    override val reason =
+      s"Claims are only valid for applicationId $authorized, actual applicationId is $actual."
+  }
+
+  case object MissingPublicClaim extends AuthorizationError {
+    override val reason = "Claims do not authorize the use of public services."
+  }
+
+  case object MissingAdminClaim extends AuthorizationError {
+    override val reason = "Claims do not authorize the use of administrative services."
+  }
+
+  final case class MissingReadClaim(party: String) extends AuthorizationError {
+    override val reason = s"Claims do not authorize to read data for party $party"
+  }
+
+  final case class MissingActClaim(party: String) extends AuthorizationError {
+    override val reason = s"Claims do not authorize to act as party $party"
+  }
+}
diff --git a/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Authorizer.scala b/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Authorizer.scala
@@ -9,6 +9,7 @@ import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
 import com.daml.ledger.api.v1.transaction_filter.TransactionFilter
 import com.daml.platform.server.api.validation.ErrorFactories.permissionDenied
 import io.grpc.stub.{ServerCallStreamObserver, StreamObserver}
+import org.slf4j.LoggerFactory
 
 import scala.concurrent.Future
 
@@ -17,29 +18,56 @@ import scala.concurrent.Future
   */
 final class Authorizer(now: () => Instant, ledgerId: String, participantId: String) {
 
+  private val logger = LoggerFactory.getLogger(this.getClass)
+
   /** Validates all properties of claims that do not depend on the request,
     * such as expiration time or ledger ID. */
-  private def valid(claims: Claims): Boolean =
-    claims.notExpired(now()) &&
-      claims.validForLedger(ledgerId) &&
-      claims.validForParticipant(participantId)
+  private def valid(claims: Claims): Either[AuthorizationError, Unit] =
+    for {
+      _ <- claims.notExpired(now())
+      _ <- claims.validForLedger(ledgerId)
+      _ <- claims.validForParticipant(participantId)
+    } yield {
+      ()
+    }
 
   def requirePublicClaimsOnStream[Req, Res](
       call: (Req, StreamObserver[Res]) => Unit): (Req, StreamObserver[Res]) => Unit =
     authorize(call) { claims =>
-      valid(claims) && claims.isPublic
+      for {
+        _ <- valid(claims)
+        _ <- claims.isPublic
+      } yield {
+        ()
+      }
     }
 
   def requirePublicClaims[Req, Res](call: Req => Future[Res]): Req => Future[Res] =
     authorize(call) { claims =>
-      valid(claims) && claims.isPublic
+      for {
+        _ <- valid(claims)
+        _ <- claims.isPublic
+      } yield {
+        ()
+      }
     }
 
   def requireAdminClaims[Req, Res](call: Req => Future[Res]): Req => Future[Res] =
     authorize(call) { claims =>
-      valid(claims) && claims.isAdmin
+      for {
+        _ <- valid(claims)
+        _ <- claims.isAdmin
+      } yield {
+        ()
+      }
     }
 
+  private[this] def requireForAll[T](
+      xs: TraversableOnce[T],
+      f: T => Either[AuthorizationError, Unit]): Either[AuthorizationError, Unit] = {
+    xs.foldLeft[Either[AuthorizationError, Unit]](Right(()))((acc, x) => acc.flatMap(_ => f(x)))
+  }
+
   /** Wraps a streaming call to verify whether some Claims authorize to read as all parties
     * of the given set. Authorization is always granted for an empty collection of parties.
     */
@@ -48,9 +76,13 @@ final class Authorizer(now: () => Instant, ledgerId: String, participantId: Stri
       applicationId: Option[String],
       call: (Req, StreamObserver[Res]) => Unit): (Req, StreamObserver[Res]) => Unit =
     authorize(call) { claims =>
-      valid(claims) &&
-      parties.forall(claims.canReadAs) &&
-      applicationId.forall(claims.validForApplication)
+      for {
+        _ <- valid(claims)
+        _ <- requireForAll(parties, party => claims.canReadAs(party))
+        _ <- applicationId.map(claims.validForApplication).getOrElse(Right(()))
+      } yield {
+        ()
+      }
     }
 
   /** Wraps a single call to verify whether some Claims authorize to read as all parties
@@ -60,8 +92,12 @@ final class Authorizer(now: () => Instant, ledgerId: String, participantId: Stri
       parties: Iterable[String],
       call: Req => Future[Res]): Req => Future[Res] =
     authorize(call) { claims =>
-      valid(claims) &&
-      parties.forall(claims.canReadAs)
+      for {
+        _ <- valid(claims)
+        _ <- requireForAll(parties, party => claims.canReadAs(party))
+      } yield {
+        ()
+      }
     }
 
   /** Checks whether the current Claims authorize to act as the given party, if any.
@@ -72,9 +108,13 @@ final class Authorizer(now: () => Instant, ledgerId: String, participantId: Stri
       applicationId: Option[String],
       call: Req => Future[Res]): Req => Future[Res] =
     authorize(call) { claims =>
-      valid(claims) &&
-      party.forall(claims.canActAs) &&
-      applicationId.forall(claims.validForApplication)
+      for {
+        _ <- valid(claims)
+        _ <- party.map(claims.canActAs).getOrElse(Right(()))
+        _ <- applicationId.map(claims.validForApplication).getOrElse(Right(()))
+      } yield {
+        ()
+      }
     }
 
   /** Checks whether the current Claims authorize to read data for all parties mentioned in the given transaction filter */
@@ -96,10 +136,18 @@ final class Authorizer(now: () => Instant, ledgerId: String, participantId: Stri
     }
 
   private def ongoingAuthorization[Res](scso: ServerCallStreamObserver[Res], claims: Claims) =
-    new OngoingAuthorizationObserver[Res](scso, claims, _.notExpired(now()), permissionDenied())
+    new OngoingAuthorizationObserver[Res](
+      scso,
+      claims,
+      _.notExpired(now()),
+      authorizationError => {
+        logger.warn(s"Permission denied. Reason: ${authorizationError.reason}.")
+        permissionDenied()
+      }
+    )
 
   private def authorize[Req, Res](call: (Req, ServerCallStreamObserver[Res]) => Unit)(
-      authorized: Claims => Boolean,
+      authorized: Claims => Either[AuthorizationError, Unit],
   ): (Req, StreamObserver[Res]) => Unit =
     (request, observer) => {
       val scso = assertServerCall(observer)
@@ -108,29 +156,37 @@ final class Authorizer(now: () => Instant, ledgerId: String, participantId: Stri
         .fold(
           observer.onError(_),
           claims =>
-            if (authorized(claims))
-              call(
-                request,
-                if (claims.expiration.isDefined)
-                  ongoingAuthorization(scso, claims)
-                else
-                  scso
-              )
-            else observer.onError(permissionDenied())
+            authorized(claims) match {
+              case Right(_) =>
+                call(
+                  request,
+                  if (claims.expiration.isDefined)
+                    ongoingAuthorization(scso, claims)
+                  else
+                    scso
+                )
+              case Left(authorizationError) =>
+                logger.warn(s"Permission denied. Reason: ${authorizationError.reason}.")
+                observer.onError(permissionDenied())
+          }
         )
     }
 
   private def authorize[Req, Res](call: Req => Future[Res])(
-      authorized: Claims => Boolean,
+      authorized: Claims => Either[AuthorizationError, Unit],
   ): Req => Future[Res] =
     request =>
       AuthorizationInterceptor
         .extractClaimsFromContext()
         .fold(
           Future.failed,
           claims =>
-            if (authorized(claims)) call(request)
-            else Future.failed(permissionDenied())
+            authorized(claims) match {
+              case Right(_) => call(request)
+              case Left(authorizationError) =>
+                logger.warn(s"Permission denied. Reason: ${authorizationError.reason}.")
+                Future.failed(permissionDenied())
+          }
       )
 
 }
diff --git a/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Claims.scala b/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Claims.scala
@@ -87,44 +87,57 @@ final case class Claims(
     applicationId: Option[String] = None,
     expiration: Option[Instant] = None,
 ) {
-  def validForLedger(id: String): Boolean =
-    ledgerId.forall(_ == id)
+  def validForLedger(id: String): Either[AuthorizationError, Unit] =
+    Either.cond(ledgerId.forall(_ == id), (), AuthorizationError.InvalidLedger(ledgerId.get, id))
 
-  def validForParticipant(id: String): Boolean =
-    participantId.forall(_ == id)
+  def validForParticipant(id: String): Either[AuthorizationError, Unit] =
+    Either.cond(
+      participantId.forall(_ == id),
+      (),
+      AuthorizationError.InvalidParticipant(participantId.get, id))
 
-  def validForApplication(id: String): Boolean =
-    applicationId.forall(_ == id)
+  def validForApplication(id: String): Either[AuthorizationError, Unit] =
+    Either.cond(
+      applicationId.forall(_ == id),
+      (),
+      AuthorizationError.InvalidApplication(applicationId.get, id))
 
   /** Returns false if the expiration timestamp exists and is greater than or equal to the current time */
-  def notExpired(now: Instant): Boolean =
-    expiration.forall(now.isBefore)
+  def notExpired(now: Instant): Either[AuthorizationError, Unit] =
+    Either.cond(
+      expiration.forall(now.isBefore),
+      (),
+      AuthorizationError.Expired(expiration.get, now))
 
   /** Returns true if the set of claims authorizes the user to use admin services, unless the claims expired */
-  def isAdmin: Boolean =
-    claims.contains(ClaimAdmin)
+  def isAdmin: Either[AuthorizationError, Unit] =
+    Either.cond(claims.contains(ClaimAdmin), (), AuthorizationError.MissingAdminClaim)
 
   /** Returns true if the set of claims authorizes the user to use public services, unless the claims expired */
-  def isPublic: Boolean =
-    claims.contains(ClaimPublic)
+  def isPublic: Either[AuthorizationError, Unit] =
+    Either.cond(claims.contains(ClaimPublic), (), AuthorizationError.MissingPublicClaim)
 
   /** Returns true if the set of claims authorizes the user to act as the given party, unless the claims expired */
-  def canActAs(party: String): Boolean = {
-    claims.exists {
+  def canActAs(party: String): Either[AuthorizationError, Unit] = {
+    Either.cond(claims.exists {
       case ClaimActAsAnyParty => true
       case ClaimActAsParty(p) if p == party => true
       case _ => false
-    }
+    }, (), AuthorizationError.MissingActClaim(party))
   }
 
   /** Returns true if the set of claims authorizes the user to read data for the given party, unless the claims expired */
-  def canReadAs(party: String): Boolean = {
-    claims.exists {
-      case ClaimActAsAnyParty => true
-      case ClaimActAsParty(p) if p == party => true
-      case ClaimReadAsParty(p) if p == party => true
-      case _ => false
-    }
+  def canReadAs(party: String): Either[AuthorizationError, Unit] = {
+    Either.cond(
+      claims.exists {
+        case ClaimActAsAnyParty => true
+        case ClaimActAsParty(p) if p == party => true
+        case ClaimReadAsParty(p) if p == party => true
+        case _ => false
+      },
+      (),
+      AuthorizationError.MissingReadClaim(party)
+    )
   }
 }
 

diff --git a/...i-auth/src/main/scala/com/digitalasset/ledger/api/auth/OngoingAuthorizationObserver.scala b/...i-auth/src/main/scala/com/digitalasset/ledger/api/auth/OngoingAuthorizationObserver.scala
@@ -8,8 +8,8 @@ import io.grpc.stub.ServerCallStreamObserver
 private[auth] final class OngoingAuthorizationObserver[A](
     observer: ServerCallStreamObserver[A],
     claims: Claims,
-    authorized: Claims => Boolean,
-    throwOnFailure: => Throwable)
+    authorized: Claims => Either[AuthorizationError, Unit],
+    throwOnFailure: AuthorizationError => Throwable)
     extends ServerCallStreamObserver[A] {
 
   override def isCancelled: Boolean = observer.isCancelled
@@ -29,8 +29,10 @@ private[auth] final class OngoingAuthorizationObserver[A](
   override def setMessageCompression(b: Boolean): Unit = observer.setMessageCompression(b)
 
   override def onNext(v: A): Unit =
-    if (authorized(claims)) observer.onNext(v)
-    else observer.onError(throwOnFailure)
+    authorized(claims) match {
+      case Right(_) => observer.onNext(v)
+      case Left(authorizationError) => observer.onError(throwOnFailure(authorizationError))
+    }
 
   override def onError(throwable: Throwable): Unit = observer.onError(throwable)