[8.19] fix: restrict unpublished SharePoint Online pages to owners/editors (#3645) (#4437)

connectors/sources/sharepoint_online.py

@@ -103,6 +103,7 @@ def _is_excluded_sharepoint_url(url: str) -> bool:
 # See also: https://learn.microsoft.com/en-us/previous-versions/office/sharepoint-csom/ee536458(v=office.15)
 VIEW_ITEM_MASK = 0x1  # View items in lists, documents in document libraries, and Web discussion comments.
 VIEW_PAGE_MASK = 0x20000  # View pages in a Site.
+EDIT_ITEM_MASK = 0x4  # Edit items in lists, edit documents in document libraries, and customize Web Part Pages in document libraries.
 
 # See https://github.com/pnp/pnpcore/blob/dev/src/sdk/PnP.Core/Model/SharePoint/Core/Public/Enums/RoleType.cs
 # See also: https://learn.microsoft.com/en-us/dotnet/api/microsoft.sharepoint.client.roletype?view=sharepoint-csom
@@ -123,6 +124,13 @@ def _is_excluded_sharepoint_url(url: str) -> bool:
     REVIEWER,
     SYSTEM,
 ]
+EDIT_ROLE_TYPES = [
+    CONTRIBUTOR,
+    WEB_DESIGNER,
+    ADMINISTRATOR,
+    EDITOR,
+    SYSTEM,
+]
 
 # $expand param has a max of 20: see: https://developer.microsoft.com/en-us/graph/known-issues/?search=expand
 SPO_MAX_EXPAND_SIZE = 20
@@ -1296,6 +1304,23 @@ def _parse_created_date_time(created_date_time):
     return datetime.strptime(created_date_time, TIMESTAMP_FORMAT)
 
 
+def _is_page_published(version_string):
+    """True when the page version is a major release (e.g. "3.0")."""
+    if not version_string:
+        return True
+
+    _, _, minor = str(version_string).partition(".")
+
+    if not minor:
+        return True
+
+    try:
+        return int(minor) == 0
+    except ValueError:
+        # An unparsable minor version is no evidence of a published page
+        return False
+
+
 class SharepointOnlineDataSource(BaseDataSource):
     """Sharepoint Online"""
 
@@ -1554,17 +1579,23 @@ async def _site_access_control(self, site):
                 [
                   "user":spo-admin"
                 ]
+            - list: site members with edit-or-higher access, plus site-admins
+                [
+                    "user:spo-admin",
+                    "user:spo-editor"
+                ]
         """
 
         self._logger.debug(f"Looking at site: {site['id']} with url {site['webUrl']}")
         if not self._dls_enabled():
-            return [], []
+            return [], [], []
 
         def _is_site_admin(user):
             return user.get("IsSiteAdmin", False)
 
         access_control = set()
         site_admins_access_control = set()
+        editors_access_control = set()
 
         try:
             async for role_assignment in self.client.site_role_assignments(
@@ -1576,6 +1607,12 @@ def _is_site_admin(user):
                     await self._get_access_control_from_role_assignment(role_assignment)
                 )
 
+                editors_access_control.update(
+                    await self._get_access_control_from_role_assignment(
+                        role_assignment, require_edit_access=True
+                    )
+                )
+
                 if _is_site_admin(member):
                     # These are likely in the "Owners" group for the site
                     site_admins_access_control |= member_access_control
@@ -1600,7 +1637,13 @@ def _is_site_admin(user):
             )
             raise PermissionsMissing(msg) from e
 
-        return list(access_control), list(site_admins_access_control)
+        editors_access_control |= site_admins_access_control
+
+        return (
+            list(access_control),
+            list(site_admins_access_control),
+            list(editors_access_control),
+        )
 
     def _dls_enabled(self):
         if self._features is None:
@@ -1845,6 +1888,7 @@ async def get_docs(self, filtering=None):
                 (
                     site_access_control,
                     site_admin_access_control,
+                    site_editors_access_control,
                 ) = await self._site_access_control(site)
 
                 yield (
@@ -1916,7 +1960,9 @@ async def get_docs(self, filtering=None):
                         yield list_item, download_func
 
                 # Sync site pages
-                async for site_page in self.site_pages(site, site_access_control):
+                async for site_page in self.site_pages(
+                    site, site_access_control, site_editors_access_control
+                ):
                     # Always include site admins in site page access controls
                     site_page = self._decorate_with_access_control(
                         site_page, site_admin_access_control
@@ -1948,6 +1994,7 @@ async def get_docs_incrementally(self, sync_cursor, filtering=None):
                 (
                     site_access_control,
                     site_admin_access_control,
+                    site_editors_access_control,
                 ) = await self._site_access_control(site)
 
                 yield (
@@ -2029,7 +2076,10 @@ async def get_docs_incrementally(self, sync_cursor, filtering=None):
 
                 # Sync site pages
                 async for site_page in self.site_pages(
-                    site, site_access_control, check_timestamp=True
+                    site,
+                    site_access_control,
+                    site_editors_access_control,
+                    check_timestamp=True,
                 ):
                     # Always include site admins in site page access controls
                     site_page = self._decorate_with_access_control(
@@ -2362,11 +2412,14 @@ async def site_lists(self, site, site_access_control, check_timestamp=False):
 
                 yield site_list
 
-    async def _get_access_control_from_role_assignment(self, role_assignment):
+    async def _get_access_control_from_role_assignment(
+        self, role_assignment, require_edit_access=False
+    ):
         """Extracts access control from a role assignment.
 
         Args:
             role_assignment (dict): dictionary representing a role assignment.
