Merge pull request #252 from specklesystems/dogukan/implicit-access

fix: checks for permissions

bpy_speckle/connector/blender_operators/add_project_by_url.py

@@ -1,9 +1,11 @@
 import bpy
 from bpy.types import Context, Event, UILayout, WindowManager
-from specklepy.api.wrapper import StreamWrapper
-from typing import Tuple
+from ..utils.account_manager import (
+    get_model_details_by_wrapper,
+    check_project_permissions,
+    get_project_from_url,
+)
 
-from ...connector.utils.version_manager import get_latest_version
 
 class SPECKLE_OT_add_project_by_url(bpy.types.Operator):
     """
@@ -19,18 +21,35 @@ class SPECKLE_OT_add_project_by_url(bpy.types.Operator):
     )
 
     def execute(self, context: Context) -> set[str]:
-        # TODO: Implement logic to add project using the URL
         self.report({"INFO"}, f"Adding project from URL: {self.url}")
 
         wm = context.window_manager
-        try:
-            wrapper = StreamWrapper(self.url)
-        except Exception as e:
-            self.report({"ERROR"}, f"Failed to process URL: {str(e)}")
-            return {"CANCELLED"}
-        # Get model details from the wrapper
-        account_id, project_id, project_name, model_id, model_name, version_id, load_option = get_model_details_by_wrapper(wrapper)
 
+        # Get project from URL
+        wrapper, client, project, error_message = get_project_from_url(self.url)
+
+        if error_message:
+            self.report({"ERROR"}, error_message)
+            return {"CANCELLED"}
+
+        # Get model details from the wrapper
+        (
+            account_id,
+            project_id,
+            project_name,
+            model_id,
+            model_name,
+            version_id,
+            load_option,
+        ) = get_model_details_by_wrapper(wrapper)
+
+        # Check permissions
+        can_receive, permission_error = check_project_permissions(client, project)
+        if not can_receive:
+            self.report({"ERROR"}, permission_error)
+            return {"CANCELLED"}
+
+        # Update the window manager with the selected project/model/version
         wm.selected_account_id = account_id
 
         if project_id:
@@ -43,6 +62,7 @@ class SPECKLE_OT_add_project_by_url(bpy.types.Operator):
                     wm.selected_version_id = version_id
             wm.selected_version_id = version_id
             wm.selected_version_load_option = load_option
+
         context.window.screen = context.window.screen
         context.area.tag_redraw()
         return {"FINISHED"}
@@ -85,25 +105,3 @@ class SPECKLE_OT_add_project_by_url(bpy.types.Operator):
     def draw(self, context: Context) -> None:
         layout: UILayout = self.layout
         layout.prop(self, "url", text="")
-
-def register() -> None:
-    bpy.utils.register_class(SPECKLE_OT_add_project_by_url)
-
-def unregister() -> None:
-    bpy.utils.unregister_class(SPECKLE_OT_add_project_by_url)
-
-def get_model_details_by_wrapper(wrapper: StreamWrapper) -> Tuple[str, str, str, str, str, str]:
-    client = wrapper.get_client()
-    client.authenticate_with_account(wrapper.get_account())
-    account_id, project_id, project_name, model_id, model_name, version_id, load_option = "", "", "", "", "", "", ""
-    account_id = wrapper.get_account().id
-    if wrapper.stream_id:
-        project_id = wrapper.stream_id
-        project_name = client.project.get(project_id).name
-    if wrapper.model_id:
-        model_id = wrapper.model_id
-        model = client.model.get(model_id, project_id)
-        model_name = model.name
-        load_option = "LATEST" if not wrapper.commit_id else "SPECIFIC"
-        version_id = wrapper.commit_id if wrapper.commit_id else client.version.get_versions(wrapper.model_id, wrapper.stream_id, limit = 1).items[0].id
-    return (account_id, project_id, project_name, model_id, model_name, version_id, load_option)

bpy_speckle/connector/ui/main_panel.py

@@ -99,17 +99,19 @@ class SPECKLE_PT_main_panel(bpy.types.Panel):
                 box: UILayout = project_box.box()
                 row: UILayout = box.row()
                 icon: str = "EXPORT" if model_card.is_publish else "IMPORT"
-                
+
                 # Load latest button in the model card
-                row.operator("speckle.load_latest", text="", icon=icon).model_card_id = model_card.get_model_card_id()
+                row.operator(
+                    "speckle.load_latest", text="", icon=icon
+                ).model_card_id = model_card.get_model_card_id()
                 row.label(text=f"{model_card.model_name}")
-                
+
                 # Select button in the model card
                 select_op = row.operator(
                     "speckle.select_objects", text="", icon="RESTRICT_SELECT_OFF"
                 )
                 select_op.model_card_id = model_card.get_model_card_id()
-                
+
                 # Settings button in the model card
                 row.operator(
                     "speckle.model_card_settings", text="", icon="PREFERENCES"
@@ -125,7 +127,9 @@ class SPECKLE_PT_main_panel(bpy.types.Panel):
                     # TODO: Connect to version operator
                     if model_card.load_option == "LATEST":
                         split.operator("speckle.load", text="Latest")
+                        split.enabled = False
                     if model_card.load_option == "SPECIFIC":
                         split.operator("speckle.load", text=f"{model_card.version_id}")
+                        split.enabled = False
                     # TODO: Get last updated time
                     split.label(text="Last updated: 2 days ago")

bpy_speckle/connector/ui/project_selection_dialog.py

@@ -1,38 +1,39 @@
 import bpy
 from bpy.types import UILayout, Context, PropertyGroup, Event
 from typing import List, Tuple
-from ..utils.account_manager import get_account_enum_items, speckle_account, get_workspaces, speckle_workspace, get_account_from_id
+from ..utils.account_manager import (
+    get_account_enum_items,
+    speckle_account,
+    get_workspaces,
+    speckle_workspace,
+)
 from ..utils.project_manager import get_projects_for_account
 
+
 def get_accounts_callback(self, context):
-    """Callback to dynamically fetch account enum items.
-    """
+    """Callback to dynamically fetch account enum items."""
     wm = context.window_manager
     return [
         (
             account.id,
             f"{account.user_name} - {account.user_email} - {account.server_url}",
-            ""
+            "",
         )
         for account in wm.speckle_accounts
     ]
 
+
 def get_workspaces_callback(self, context):
     """
     Callback to dynamically fetch workspace enum items.
     """
     wm = context.window_manager
     return [
-        (
-            workspace.id,
-            workspace.name,
-            "",
-            "WORKSPACE",
-            i
-        )
+        (workspace.id, workspace.name, "", "WORKSPACE", i)
         for i, workspace in enumerate(wm.speckle_workspaces)
     ]
 
+
 class speckle_project(bpy.types.PropertyGroup):
     """
     PropertyGroup for storing project information
