Added new boto api and upgraded boto version

requirements.txt

@@ -15,3 +15,4 @@ setuptools>=56.0.0
 lockfile>=0.10
 urllib3<1.27,>=1.26.2
 distro>=1.8.0
+boto3==1.34.28

src/wazuh_testing/modules/aws/utils.py

@@ -30,10 +30,9 @@
 
 # Session setup
 session = boto3.Session(profile_name=f'{aws_profile}')
-s3 = session.client(service_name="s3")
+s3 = session.resource(service_name="s3")
 logs = session.client(service_name="logs", region_name=US_EAST_1_REGION)
 sqs = session.client(service_name="sqs", region_name=US_EAST_1_REGION)
-sts = session.client(service_name="sts")
 
 
 # Custom exception
@@ -80,9 +79,12 @@ def delete_bucket(bucket_name: str):
         bool: True if bucket is deleted else False.
     """
     try:
-        s3.delete_bucket(
-            Bucket=bucket_name
-        )
+        # Get bucket
+        bucket = s3.Bucket(name=bucket_name)
+
+        # Delete bucket
+        bucket.delete()
+
     except ClientError as error:
         if error.response['Error']['Code'] == 'ResourceNotFound':
             logger.error(f"Bucket {bucket_name} not found.")
@@ -91,6 +93,20 @@ def delete_bucket(bucket_name: str):
             raise error
 
 
+def delete_bucket_files(bucket_name: str):
+    """Delete all files i n the bucket"""
+    try:
+        # Get bucket
+        bucket = s3.Bucket(name=bucket_name)
+
+        # Delete all objects
+        bucket.objects.all().delete()
+    except ClientError as error:
+        raise error
+    except Exception as error:
+        raise error
+
+
 def generate_file(bucket_type: str, bucket_name: str):
     """ Generate a file for a specific bucket type.
 
@@ -110,25 +126,31 @@ def generate_file(bucket_type: str, bucket_name: str):
     return data, filename
 
 
-def upload_bucket_file(bucket_name: str, data: str, filename: str):
+def upload_bucket_file(bucket_name: str, data: str, key: str):
     """Upload a file to an S3 bucket.
 
-    Args:
-        bucket_name (str): Bucket to upload.
-        data (str): Data to upload in bucket
-        filename (str): Name of file uploaded.
-
-    Returns:
-        bool: True if uploaded.
+    Parameters
+    ----------
+    bucket_name : str
+        Bucket to upload.
+    data : str
+        Data to upload in bucket
+    key : str
+        Key
     """
-    obj = s3.Object(bucket_name, filename)
-
-    # Upload the file
+    # Set bucket
     try:
-        obj.put(Body=data)
-        return True
+        # Get bucket
+        bucket = s3.Bucket(name=bucket_name)
+
+        # Upload the file
+        bucket.put_object(
+            Key=key,
+            Body=data
+        )
     except ClientError as error:
-        logger.error(error)
+        raise error
+    except Exception as error:
         raise error
 
 
@@ -142,25 +164,6 @@ def delete_bucket_file(filename: str, bucket_name: str):
     s3.Object(bucket_name, filename).delete()
 
 
-# def file_exists(filename: str, bucket_name: str):
-#     """Check if a file exists in a bucket.
-#
-#     Args:
-#         filename (str): Full filename to check.
-#         bucket_name (str): Bucket that contains the file.
-#     Returns:
-#         bool: True if exists else False.
-#     """
-#     exists = True
-#     try:
-#         s3.Object(bucket_name, filename).load()
-#     except ClientError as error:
-#         if error.response['Error']['Code'] == '404':
-#             exists = False
-#
-#     return exists
-
-
 def get_last_file_key(bucket_type: str, bucket_name: str, execution_datetime: datetime):
     """Return the last file key contained in a default path of a bucket.
 
@@ -404,7 +407,7 @@ def get_sqs_queue_arn(sqs_url: str) -> str:
         raise error
 
 
-def set_sqs_policy(bucket_name: str, sqs_queue_url: str) -> None:
+def set_sqs_policy(bucket_name: str, sqs_queue_url: str, sqs_queue_arn: str) -> None:
     """Set a policy for the SQS queue
 
     Parameters
@@ -413,16 +416,15 @@ def set_sqs_policy(bucket_name: str, sqs_queue_url: str) -> None:
         The bucket name.
     sqs_queue_url : str
         The SQS queue Url to apply policy.
+    sqs_queue_arn : str
+        The SQS queue ARN.
     """
     # Get account id
-    account_id = sts.get_caller_identiity()["Account"]
-
-    # Get date
-    today_date = datetime.now().date().strftime("%Y-%m-%d")
+    account_id = sqs_queue_arn.split(':')[4]
 
     # Create Policy
     policy = {
-        "Version": today_date,
+        "Version": "2012-10-17",
         "Id": "wazuh-integration-test-policy-ID",
         "Statement": [
             {
@@ -432,7 +434,7 @@ def set_sqs_policy(bucket_name: str, sqs_queue_url: str) -> None:
                     "Service": "s3.amazonaws.com"
                 },
                 "Action": "SQS:SendMessage",
-                "Resource": f"arn:aws:sqs:{US_EAST_1_REGION}:{account_id}:{bucket_name}",
+                "Resource": sqs_queue_arn,
                 "Condition": {
                     "StringEquals": {
                         "aws:SourceAccount": account_id
@@ -480,10 +482,10 @@ def set_bucket_event_notification_configuration(bucket_name: str, sqs_queue_arn:
         ]
     }
     try:
-        s3.put_bucket_notification_configuration(
-            bucket=bucket_name,
-            NotificationConfiguration=notification_configuration
-        )
+        # Get bucket
+        bucket = s3.Bucket(name=bucket_name)
+
+        bucket.Notification().put(NotificationConfiguration=notification_configuration)
 
     except ClientError as error:
         raise error