Add ACL support for cloud ingestion pipeline (#2917)

* add custom role

* add x ms elevated read

* conditionally provision role

* fall back to AZURE_TENANT_ID if auth tenant unavailable

* Implement ACLs for cloud ingestion

* Address PR review comments for ACL cloud ingestion

- Remove unused AZURE_ADLS_GEN2_FILESYSTEM_PATH from CI/CD files and docs
- Make ACL fields conditional on use_acls in cloudingestionstrategy.py
- Add use_acls setting to text_processor for conditional ACL output
- Make local_files required parameter in setup_list_file_strategy()
- Add legacy fallback comments for AZURE_CLOUD_INGESTION_STORAGE_ACCOUNT
- Rename is_adls to storage_is_adls for clarity
- Update tests for all changes

* Fix markdown error

* Address Copilot PR review comments

- Fix print statement to use f-string instead of %s placeholder
- Fix typo in comment (hyphen to colon)
- Include empty ACL arrays when use_acls=True to distinguish 'no ACLs' from 'not extracted'
- Add 3 tests for text_processor ACL passthrough behavior

* Upgrade black to 26.1.0 and reformat files

- Update pre-commit config to black 26.1.0
- Reformat test_htmlparser.py and test_textparser.py
- Add github-pr-inline-reply skill

* Add issue links to skill

* Add docs on using verify

* Fix ACL config issues from code review

- Fix env var: use USE_CLOUD_INGESTION_ACLS instead of AZURE_USE_AUTHENTICATION in setup_cloud_ingestion.py
- Add warning when AZURE_ENFORCE_ACCESS_CONTROL enabled without USE_CLOUD_INGESTION_ACLS
- Move DataLakeServiceClient initialization to GlobalSettings for better performance

* Update tests for GlobalSettings data_lake_service_client field

* Address PR review comments: add permission note and USE_CLOUD_INGESTION_ACLS to env vars reference

* Fix: use consistent USE_CLOUD_INGESTION_ACLS env var name across functions and Bicep

* Reorganize login_and_acl.md: remove outdated Testing section, move ACL verification to cloud ingestion, move Programmatic access to top-level, rename Troubleshooting

* Address PR review comments: add Bicep parameter descriptions and documentation

- Add comment on adlsStorageAccountName explaining it must be specified when
  useExistingAdlsStorage is true (Bicep assert is experimental)
- Add @description() decorators to storage-role.bicep parameters
- Add comment explaining useCloudIngestionAcls requires useCloudIngestion

* Update docs/login_and_acl.md

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Fix markdown linting: add blank line above heading

---------

Co-authored-by: Matt Gotteiner <matthew.gotteiner@microsoft.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 3 people

.azdo/pipelines/azure-dev.yml

@@ -87,6 +87,11 @@ steps:
       AZURE_LOG_ANALYTICS: $(AZURE_LOG_ANALYTICS)
       USE_VECTORS: $(USE_VECTORS)
       USE_MULTIMODAL: $(USE_MULTIMODAL)
+      USE_CLOUD_INGESTION: $(USE_CLOUD_INGESTION)
+      USE_CLOUD_INGESTION_ACLS: $(USE_CLOUD_INGESTION_ACLS)
+      USE_EXISTING_ADLS_STORAGE: $(USE_EXISTING_ADLS_STORAGE)
+      AZURE_ADLS_GEN2_STORAGE_ACCOUNT: $(AZURE_ADLS_GEN2_STORAGE_ACCOUNT)
+      AZURE_ADLS_GEN2_STORAGE_RESOURCE_GROUP: $(AZURE_ADLS_GEN2_STORAGE_RESOURCE_GROUP)
       AZURE_VISION_ENDPOINT: $(AZURE_VISION_ENDPOINT)
       VISION_SECRET_NAME: $(VISION_SECRET_NAME)
       AZURE_VISION_SERVICE: $(AZURE_VISION_SERVICE)
@@ -114,8 +119,6 @@ steps:
       ALLOWED_ORIGIN: $(ALLOWED_ORIGIN)
       AZURE_SERVER_APP_SECRET: $(AZURE_SERVER_APP_SECRET)
       AZURE_CLIENT_APP_SECRET: $(AZURE_CLIENT_APP_SECRET)
-      AZURE_ADLS_GEN2_STORAGE_ACCOUNT: $(AZURE_ADLS_GEN2_STORAGE_ACCOUNT)
-      AZURE_ADLS_GEN2_FILESYSTEM_PATH: $(AZURE_ADLS_GEN2_FILESYSTEM_PATH)
       AZURE_ADLS_GEN2_FILESYSTEM: $(AZURE_ADLS_GEN2_FILESYSTEM)
       DEPLOYMENT_TARGET: $(DEPLOYMENT_TARGET)
       AZURE_CONTAINER_APPS_WORKLOAD_PROFILE: $(AZURE_CONTAINER_APPS_WORKLOAD_PROFILE)

.github/workflows/azure-dev.yml

@@ -83,6 +83,11 @@ jobs:
       AZURE_LOG_ANALYTICS: ${{ vars.AZURE_LOG_ANALYTICS }}
       USE_VECTORS: ${{ vars.USE_VECTORS }}
       USE_MULTIMODAL: ${{ vars.USE_MULTIMODAL }}
+      USE_CLOUD_INGESTION: ${{ vars.USE_CLOUD_INGESTION }}
+      USE_CLOUD_INGESTION_ACLS: ${{ vars.USE_CLOUD_INGESTION_ACLS }}
+      USE_EXISTING_ADLS_STORAGE: ${{ vars.USE_EXISTING_ADLS_STORAGE }}
+      AZURE_ADLS_GEN2_STORAGE_ACCOUNT: ${{ vars.AZURE_ADLS_GEN2_STORAGE_ACCOUNT }}
+      AZURE_ADLS_GEN2_STORAGE_RESOURCE_GROUP: ${{ vars.AZURE_ADLS_GEN2_STORAGE_RESOURCE_GROUP }}
       AZURE_VISION_ENDPOINT: ${{ vars.AZURE_VISION_ENDPOINT }}
       VISION_SECRET_NAME: ${{ vars.VISION_SECRET_NAME }}
       ENABLE_LANGUAGE_PICKER: ${{ vars.ENABLE_LANGUAGE_PICKER }}
@@ -103,8 +108,6 @@ jobs:
       AZURE_SERVER_APP_ID: ${{ vars.AZURE_SERVER_APP_ID }}
       AZURE_CLIENT_APP_ID: ${{ vars.AZURE_CLIENT_APP_ID }}
       ALLOWED_ORIGIN: ${{ vars.ALLOWED_ORIGIN }}
-      AZURE_ADLS_GEN2_STORAGE_ACCOUNT: ${{ vars.AZURE_ADLS_GEN2_STORAGE_ACCOUNT }}
-      AZURE_ADLS_GEN2_FILESYSTEM_PATH: ${{ vars.AZURE_ADLS_GEN2_FILESYSTEM_PATH }}
       AZURE_ADLS_GEN2_FILESYSTEM: ${{ vars.AZURE_ADLS_GEN2_FILESYSTEM }}
       DEPLOYMENT_TARGET: ${{ vars.DEPLOYMENT_TARGET }}
       AZURE_CONTAINER_APPS_WORKLOAD_PROFILE: ${{ vars.AZURE_CONTAINER_APPS_WORKLOAD_PROFILE }}

.github/workflows/evaluate.yaml

@@ -98,7 +98,6 @@ jobs:
       AZURE_CLIENT_APP_ID: ${{ vars.AZURE_CLIENT_APP_ID }}
       ALLOWED_ORIGIN: ${{ vars.ALLOWED_ORIGIN }}
       AZURE_ADLS_GEN2_STORAGE_ACCOUNT: ${{ vars.AZURE_ADLS_GEN2_STORAGE_ACCOUNT }}
-      AZURE_ADLS_GEN2_FILESYSTEM_PATH: ${{ vars.AZURE_ADLS_GEN2_FILESYSTEM_PATH }}
       AZURE_ADLS_GEN2_FILESYSTEM: ${{ vars.AZURE_ADLS_GEN2_FILESYSTEM }}
       DEPLOYMENT_TARGET: ${{ vars.DEPLOYMENT_TARGET }}
       AZURE_CONTAINER_APPS_WORKLOAD_PROFILE: ${{ vars.AZURE_CONTAINER_APPS_WORKLOAD_PROFILE }}

app/backend/prepdocs.py

@@ -16,8 +16,6 @@
     IntegratedVectorizerStrategy,
 )
 from prepdocslib.listfilestrategy import (
-    ADLSGen2ListFileStrategy,
-    ListFileStrategy,
     LocalListFileStrategy,
 )
 from prepdocslib.servicesetup import (
@@ -51,33 +49,13 @@ async def check_search_service_connectivity(search_service: str) -> bool:
 
 def setup_list_file_strategy(
     azure_credential: AsyncTokenCredential,
-    local_files: Optional[str],
-    datalake_storage_account: Optional[str],
-    datalake_filesystem: Optional[str],
-    datalake_path: Optional[str],
-    datalake_key: Optional[str],
+    local_files: str,
     enable_global_documents: bool = False,
 ):
-    list_file_strategy: ListFileStrategy
-    if datalake_storage_account:
-        if datalake_filesystem is None or datalake_path is None:
-            raise ValueError("DataLake file system and path are required when using Azure Data Lake Gen2")
-        adls_gen2_creds: AsyncTokenCredential | str = azure_credential if datalake_key is None else datalake_key
-        logger.info("Using Data Lake Gen2 Storage Account: %s", datalake_storage_account)
-        list_file_strategy = ADLSGen2ListFileStrategy(
-            data_lake_storage_account=datalake_storage_account,
-            data_lake_filesystem=datalake_filesystem,
-            data_lake_path=datalake_path,
-            credential=adls_gen2_creds,
-            enable_global_documents=enable_global_documents,
-        )
-    elif local_files:
-        logger.info("Using local files: %s", local_files)
-        list_file_strategy = LocalListFileStrategy(
-            path_pattern=local_files, enable_global_documents=enable_global_documents
-        )
-    else:
-        raise ValueError("Either local_files or datalake_storage_account must be provided.")
+    logger.info("Using local files: %s", local_files)
+    list_file_strategy = LocalListFileStrategy(
+        path_pattern=local_files, enable_global_documents=enable_global_documents
+    )
     return list_file_strategy
 
 
@@ -165,9 +143,6 @@ async def main(strategy: Strategy, setup_index: bool = True):
         required=False,
         help="Optional. Use this Azure Blob Storage account key instead of the current user identity to login (use az login to set current user for Azure)",
     )