@@ -42,6 +43,7 @@ class speckle_project(bpy.types.PropertyGroup):
     role: bpy.props.StringProperty(name="Role")  # type: ignore
     updated: bpy.props.StringProperty(name="Updated")  # type: ignore
     id: bpy.props.StringProperty(name="ID")  # type: ignore
+    can_receive: bpy.props.BoolProperty(name="Can Receive", default=False)  # type: ignore
 
 
 class SPECKLE_UL_projects_list(bpy.types.UIList):
@@ -61,6 +63,9 @@ class SPECKLE_UL_projects_list(bpy.types.UIList):
     ) -> None:
         if self.layout_type in {"DEFAULT", "COMPACT"}:
             row = layout.row(align=True)
+            # enable/disable the row based on permission
+            row.enabled = item.can_receive
+
             split = row.split(factor=0.5)
             split.label(text=item.name)
 
@@ -71,6 +76,7 @@ class SPECKLE_UL_projects_list(bpy.types.UIList):
         # handles when the list is in a grid layout
         elif self.layout_type == "GRID":
             layout.alignment = "CENTER"
+            layout.enabled = item.can_receive
             layout.label(text=item.name)
 
 
@@ -97,16 +103,17 @@ class SPECKLE_OT_project_selection_dialog(bpy.types.Operator):
 
         # get projects for the selected account, using search if provided
         search = self.search_query if self.search_query.strip() else None
