HARMONY-1938: Update code to omit Authorization header when redirected to a presigned S3 URL

harmony_service_lib/earthdata.py

@@ -1,7 +1,15 @@
+from urllib.parse import urlparse, parse_qs
 from requests.auth import AuthBase
 from requests import Session
 
 
+def _is_presigned_url(url: str) -> bool:
+    """Check if the URL is an AWS presigned URL. For AWS presigned URLs we do not
+    want to pass in an authorization header."""
+    query_params = parse_qs(urlparse(url).query)
+    return "X-Amz-Algorithm" in query_params or "Signature" in query_params
+
+
 class EarthdataSession(Session):
     """Session which ensures the Authorization header is sent to correct
     servers.
@@ -15,39 +23,24 @@ class EarthdataSession(Session):
     redirect requests.
     """
     def rebuild_auth(self, prepared_request, response):
-        # If not configured with an EarthdataAuth instance, defer to
-        # default behavior
+        # If not configured with an EarthdataAuth instance, defer to default behavior
         if not self.auth:
             return super().rebuild_auth(prepared_request, response)
 
         self.auth(prepared_request)
 
 
 class EarthdataAuth(AuthBase):
-    """Custom Earthdata Auth provider to add EDL Authorization headers to
-    requests when required for token sharing and federated
-    authentication.
-
-    When instantiated with an EDL application's credentials and a
-    user's access token, the resulting HTTP Authorization header will
-    include the properly-encoded app credentials as a Basic auth
-    header, and the user's access token as a Bearer auth header.
+    """Custom Earthdata Auth provider to add EDL Authorization headers."""
 
-    """
     def __init__(self, user_access_token: str):
-        """Instantiate the Earthdata Auth provider.
-
-        Parameters
-        ----------
-        user_access_token:
-            The EDL-issued token for the user making the request.
-        """
-        self.authorization_header = f'Bearer {user_access_token}'
+        self.authorization_header = f"Bearer {user_access_token}"
 
     def __call__(self, r):
-        """The EarthdataAuth is a callable which adds Authorization headers
-        when handling a request for sites backed by Earthdata Login.
+        """Add Authorization headers unless the request is to an AWS presigned URL."""
+        if _is_presigned_url(r.url):
+            r.headers.pop('Authorization', None)
+        else:
+            r.headers["Authorization"] = self.authorization_header
 
-        """
-        r.headers['Authorization'] = self.authorization_header
         return r

tests/test_earthdata.py

@@ -31,13 +31,27 @@ def test_authdata_auth_creates_correct_header(faker):
     assert token in request.headers['Authorization']
 
 
-def test_earthdata_auth_given_edl_url_adds_auth_header(earthdata_auth):
+def test_earthdata_auth_adds_auth_header_(earthdata_auth):
     request = FakeRequest()
 
     earthdata_auth(request)
 
     assert 'Authorization' in request.headers
 
+def test_earthdata_auth_removes_auth_header_when_X_Amz_Algorithm_is_set(earthdata_auth):
+    request = FakeRequest(url='https://presigned.s3.url.com?X-Amz-Algorithm=foo')
+
+    earthdata_auth(request)
+
+    assert 'Authorization' not in request.headers
+
+def test_earthdata_auth_removes_auth_header_when_signature_is_set(earthdata_auth):
+    request = FakeRequest(url='https://presigned.s3.url.com?Signature=bar')
+
+    earthdata_auth(request)
+
+    assert 'Authorization' not in request.headers
+
 def test_earthdata_session_given_no_auth_delegates_to_super(monkeypatch):
     called = False