🔒 Replace mktemp with NamedTemporaryFile (CWE-377)

Author: shnizzedy

indi_aws/aws_utils.py

@@ -431,49 +431,52 @@ def test_bucket_access(creds_path, output_directory):
     """
 
     # Import packages
+    import logging
     import os
-    import tempfile
+    from tempfile import NamedTemporaryFile
 
-    import botocore.exceptions as bexc
+    from botocore.exceptions import ClientError
+    from boto3.exceptions import S3UploadFailedError
     from indi_aws import fetch_creds
 
     # Init variables
     s3_str = 's3://'
-    test_file = tempfile.mktemp()
-
-    # Explicitly lower-case the "s3"
-    if output_directory.lower().startswith(s3_str):
-        out_dir_sp = output_directory.split('/')
-        out_dir_sp[0] = out_dir_sp[0].lower()
-        output_directory = '/'.join(out_dir_sp)
-
-    # Get bucket name
-    bucket_name = output_directory.replace(s3_str, '').split('/')[0]
-
-    # Get bucket
-    bucket = fetch_creds.return_bucket(creds_path, bucket_name)
-
-    # Create local file
-    with open(test_file, 'w') as f:
-        f.write('test123')
-    f.close()
-
-    # Formulate test ouput key in bucket path output directory
-    rel_key_path = output_directory.replace(
-        os.path.join(s3_str, bucket_name), '').lstrip('/')
-    write_test_key = os.path.join(rel_key_path, os.path.basename(test_file))
-
-    # Attempt a write to bucket
-    try:
-        bucket.upload_file(test_file, write_test_key)
-        print('S3 write access confirmed!')
-        test_key = bucket.Object(key=write_test_key)
-        test_key.delete()
-        s3_write_access = True
-    # Otherwise we set the access flag to false
-    except bexc.ClientError:
-        print('S3 write access is not available!')
-        s3_write_access = False
+    with NamedTemporaryFile('w', encoding='utf8') as _temp_file:
+        test_file = _temp_file.name
+
+        # Explicitly lower-case the "s3"
+        if output_directory.lower().startswith(s3_str):
+            out_dir_sp = output_directory.split('/')
+            out_dir_sp[0] = out_dir_sp[0].lower()
+            output_directory = '/'.join(out_dir_sp)
+
+        # Get bucket name
+        bucket_name = output_directory.replace(s3_str, '').split('/')[0]
+
+        # Get bucket
+        bucket = fetch_creds.return_bucket(creds_path, bucket_name)
+
+        # Create local file
+        with open(test_file, 'w') as f:
+            f.write('test123')
+        f.close()
+
+        # Formulate test ouput key in bucket path output directory
+        rel_key_path = output_directory.replace(
+            os.path.join(s3_str, bucket_name), '').lstrip('/')
+        write_test_key = os.path.join(rel_key_path, os.path.basename(test_file))
+
+        # Attempt a write to bucket
+        try:
+            bucket.upload_file(test_file, write_test_key)
+            logging.info('S3 write access confirmed!')
+            test_key = bucket.Object(key=write_test_key)
+            test_key.delete()
+            s3_write_access = True
+        # Otherwise we set the access flag to false
+        except (ClientError, S3UploadFailedError) as exc:
+            logging.warning('S3 write access is not available!')
+            s3_write_access = False
 
     # Return the access flag
     return s3_write_access