feat: add file-based access permissions for SharePoint ingest

CHANGELOG.md

@@ -1,3 +1,14 @@
+## 0.10.20-dev0
+
+### Enhancements
+
+### Fixes
+
+### Features
+
+* **Adds permissions(RBAC) data ingestion functionality for the Sharepoint connector.** Problem: Role based access control is an important component in many data storage systems. Users may need to pass permissions (RBAC) data to downstream systems when ingesting data. Feature: Added permissions data ingestion functionality to the Sharepoint connector.
+
+
 ## 0.10.19
 
 ### Enhancements

unstructured/__version__.py

@@ -1 +1 @@
-__version__ = "0.10.19"  # pragma: no cover
+__version__ = "0.10.20-dev0"  # pragma: no cover

unstructured/documents/elements.py

@@ -44,6 +44,7 @@ class DataSourceMetadata:
     date_created: Optional[str] = None
     date_modified: Optional[str] = None
     date_processed: Optional[str] = None
+    permissions_data: Optional[List[Dict[str, Any]]] = None
 
     def to_dict(self):
         return {key: value for key, value in self.__dict__.items() if value is not None}

unstructured/ingest/cli/cmds/sharepoint.py

@@ -25,6 +25,9 @@
 class SharepointCliConfig(BaseConfig, CliMixin):
     client_id: t.Optional[str] = None
     client_cred: t.Optional[str] = None
+    permissions_application_id: t.Optional[str] = None
+    permissions_client_cred: t.Optional[str] = None
+    permissions_tenant: t.Optional[str] = None
     site: t.Optional[str] = None
     path: str = "Shared Documents"
     files_only: bool = False
@@ -55,6 +58,24 @@ def add_cli_options(cmd: click.Command) -> None:
                     https://[tenant]-admin.sharepoint.com.\
                     This requires the app to be registered at a tenant level",
             ),
