Shortcut Acl permission check (project -> org -> root) (#2917)

Shortcut Acl permission check (project -> org -> root) when address matches

Fixes #1916

delta/app/src/main/resources/akka.conf

@@ -80,9 +80,17 @@ akka {
 
 }
 
+# reference configuration at: https://github.com/altoo-ag/akka-kryo-serialization/blob/master/akka-kryo-serialization/src/main/resources/reference.conf
 akka-kryo-serialization {
   id-strategy = "automatic"
   implicit-registration-logging = true
   resolve-subclasses = false
   kryo-initializer = "ch.epfl.bluebrain.nexus.delta.service.serialization.KryoSerializerInit"
+  # The transformations that have be done while serialization
+  # Supported transformations: compression and encryption
+  # accepted values(comma separated if multiple): off | lz4 | deflate | aes
+  # Transformations occur in the order they are specified on serialization
+  # and reverse order on deserialization. For example: "lz4,aes"
+  # lz4 usually offers a good middle ground between size and performance.
+  post-serialization-transformations = "lz4"
 }

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/Acls.scala

@@ -239,9 +239,12 @@ trait Acls {
     * ''onError'' when it doesn't
     */
   def authorizeForOr[E](path: AclAddress, permission: Permission)(onError: => E)(implicit caller: Caller): IO[E, Unit] =
-    fetchWithAncestors(path).flatMap { acls =>
-      IO.raiseWhen(!acls.exists(caller.identities, permission, path))(onError)
-    }
+    path.ancestors
+      .foldM(false) {
+        case (false, address) => fetch(address).redeem(_ => false, _.value.hasPermission(caller.identities, permission))
+        case (true, _)        => UIO.pure(true)
+      }
+      .flatMap { result => IO.raiseWhen(!result)(onError) }
 
   /**
     * Checks whether a given [[Caller]] has the passed ''permission'' on the passed ''path''.
@@ -255,13 +258,26 @@ trait Acls {
     */
   def authorizeForEveryOr[E](path: AclAddress, permissions: Set[Permission])(
       onError: => E
-  )(implicit caller: Caller): IO[E, Unit] =
-    fetchWithAncestors(path).flatMap { acls =>
-      val hasAccess = permissions.toList
-        .foldM(false)((_, perm) => Option.when(acls.exists(caller.identities, perm, path))(true))
-        .getOrElse(false)
-      IO.raiseWhen(!hasAccess)(onError)
-    }
+  )(implicit caller: Caller): IO[E, Unit]                   =
+    path.ancestors
+      .foldM((false, Set.empty[Permission])) {
+        case ((true, set), _)        => UIO.pure((true, set))
+        case ((false, set), address) =>
+          fetch(address)
+            .redeem(
+              _ => (false, set),
+              { res =>
+                val resSet =
+                  res.value.value // fetch the set of permissions defined in the resource that apply to the caller
+                    .filter { case (identity, _) => caller.identities.contains(identity) }
+                    .values
+                    .foldLeft(Set.empty[Permission])(_ ++ _)
+                val sum = set ++ resSet // add it to the accumulated set
+                (permissions.forall(sum.contains), sum) // true if all permissions are found, recurse otherwise
+              }
+            )
+      }
+      .flatMap { case (result, _) => IO.raiseWhen(!result)(onError) }
 
   /**
     * Checks whether a given [[Caller]] has the ''permissions'' on the passed ''paths''.

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/model/acls/AclAddress.scala

@@ -22,6 +22,12 @@ sealed trait AclAddress extends Product with Serializable {
     */
   def parent: Option[AclAddress]
 
+  /**
+    * @return
+    *   an ordered list of ancestors (that includes this address) first to last being project to root
+    */
+  def ancestors: List[AclAddress]
+
   override def toString: String = string
 }
 
@@ -55,28 +61,29 @@ object AclAddress {
     */
   final case object Root extends AclAddress {
 
-    val string: String             = "/"
-    val parent: Option[AclAddress] = None
+    val string: String              = "/"
+    val parent: Option[AclAddress]  = None
+    val ancestors: List[AclAddress] = List(this)
   }
 
   /**
     * The organization level address.
     */
   final case class Organization(org: Label) extends AclAddress {
 
-    val string                     = s"/$org"
-    val parent: Option[AclAddress] = Some(Root)
-
+    val string                      = s"/$org"
+    val parent: Option[AclAddress]  = Some(Root)
+    val ancestors: List[AclAddress] = List(this, Root)
   }
 
   /**
     * The project level address.
     */
   final case class Project(org: Label, project: Label) extends AclAddress {
 
-    val string                     = s"/$org/$project"
-    val parent: Option[AclAddress] = Some(Organization(org))
-
+    val string                      = s"/$org/$project"
+    val parent: Option[AclAddress]  = Some(Organization(org))
+    val ancestors: List[AclAddress] = List(this, Organization(org), Root)
   }
 
   object Project {

delta/sdk/src/test/scala/ch/epfl/bluebrain/nexus/delta/sdk/model/acls/AclAddressSpec.scala

@@ -38,5 +38,16 @@ class AclAddressSpec extends AnyWordSpecLike with Matchers with AclFixtures with
         AclAddress.fromString(string).leftValue
       }
     }
+
+    "return the correct ancestor list" in {
+      val list = List(
+        Root        -> List(Root),
+        orgAddress  -> List(orgAddress, Root),
+        projAddress -> List(projAddress, orgAddress, Root)
+      )
+      forAll(list) { case (address, ancestors) =>
+        address.ancestors shouldEqual ancestors
+      }
+    }
   }
 }