-    parser.add_argument(
-        "--datalakekey", required=False, help="Optional. Use this key when authenticating to Azure Data Lake Gen2"
-    )
     parser.add_argument(
         "--documentintelligencekey",
         required=False,
@@ -274,10 +249,6 @@ async def main(strategy: Strategy, setup_index: bool = True):
     list_file_strategy = setup_list_file_strategy(
         azure_credential=azd_credential,
         local_files=args.files,
-        datalake_storage_account=os.getenv("AZURE_ADLS_GEN2_STORAGE_ACCOUNT"),
-        datalake_filesystem=os.getenv("AZURE_ADLS_GEN2_FILESYSTEM"),
-        datalake_path=os.getenv("AZURE_ADLS_GEN2_FILESYSTEM_PATH"),
-        datalake_key=clean_key_if_exists(args.datalakekey),
         enable_global_documents=enable_global_documents,
     )
 

app/backend/prepdocslib/cloudingestionstrategy.py

@@ -131,6 +131,9 @@ def _build_skillset(self) -> SearchIndexerSkillset:
         ]
         if self.use_multimodal:
             mappings.append(InputFieldMappingEntry(name="images", source="/document/chunks/*/images"))
+        if self.use_acls:
+            mappings.append(InputFieldMappingEntry(name="oids", source="/document/chunks/*/oids"))
+            mappings.append(InputFieldMappingEntry(name="groups", source="/document/chunks/*/groups"))
 
         index_projection = SearchIndexerIndexProjection(
             selectors=[
@@ -168,7 +171,16 @@ def _build_skillset(self) -> SearchIndexerSkillset:
             outputs=[
                 OutputFieldMappingEntry(name="pages", target_name="pages"),
                 OutputFieldMappingEntry(name="figures", target_name="figures"),
-            ],
+            ]
+            + (
+                [
+                    # ACL outputs for document-level access control (populated by manual ADLS Gen2 extraction)
+                    OutputFieldMappingEntry(name="oids", target_name="oids"),
+                    OutputFieldMappingEntry(name="groups", target_name="groups"),
+                ]
+                if self.use_acls
+                else []
+            ),
         )
 
         figure_processor_skill = WebApiSkill(
@@ -232,7 +244,16 @@ def _build_skillset(self) -> SearchIndexerSkillset:
                 ),
                 InputFieldMappingEntry(name="file_name", source="/document/metadata_storage_name"),
                 InputFieldMappingEntry(name="storageUrl", source="/document/metadata_storage_path"),
