Type access policy methods, and introduce aliases.

Author: danielballan

tiled/_tests/test_protocols.py

@@ -28,7 +28,7 @@
 from ..structures.core import Spec, StructureFamily
 from ..structures.sparse import COOStructure
 from ..structures.table import TableStructure
-from ..type_aliases import JSON, Filters, Scopes
+from ..type_aliases import JSON, AccessBlob, AccessTags, Filters, Scopes
 
 
 class CustomArrayAdapter:
@@ -379,11 +379,30 @@ def __init__(self, scopes: Optional[Scopes] = None) -> None:
     def _get_id(self, principal: Principal) -> None:
         return None
 
+    async def init_node(
+        self,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        access_blob: Optional[AccessBlob] = None,
+    ) -> Tuple[bool, Optional[AccessBlob]]:
+        return (False, access_blob)
+
+    async def modify_node(
+        self,
+        node: BaseAdapter,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        access_blob: Optional[AccessBlob] = None,
+    ) -> Tuple[bool, Optional[AccessBlob]]:
+        return (False, access_blob)
+
     async def allowed_scopes(
         self,
         node: BaseAdapter,
         principal: Principal,
-        authn_access_tags: Optional[Set[str]],
+        authn_access_tags: Optional[AccessTags],
         authn_scopes: Scopes,
     ) -> Scopes:
         allowed = self.scopes
@@ -394,7 +413,7 @@ async def filters(
         self,
         node: BaseAdapter,
         principal: Principal,
-        authn_access_tags: Optional[Set[str]],
+        authn_access_tags: Optional[AccessTags],
         authn_scopes: Scopes,
         scopes: Scopes,
     ) -> Filters:

tiled/access_control/access_policies.py

@@ -1,7 +1,11 @@
 import logging
 import os
+from types import Optional, Tuple
 
+from ..adapters.protocols import BaseAdapter
 from ..queries import AccessBlobFilter
+from ..server.schemas import Principal
+from ..type_aliases import AccessBlob, AccessTags, Filters, Scopes
 from ..utils import Sentinel, import_object
 from .protocols import AccessPolicy
 from .scopes import ALL_SCOPES, PUBLIC_SCOPES
@@ -24,10 +28,35 @@
 class DummyAccessPolicy(AccessPolicy):
     "Impose no access restrictions."
 
-    async def allowed_scopes(self, node, principal, authn_access_tags, authn_scopes):
+    async def init_node(
+        self,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        access_blob: Optional[AccessBlob] = None,
+    ) -> Tuple[bool, AccessBlob]:
+        "Do nothing; there is no persistent state to initialize."
+        return (False, access_blob)
+
+    async def allowed_scopes(
+        self,
+        node: BaseAdapter,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+    ) -> Scopes:
+        "Always allow all scopes."
         return ALL_SCOPES
 
-    async def filters(self, node, principal, authn_access_tags, authn_scopes, scopes):
+    async def filters(
+        self,
+        node: BaseAdapter,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        scopes: Scopes,
+    ) -> Filters:
+        "Always impose no filtering on results."
         return []
 
 
@@ -74,8 +103,12 @@ def _is_admin(self, authn_scopes):
         return False
 
     async def init_node(
-        self, principal, authn_access_tags, authn_scopes, access_blob=None
-    ):
+        self,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        access_blob: Optional[AccessBlob] = None,
+    ) -> Tuple[bool, AccessBlob]:
         if principal.type == "service":
             identifier = str(principal.uuid)
         else:
@@ -157,8 +190,13 @@ async def init_node(
         return access_blob_modified, access_blob_from_policy
 
     async def modify_node(
-        self, node, principal, authn_access_tags, authn_scopes, access_blob
-    ):
+        self,
+        node: BaseAdapter,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        access_blob: Optional[AccessBlob],
+    ) -> Tuple[bool, AccessBlob]:
         if principal.type == "service":
             identifier = str(principal.uuid)
         else:
@@ -279,7 +317,13 @@ async def modify_node(
         # modified means the blob to-be-used was changed in comparison to the user input
         return access_blob_modified, access_blob_from_policy
 
-    async def allowed_scopes(self, node, principal, authn_access_tags, authn_scopes):
+    async def allowed_scopes(
+        self,
+        node: BaseAdapter,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+    ) -> Scopes:
         # If this is being called, filter_for_access has let us get this far.
         # However, filters and allowed_scopes should always be implemented to
         # give answers consistent with each other.
@@ -318,7 +362,14 @@ async def allowed_scopes(self, node, principal, authn_access_tags, authn_scopes)
 
         return allowed
 
-    async def filters(self, node, principal, authn_access_tags, authn_scopes, scopes):
+    async def filters(
+        self,
+        node: BaseAdapter,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        scopes: Scopes,
+    ) -> Filters:
         queries = []
         query_filter = AccessBlobFilter
 

tiled/access_control/protocols.py

@@ -1,18 +1,38 @@
 from abc import ABC, abstractmethod
-from typing import Any, Optional, Set
+from typing import Optional, Tuple
 
 from ..adapters.protocols import BaseAdapter
 from ..server.schemas import Principal
-from ..type_aliases import Filters, Scopes
+from ..type_aliases import AccessBlob, AccessTags, Filters, Scopes
 
 
 class AccessPolicy(ABC):
+    @abstractmethod
+    async def init_node(
+        self,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        access_blob: Optional[AccessBlob] = None,
+    ) -> Tuple[bool, Optional[AccessBlob]]:
+        pass
+
+    async def modify_node(
+        self,
+        node: BaseAdapter,
+        principal: Principal,
+        authn_access_tags: Optional[AccessTags],
+        authn_scopes: Scopes,
+        access_blob: Optional[AccessBlob],
+    ) -> Tuple[bool, Optional[AccessBlob]]:
+        return (False, access_blob)
+
     @abstractmethod
     async def allowed_scopes(
         self,
         node: BaseAdapter,
         principal: Principal,
-        authn_access_tags: Optional[Set[str]],
+        authn_access_tags: Optional[AccessTags],
         authn_scopes: Scopes,
     ) -> Scopes:
         pass
@@ -22,17 +42,8 @@ async def filters(
         self,
         node: BaseAdapter,
         principal: Principal,
-        authn_access_tags: Optional[Set[str]],
+        authn_access_tags: Optional[AccessTags],
         authn_scopes: Scopes,
         scopes: Scopes,
     ) -> Filters:
         pass
-
-    async def modify_node(
-        self,
-        node: BaseAdapter,
-        principal: Principal,
-        authn_scopes: Scopes,
-        access_blob: Optional[dict[str, Any]],
-    ) -> tuple[bool, Optional[dict[str, Any]]]:
-        return (False, access_blob)