+            click.Option(
+                ["--permissions-application-id"],
+                default=None,
+                type=str,
+                help="Application id for ingesting permission (rbac) data",
+            ),
+            click.Option(
+                ["--permissions-client-cred"],
+                default=None,
+                type=str,
+                help="Credentials for ingesting permission (rbac) data",
+            ),
+            click.Option(
+                ["--permissions-tenant"],
+                default=None,
+                type=str,
+                help="Sharepoint permission (rbac) tenant name, such as: abcde.onmicrosoft.com",
+            ),
             click.Option(
                 ["--path"],
                 default="Shared Documents",

unstructured/ingest/connector/sharepoint.py

@@ -1,3 +1,5 @@
+import json
+import os
 import typing as t
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -10,6 +12,7 @@
 from unstructured.file_utils.filetype import EXT_TO_FILETYPE
 from unstructured.ingest.error import SourceConnectionError
 from unstructured.ingest.interfaces import (
+    BaseConfig,
     BaseConnectorConfig,
     BaseIngestDoc,
     BaseSourceConnector,
@@ -31,6 +34,26 @@
 CONTENT_LABELS = ["CanvasContent1", "LayoutWebpartsContent1", "TimeCreated"]
 
 
+@dataclass
+class SharepointPermissionsConfig(BaseConfig):
+    application_id: str = None
+    client_credential: str = None
+    tenant: str = None
+
+    def __post_init__(self):
+        self.provided = False
+        if any([self.application_id or self.client_credential or self.tenant]):
+            if not all([self.application_id and self.client_credential and self.tenant]):
+                raise ValueError(
+                    "Please provide either none or all of the following optional values:\n"
+                    "--permissions-application-id\n"
+                    "--permissions-client-cred\n"
+                    "--permissions-tenant",
+                )
+            else:
+                self.provided = True
+
+
 @dataclass
 class SimpleSharepointConfig(BaseConnectorConfig):
     client_id: str
@@ -39,6 +62,7 @@ class SimpleSharepointConfig(BaseConnectorConfig):
     path: str
     process_pages: bool = False
     recursive: bool = False
+    permissions_config: t.Optional[SharepointPermissionsConfig] = None
 
     def __post_init__(self):
         if not (self.client_id and self.client_credential and self.site_url):
@@ -61,6 +85,14 @@ def get_site_client(self, site_url: str = "") -> "ClientContext":
             raise
         return site_client
 
+    def get_permissions_client(self):
+        try:
+            permissions_connector = SharepointPermissionsConnector(self.permissions_config)
+            assert permissions_connector.access_token
+            return permissions_connector
+        except Exception as e:
+            logger.error("Couldn't obtain Sharepoint permissions ingestion access token:", e)
+
 
 @dataclass
 class SharepointIngestDoc(IngestDocCleanupMixin, BaseIngestDoc):
@@ -146,7 +178,6 @@ def _fetch_file(self, properties_only: bool = False):
                 file = site_client.web.get_file_by_server_relative_url(self.server_path)
                 if properties_only:
                     file = file.get().execute_query()
-
         except ClientRequestException as e:
             if e.response.status_code == 404:
                 return None
@@ -168,6 +199,42 @@ def _fetch_page(self):
             return None
         return page
 
+    def update_permissions_data(self):
+        def parent_name_matches(parent_type, permissions_filename, ingest_doc_filepath):
+            permissions_filename = permissions_filename.split("_SEP_")
+            ingest_doc_filepath = ingest_doc_filepath.split("/")
+
+            if parent_type == "sites":
+                return permissions_filename[0] == ingest_doc_filepath[1]
+
+            elif parent_type == "SitePages" or parent_type == "Shared Documents":
+                return True
+
+        permissions_data = None
+        permissions_dir = Path(self.partition_config.output_dir) / "permissions_data"
+
+        if permissions_dir.is_dir():
+            parent_type = self.file_path.split("/")[0]
+
+            if parent_type == "sites":
+                read_dir = permissions_dir / "sites"
+            elif parent_type == "SitePages" or parent_type == "Shared Documents":
+                read_dir = permissions_dir / "other"
+
+            for filename in os.listdir(read_dir):
+                permissions_docname = os.path.splitext(filename)[0].split("_SEP_")[1]
+                ingestdoc_docname = self.file_path.split("/")[-1]
+
+                if ingestdoc_docname == permissions_docname and parent_name_matches(
+                    parent_type=parent_type,
+                    permissions_filename=filename,
+                    ingest_doc_filepath=self.file_path,
+                ):
+                    with open(read_dir / filename) as f:
+                        permissions_data = json.loads(f.read())
+
+        return permissions_data
+
     def update_source_metadata(self, **kwargs):
         if self.is_page:
             page = self._fetch_page()
@@ -182,6 +249,9 @@ def update_source_metadata(self, **kwargs):
                 version=page.get_property("Version", ""),
                 source_url=page.absolute_url,
                 exists=True,
+                permissions_data=self.update_permissions_data()
+                if self.connector_config.permissions_config
+                else None,
             )
             return
 
@@ -200,6 +270,9 @@ def update_source_metadata(self, **kwargs):
             version=file.major_version,
             source_url=file.properties.get("LinkingUrl", None),
             exists=True,
+            permissions_data=self.update_permissions_data()
+            if self.connector_config.permissions_config
+            else None,
         )
 
     def _download_page(self):
@@ -345,6 +418,12 @@ def initialize(self):
 
     def get_ingest_docs(self):
         base_site_client = self.connector_config.get_site_client()
+
+        if self.connector_config.permissions_config:
+            permissions_client = self.connector_config.get_permissions_client()
+            if permissions_client:
+                permissions_client.write_all_permissions(self.partition_config.output_dir)
+
         if not base_site_client.is_tenant:
             return self._ingest_site_docs(base_site_client)
         tenant = base_site_client.tenant
@@ -356,3 +435,160 @@ def get_ingest_docs(self):
             site_client = self.connector_config.get_site_client(site_url)
             ingest_docs = ingest_docs + self._ingest_site_docs(site_client)
         return ingest_docs
