Merge pull request #113 from sa7mon/enhancements

Improve endpoint validation and add service tests

S3Scanner/S3Service.py

@@ -450,8 +450,7 @@ def validate_endpoint_url(self, use_ssl=True, verify_ssl=True, endpoint_address_
         try:
             validation_client.list_objects_v2(Bucket=non_existent_bucket, MaxKeys=0)
         except ClientError as e:
-            if (e.response['Error']['Code'] == 'NoSuchBucket' or e.response['Error']['Code'] == 'AccessDenied') and \
-                    'BucketName' in e.response['Error']:
+            if e.response['Error']['Code'] == 'NoSuchBucket' or e.response['Error']['Code'] == 'AccessDenied':
                 return True
             return False
         except botocore.exceptions.ConnectTimeoutError:

S3Scanner/__main__.py

@@ -16,7 +16,7 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from .exceptions import InvalidEndpointException
 
-CURRENT_VERSION = '2.0.0'
+CURRENT_VERSION = '2.0.1'
 AWS_ENDPOINT = 'https://s3.amazonaws.com'
 
 

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = S3Scanner
-version = 2.0.0
+version = 2.0.1
 author = Dan Salmon
 author_email = dan@salmon.cat
 description = Scan for open S3 buckets and dump the contents

tests/test_scanner.py

@@ -11,7 +11,7 @@ def test_arguments():
     s = S3Service()
 
     a = subprocess.run([sys.executable, '-m', 'S3Scanner', '--version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    assert a.stdout.decode('utf-8').strip() == '2.0.0'
+    assert a.stdout.decode('utf-8').strip() == '2.0.1'
 
     b = subprocess.run([sys.executable, '-m', 'S3Scanner', 'scan', '--bucket', 'flaws.cloud'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     assert_scanner_output(s, 'flaws.cloud | bucket_exists | AuthUsers: [], AllUsers: [Read]', b.stdout.decode('utf-8').strip())

tests/test_service.py

@@ -7,6 +7,7 @@
 from TestUtils import TestBucketService
 from S3Scanner.exceptions import AccessDeniedException, BucketMightNotExistException
 from pathlib import Path
+from urllib3 import disable_warnings
 
 testingFolder = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), 'test/')
 setupRan = False
@@ -543,3 +544,44 @@ def test_download_file():
 
     b = S3Bucket("bucket-no-existo")
     s.download_file(os.path.join(dest_folder, ''), b, True, o)
+
+
+def test_validate_endpoint_url_nonaws():
+    disable_warnings()
+    s = S3Service()
+
+    # Test CenturyLink_Lumen
+    s.endpoint_url = 'https://useast.os.ctl.io'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True
+
+    # Test DigitalOcean
+    s.endpoint_url = 'https://sfo2.digitaloceanspaces.com'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True
+
+    # Test Dreamhost
+    s.endpoint_url = 'https://objects.dreamhost.com'
+    assert s.validate_endpoint_url(use_ssl=False, verify_ssl=False, endpoint_address_style='vhost') is True
+
+    # Test GCP
+    s.endpoint_url = 'https://storage.googleapis.com'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True
+
+    # Test IBM
+    s.endpoint_url = 'https://s3.us-east.cloud-object-storage.appdomain.cloud'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True
+
+    # Test Linode
+    s.endpoint_url = 'https://eu-central-1.linodeobjects.com'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True
+
+    # Test Scaleway
+    s.endpoint_url = 'https://s3.nl-ams.scw.cloud'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True
+
+    # Test Vultr
+    s.endpoint_url = 'https://ewr1.vultrobjects.com'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True
+
+    # Test Wasabi
+    s.endpoint_url = 'https://s3.wasabisys.com'
+    assert s.validate_endpoint_url(use_ssl=True, verify_ssl=True, endpoint_address_style='path') is True