Bugfix on ACLs

core/schemas/dfiq.py

@@ -13,7 +13,7 @@
 
 from core import database_arango
 from core.helpers import now
-from core.schemas import audit, indicator
+from core.schemas import audit, indicator, rbac
 from core.schemas.model import YetiAclModel, YetiModel
 
 LATEST_SUPPORTED_DFIQ_VERSION = "1.1.0"
@@ -88,6 +88,7 @@ def read_from_data_directory(
                 if not dfiq_object.uuid:
                     dfiq_object.uuid = str(uuid.uuid4())
                 dfiq_object = dfiq_object.save()
+                rbac.set_acls(dfiq_object)
                 dfiq_addition.dfiq.append(dfiq_object)
                 audit.log_timeline(user, dfiq_object, old=db_dfiq)
                 total_added += 1

core/schemas/model.py

@@ -96,18 +96,19 @@ def __init__(self, **data):
     def acls(self):
         return self._acls
 
-    def get_acls(self) -> None:
+    def get_acls(self, direct=False) -> None:
         """Returns the permissions assigned to a user.
         Args:
             user: The user to check permissions for.
         """
         vertices, paths, total = self.neighbors(
-            graph="acls", direction="inbound", max_hops=2
+            graph="acls", direction="inbound", max_hops=1 if direct else 2
         )
         for path in paths:
+            source = path[-1].source  # inbound, so source is last.
             for edge in path:
                 if edge.target == self.extended_id:
-                    identity = vertices[edge.source]
+                    identity = vertices[source]
                     if identity.root_type == "rbacgroup":
                         self._acls[identity.name] = edge
                     if identity.root_type == "user":

core/web/apiv2/rbac.py

@@ -62,7 +62,8 @@ def get_acl_details(httpreq: Request, type: str, id: str):
     if not db_entity:
         raise HTTPException(status_code=404, detail=f"{type} {id} not found")
 
-    return db_entity.acl
+    db_entity.get_acls(direct=True)
+    return db_entity
 
 
 @router.post("/{type}/{id}/update-members")

tests/schemas/rbac.py

@@ -11,12 +11,12 @@ def setUp(self) -> None:
         database_arango.RBAC_ENABLED = True
 
         self.yeti_user = user.User(username="yeti", admin=True).save()
-        self.user1 = user.User(username="test1").save()
-        self.user2 = user.User(username="test2").save()
-        self.group1 = rbac.Group(name="test1").save()
-        self.group2 = rbac.Group(name="test2").save()
-        self.entity1 = entity.Malware(name="test1").save()
-        self.entity2 = entity.Malware(name="test2").save()
+        self.user1 = user.User(username="user1").save()
+        self.user2 = user.User(username="user2").save()
+        self.group1 = rbac.Group(name="group1").save()
+        self.group2 = rbac.Group(name="group2").save()
+        self.entity1 = entity.Malware(name="malware1").save()
+        self.entity2 = entity.Malware(name="malware2").save()
         self.observable1 = observable.Hostname(value="test.com").save()
         self.observable1.link_to(self.entity1, "test", description="test")
 
@@ -115,3 +115,12 @@ def test_neighbors_filter_when_passing_username(self):
         vertices, edges, total = self.observable1.neighbors(user=self.user1)
         self.assertEqual(total, 1)
         self.assertEqual(len(vertices), 1)
+
+    def test_get_acls(self):
+        """Test that get_acls() returns the correct ACLs"""
+        self.user1.link_to_acl(self.group1, roles.Role.OWNER)
+        self.group1.link_to_acl(self.entity1, roles.Role.OWNER)
+        self.entity1.get_acls()
+        self.assertEqual(len(self.entity1._acls), 2)
+        self.assertIn(self.group1.name, self.entity1._acls)
+        self.assertIn(self.user1.username, self.entity1._acls)