Bug fix search table filter invalid char (#100)

* BugFixSearchTableFilterInvalidChar

* Linting

* Move the Check of the filter values in ES proxy file

* Fix linting

* Fix linting

* Fix linting

* Fix linting

search_service/api/table.py

@@ -4,7 +4,6 @@
 from flask_restful import Resource, fields, marshal_with, reqparse
 from flasgger import swag_from
 
-
 from search_service.proxy import get_proxy_client
 
 tag_fields = {
@@ -88,6 +87,7 @@ class SearchTableFilterAPI(Resource):
     This API should be generic enough to support every search filter use case.
     TODO: Deprecate the SearchTableFieldAPI for this more flexible API
     """
+
     def __init__(self) -> None:
         self.proxy = get_proxy_client()
 

search_service/proxy/elasticsearch.py

@@ -1,5 +1,6 @@
 import logging
 import uuid
+import itertools
 from typing import Any, List, Dict
 
 from elasticsearch import Elasticsearch
@@ -346,7 +347,7 @@ def parse_filters(filter_list: Dict) -> str:
             if mapped_category is None:
                 LOGGING.warn(f'Unsupported filter category: {category} passed in list of filters')
             elif item_list is '' or item_list == ['']:
-                LOGGING.warn(f'The filter value cannot be empty.In this case the filter {category} gets ignored')
+                LOGGING.warn(f'The filter value cannot be empty.In this case the filter {category} is ignored')
             else:
                 query_list.append(mapped_category + ':' + '(' + ' OR '.join(item_list) + ')')
 
@@ -355,6 +356,20 @@ def parse_filters(filter_list: Dict) -> str:
 
         return ' AND '.join(query_list)
 
+    @staticmethod
+    def validate_filter_values(search_request: dict) -> Any:
+        if 'filters' in search_request:
+            filter_values_list = search_request['filters'].values()
+            # Ensure all values are arrays
+            filter_values_list = list(
+                map(lambda x: x if type(x) == list else [x], filter_values_list))
+            # Flatten the array of arrays
+            filter_values_list = list(itertools.chain.from_iterable(filter_values_list))
+            # Check if / or : exist in any of the values
+            if any(("/" in str(item) or ":" in str(item)) for item in (filter_values_list)):
+                return False
+            return True
+
     @staticmethod
     def parse_query_term(query_term: str) -> str:
         # TODO: Might be some issue with using wildcard & underscore
@@ -400,8 +415,11 @@ def convert_query_json_to_query_dsl(self, *,
         filter_list = search_request.get('filters')
         add_query = ''
         query_dsl = ''
-
         if filter_list:
+            valid_filters = self.validate_filter_values(search_request)
+            if valid_filters is False:
+                raise Exception(
+                    'The search filters contain invalid characters and thus cannot be handled by ES')
             query_dsl = self.parse_filters(filter_list)
 
         if query_term:

tests/unit/proxy/test_elasticsearch.py

@@ -367,6 +367,30 @@ def test_parse_filters_return_no_results(self) -> None:
         }
         self.assertEquals(self.es_proxy.parse_filters(filter_list), '')
 
+    def test_validate_wrong_filters_values(self) -> None:
+        search_request = {
+            "type": "AND",
+            "filters": {
+                "schema": ["test_schema:test_schema"],
+                "table": ["test/table"]
+            },
+            "query_term": "",
+            "page_index": 0
+        }
+        self.assertEquals(self.es_proxy.validate_filter_values(search_request), False)
+
+    def test_validate_accepted_filters_values(self) -> None:
+        search_request = {
+            "type": "AND",
+            "filters": {
+                "schema": ["test_schema"],
+                "table": ["test_table"]
+            },
+            "query_term": "a",
+            "page_index": 0
+        }
+        self.assertEquals(self.es_proxy.validate_filter_values(search_request), True)
+
     def test_parse_query_term(self) -> None:
         term = 'test'
         expected_result = "(name:(*test*) OR name:(test) OR schema:(*test*) OR " \