issues/151: Added ExpectedBucketOwner parameter when making S3 requests

Author: jackiryan

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Removed
 ### Fixed
 ### Security
+- [issues/151](https://github.com/podaac/bignbit/issues/151): Added ExpectedBucketOwner parameter when making S3 requests.
 
 ## [0.7.2]
 ### Added

bignbit/utils.py

@@ -15,10 +15,27 @@
 ED_USER = ED_PASS = None
 EDL_USER_TOKEN: dict[str, str] = {}
 HARMONY_CLIENT: Client | None = None
+AWS_ACCOUNT_ID: str | None = None
 
 HARMONY_SHOULD_VALIDATE_AUTH = os.environ.get('HARMONY_SHOULD_VALIDATE_AUTH', default='False').upper() == 'TRUE'
 
 
+def get_aws_account_id() -> str:
+    """
+    Get and cache the current AWS account ID via STS.
+
+    Returns
+    -------
+    str
+        The AWS account ID for the current caller identity
+    """
+    global AWS_ACCOUNT_ID  # pylint: disable=W0603
+    if not AWS_ACCOUNT_ID:
+        sts_client = boto3.client('sts')
+        AWS_ACCOUNT_ID = sts_client.get_caller_identity()['Account']
+    return AWS_ACCOUNT_ID
+
+
 def get_edl_creds() -> tuple[str, str]:
     """
     Get EDL username and password from SSM.
@@ -215,7 +232,8 @@ def upload_string_as_object(bucket_name: str, key_name: str, object_content: str
     s3_client.put_object(
         Body=object_content.encode(),
         Bucket=bucket_name,
-        Key=key_name
+        Key=key_name,
+        ExpectedBucketOwner=get_aws_account_id()
     )
     return f's3://{bucket_name}/{key_name}'
 
@@ -235,6 +253,7 @@ def upload_object(
         Key=key,
         Body=body_content,
         ContentType=content_type,
+        ExpectedBucketOwner=get_aws_account_id()
     )
     return f's3://{bucket}/{key}'
 
@@ -258,7 +277,8 @@ def upload_to_s3(filepath: pathlib.Path, bucket_name: str, object_key: str):
       s3 uri of new object
     """
     s3_client = boto3.client('s3')
-    s3_client.upload_file(str(filepath), bucket_name, object_key)
+    s3_client.upload_file(str(filepath), bucket_name, object_key,
+                          ExtraArgs={'ExpectedBucketOwner': get_aws_account_id()})
 
     return f's3://{bucket_name}/{object_key}'
 

tests/test_handle_big_result.py

@@ -4,7 +4,7 @@
 import xml.etree.ElementTree as ET
 import boto3
 import pytest
-from moto import mock_s3
+from moto import mock_s3, mock_sts
 
 import bignbit.utils
 from bignbit.handle_big_result import (
@@ -73,9 +73,11 @@ def test_process_harmony_results_no_data():
     file_dicts = process_harmony_results(harmony_job, cmr_environment)
     assert file_dicts == []
 
+@mock_sts
 @mock_s3
 def test_generate_metadata():
     """Test generating image metadata xml end-to-end for a single image set."""
+    bignbit.utils.AWS_ACCOUNT_ID = None
     # image metadata xml will be uploaded within the function, so the bucket needs to be mocked
     s3_client = boto3.client('s3', region_name='us-west-2')
     bucket_name = 'svc-bignbit-podaac-sit-svc-staging'
@@ -234,9 +236,11 @@ def test_get_mdxml_cnm_file_meta():
     assert result['size'] == len(image_metadata_xml)
 
 
+@mock_sts
 @mock_s3
 def test_write_cnm_message():
     """Test writing CNM message to S3"""
+    bignbit.utils.AWS_ACCOUNT_ID = None
     # Create mock S3 bucket
     s3_client = boto3.client('s3', region_name='us-west-2')
     bucket_name = 'test-audit-bucket'