Double auth

api/src/adapters/search/opensearch_client.py

@@ -3,6 +3,7 @@
 
 import boto3
 import opensearchpy
+from requests.auth import HTTPBasicAuth
 
 from src.adapters.search.opensearch_config import OpensearchConfig, get_opensearch_config
 from src.adapters.search.opensearch_response import SearchResponse
@@ -252,6 +253,22 @@ def scroll(
         self._client.clear_scroll(scroll_id=scroll_id)
 
 
+class CustomAuth(opensearchpy.RequestsAWSV4SignerAuth):
+    # TODO - better name  once we see if this works
+    def __init__(self, credentials, region, username: str, password: str, service: str = "es") -> None:  # type: ignore
+        super().__init__(credentials, region, service)
+
+        self.username = username
+        self.password = password
+
+    def __call__(self, request):  # type: ignore
+        # Use HTTPBasicAuth's __call__ method to add a
+        # username+password authorization header
+        request = HTTPBasicAuth(self.username, self.password)(request)
+
+        return self._sign_request(request)
+
+
 def _get_connection_parameters(opensearch_config: OpensearchConfig) -> dict[str, Any]:
     # See: https://opensearch.org/docs/latest/clients/python-low-level/#connecting-to-opensearch
     # for further details on configuring the connection to OpenSearch
@@ -268,13 +285,13 @@ def _get_connection_parameters(opensearch_config: OpensearchConfig) -> dict[str,
     # and should connect using the session credentials
     if opensearch_config.search_username and opensearch_config.search_password:
         # Get credentials and authorize with AWS Opensearch Serverless (es)
-        #credentials = boto3.Session().get_credentials()
-        #auth = opensearchpy.AWSV4SignerAuth(credentials, opensearch_config.aws_region, "es")
-        auth = (opensearch_config.search_username, opensearch_config.search_password)
+        credentials = boto3.Session().get_credentials()
+        auth = CustomAuth(
+            credentials=credentials,
+            region=opensearch_config.aws_region,
+            username=opensearch_config.search_username,
+            password=opensearch_config.search_password,
+        )
         params["http_auth"] = auth
 
-        # TODO hacky - do this a better way
-        #path = f"https://{opensearch_config.search_username}:{opensearch_config.search_password}@{opensearch_config.search_endpoint}:{opensearch_config.search_port}"
-        #params["hosts"] = [path]
-
     return params