Replace requests-auth-aws-sigv4 with boto3 signing

Signed-off-by: Andre Kurait <akurait@amazon.com>

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/Pipfile

@@ -10,7 +10,6 @@ pyyaml = "*"
 Click = "*"
 cerberus = "*"
 awscli = "*"
-requests-auth-aws-sigv4 = ">=0.7"
 
 [dev-packages]
 pytest = "*"

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/console_link/models/cluster.py

@@ -8,11 +8,10 @@
 import requests
 import requests.auth
 from requests.auth import HTTPBasicAuth
-from requests_auth_aws_sigv4 import AWSSigV4
-
+import boto3
 from console_link.models.client_options import ClientOptions
 from console_link.models.schema_tools import contains_one_of
-from console_link.models.utils import create_boto3_client, append_user_agent_header_for_requests
+from console_link.models.utils import SigV4AuthPlugin, create_boto3_client, append_user_agent_header_for_requests
 
 requests.packages.urllib3.disable_warnings()  # ignore: type
 
@@ -136,8 +135,8 @@ def _generate_auth_object(self) -> requests.auth.AuthBase | None:
                 password
             )
         elif self.auth_type == AuthMethod.SIGV4:
-            sigv4_details = self._get_sigv4_details()
-            return AWSSigV4(sigv4_details[0], region=sigv4_details[1])
+            service_name, region_name = self._get_sigv4_details(force_region=True)
+            return SigV4AuthPlugin(service_name, region_name)
         elif self.auth_type is AuthMethod.NO_AUTH:
             return None
         raise NotImplementedError(f"Auth type {self.auth_type} not implemented")
