Added support for AWS Sigv4 for UrlLib3.

CHANGELOG.md

@@ -9,6 +9,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Added `pool_maxsize` for `Urllib3HttpConnection` ([#535](https://github.com/opensearch-project/opensearch-py/pull/535))
 - Added benchmarks ([#537](https://github.com/opensearch-project/opensearch-py/pull/537))
 - Added guide on making raw JSON REST requests ([#542](https://github.com/opensearch-project/opensearch-py/pull/542))
+- Added support for AWS SigV4 for urllib3 ([#547](https://github.com/opensearch-project/opensearch-py/pull/547))
 ### Changed
 - Generate `tasks` client from API specs ([#508](https://github.com/opensearch-project/opensearch-py/pull/508))
 - Generate `ingest` client from API specs ([#513](https://github.com/opensearch-project/opensearch-py/pull/513))

DEVELOPER_GUIDE.md

@@ -45,7 +45,19 @@ docker run -d -p 9200:9200 -p 9600:9600 -e "discovery.type=single-node" opensear
 
 Tests require a live instance of OpenSearch running in docker.
 
-This will start a new instance and run tests against the latest version of OpenSearch.
+If you have one running.
+
+```
+python setup.py test
+```
+
+To run tests in a specific test file.
+
+```
+python setup.py test -s test_opensearchpy/test_connection.py
+```
+
+If you want to auto-start one, the following will start a new instance and run tests against the latest version of OpenSearch.
 
 ```
 ./.ci/run-tests
@@ -76,7 +88,7 @@ You can also run individual tests matching a pattern (`pytest -k [pattern]`).
 ```
 ./.ci/run-tests true 1.3.0 test_no_http_compression
 
-test_opensearchpy/test_connection.py::TestUrllib3Connection::test_no_http_compression PASSED [ 33%]
+test_opensearchpy/test_connection.py::TestUrllib3HttpConnection::test_no_http_compression PASSED [ 33%]
 test_opensearchpy/test_connection.py::TestRequestsConnection::test_no_http_compression PASSED [ 66%]
 test_opensearchpy/test_async/test_connection.py::TestAIOHttpConnection::test_no_http_compression PASSED [100%]
 ```

guides/auth.md

@@ -9,24 +9,24 @@ OpenSearch allows you to use different methods for the authentication via `conne
 
 ## IAM Authentication
 
-Opensearch-py supports IAM-based authentication via `AWSV4SignerAuth`, which uses `RequestHttpConnection` as the transport class for communicating with OpenSearch clusters running in Amazon Managed OpenSearch and OpenSearch Serverless, and works in conjunction with [botocore](https://pypi.org/project/botocore/).
+Opensearch-py supports IAM-based authentication via `RequestsAWSV4SignerAuth` and `Urllib3AWSV4SignerAuth`, which use `RequestHttpConnection` and `Urllib3HttpConnection` respectively, as the transport classes for communicating with OpenSearch clusters running in Amazon Managed OpenSearch and OpenSearch Serverless, and works in conjunction with [botocore](https://pypi.org/project/botocore/).
 
 ```python
-from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth
+from opensearchpy import OpenSearch, Urllib3HttpConnection, Urllib3AWSV4SignerAuth
 import boto3
 
 host = '' # cluster endpoint, for example: my-test-domain.us-east-1.es.amazonaws.com
 region = 'us-west-2'
 service = 'es' # 'aoss' for OpenSearch Serverless
 credentials = boto3.Session().get_credentials()
-auth = AWSV4SignerAuth(credentials, region, service)
+auth = Urllib3AWSV4SignerAuth(credentials, region, service)
 
 client = OpenSearch(
     hosts = [{'host': host, 'port': 443}],
     http_auth = auth,
     use_ssl = True,
     verify_certs = True,
-    connection_class = RequestsHttpConnection,
+    connection_class = Urllib3HttpConnection,
     pool_maxsize = 20
 )
 

opensearchpy/__init__.py

@@ -71,7 +71,12 @@
     UnknownDslObject,
     ValidationException,
 )
-from .helpers import AWSV4SignerAsyncAuth, AWSV4SignerAuth
+from .helpers import (
+    AWSV4SignerAsyncAuth,
+    AWSV4SignerAuth,
+    RequestsAWSV4SignerAuth,
+    Urllib3AWSV4SignerAuth,
+)
 from .helpers.aggs import A
 from .helpers.analysis import analyzer, char_filter, normalizer, token_filter, tokenizer
 from .helpers.document import Document, InnerDoc, MetaField
@@ -166,6 +171,8 @@
     "OpenSearchWarning",
     "OpenSearchDeprecationWarning",
     "AWSV4SignerAuth",
+    "Urllib3AWSV4SignerAuth",
+    "RequestsAWSV4SignerAuth",
     "AWSV4SignerAsyncAuth",
     "A",
     "AttrDict",

opensearchpy/connection/http_urllib3.py

@@ -27,6 +27,7 @@
 import ssl
 import time
 import warnings
+from typing import Callable
 
 import urllib3  # type: ignore
 from urllib3.exceptions import ReadTimeoutError
@@ -128,10 +129,17 @@ def __init__(
             opaque_id=opaque_id,
             **kwargs
         )
-        if http_auth is not None:
-            if isinstance(http_auth, (tuple, list)):
-                http_auth = ":".join(http_auth)
-            self.headers.update(urllib3.make_headers(basic_auth=http_auth))
+
+        self.http_auth = http_auth
+        if self.http_auth is not None:
+            if isinstance(self.http_auth, Callable):
+                pass
+            elif isinstance(self.http_auth, (tuple, list)):
+                self.headers.update(
+                    urllib3.make_headers(basic_auth=":".join(http_auth))
+                )
+            else:
+                self.headers.update(urllib3.make_headers(basic_auth=http_auth))
 
         pool_class = urllib3.HTTPConnectionPool
         kw = {}
@@ -218,6 +226,7 @@ def perform_request(
             url = "%s?%s" % (url, urlencode(params))
 
         full_url = self.host + url
+
         start = time.time()
         orig_body = body
         try:
@@ -240,6 +249,10 @@ def perform_request(
                 body = self._gzip_compress(body)
                 request_headers["content-encoding"] = "gzip"
 
+            if self.http_auth is not None:
+                if isinstance(self.http_auth, Callable):
+                    request_headers.update(self.http_auth(method, full_url, body))
+
             response = self.pool.urlopen(
                 method, url, body, retries=Retry(False), headers=request_headers, **kw
             )

opensearchpy/helpers/__init__.py

@@ -39,7 +39,7 @@
 )
 from .asyncsigner import AWSV4SignerAsyncAuth
 from .errors import BulkIndexError, ScanError
-from .signer import AWSV4SignerAuth
+from .signer import AWSV4SignerAuth, RequestsAWSV4SignerAuth, Urllib3AWSV4SignerAuth
 
 __all__ = [
     "BulkIndexError",
@@ -54,6 +54,8 @@
     "_process_bulk_chunk",
     "AWSV4SignerAuth",
     "AWSV4SignerAsyncAuth",
+    "RequestsAWSV4SignerAuth",
+    "Urllib3AWSV4SignerAuth",
 ]
 
 

opensearchpy/helpers/__init__.pyi

@@ -48,5 +48,6 @@ try:
     from .._async.helpers.actions import async_streaming_bulk as async_streaming_bulk
     from .asyncsigner import AWSV4SignerAsyncAuth as AWSV4SignerAsyncAuth
     from .signer import AWSV4SignerAuth as AWSV4SignerAuth
+    from .signer import RequestsAWSV4SignerAuth, Urllib3AWSV4SignerAuth
 except (ImportError, SyntaxError):
     pass

opensearchpy/helpers/signer.py

@@ -8,6 +8,7 @@
 # GitHub history for details.
 
 import sys
+from typing import Any, Callable, Dict
 
 import requests
 
@@ -43,12 +44,12 @@ def fetch_url(prepared_request):  # type: ignore
     return url.scheme + "://" + location + path + querystring
 
 
-class AWSV4SignerAuth(requests.auth.AuthBase):
+class AWSV4Signer:
     """
-    AWS V4 Request Signer for Requests.
+    Generic AWS V4 Request Signer.
     """
 
-    def __init__(self, credentials, region, service="es"):  # type: ignore
+    def __init__(self, credentials, region: str, service: str = "es") -> Any:  # type: ignore
         if not credentials:
             raise ValueError("Credentials cannot be empty")
         self.credentials = credentials
@@ -61,27 +62,20 @@ def __init__(self, credentials, region, service="es"):  # type: ignore
             raise ValueError("Service name cannot be empty")
         self.service = service
 
-    def __call__(self, request):  # type: ignore
-        return self._sign_request(request)  # type: ignore
-
-    def _sign_request(self, prepared_request):  # type: ignore
+    def sign(self, method: str, url: str, body: Any) -> Dict[str, str]:
         """
-        This method helps in signing the request by injecting the required headers.
-        :param prepared_request: unsigned request
-        :return: signed request
+        This method signs the request and returns headers.
+        :param method: HTTP method
+        :param url: url
+        :param body: body
+        :return: headers
         """
 
         from botocore.auth import SigV4Auth
         from botocore.awsrequest import AWSRequest
 
-        url = fetch_url(prepared_request)  # type: ignore
-
         # create an AWS request object and sign it using SigV4Auth
-        aws_request = AWSRequest(
-            method=prepared_request.method.upper(),
-            url=url,
-            data=prepared_request.body,
-        )
+        aws_request = AWSRequest(method=method.upper(), url=url, data=body)
 
         # credentials objects expose access_key, secret_key and token attributes
         # via @property annotations that call _refresh() on every access,
@@ -101,9 +95,49 @@ def _sign_request(self, prepared_request):  # type: ignore
         sig_v4_auth.add_auth(aws_request)
 
         # copy the headers from AWS request object into the prepared_request
-        prepared_request.headers.update(dict(aws_request.headers.items()))
-        prepared_request.headers["X-Amz-Content-SHA256"] = sig_v4_auth.payload(
-            aws_request
+        headers = dict(aws_request.headers.items())
+        headers["X-Amz-Content-SHA256"] = sig_v4_auth.payload(aws_request)
+
+        return headers
+
+
+class RequestsAWSV4SignerAuth(requests.auth.AuthBase):
+    """
+    AWS V4 Request Signer for Requests.
+    """
+
+    def __init__(self, credentials, region, service="es"):  # type: ignore
+        self.signer = AWSV4Signer(credentials, region, service)
+
+    def __call__(self, request):  # type: ignore
+        return self._sign_request(request)  # type: ignore
+
+    def _sign_request(self, prepared_request):  # type: ignore
+        """
+        This method helps in signing the request by injecting the required headers.
+        :param prepared_request: unsigned request
+        :return: signed request
+        """
+
+        prepared_request.headers.update(
+            self.signer.sign(
+                prepared_request.method,
+                fetch_url(prepared_request),  # type: ignore
+                prepared_request.body,
+            )
         )
 
         return prepared_request
+
+
+# Deprecated: use RequestsAWSV4SignerAuth
+class AWSV4SignerAuth(RequestsAWSV4SignerAuth):
+    pass
+
+
+class Urllib3AWSV4SignerAuth(Callable):  # type: ignore
+    def __init__(self, credentials, region, service="es"):  # type: ignore
+        self.signer = AWSV4Signer(credentials, region, service)
+
+    def __call__(self, method: str, url: str, body: Any) -> Dict[str, str]:
+        return self.signer.sign(method, url, body)

samples/aws/README.md

@@ -0,0 +1,22 @@
+## AWS SigV4 Samples
+
+Create an OpenSearch domain in (AWS) which support IAM based AuthN/AuthZ.
+
+```
+export AWS_ACCESS_KEY_ID=
+export AWS_SECRET_ACCESS_KEY=
+export AWS_SESSION_TOKEN=
+export AWS_REGION=us-west-2
+
+export SERVICE=es # use "aoss" for OpenSearch Serverless.
+export ENDPOINT=https://....us-west-2.es.amazonaws.com
+
+poetry run aws/search-urllib.py
+```
+
+This will output the version of OpenSearch and a search result.
+
+```
+opensearch: 2.3.0
+{'director': 'Bennett Miller', 'title': 'Moneyball', 'year': 2011}
+```

samples/aws/search-requests.py

@@ -0,0 +1,69 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+import logging
+
+from os import environ
+from time import sleep
+from urllib.parse import urlparse
+
+from boto3 import Session
+from opensearchpy import RequestsAWSV4SignerAuth, OpenSearch, RequestsHttpConnection
+
+# verbose logging
+logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.INFO)
+
+# cluster endpoint, for example: my-test-domain.us-east-1.es.amazonaws.com
+url = urlparse(environ['ENDPOINT'])
+region = environ.get('AWS_REGION', 'us-east-1')
+service = environ.get('SERVICE', 'es')
+
+credentials = Session().get_credentials()
+
+auth = RequestsAWSV4SignerAuth(credentials, region, service)
+
+client = OpenSearch(
+  hosts=[{
+    'host': url.netloc,
+    'port': url.port or 443
+  }],
+  http_auth=auth,
+  use_ssl=True,
+  verify_certs=True,
+  connection_class=RequestsHttpConnection,
+  timeout=30
+)
+
+# TODO: remove when OpenSearch Serverless adds support for /
+if service == 'es':
+  info = client.info()
+  print(f"{info['version']['distribution']}: {info['version']['number']}")
+
+# create an index
+index = 'movies'
+client.indices.create(index=index)
+
+try:
+  # index data
+  document = {'director': 'Bennett Miller', 'title': 'Moneyball', 'year': 2011}
+  client.index(index=index, body=document, id='1')
+
+  # wait for the document to index
+  sleep(1)
+
+  # search for the document
+  results = client.search(body={'query': {'match': {'director': 'miller'}}})
+  for hit in results['hits']['hits']:
+    print(hit['_source'])
+
+  # delete the document
+  client.delete(index=index, id='1')
+finally:
+  # delete the index
+  client.indices.delete(index=index)