-        projects: List[Tuple[str, str, str, str]] = get_projects_for_account(
+        projects: List[Tuple[str, str, str, str, bool]] = get_projects_for_account(
             self.accounts, search=search, workspace_id=self.workspaces
         )
 
-        for name, role, updated, id in projects:
+        for name, role, updated, id, can_receive in projects:
             project: speckle_project = wm.speckle_projects.add()
             project.name = name
             project.role = role
             project.updated = updated
             project.id = id
+            project.can_receive = can_receive
         print("Updated Projects List!")
 
         return None
@@ -124,16 +131,17 @@ class SPECKLE_OT_project_selection_dialog(bpy.types.Operator):
 
         # get projects for the selected account, using search if provided
         search = self.search_query if self.search_query.strip() else None
-        projects: List[Tuple[str, str, str, str]] = get_projects_for_account(
+        projects: List[Tuple[str, str, str, str, bool]] = get_projects_for_account(
             self.accounts, search=search, workspace_id=self.workspaces
         )
 
-        for name, role, updated, id in projects:
+        for name, role, updated, id, can_receive in projects:
             project: speckle_project = wm.speckle_projects.add()
             project.name = name
             project.role = role
             project.updated = updated
             project.id = id
+            project.can_receive = can_receive
         print("Updated Projects List!")
         return None
 
@@ -155,9 +163,9 @@ class SPECKLE_OT_project_selection_dialog(bpy.types.Operator):
         name="Workspace",
         description="Selected workspace to filter projects by",
         items=get_workspaces_callback,
-        update=update_projects_list
+        update=update_projects_list,
     )
-    
+
     project_index: bpy.props.IntProperty(name="Project Index", default=0)  # type: ignore
 
     def execute(self, context: Context) -> set[str]:
@@ -165,6 +173,14 @@ class SPECKLE_OT_project_selection_dialog(bpy.types.Operator):
         if 0 <= self.project_index < len(wm.speckle_projects):
             selected_project = wm.speckle_projects[self.project_index]
 
+            # verify the user has permission to receive from this project
+            if not selected_project.can_receive:
+                self.report(
+                    {"ERROR"},
+                    "Your role on this project doesn't give you permission to load.",
+                )
+                return {"CANCELLED"}
+
             wm.selected_project_id = selected_project.id
             wm.selected_project_name = selected_project.name
 
@@ -201,31 +217,40 @@ class SPECKLE_OT_project_selection_dialog(bpy.types.Operator):
         wm.selected_workspace_id = selected_workspace_id
 
         # Fetch projects from server
-        projects: List[Tuple[str, str, str, str]] = get_projects_for_account(
+        projects: List[Tuple[str, str, str, str, bool]] = get_projects_for_account(
             selected_account_id, workspace_id=selected_workspace_id
         )
 
-        for name, role, updated, id in projects:
+        for name, role, updated, id, can_receive in projects:
             project: speckle_project = wm.speckle_projects.add()
             project.name = name
             project.role = role
             project.updated = updated
             project.id = id
+            project.can_receive = can_receive
 
         return context.window_manager.invoke_props_dialog(self)
 
     def draw(self, context: Context) -> None:
         layout: UILayout = self.layout
         wm = context.window_manager
-        
+
         # Account selection
         row = layout.row()
         if wm.selected_account_id != "NO_ACCOUNTS":
             row.prop(self, "accounts", text="")
-        add_account_button_text = "Sign In" if wm.selected_account_id == "NO_ACCOUNTS" else ""
-        add_account_button_icon = 'WORLD' if wm.selected_account_id == "NO_ACCOUNTS" else 'ADD'
-        row.operator("speckle.add_account", icon=add_account_button_icon, text=add_account_button_text)
-        
+        add_account_button_text = (
+            "Sign In" if wm.selected_account_id == "NO_ACCOUNTS" else ""
+        )
+        add_account_button_icon = (
+            "WORLD" if wm.selected_account_id == "NO_ACCOUNTS" else "ADD"
+        )
+        row.operator(
+            "speckle.add_account",
+            icon=add_account_button_icon,
+            text=add_account_button_text,
+        )
+
         # Workspace selection
         row = layout.row()
         if wm.selected_workspace_id != "NO_WORKSPACES":
@@ -234,7 +259,7 @@ class SPECKLE_OT_project_selection_dialog(bpy.types.Operator):
         # Search field
         row = layout.row(align=True)
         row.prop(self, "search_query", icon="VIEWZOOM", text="")
-        row.operator("speckle.add_project_by_url", icon='LINKED', text="")
+        row.operator("speckle.add_project_by_url", icon="LINKED", text="")
 
         layout.template_list(
             "SPECKLE_UL_projects_list",
@@ -247,7 +272,6 @@ class SPECKLE_OT_project_selection_dialog(bpy.types.Operator):
         layout.separator()
 
 
-
 def register() -> None:
     bpy.utils.register_class(speckle_project)
     bpy.utils.register_class(SPECKLE_UL_projects_list)
@@ -255,7 +279,6 @@ def register() -> None:
 
 
 def unregister() -> None:
-
     bpy.utils.unregister_class(SPECKLE_OT_project_selection_dialog)
     bpy.utils.unregister_class(SPECKLE_UL_projects_list)
     bpy.utils.unregister_class(speckle_project)

bpy_speckle/connector/utils/account_manager.py

@@ -3,6 +3,8 @@ from specklepy.api.credentials import get_local_accounts
 from typing import List, Tuple, Optional
 from specklepy.core.api.credentials import Account
 from specklepy.api.client import SpeckleClient
+from specklepy.api.wrapper import StreamWrapper
+
 from .misc import strip_non_ascii
 
 
@@ -55,7 +57,7 @@ def get_workspaces(account_id: str) -> List[Tuple[str, str]]:
         workspace_list = [
             (ws.id, strip_non_ascii(ws.name))
             for ws in workspaces
-            if ws.creation_state == None or ws.creation_state.completed
+            if ws.creation_state is None or ws.creation_state.completed
         ]
         personal_projects_text = "Personal Projects (Legacy)"
     else:
@@ -123,3 +125,116 @@ def reorder_tuple(tuple_list, target_id):
     # If the target_id wasn't found
     print(f"Tuple with ID {target_id} not found in the list")
     return tuple_list
+
+
+def get_project_from_url(
+    url: str,
+) -> Tuple[Optional[StreamWrapper], Optional[object], Optional[object], str]:
+    """
+    get a project from a URL, handling all the client setup.
+    """
+    try:
+        wrapper = StreamWrapper(url)
+        client = wrapper.get_client()
+        client.authenticate_with_account(wrapper.get_account())
+
+        # get the stream_id (project_id) from the wrapper
+        if not wrapper.stream_id:
+            return wrapper, client, None, "Could not extract project ID from URL"
+
+        project = client.project.get(wrapper.stream_id)
+
+        if not project:
+            return wrapper, client, None, "Could not access project"
+
+        return wrapper, client, project, ""
+
+    except Exception as e:
+        return None, None, None, f"Failed to process URL: {str(e)}"
+
+
+def get_model_details_by_wrapper(
+    wrapper: StreamWrapper,
+) -> Tuple[str, str, str, str, str, str, str]:
+    """
+    extract model details from a StreamWrapper object.
+    """
+    client = wrapper.get_client()
+    client.authenticate_with_account(wrapper.get_account())
+    (
+        account_id,
+        project_id,
+        project_name,
+        model_id,
+        model_name,
+        version_id,
+        load_option,
+    ) = "", "", "", "", "", "", ""
+    account_id = wrapper.get_account().id
+    if wrapper.stream_id:
+        project_id = wrapper.stream_id
+        project_name = client.project.get(project_id).name
+    if wrapper.model_id:
+        model_id = wrapper.model_id
+        model = client.model.get(model_id, project_id)
+        model_name = model.name
+        load_option = "LATEST" if not wrapper.commit_id else "SPECIFIC"
+        version_id = (
+            wrapper.commit_id
+            if wrapper.commit_id
+            else client.version.get_versions(
+                wrapper.model_id, wrapper.stream_id, limit=1
+            )
+            .items[0]
+            .id
+        )
+    return (
+        account_id,
+        project_id,
+        project_name,
+        model_id,
+        model_name,
+        version_id,
+        load_option,
+    )
+
+
+def check_project_permissions(client, project, workspace_id=None) -> Tuple[bool, str]:
+    """
+    check if the user has permission to receive from a project.
+    """
+    # check if user is workspace admin
+    is_workspace_admin = False
+    if workspace_id and workspace_id != "personal":
+        try:
+            workspace = client.workspace.get(workspace_id)
+            if workspace and workspace.role:
+                is_workspace_admin = "workspace:admin" in workspace.role
+        except Exception as e:
+            print(f"Cannot access workspace: {e}")
+
+    elif not workspace_id and hasattr(project, "workspace_id") and project.workspace_id:
+        try:
+            workspace = client.workspace.get(project.workspace_id)
+            if workspace and workspace.role:
+                is_workspace_admin = "workspace:admin" in workspace.role
+        except Exception as e:
+            print(f"Cannot access workspace: {e}")
+
+    # check permission
+    role = getattr(project, "role", "")
+    can_receive = False
+    error_message = ""
+
+    if role:
+        if is_workspace_admin:
+            can_receive = "stream:reviewer" not in role
+        else:
+            can_receive = any(r in role for r in ["stream:owner", "stream:contributor"])
+    else:
+        can_receive = is_workspace_admin
+
+    if not can_receive:
+        error_message = "Your role on this project doesn't give you permission to load."
+
+    return can_receive, error_message

bpy_speckle/connector/utils/project_manager.py

@@ -4,10 +4,12 @@ from specklepy.core.api.inputs.user_inputs import UserProjectsFilter
 from typing import List, Tuple, Optional
 from specklepy.core.api.credentials import Account
 from .misc import format_relative_time, format_role, strip_non_ascii
+from .account_manager import check_project_permissions
+
 
 def get_projects_for_account(
     account_id: str, workspace_id: str = None, search: Optional[str] = None
-) -> List[Tuple[str, str, str, str]]:
+) -> List[Tuple[str, str, str, str, bool]]:
     """
     fetches projects for a given account from the Speckle server
     """
@@ -24,20 +26,36 @@ def get_projects_for_account(
         client.authenticate_with_account(account)
 
         personal_only = workspace_id == "personal"
-        workspace_id = None if personal_only else workspace_id
-        filter = UserProjectsFilter(search=search, workspaceId=workspace_id, personalOnly=personal_only)
+        workspace_id_query = None if personal_only else workspace_id
+
+        # set include_implicit_access to True to get all projects
+        filter = UserProjectsFilter(
+            search=search,
+            workspaceId=workspace_id_query,
+            personalOnly=personal_only,
+            include_implicit_access=True,
+        )
 
         projects = client.active_user.get_projects(limit=10, filter=filter).items
 
-        return [
-            (
-                strip_non_ascii(project.name),
-                format_role(project.role),
-                format_relative_time(project.updated_at),
-                project.id,
+        # determine if user can receive from project based on role
+        result = []
+        for project in projects:
+            can_receive, _ = check_project_permissions(client, project, workspace_id)
+
+            result.append(
+                (
+                    strip_non_ascii(project.name),
+                    format_role(getattr(project, "role", ""))
+                    if hasattr(project, "role") and project.role
+                    else "",
+                    format_relative_time(project.updated_at),
+                    project.id,
+                    can_receive,
+                )
             )
-            for project in projects
-        ]
+
+        return result
 
     except Exception as e:
         import traceback