-            ],
+            ]
+            + (
+                [
+                    # ACL fields from document_extractor's manual ADLS Gen2 ACL extraction
+                    InputFieldMappingEntry(name="oids", source="/document/oids"),
+                    InputFieldMappingEntry(name="groups", source="/document/groups"),
+                ]
+                if self.use_acls
+                else []
+            ),
             outputs=[OutputFieldMappingEntry(name="output", target_name="consolidated_document")],
         )
 
@@ -272,6 +293,23 @@ async def setup(self) -> None:
         if not isinstance(self.embeddings, OpenAIEmbeddings):
             raise TypeError("Cloud ingestion requires Azure OpenAI embeddings to configure the search index.")
 
+        # Warn if access control is enforced but ACL extraction is not enabled
+        if self.enforce_access_control and not self.use_acls:
+            logger.warning(
+                "AZURE_ENFORCE_ACCESS_CONTROL is enabled but USE_CLOUD_INGESTION_ACLS is not. "
+                "Documents will not have ACLs extracted automatically from ADLS Gen2. "
+                "If you intend to use document-level access control, either set USE_CLOUD_INGESTION_ACLS=true "
+                "or manually set ACLs using scripts/manageacl.py after ingestion."
+            )
+
+        # Verify the storage container exists before attempting to create the data source
+        container_client = self.blob_manager.blob_service_client.get_container_client(self.blob_manager.container)
+        if not await container_client.exists():
+            raise ValueError(
+                f"Storage container '{self.blob_manager.container}' does not exist in storage account '{self.blob_manager.account}'. "
+                f"Please create the container first, or set AZURE_STORAGE_CONTAINER to an existing container name."
+            )
+
         self._search_manager = SearchManager(
             search_info=self.search_info,
             search_analyzer_name=self.search_analyzer_name,
@@ -287,9 +325,15 @@ async def setup(self) -> None:
         await self._search_manager.create_index()
 
         async with self.search_info.create_search_indexer_client() as indexer_client:
+            # Use ADLS_GEN2 when ACLs are enabled (requires hierarchical namespace storage)
+            # Note: We do NOT use indexer_permission_options because that's incompatible with
+            # Custom WebAPI skills. Instead, ACLs are extracted manually in document_extractor.
+            data_source_type = (
+                SearchIndexerDataSourceType.ADLS_GEN2 if self.use_acls else SearchIndexerDataSourceType.AZURE_BLOB
+            )
             data_source_connection = SearchIndexerDataSourceConnection(
                 name=self.data_source_name,
-                type=SearchIndexerDataSourceType.AZURE_BLOB,
+                type=data_source_type,
                 connection_string=self.blob_manager.get_managedidentity_connectionstring(),
                 container=SearchIndexerDataContainer(name=self.blob_manager.container),
                 data_deletion_detection_policy=NativeBlobSoftDeleteDeletionDetectionPolicy(),

app/backend/prepdocslib/listfilestrategy.py

@@ -3,17 +3,11 @@
 import logging
 import os
 import re
-import tempfile
 from abc import ABC
 from collections.abc import AsyncGenerator
 from glob import glob
 from typing import IO, Optional
 
-from azure.core.credentials_async import AsyncTokenCredential
-from azure.storage.filedatalake.aio import (
-    DataLakeServiceClient,
-)
-
 logger = logging.getLogger("scripts")
 
 
@@ -136,75 +130,3 @@ def check_md5(self, path: str) -> bool:
             md5_f.write(existing_hash)
 
         return False
-
-
-class ADLSGen2ListFileStrategy(ListFileStrategy):
-    """
-    Concrete strategy for listing files that are located in a data lake storage account
-    """
-
-    def __init__(
-        self,
-        data_lake_storage_account: str,
-        data_lake_filesystem: str,
-        data_lake_path: str,
-        credential: AsyncTokenCredential | str,
-        enable_global_documents: bool = False,
-    ):
-        self.data_lake_storage_account = data_lake_storage_account
-        self.data_lake_filesystem = data_lake_filesystem
-        self.data_lake_path = data_lake_path
-        self.credential = credential
-        self.enable_global_documents = enable_global_documents
-
-    async def list_paths(self) -> AsyncGenerator[str, None]:
-        async with DataLakeServiceClient(
-            account_url=f"https://{self.data_lake_storage_account}.dfs.core.windows.net", credential=self.credential
-        ) as service_client, service_client.get_file_system_client(self.data_lake_filesystem) as filesystem_client:
-            async for path in filesystem_client.get_paths(path=self.data_lake_path, recursive=True):
-                if path.is_directory:
-                    continue
-
-                yield path.name
-
-    async def list(self) -> AsyncGenerator[File, None]:
-        async with DataLakeServiceClient(
-            account_url=f"https://{self.data_lake_storage_account}.dfs.core.windows.net", credential=self.credential
-        ) as service_client, service_client.get_file_system_client(self.data_lake_filesystem) as filesystem_client:
-            async for path in self.list_paths():
-                temp_file_path = os.path.join(tempfile.gettempdir(), os.path.basename(path))
-                try:
-                    async with filesystem_client.get_file_client(path) as file_client:
-                        with open(temp_file_path, "wb") as temp_file:
-                            downloader = await file_client.download_file()
-                            await downloader.readinto(temp_file)
-                    # Parse out user ids and group ids
-                    acls: dict[str, list[str]] = {"oids": [], "groups": []}
-                    # https://learn.microsoft.com/python/api/azure-storage-file-datalake/azure.storage.filedatalake.datalakefileclient?view=azure-python#azure-storage-filedatalake-datalakefileclient-get-access-control
-                    # Request ACLs as GUIDs
-                    access_control = await file_client.get_access_control(upn=False)
-                    acl_list = access_control["acl"]
-                    # https://learn.microsoft.com/azure/storage/blobs/data-lake-storage-access-control
-                    # ACL Format: user::rwx,group::r-x,other::r--,user:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx:r--
-                    acl_list = acl_list.split(",")
-                    for acl in acl_list:
-                        acl_parts: list = acl.split(":")
-                        if len(acl_parts) != 3:
-                            continue
-                        if len(acl_parts[1]) == 0:
-                            continue
-                        if acl_parts[0] == "user" and "r" in acl_parts[2]:
-                            acls["oids"].append(acl_parts[1])
-                        if acl_parts[0] == "group" and "r" in acl_parts[2]:
-                            acls["groups"].append(acl_parts[1])
-
-                    if self.enable_global_documents and len(acls["oids"]) == 0 and len(acls["groups"]) == 0:
-                        acls = {"oids": ["all"], "groups": ["all"]}
-
-                    yield File(content=open(temp_file_path, "rb"), acls=acls, url=file_client.url)
-                except Exception as data_lake_exception:
-                    logger.error(f"\tGot an error while reading {path} -> {data_lake_exception} --> skipping file")
-                    try:
-                        os.remove(temp_file_path)
-                    except Exception as file_delete_exception:
-                        logger.error(f"\tGot an error while deleting {temp_file_path} -> {file_delete_exception}")

app/backend/setup_cloud_ingestion.py

@@ -36,13 +36,18 @@ async def setup_cloud_ingestion_strategy(
     search_service = os.environ["AZURE_SEARCH_SERVICE"]
     index_name = os.environ["AZURE_SEARCH_INDEX"]
     search_user_assigned_identity_resource_id = os.environ["AZURE_SEARCH_USER_ASSIGNED_IDENTITY_RESOURCE_ID"]
-    storage_account = os.environ["AZURE_STORAGE_ACCOUNT"]
     storage_container = os.environ["AZURE_STORAGE_CONTAINER"]
-    storage_resource_group = os.environ["AZURE_STORAGE_RESOURCE_GROUP"]
     subscription_id = os.environ["AZURE_SUBSCRIPTION_ID"]
     image_storage_container = os.environ.get("AZURE_IMAGESTORAGE_CONTAINER")
     search_embedding_field = os.environ["AZURE_SEARCH_FIELD_NAME_EMBEDDING"]
 
+    # Cloud ingestion storage account (ADLS Gen2 when ACLs enabled, standard blob otherwise)
+    # Fallback to AZURE_STORAGE_ACCOUNT is for legacy deployments only - may be removed in future
+    storage_account = os.getenv("AZURE_CLOUD_INGESTION_STORAGE_ACCOUNT") or os.environ["AZURE_STORAGE_ACCOUNT"]
+    storage_resource_group = (
+        os.getenv("AZURE_CLOUD_INGESTION_STORAGE_RESOURCE_GROUP") or os.environ["AZURE_STORAGE_RESOURCE_GROUP"]
+    )
+
     # Cloud ingestion specific endpoints
     document_extractor_uri = os.environ["DOCUMENT_EXTRACTOR_SKILL_ENDPOINT"]
     document_extractor_resource_id = os.environ["DOCUMENT_EXTRACTOR_SKILL_AUTH_RESOURCE_ID"]
@@ -53,10 +58,19 @@ async def setup_cloud_ingestion_strategy(
 
     # Feature flags
     use_multimodal = os.getenv("USE_MULTIMODAL", "").lower() == "true"
-    use_acls = os.getenv("AZURE_USE_AUTHENTICATION", "").lower() == "true"
+    use_acls = os.getenv("USE_CLOUD_INGESTION_ACLS", "").lower() == "true"
     enforce_access_control = os.getenv("AZURE_ENFORCE_ACCESS_CONTROL", "").lower() == "true"
     use_web_source = os.getenv("USE_WEB_SOURCE", "").lower() == "true"
 
+    # Warn if access control is enforced but ACL extraction is not enabled
+    if enforce_access_control and not use_acls:
+        logger.warning(
+            "AZURE_ENFORCE_ACCESS_CONTROL is enabled but USE_CLOUD_INGESTION_ACLS is not. "
+            "Documents will be indexed without ACLs, so access control filtering will not work. "
+            "Either set USE_CLOUD_INGESTION_ACLS=true to extract ACLs from ADLS Gen2, "
+            "or manually set ACLs using scripts/manageacl.py after ingestion."
+        )
+
     # Setup search info
     search_info = setup_search_info(
         search_service=search_service,