+
+
+@dataclass
+class SharepointPermissionsConnector:
+    def __init__(self, permissions_config):
+        self.permissions_config: SharepointPermissionsConfig = permissions_config
+        self.initialize()
+
+    def initialize(self):
+        self.access_token: str = self.get_access_token()
+
+    @requires_dependencies(["requests"], extras="sharepoint")
+    def get_access_token(self) -> str:
+        import requests
+
+        url = (
+            f"https://login.microsoftonline.com/{self.permissions_config.tenant}/oauth2/v2.0/token"
+        )
+        headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        data = {
+            "client_id": self.permissions_config.application_id,
+            "scope": "https://graph.microsoft.com/.default",
+            "client_secret": self.permissions_config.client_credential,
+            "grant_type": "client_credentials",
+        }
+        response = requests.post(url, headers=headers, data=data)
+        return response.json()["access_token"]
+
+    def validated_response(self, response):
+        if response.status_code == 200:
+            return response.json()
+        else:
+            print(f"Request failed with status code {response.status_code}:")
+            print(response.text)
+
+    @requires_dependencies(["requests"], extras="sharepoint")
+    def get_sites(self):
+        import requests
+
+        url = "https://graph.microsoft.com/v1.0/sites"
+        params = {
+            "$select": "webUrl, id",
+        }
+
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+        }
+
+        response = requests.get(url, params=params, headers=headers)
+        return self.validated_response(response)
+
+    @requires_dependencies(["requests"], extras="sharepoint")
+    def get_drives(self, site):
+        import requests
+
+        url = f"https://graph.microsoft.com/v1.0/sites/{site}/drives"
+
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+        }
+
+        response = requests.get(url, headers=headers)
+
+        return self.validated_response(response)
+
+    @requires_dependencies(["requests"], extras="sharepoint")
+    def get_drive_items(self, site, drive_id):
+        import requests
+
+        url = f"https://graph.microsoft.com/v1.0/sites/{site}/drives/{drive_id}/root/children"
+
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+        }
+
+        response = requests.get(url, headers=headers)
+
+        return self.validated_response(response)
+
+    def extract_site_name_from_weburl(self, weburl):
+        split_path = urlparse(weburl).path.lstrip("/").split("/")
+
+        if split_path[0] == "sites":
+            return "sites", split_path[1]
+
+        elif split_path[0] == "Shared%20Documents":
+            return "Shared Documents", "Shared Documents"
+
+        elif split_path[0] == "personal":
+            return "Personal", "Personal"
+
+        # if other weburl structures are found, additional logic might need to be implemented
+
+        logger.warning(
+            "Couldn't extract sitename, skipping RBAC ingestion \
+                           for the document with the URL:",
+            weburl,
+        )
+
+    @requires_dependencies(["requests"], extras="sharepoint")
+    def get_permissions_for_drive_item(self, site, drive_id, item_id):
+        import requests
+
+        url = f"https://graph.microsoft.com/v1.0/sites/ \
+        {site}/drives/{drive_id}/items/{item_id}/permissions"
+
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+        }
+
+        response = requests.get(url, headers=headers)
+
+        return self.validated_response(response)
+
+    def write_all_permissions(self, output_dir):
+        sites = [(site["id"], site["webUrl"]) for site in self.get_sites()["value"]]
+        drive_ids = []
+
+        print("Obtaining drive data for sites for permissions (rbac)")
+        for site_id, site_url in sites:
+            drives = self.get_drives(site_id)
+            if drives:
+                drives_for_site = drives["value"]
+                drive_ids.extend([(site_id, drive["id"]) for drive in drives_for_site])
+
+        print("Obtaining item data from drives for permissions (rbac)")
+        item_ids = []
+        for site, drive_id in drive_ids:
+            drive_items = self.get_drive_items(site, drive_id)
+            if drive_items:
+                item_ids.extend(
+                    # [(site, drive_id, item["id"], item["name"]) for item in drive_items["value"]],
+                    [
+                        (site, drive_id, item["id"], item["name"], item["webUrl"])
+                        for item in drive_items["value"]
+                    ],
+                )
+
+        permissions_dir = Path(output_dir) / "permissions_data"
+
+        print("Writing permissions data to disk")
+        for site, drive_id, item_id, item_name, item_web_url in item_ids:
+            res = self.get_permissions_for_drive_item(site, drive_id, item_id)
+            if res:
+                parent_type, parent_name = self.extract_site_name_from_weburl(item_web_url)
+
+                if parent_type == "sites":
+                    write_path = permissions_dir / "sites" / f"{parent_name}_SEP_{item_name}.json"
+
+                if parent_type == "Personal" or parent_type == "Shared Documents":
+                    write_path = permissions_dir / "other" / f"{parent_name}_SEP_{item_name}.json"
+
+                if not Path(os.path.dirname(write_path)).is_dir():
+                    os.makedirs(os.path.dirname(write_path))
+
+                with open(write_path, "w") as f:
+                    json.dump(res["value"], f)