@@ -200,3 +199,4 @@ def execute_benchmark_workload(self, workload: str,
         display_command = command.replace(f"basic_auth_password:{password_to_censor}", "basic_auth_password:********")
         logger.info(f"Executing command: {display_command}")
         subprocess.run(command, shell=True)
+

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/console_link/models/utils.py

@@ -4,6 +4,10 @@
 from datetime import datetime
 import boto3
 import requests.utils
+from botocore.auth import SigV4Auth
+from botocore.awsrequest import AWSRequest
+from requests.models import PreparedRequest
+
 
 from console_link.models.client_options import ClientOptions
 
@@ -54,3 +58,21 @@ def append_user_agent_header_for_requests(headers: Optional[dict], user_agent_ex
     else:
         adjusted_headers["User-Agent"] = f"{requests.utils.default_user_agent()} {user_agent_extra}"
     return adjusted_headers
+
+# The SigV4AuthPlugin allows us to use boto3 with the requests library by integrating
+# AWS Signature Version 4 signing. This enables the requests library to authenticate
+# requests to AWS services using SigV4.
+class SigV4AuthPlugin(requests.auth.AuthBase):
+    def __init__(self, service, region):
+        self.service = service
+        self.region = region
+        session = boto3.Session()
+        self.credentials = session.get_credentials()
+
+    def __call__(self, r: PreparedRequest) -> PreparedRequest:
+        aws_request = AWSRequest(method=r.method, url=r.url, data=r.body, headers=r.headers)
+        signer = SigV4Auth(self.credentials, self.service, self.region)
+        if aws_request.body is not None:
+            aws_request.headers['x-amz-content-sha256'] = signer.payload(aws_request)
+        signer.add_auth(aws_request)
+        return aws_request

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/setup.py

@@ -5,8 +5,7 @@
     version="1.0.0",
     description="A Python module to create a console application from a Python script",
     packages=find_packages(exclude=("tests")),
-    install_requires=["requests", "boto3", "pyyaml", "Click", "cerberus",
-                      "requests-auth-aws-sigv4"],
+    install_requires=["requests", "boto3", "pyyaml", "Click", "cerberus"],
     entry_points={
         "console_scripts": [
             "console = console_link.cli:cli",

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/tests/test_cluster.py

@@ -1,13 +1,17 @@
-from base64 import b64encode
+import boto3
+import console_link.middleware.clusters as clusters_
+import hashlib
 import os
-
 import pytest
-from moto import mock_aws
-import boto3
+import re
+import json
 
-import console_link.middleware.clusters as clusters_
+from base64 import b64encode
+from botocore.auth import SigV4Auth
+from botocore.awsrequest import AWSRequest
 from console_link.models.client_options import ClientOptions
-from console_link.models.cluster import AuthMethod, Cluster
+from console_link.models.cluster import AuthMethod, Cluster, HttpMethod
+from moto import mock_aws
 from tests.utils import create_valid_cluster
 
 
@@ -383,3 +387,92 @@ def test_run_benchmark_executes_correctly_basic_auth_and_https(mocker):
                                  "--client-options=verify_certs:false,use_ssl:true,"
                                  f"basic_auth_user:{auth_details['username']},"
                                  f"basic_auth_password:{auth_details['password']}", shell=True)
+
+@pytest.mark.parametrize("method, endpoint, data, has_body", [
+    (HttpMethod.GET, "/_cluster/health", None, False),
+    (HttpMethod.POST, "/_search", {"query": {"match_all": {}}}, True)
+])
+def test_sigv4_authentication_signature(requests_mock, method, endpoint, data, has_body):
+    # Set up a valid cluster configuration with SIGV4 authentication
+    sigv4_cluster_config = {
+        "endpoint": "https://opensearchtarget:9200",
+        "allow_insecure": True,
+        "sigv4": {
+            "region": "us-east-1",
+            "service": "es"
+        }
+    }
+    cluster = Cluster(sigv4_cluster_config)
+
+    # Prepare the mocked API response
+    url = f"{cluster.endpoint}{endpoint}"
+    if method == HttpMethod.GET:
+        requests_mock.get(url, json={'status': 'green'})
+    elif method == HttpMethod.POST:
+        requests_mock.post(url, json={'hits': {'total': 0, 'hits': []}})
+
+    with mock_aws():
+        if data is not None:
+            response = cluster.call_api(endpoint, method=method, data=json.dumps(data))
+        else:
+            response = cluster.call_api(endpoint, method=method)
+        assert response.status_code == 200
+
+        # Retrieve the last request made
+        last_request = requests_mock.last_request
+
+        # Verify the Authorization header
+        auth_header = last_request.headers.get('Authorization')
+        assert auth_header is not None, "Authorization header is missing"
+        assert auth_header.startswith("AWS4-HMAC-SHA256"), "Incorrect Authorization header format"
+
+        # Extract SignedHeaders and Signature from the Authorization header
+        signed_headers_match = re.search(r"SignedHeaders=([^,]+)", auth_header)
+        signature_match = re.search(r"Signature=([a-f0-9]+)", auth_header)
+        assert signed_headers_match is not None, "SignedHeaders not found in Authorization header"
+        assert signature_match is not None, "Signature not found in Authorization header"
+
+        # Verify that essential headers are included in SignedHeaders
+        signed_headers = signed_headers_match.group(1).split(';')
+        required_headers = ['host', 'x-amz-date']
+        for header in required_headers:
+            assert header in signed_headers, f"Header '{header}' not found in SignedHeaders"
+
+        # Check that the x-amz-date header is present
+        amz_date_header = last_request.headers.get('x-amz-date')
+        assert amz_date_header is not None, "x-amz-date header is missing"
+
+        if has_body:
+            # Verify that the 'x-amz-content-sha256' header is present
+            content_sha256 = last_request.headers.get('x-amz-content-sha256')
+            assert content_sha256 is not None, f"x-amz-content-sha256 header is missing. Headers: {last_request.headers}"
+            # Compute the SHA256 hash of the body
+            body_hash = hashlib.sha256(last_request.body.encode('utf-8')).hexdigest()
+            assert content_sha256 == body_hash, "x-amz-content-sha256 does not match body hash"
+
+        # Re-sign the request using botocore to verify the signature
+        session = boto3.Session()
+        credentials = session.get_credentials().get_frozen_credentials()
+        service_name = cluster.auth_details.get("service", "es")
+        region_name = cluster.auth_details.get("region", "us-east-1")
+        # Create a new AWSRequest
+        aws_request = AWSRequest(
+            method=last_request.method,
+            url=last_request.url,
+            data=last_request.body,
+            headers=dict(last_request.headers)
+        )
+        # Sign the request
+        SigV4Auth(credentials, service_name, region_name).add_auth(aws_request)
+
+        # Extract the new signature
+        new_auth_header = aws_request.headers.get('Authorization')
+        assert new_auth_header is not None, "Failed to generate new Authorization header"
+
+        # Compare signatures
+        original_signature = signature_match.group(1)
+        new_signature_match = re.search(r"Signature=([a-f0-9]+)", new_auth_header)
+        assert new_signature_match is not None, "New signature not found in Authorization header"
+        new_signature = new_signature_match.group(1)
+
+        assert original_signature == new_signature, "Signatures do not match"