Escape single quote marks for search filters (#1599)

* Escape filters

* Fix for mypy

Author: pamelafox

app/backend/core/authentication.py

@@ -253,7 +253,10 @@ async def check_path_auth(self, path: str, auth_claims: dict[str, Any], search_c
 
         # Filter down to only chunks that are from the specific source file
         # Sourcepage is used for GPT-4V
-        filter = f"{security_filter} and ((sourcefile eq '{path}') or (sourcepage eq '{path}'))"
+        # Replace ' with '' to escape the single quote for the filter
+        # https://learn.microsoft.com/azure/search/query-odata-filter-orderby-syntax#escaping-special-characters-in-string-constants
+        path_for_filter = path.replace("'", "''")
+        filter = f"{security_filter} and ((sourcefile eq '{path_for_filter}') or (sourcepage eq '{path_for_filter}'))"
 
         # If the filter returns any results, the user is allowed to access the document
         # Otherwise, access is denied

app/backend/prepdocslib/searchmanager.py

@@ -250,7 +250,12 @@ async def remove_content(self, path: Optional[str] = None, only_oid: Optional[st
         )
         async with self.search_info.create_search_client() as search_client:
             while True:
-                filter = None if path is None else f"sourcefile eq '{os.path.basename(path)}'"
+                filter = None
+                if path is not None:
+                    # Replace ' with '' to escape the single quote for the filter
+                    # https://learn.microsoft.com/azure/search/query-odata-filter-orderby-syntax#escaping-special-characters-in-string-constants
+                    path_for_filter = os.path.basename(path).replace("'", "''")
+                    filter = f"sourcefile eq '{path_for_filter}'"
                 max_results = 1000
                 result = await search_client.search(
                     search_text="", filter=filter, top=max_results, include_total_count=True

tests/test_authenticationhelper.py

@@ -206,15 +206,15 @@ async def mock_search(self, *args, **kwargs):
 
     assert (
         await auth_helper_require_access_control.check_path_auth(
-            path="Benefit_Options-2.pdf",
+            path="Benefit_Options-2's complement.pdf",
             auth_claims={"oid": "OID_X", "groups": ["GROUP_Y", "GROUP_Z"]},
             search_client=create_search_client(),
         )
         is True
     )
     assert (
         filter
-        == "(oids/any(g:search.in(g, 'OID_X')) or groups/any(g:search.in(g, 'GROUP_Y, GROUP_Z'))) and ((sourcefile eq 'Benefit_Options-2.pdf') or (sourcepage eq 'Benefit_Options-2.pdf'))"
+        == "(oids/any(g:search.in(g, 'OID_X')) or groups/any(g:search.in(g, 'GROUP_Y, GROUP_Z'))) and ((sourcefile eq 'Benefit_Options-2''s complement.pdf') or (sourcepage eq 'Benefit_Options-2''s complement.pdf'))"
     )
 
 

tests/test_searchmanager.py

@@ -334,8 +334,8 @@ async def test_remove_content(monkeypatch, search_info):
                 "id": "file-foo_pdf-666F6F2E706466-page-0",
                 "content": "test content",
                 "category": "test",
-                "sourcepage": "foo.pdf#page=1",
-                "sourcefile": "foo.pdf",
+                "sourcepage": "foo's bar.pdf#page=1",
+                "sourcefile": "foo's bar.pdf",
             }
         ]
     )
@@ -359,10 +359,10 @@ async def mock_delete_documents(self, documents):
 
     manager = SearchManager(search_info)
 
-    await manager.remove_content("foo.pdf")
+    await manager.remove_content("foo's bar.pdf")
 
     assert len(searched_filters) == 2, "It should have searched twice (with no results on second try)"
-    assert searched_filters[0] == "sourcefile eq 'foo.pdf'"
+    assert searched_filters[0] == "sourcefile eq 'foo''s bar.pdf'"
     assert len(deleted_documents) == 1, "It should have deleted one document"
     assert deleted_documents[0]["id"] == "file-foo_pdf-666F6F2E706466-page-0"
 

tests/test_upload.py

@@ -159,8 +159,8 @@ class AsyncSearchResultsIterator:
         def __init__(self):
             self.results = [
                 {
-                    "sourcepage": "a.txt",
-                    "sourcefile": "a.txt",
+                    "sourcepage": "a's doc.txt",
+                    "sourcefile": "a's doc.txt",
                     "content": "This is a test document.",
                     "embedding": [],
                     "category": None,
@@ -170,8 +170,8 @@ def __init__(self):
                     "@search.reranker_score": 3.4577205181121826,
                 },
                 {
-                    "sourcepage": "a.txt",
-                    "sourcefile": "a.txt",
+                    "sourcepage": "a's doc.txt",
+                    "sourcefile": "a's doc.txt",
                     "content": "This is a test document.",
                     "embedding": [],
                     "category": None,
@@ -181,8 +181,8 @@ def __init__(self):
                     "@search.reranker_score": 3.4577205181121826,
                 },
                 {
-                    "sourcepage": "a.txt",
-                    "sourcefile": "a.txt",
+                    "sourcepage": "a's doc.txt",
+                    "sourcefile": "a's doc.txt",
                     "content": "This is a test document.",
                     "embedding": [],
                     "category": None,
@@ -224,10 +224,10 @@ async def mock_delete_documents(self, documents):
     monkeypatch.setattr(SearchClient, "delete_documents", mock_delete_documents)
 
     response = await auth_client.post(
-        "/delete_uploaded", headers={"Authorization": "Bearer test"}, json={"filename": "a.txt"}
+        "/delete_uploaded", headers={"Authorization": "Bearer test"}, json={"filename": "a's doc.txt"}
     )
     assert response.status_code == 200
     assert len(searched_filters) == 2, "It should have searched twice (with no results on second try)"
-    assert searched_filters[0] == "sourcefile eq 'a.txt'"
+    assert searched_filters[0] == "sourcefile eq 'a''s doc.txt'"
     assert len(deleted_documents) == 1, "It should have only deleted the document solely owned by OID_X"
     assert deleted_documents[0]["id"] == "file-a_txt-7465737420646F63756D656E742E706466"