+            require_edit_access (bool): when True, return only members with edit access.
 
         Returns:
             access_control (list): list of usernames and dynamic group ids, which have the role assigned.
@@ -2375,36 +2428,38 @@ async def _get_access_control_from_role_assignment(self, role_assignment):
         If any role is assigned to a user this means at least "read" access.
         """
 
-        def _has_limited_access(role_assignment):
+        def _grants_access(role_assignment):
             bindings = role_assignment.get("RoleDefinitionBindings", [])
 
             # If there is no permission information, default to restrict access
             if not bindings:
                 self._logger.debug(
                     f"No RoleDefinitionBindings found for '{role_assignment.get('odata.id')}'"
                 )
-                return True
+                return False
 
-            # if any binding grants view access, this role assignment's member has view access
             for binding in bindings:
                 # full explanation of the bit-math: https://stackoverflow.com/questions/51897160/how-to-parse-getusereffectivepermissions-sharepoint-response-in-java
                 # this approach was confirmed as valid by a Microsoft Sr. Support Escalation Engineer
                 base_permission_low = int(
                     nested_get_from_dict(binding, ["BasePermissions", "Low"], "0")  # pyright: ignore
                 )
                 role_type_kind = binding.get("RoleTypeKind", 0)
-                if (
+                if require_edit_access:
+                    if (base_permission_low & EDIT_ITEM_MASK) or (
+                        role_type_kind in EDIT_ROLE_TYPES
+                    ):
+                        return True
+                elif (
                     (base_permission_low & VIEW_ITEM_MASK)
                     or (base_permission_low & VIEW_PAGE_MASK)
                     or (role_type_kind in VIEW_ROLE_TYPES)
                 ):
-                    return False
+                    return True
 
-            return (
-                True  # no evidence of view access was found, so assuming limited access
-            )
+            return False
 
-        if _has_limited_access(role_assignment):
+        if not _grants_access(role_assignment):
             return []
 
         access_control = []
@@ -2429,7 +2484,13 @@ def _has_limited_access(role_assignment):
 
         return access_control
 
-    async def site_pages(self, site, site_access_control, check_timestamp=False):
+    async def site_pages(
+        self,
+        site,
+        site_access_control,
+        site_editors_access_control,
+        check_timestamp=False,
+    ):
         site_id = site["id"]
         url = site["webUrl"]
         async for site_page in self.client.site_pages(url):
@@ -2446,6 +2507,13 @@ async def site_pages(self, site, site_access_control, check_timestamp=False):
                 site_page["_id"] = f"{site_id}-site_page-{site_page['Id']}"
                 site_page["object_type"] = "site_page"
 
+                published = _is_page_published(site_page.get("OData__UIVersionString"))
+                site_page["published"] = published
+                if not published:
+                    self._logger.debug(
+                        f"Unpublished site page '{site_page['_id']}'; restricting ACL to owners/editors."
+                    )
+
                 has_unique_role_assignments = False
 
                 # ignore parent site permissions and use unique per page permissions ("unique permissions" means breaking the inheritance to the parent site)
@@ -2473,18 +2541,23 @@ async def site_pages(self, site, site_access_control, check_timestamp=False):
                         ):
                             page_access_control.extend(
                                 await self._get_access_control_from_role_assignment(
-                                    role_assignment
+                                    role_assignment,
+                                    require_edit_access=not published,
                                 )
                             )
 
                         site_page = self._decorate_with_access_control(
                             site_page, page_access_control
                         )
 
-                # set parent site access control
                 if not has_unique_role_assignments:
+                    inherited_access_control = (
+                        site_access_control
+                        if published
+                        else site_editors_access_control
+                    )
                     site_page = self._decorate_with_access_control(
-                        site_page, site_access_control
+                        site_page, inherited_access_control
                     )
 
                 for html_field in [