diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 0000000..717d2f4
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,4 @@
+[run]
+omit =
+ venv*
+ tests/TestUtils.py
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
index 98cb83d..78c43e7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,10 +1,24 @@
-venv/
-venv2.7/
+# IDE
.idea/
+.vscode/
+
+# Virtualenv
+venv*/
+
+# Cache
__pycache__
*.pyc
.cache/
+
+# Testing
.pytest_cache
-list-buckets/
+test/
+.coverage
+htmlcov/
+temp-*/
buckets.txt
-test/
\ No newline at end of file
+
+# Pip build
+build/
+dist/
+S3Scanner.egg-info/
\ No newline at end of file
diff --git a/.travis.yml b/.travis.yml
index d839301..1d3c68e 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,16 +1,28 @@
language: python
-matrix:
+jobs:
include:
- - python: '3.6'
- env:
- - secure: QCHu2+V5MO4OtZfBmB9yZVof0NeHK/3CGmErdK/4ICnvfkf4wteYKdDgFsZyzt/ns3BCuP3dZvXEU9p4EEudjMEv66RvPyr+q29JhK42a+HjHP7XAHBMLEENkihtBtrsnfr7NrSgPtfLkG4XaJVyn9mPCvo4JuDotldC4jlAEJqkByKbh5koI4533WOG36pmvO7jcsuqm3+ii7J9PYCRqqbqjvWjt1eJx8ITiBPbJ8+1e32HGWKSMJMJhe2RlxuhglQmOMlZDKrtQ/5u+pgMWtItjP798zJRyKEG8zrt+PM2vgYBeHB9rFMqCmtEO68zpB+HhYXb21R7qkDSTocL6uMKkR1cYplKwilFTLMutfUfCv3qRT77ciW8GGfHEbB7mukeeu8Jvz2lphFvbiRPiReQlWbmn0zJaSOW1SbqrGYrAy6WM4IyJR4v1sDPXZHUq6wYFywKqpbA4sF59RAVrPvMu/Z8Zhdol+P5CV6FZnTr1fz3JTEea/CMnbExKQfFUNgvAd11WncIyrxV0Pr26iR6HD8ajRQh3Uk8cvYtUZnnvTvSIF2Rp913RdJVaBjy1mZ/IarkhZIGp4F6y0892Ok6bRXkMsDLxkMMWWRWWUYgM/8UY9tckv4mBPDi8J9u2UAjYpkgxdyweCjI2F3aRBbRMiFZJPmepGbiUipk0zg=
- - secure: Phv6W+P19/wJQfY2frJNuqldZc4snt8gUaI8ZtXxTLBFYaUpsbM4Pl2lJErp+W4k0zZ9I0i6EKs+WWj4jLkmg5tauXL7O3C4QNK3u5RAD154bcG1up8U5XPrLvz0ZRzvbAJilQQpjLrMyLO7GXaZIzZIN4MaUCuTTlvKls5fCp3Q2UJZdCpW94zlyFKInkk5Ox1wZsCbjptG5i6eGm/Q+g/ObuyfMMykUISjP7/BElb3lbVnIe2MCWQxoNlt1d0T6oaUSkIbnaYD384uEkYn5bAlEoOyUlurgMyifPWoBI4vLJHBj8votdgjxcgThGzqjfsB29U/eaaP3VUc77CeBLMGwoYzqIPcoicUBkK5GowvIXkukoUqN24JK5kTROv/aYOdVqoLCrT8fYruVdS0kkJs3DZbBoMDaWte36PhvJouh7MPaWgdm1ZcfLP2juLe5UGaiuN5iY95L/XBXbrnd6+adJYrJd7TYX6+zJE2trmwaxqZCELQtUVUcrYJ85KljjQS7QJQTE1uXQRlPCzGQlhziC8ePJI1zOuHQV7xZWbi04Jg6Pe9zE3UXnBU6KvB/q3fLdrFT5F8mqI2lUXFl+kq28sucqgUMM1rfmW30OKkyHfiRuTZL43Img0Xduql++DgLz6WweFhXf8oH18b8XuV7KUZz5FFJgjs86FM8AE=
- python: '3.6'
- python: '3.7'
+ - python: '3.8'
+ - python: '3.9'
+ env:
+ - secure: "JShcKAHn4y57mTHDIV5+8dTRjE2cREJSswXAxhFf8jha+r58zF/uBfgXapzNh9u+dpvbVjF/N0/KxREubMTd4fduYTsxMOXyqENHnq7kVmRK6HXAAnM75JZzl1sonlHsIHHXxv45SuwYWX/fk6aMeBmkGukuvM8DGi4BEBzv0CnzEUmHlb5ZPKmQteemjhbn2d3yKPKagcieeDbSRhevGKPPmfnt0TqzpF/xrbtIL05yC+038Tesa0mZqV/HBrfZgSEtcMydIhbszhDjBwC3nzhhiC8AQJ8JGRPqR3nfTZRrHA0QMT3hr8XGpLouphvpDDwiotmOTRsGiBfONX+b2JDTx989eswIXmBsdua3pxjUNuLVTiRjl63+6zJSvT3mrJ1cZMRJPvbqYTY+mvckSMeDQv4oFZeD8QCD+z8zLa39GYfKBnapo0s+rvvxYyiVNZ9HQ1MExJyVleJWRMlmKtuNUHzHaCq+B8omcGZxhEfX4dVQ/RHwNRwkKbdUKOZy4muardhYorhVO+eLt4+bAipk8BEAXvIBwaAqbIN3+01a8TbTGKkxJUTllkf2Y7wFeF6IPtxvfpJ6Bgj4BNSpDrR/eoyIodG42J6Qdl4aK6/RQbI9vzUQ8yoSxQxzHHFZeclU2Qe5KZem3ztbexkiYB+Mv7oV/rr1LGixvbBsLzI="
+ - secure: "SLjBJJsmtbHZGwmZHHJTYk0qmlS6kcbur1SMM70+n/UEp55hAFo6Ae/n75G4RR0bVIVkgrJp3ZE9V/wZKVbOOaUaepyjZXfgRBjL/zBYjFgxgQhrLis3Bg+lR6qBoWifm/mfM+mUqHLDqelSbvpgE/oZLeM9vuYBYvI9LIZSeM3C6m+4ytKoayUgggq87lQRDr9d/YPpAZEYnT13mAqkd3zbovjLAEtALx6BOg5xZv7bHCx5WS5gz79CA+jFRjWU9q4ng5zyCERWFOeTcCYAHjxKXYOJYew8N/NYA2PFd+BiedQRHuIJAHg/auofchBewmtfHG6rgMZSuE+jzl1aB344zwpVJocR09kBXi6tk9KiASTrZMSHf31LEJFwciFnSsCTVQG7kVL0NdjZBEGO7zE1u5c4dxYEctDVvCz/kmH7hl70zajot9cYihh8VLvLwpGBYepTf2a+vhRwdxeZ81KuI3SeOBqNJwyT6wZMw9AfEOmK9LqyS9vBqusujwua+W/DDeqYFo99HkS2uMdX4/IfAB5DDhVakMrff8rrUuf1K1H6rtV7qckOHDET+wdjqymZkPD/mjjW+ibAattls4cZU3I7NRVsnNiZmXAT410A92y6JEiZPuG1djz/57yrqQ3S4KzVhgBq1t5WJRc84dUKCbYnwY4fDL4BH8lkync="
+ - name: "Python: 3.6"
+ os: windows
+ language: shell
+ before_install:
+ - choco install python --version=3.6.8
+ - python -m pip install -U pip setuptools
+ env: PATH=/c/Python36:/c/Python36/Scripts:$PATH
+ cache:
+ directories:
+ - $LOCALAPPDATA/pip/Cache
+
cache: pip
install:
- pip install -r requirements.txt
script:
-- pytest ./test_scanner.py -s
+- pytest -s
notifications:
- email: false
\ No newline at end of file
+ email: false
diff --git a/Dockerfile b/Dockerfile
index d0a80a9..aee9262 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,5 +1,6 @@
-FROM python:3-alpine
+FROM python:3.8-alpine
COPY . /app
WORKDIR /app
-RUN pip install -r requirements.txt
-ENTRYPOINT ["python", "s3scanner.py"]
+RUN pip install boto3
+RUN pip install .
+ENTRYPOINT ["python", "-m", "S3Scanner"]
\ No newline at end of file
diff --git a/LICENSE.txt b/LICENSE
similarity index 100%
rename from LICENSE.txt
rename to LICENSE
diff --git a/README.md b/README.md
index e7e6884..15f4e9c 100644
--- a/README.md
+++ b/README.md
@@ -1,134 +1,136 @@
# S3Scanner
[](https://opensource.org/licenses/MIT) [](https://travis-ci.org/sa7mon/S3Scanner)
-A tool to find open S3 buckets and dump their contents :droplet:
-
-
-
-### If you've earned a bug bounty using this tool, please consider donating to support it's development
-
-[](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=XG5BGLQZPJ9H8)
+A tool to find open S3 buckets and dump their contents๐ง
+
## Usage
-
-usage: s3scanner [-h] [-o OUTFILE] [-d] [-l] [--version] buckets
-
-# s3scanner - Find S3 buckets and dump!
-#
-# Author: Dan Salmon - @bltjetpack, github.com/sa7mon
+usage: s3scanner [-h] [--version] [--threads n] [--endpoint-url ENDPOINT_URL] [--endpoint-address-style {path,vhost}] [--insecure] {scan,dump} ...
-positional arguments:
- buckets Name of text file containing buckets to check
+s3scanner: Audit unsecured S3 buckets
+ by Dan Salmon - github.com/sa7mon, @bltjetpack
optional arguments:
-h, --help show this help message and exit
- -o OUTFILE, --out-file OUTFILE
- Name of file to save the successfully checked buckets in (Default: buckets.txt)
- -d, --dump Dump all found open buckets locally
- -l, --list Save bucket file listing to local file: ./list-buckets/${bucket}.txt
--version Display the current version of this tool
+ --threads n, -t n Number of threads to use. Default: 4
+ --endpoint-url ENDPOINT_URL, -u ENDPOINT_URL
+ URL of S3-compliant API. Default: https://s3.amazonaws.com
+ --endpoint-address-style {path,vhost}, -s {path,vhost}
+ Address style to use for the endpoint. Default: path
+ --insecure, -i Do not verify SSL
+
+mode:
+ {scan,dump} (Must choose one)
+ scan Scan bucket permissions
+ dump Dump the contents of buckets
-The tool takes in a list of bucket names to check. Found S3 buckets are output to file. The tool will also dump or list the contents of 'open' buckets locally.
-
-### Interpreting Results
-
-This tool will attempt to get all available information about a bucket, but it's up to you to interpret the results.
+## Support
+๐ If you've found this tool useful, please consider donating to support its development
-[Settings available](https://docs.aws.amazon.com/AmazonS3/latest/user-guide/set-bucket-permissions.html) for buckets:
-* Object Access (object in this case refers to files stored in the bucket)
- * List Objects
- * Write Objects
-* ACL Access
- * Read Permissions
- * Write Permissions
-
-Any or all of these permissions can be set for the 2 main user groups:
-* Authenticated Users
-* Public Users (those without AWS credentials set)
-* (They can also be applied to specific users, but that's out of scope)
-
-**What this means:** Just because a bucket returns "AccessDenied" for it's ACLs doesn't mean you can't read/write to it.
-Conversely, you may be able to list ACLs but not read/write to the bucket
+[](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=XG5BGLQZPJ9H8)
+[](https://ko-fi.com/B0B54D93O)
## Installation
- 1. (Optional) `virtualenv venv && source ./venv/bin/activate`
- 2. `pip install -r requirements.txt`
- 3. `python ./s3scanner.py`
-
-(Compatibility has been tested with Python 2.7 and 3.6)
-### Using Docker
+```shell
+pip3 install s3scanner
+```
- 1. Build the [Docker](https://docs.docker.com/) image:
+or via Docker:
- ```bash
-sudo docker build -t s3scanner https://github.com/sa7mon/S3Scanner.git
+```shell
+docker build . -t s3scanner:latest
+docker run --rm s3scanner:latest scan --bucket my-buket
```
- 2. Run the Docker image:
+or from source:
- ```bash
-sudo docker run -v /input-data-dir/:/data s3scanner --out-file /data/results.txt /data/names.txt
+```shell
+git clone git@github.com:sa7mon/S3Scanner.git
+cd S3Scanner
+pip3 install -r requirements.txt
+python3 -m S3Scanner
```
-This command assumes that `names.txt` with domains to enumerate is in `/input-data-dir/` on host machine.
+
+## Features
+
+* โก๏ธ Multi-threaded scanning
+* ๐ญ Supports tons of S3-compatible APIs
+* ๐ต๏ธโโ๏ธ Scans all bucket permissions to find misconfigurations
+* ๐พ Dump bucket contents to a local folder
+* ๐ณ Docker support
## Examples
-This tool accepts the following type of bucket formats to check:
-
-- bucket name - `google-dev`
-- domain name - `uber.com`, `sub.domain.com`
-- full s3 url - `yahoo-staging.s3-us-west-2.amazonaws.com` (To easily combine with other tools like [bucket-stream](https://github.com/eth0izzle/bucket-stream))
-- bucket:region - `flaws.cloud:us-west-2`
-
-```bash
-> cat names.txt
-flaws.cloud
-google-dev
-testing.microsoft.com
-yelp-production.s3-us-west-1.amazonaws.com
-github-dev:us-east-1
-```
-
-1. Dump all open buckets, log both open and closed buckets to found.txt
-
- ```bash
- > python ./s3scanner.py --include-closed --out-file found.txt --dump names.txt
- ```
-2. Just log open buckets to the default output file (buckets.txt)
-
- ```bash
- > python ./s3scanner.py names.txt
- ```
-3. Save file listings of all open buckets to file
- ```bash
- > python ./s3scanner.py --list names.txt
-
- ```
-
-## Contributing
-Issues are welcome and Pull Requests are appreciated. All contributions should be compatible with both Python 2.7 and 3.6.
-
-| master | [](https://travis-ci.org/sa7mon/S3Scanner) |
-|:------------:|:-------------------------------------------------------------------------------------------------------------------------:|
-| enhancements | [](https://travis-ci.org/sa7mon/S3Scanner) |
-| bugs | [](https://travis-ci.org/sa7mon/S3Scanner) |
-
-### Testing
-* All test are currently in `test_scanner.py`
-* Run tests with in 2.7 and 3.6 virtual environments.
-* This project uses **pytest-xdist** to run tests. Use `pytest -n NUM` where num is number of parallel processes.
-* Run individual tests like this: `pytest -q -s test_scanner.py::test_namehere`
-
-### Contributors
+
+* Scan AWS buckets listed in a file with 8 threads
+ ```shell
+ $ s3scanner --threads 8 scan --buckets-file ./bucket-names.txt
+ ```
+* Scan a bucket in Digital Ocean Spaces
+ ```shell
+ $ s3scanner --endpoint-url https://sfo2.digitaloceanspaces.com scan --bucket my-bucket
+ ```
+* Dump a single AWS bucket
+ ```shell
+ $ s3scanner dump --bucket my-bucket-to-dump
+ ```
+* Scan a single Dreamhost Objects bucket which uses the vhost address style and an invalid SSL cert
+ ```shell
+ $ s3scanner --endpoint-url https://objects.dreamhost.com --endpoint-address-style vhost --insecure scan --bucket my-bucket
+ ```
+
+## S3-compatible APIs
+
+`S3Scanner` can scan and dump buckets in S3-compatible APIs services other than AWS by using the
+`--endpoint-url` argument. Depending on the service, you may also need the `--endpoint-address-style`
+or `--insecure` arguments as well.
+
+Some services have different endpoints corresponding to different regions
+
+**Note:** `S3Scanner` currently only supports scanning for anonymous user permissions of non-AWS services
+
+| Service | Example Endpoint | Address Style | Insecure ? |
+|---------|------------------|:-------------:|:----------:|
+| DigitalOcean Spaces (SFO2 region) | https://sfo2.digitaloceanspaces.com | path | No |
+| Dreamhost | https://objects.dreamhost.com | vhost | Yes |
+| Linode Object Storage (eu-central-1 region) | https://eu-central-1.linodeobjects.com | vhost | No |
+| Scaleway Object Storage (nl-ams region) | https://s3.nl-ams.scw.cloud | path | No |
+| Wasabi Cloud Storage | http://s3.wasabisys.com/ | path | Yes |
+
+๐ Current status of non-AWS APIs can be found [in the project wiki](https://github.com/sa7mon/S3Scanner/wiki/S3-Compatible-APIs)
+
+## Interpreting Results
+
+This tool will attempt to get all available information about a bucket, but it's up to you to interpret the results.
+
+[Possible permissions](https://docs.aws.amazon.com/AmazonS3/latest/user-guide/set-bucket-permissions.html) for buckets:
+
+* Read - List and view all files
+* Write - Write files to bucket
+* Read ACP - Read all Access Control Policies attached to bucket
+* Write ACP - Write Access Control Policies to bucket
+* Full Control - All above permissions
+
+Any or all of these permissions can be set for the 2 main user groups:
+* Authenticated Users
+* Public Users (those without AWS credentials set)
+* Individual users/groups (out of scope of this tool)
+
+**What this means:** Just because a bucket doesn't allow reading/writing ACLs doesn't mean you can't read/write files in the bucket. Conversely, you may be able to list ACLs but not read/write to the bucket
+
+## Contributors
* [Ohelig](https://github.com/Ohelig)
* [vysecurity](https://github.com/vysecurity)
* [janmasarik](https://github.com/janmasarik)
* [alanyee](https://github.com/alanyee)
+* [klau5dev](https://github.com/klau5dev)
* [hipotermia](https://github.com/hipotermia)
## License
-License: [MIT](LICENSE.txt) https://opensource.org/licenses/MIT
+
+MIT
\ No newline at end of file
diff --git a/S3Scanner/S3Bucket.py b/S3Scanner/S3Bucket.py
new file mode 100644
index 0000000..76bc79a
--- /dev/null
+++ b/S3Scanner/S3Bucket.py
@@ -0,0 +1,162 @@
+import re
+from enum import Enum
+
+
+class Permission(Enum):
+ ALLOWED = 1,
+ DENIED = 0,
+ UNKNOWN = -1
+
+
+class BucketExists(Enum):
+ YES = 1,
+ NO = 0,
+ UNKNOWN = -1
+
+
+def bytes_to_human_readable(bytes_in, suffix='B'):
+ """
+ Convert number of bytes to a "human-readable" format. i.e. 1024 -> 1KB
+ Shamelessly copied from: https://stackoverflow.com/a/1094933/2307994
+
+ :param int bytes_in: Number of bytes to convert
+ :param str suffix: Suffix to convert to - i.e. B/KB/MB
+ :return: str human-readable string
+ """
+ for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']:
+ if abs(bytes_in) < 1024.0:
+ return "%3.1f%s%s" % (bytes_in, unit, suffix)
+ bytes_in /= 1024.0
+ return "%.1f%s%s" % (bytes_in, 'Yi', suffix)
+
+
+class S3BucketObject:
+ """
+ Represents an object stored in a bucket.
+ __eq__ and __hash__ are implemented to take full advantage of the set() deduplication
+ __lt__ is implemented to enable object sorting
+ """
+ def __init__(self, size, last_modified, key):
+ self.size = size
+ self.last_modified = last_modified
+ self.key = key
+
+ def __eq__(self, other):
+ return self.key == other.key
+
+ def __hash__(self):
+ return hash(self.key)
+
+ def __lt__(self, other):
+ return self.key < other.key
+
+ def __repr__(self):
+ return "Key: %s, Size: %s, LastModified: %s" % (self.key, str(self.size), str(self.last_modified))
+
+ def get_human_readable_size(self):
+ return bytes_to_human_readable(self.size)
+
+
+class S3Bucket:
+ """
+ Represents a bucket which holds objects
+ """
+ exists = BucketExists.UNKNOWN
+ objects = set() # A collection of S3BucketObject
+ bucketSize = 0
+ objects_enumerated = False
+ foundACL = None
+
+ def __init__(self, name):
+ """
+ Constructor method
+
+ :param str name: Name of bucket
+ :raises ValueError: If bucket name is invalid according to `_check_bucket_name()`
+ """
+ check = self._check_bucket_name(name)
+ if not check['valid']:
+ raise ValueError("Invalid bucket name")
+
+ self.name = check['name']
+
+ self.AuthUsersRead = Permission.UNKNOWN
+ self.AuthUsersWrite = Permission.UNKNOWN
+ self.AuthUsersReadACP = Permission.UNKNOWN
+ self.AuthUsersWriteACP = Permission.UNKNOWN
+ self.AuthUsersFullControl = Permission.UNKNOWN
+
+ self.AllUsersRead = Permission.UNKNOWN
+ self.AllUsersWrite = Permission.UNKNOWN
+ self.AllUsersReadACP = Permission.UNKNOWN
+ self.AllUsersWriteACP = Permission.UNKNOWN
+ self.AllUsersFullControl = Permission.UNKNOWN
+
+ def _check_bucket_name(self, name):
+ """
+ Checks to make sure bucket names input are valid according to S3 naming conventions
+
+ :param str name: Name of bucket to check
+ :return: dict: ['valid'] - bool: whether or not the name is valid, ['name'] - str: extracted bucket name
+ """
+ bucket_name = ""
+ # Check if bucket name is valid and determine the format
+ if ".amazonaws.com" in name: # We were given a full s3 url
+ bucket_name = name[:name.rfind(".s3")]
+ elif ":" in name: # We were given a bucket in 'bucket:region' format
+ bucket_name = name.split(":")[0]
+ else: # We were given a regular bucket name
+ bucket_name = name
+
+ # Bucket names can be 3-63 (inclusively) characters long.
+ # Bucket names may only contain lowercase letters, numbers, periods, and hyphens
+ pattern = r'(?=^.{3,63}$)(?!^(\d+\.)+\d+$)(^(([a-z0-9]|[a-z0-9][a-z0-9\-]*[a-z0-9])\.)*([a-z0-9]|[a-z0-9][a-z0-9\-]*[a-z0-9])$)'
+ return {'valid': bool(re.match(pattern, bucket_name)), 'name': bucket_name}
+
+ def add_object(self, obj):
+ """
+ Adds object to bucket. Updates the `objects` and `bucketSize` properties of the bucket
+
+ :param S3BucketObject obj: Object to add to bucket
+ :return: None
+ """
+ self.objects.add(obj)
+ self.bucketSize += obj.size
+
+ def get_human_readable_size(self):
+ return bytes_to_human_readable(self.bucketSize)
+
+ def get_human_readable_permissions(self):
+ """
+ Returns a human-readable string of allowed permissions for this bucket
+ i.e. "AuthUsers: [Read | WriteACP], AllUsers: [FullControl]"
+
+ :return: str: Human-readable permissions
+ """
+ # Add AuthUsers permissions
+ authUsersPermissions = []
+ if self.AuthUsersFullControl == Permission.ALLOWED:
+ authUsersPermissions.append("FullControl")
+ else:
+ if self.AuthUsersRead == Permission.ALLOWED:
+ authUsersPermissions.append("Read")
+ if self.AuthUsersWrite == Permission.ALLOWED:
+ authUsersPermissions.append("Write")
+ if self.AuthUsersReadACP == Permission.ALLOWED:
+ authUsersPermissions.append("ReadACP")
+ if self.AuthUsersWriteACP == Permission.ALLOWED:
+ authUsersPermissions.append("WriteACP")
+ # Add AllUsers permissions
+ allUsersPermissions = []
+ if self.AllUsersFullControl == Permission.ALLOWED:
+ allUsersPermissions.append("FullControl")
+ else:
+ if self.AllUsersRead == Permission.ALLOWED:
+ allUsersPermissions.append("Read")
+ if self.AllUsersWrite == Permission.ALLOWED:
+ allUsersPermissions.append("Write")
+ if self.AllUsersReadACP == Permission.ALLOWED:
+ allUsersPermissions.append("ReadACP")
+ if self.AllUsersWriteACP == Permission.ALLOWED:
+ allUsersPermissions.append("WriteACP")
+ return f"AuthUsers: [{', '.join(authUsersPermissions)}], AllUsers: [{', '.join(allUsersPermissions)}]"
diff --git a/S3Scanner/S3Service.py b/S3Scanner/S3Service.py
new file mode 100644
index 0000000..151f24d
--- /dev/null
+++ b/S3Scanner/S3Service.py
@@ -0,0 +1,461 @@
+"""
+ This will be a service that the client program will instantiate to then call methods
+ passing buckets
+"""
+from boto3 import client # TODO: Limit import to just boto3.client, probably
+from S3Scanner.S3Bucket import S3Bucket, BucketExists, Permission, S3BucketObject
+from botocore.exceptions import ClientError
+import botocore.session
+from botocore import UNSIGNED
+from botocore.client import Config
+import datetime
+from S3Scanner.exceptions import AccessDeniedException, InvalidEndpointException, BucketMightNotExistException
+from os.path import normpath
+import pathlib
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from functools import partial
+from urllib3 import disable_warnings
+
+ALL_USERS_URI = 'uri=http://acs.amazonaws.com/groups/global/AllUsers'
+AUTH_USERS_URI = 'uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers'
+
+
+class S3Service:
+ def __init__(self, forceNoCreds=False, endpoint_url='https://s3.amazonaws.com', verify_ssl=True,
+ endpoint_address_style='path'):
+ """
+ Service constructor
+
+ :param forceNoCreds: (Boolean) Force the service to not use credentials, even if the user has creds configured
+ :param endpoint_url: (String) URL of S3 endpoint to use. Must include http(s):// scheme
+ :param verify_ssl: (Boolean) Whether of not to verify ssl. Set to false if endpoint is http
+ :param endpoint_address_style: (String) Addressing style of the endpoint. Must be 'path' or 'vhost'
+ :returns None
+ """
+ self.endpoint_url = endpoint_url
+ self.endpoint_address_style = 'path' if endpoint_address_style == 'path' else 'virtual'
+ use_ssl = True if self.endpoint_url.startswith('http://') else False
+
+ if not verify_ssl:
+ disable_warnings()
+
+ # DEBUG
+ # boto3.set_stream_logger(name='botocore')
+
+ # Validate the endpoint if it's not the default of AWS
+ if self.endpoint_url != 'https://s3.amazonaws.com':
+ if not self.validate_endpoint_url(use_ssl, verify_ssl, endpoint_address_style):
+ raise InvalidEndpointException(message=f"Endpoint '{self.endpoint_url}' does not appear to be S3-compliant")
+
+ # Check for AWS credentials
+ session = botocore.session.get_session()
+ if forceNoCreds or session.get_credentials() is None or session.get_credentials().access_key is None:
+ self.aws_creds_configured = False
+ self.s3_client = client('s3',
+ config=Config(signature_version=UNSIGNED, s3={'addressing_style': self.endpoint_address_style}, connect_timeout=3,
+ retries={'max_attempts': 2}),
+ endpoint_url=self.endpoint_url, use_ssl=use_ssl, verify=verify_ssl)
+ else:
+ self.aws_creds_configured = True
+ self.s3_client = client('s3', config=Config(s3={'addressing_style': self.endpoint_address_style}, connect_timeout=3,
+ retries={'max_attempts': 2}),
+ endpoint_url=self.endpoint_url, use_ssl=use_ssl, verify=verify_ssl)
+
+ del session # No longer needed
+
+ def check_bucket_exists(self, bucket):
+ """
+ Checks if a bucket exists. Sets `exists` property of `bucket`
+
+ :param S3Bucket bucket: Bucket to check
+ :raises ValueError: If `bucket` is not an s3Bucket object
+ :return: None
+ """
+ if not isinstance(bucket, S3Bucket):
+ raise ValueError("Passed object was not type S3Bucket")
+
+ bucket_exists = True
+
+ try:
+ self.s3_client.head_bucket(Bucket=bucket.name)
+ except ClientError as e:
+ if e.response['Error']['Code'] == '404':
+ bucket_exists = False
+
+ bucket.exists = BucketExists.YES if bucket_exists else BucketExists.NO
+
+ def check_perm_read_acl(self, bucket):
+ """
+ Check for the READACP permission on `bucket` by trying to get the bucket ACL
+
+ :param S3Bucket bucket: Bucket to check permission of
+ :raises BucketMightNotExistException: If `bucket` existence hasn't been checked
+ :raises ClientError: If we encounter an unexpected ClientError from boto client
+ :return: None
+ """
+
+ if bucket.exists != BucketExists.YES:
+ raise BucketMightNotExistException()
+
+ try:
+ bucket.foundACL = self.s3_client.get_bucket_acl(Bucket=bucket.name)
+ self.parse_found_acl(bucket) # If we can read ACLs, we know the rest of the permissions
+ except ClientError as e:
+ if e.response['Error']['Code'] == "AccessDenied" or e.response['Error']['Code'] == "AllAccessDisabled":
+ if self.aws_creds_configured:
+ bucket.AuthUsersReadACP = Permission.DENIED
+ else:
+ bucket.AllUsersReadACP = Permission.DENIED
+ else:
+ raise e
+
+ def check_perm_read(self, bucket):
+ """
+ Checks for the READ permission on the bucket by attempting to list the objects.
+ Sets the `AllUsersRead` and/or `AuthUsersRead` property of `bucket`.
+
+ :param S3Bucket bucket: Bucket to check permission of
+ :raises BucketMightNotExistException: If `bucket` existence hasn't been checked
+ :raises ClientError: If we encounter an unexpected ClientError from boto client
+ :return: None
+ """
+ if bucket.exists != BucketExists.YES:
+ raise BucketMightNotExistException()
+
+ list_bucket_perm_allowed = True
+ try:
+ self.s3_client.list_objects_v2(Bucket=bucket.name, MaxKeys=0) # TODO: Compare this to doing a HeadBucket
+ except ClientError as e:
+ if e.response['Error']['Code'] == "AccessDenied" or e.response['Error']['Code'] == "AllAccessDisabled":
+ list_bucket_perm_allowed = False
+ else:
+ print(f"ERROR: Error while checking bucket {bucket.name}")
+ raise e
+ if self.aws_creds_configured:
+ # Don't mark AuthUsersRead as Allowed if it's only implicitly allowed due to AllUsersRead being allowed
+ # We only want to make AuthUsersRead as Allowed if that permission is explicitly set for AuthUsers
+ if bucket.AllUsersRead != Permission.ALLOWED:
+ bucket.AuthUsersRead = Permission.ALLOWED if list_bucket_perm_allowed else Permission.DENIED
+ else:
+ bucket.AllUsersRead = Permission.ALLOWED if list_bucket_perm_allowed else Permission.DENIED
+
+ def check_perm_write(self, bucket):
+ """
+ Check for WRITE permission by trying to upload an empty file to the bucket.
+ File is named the current timestamp to ensure we're not overwriting an existing file in the bucket.
+
+ NOTE: If writing to bucket succeeds using an AuthUser, only mark AuthUsersWrite as Allowed if AllUsers are
+ Denied. Writing can succeed if AuthUsers are implicitly allowed due to AllUsers being allowed, but we only want
+ to mark AuthUsers as Allowed if they are explicitly granted. If AllUsersWrite is Allowed and the write is
+ successful by an AuthUser, we have no way of knowing if AuthUsers were granted permission explicitly
+
+ :param S3Bucket bucket: Bucket to check permission of
+ :raises BucketMightNotExistException: If `bucket` existence hasn't been checked
+ :raises ClientError: If we encounter an unexpected ClientError from boto client
+ :return: None
+ """
+ if bucket.exists != BucketExists.YES:
+ raise BucketMightNotExistException()
+
+ timestamp_file = str(datetime.datetime.now().timestamp()) + '.txt'
+
+ try:
+ # Try to create a new empty file with a key of the timestamp
+ self.s3_client.put_object(Bucket=bucket.name, Key=timestamp_file, Body=b'')
+
+ if self.aws_creds_configured:
+ if bucket.AllUsersWrite != Permission.ALLOWED: # If AllUsers have Write permission, don't mark AuthUsers as Allowed
+ bucket.AuthUsersWrite = Permission.ALLOWED
+ else:
+ bucket.AuthUsersWrite = Permission.UNKNOWN
+ else:
+ bucket.AllUsersWrite = Permission.ALLOWED
+
+ # Delete the temporary file
+ self.s3_client.delete_object(Bucket=bucket.name, Key=timestamp_file)
+ except ClientError as e:
+ if e.response['Error']['Code'] == "AccessDenied" or e.response['Error']['Code'] == "AllAccessDisabled":
+ if self.aws_creds_configured:
+ bucket.AuthUsersWrite = Permission.DENIED
+ else:
+ bucket.AllUsersWrite = Permission.DENIED
+ else:
+ raise e
+
+ def check_perm_write_acl(self, bucket):
+ """
+ Checks for WRITE_ACP permission by attempting to set an ACL on the bucket.
+ WARNING: Potentially destructive - make sure to run this check last as it will include all discovered
+ permissions in the ACL it tries to set, thus ensuring minimal disruption for the bucket owner.
+
+ :param S3Bucket bucket: Bucket to check permission of
+ :raises BucketMightNotExistException: If `bucket` existence hasn't been checked
+ :raises ClientError: If we encounter an unexpected ClientError from boto client
+ :return: None
+ """
+ if bucket.exists != BucketExists.YES:
+ raise BucketMightNotExistException()
+
+ # TODO: See if there's a way to simplify this section
+ readURIs = []
+ writeURIs = []
+ readAcpURIs = []
+ writeAcpURIs = []
+ fullControlURIs = []
+
+ if bucket.AuthUsersRead == Permission.ALLOWED:
+ readURIs.append(AUTH_USERS_URI)
+ if bucket.AuthUsersWrite == Permission.ALLOWED:
+ writeURIs.append(AUTH_USERS_URI)
+ if bucket.AuthUsersReadACP == Permission.ALLOWED:
+ readAcpURIs.append(AUTH_USERS_URI)
+ if bucket.AuthUsersWriteACP == Permission.ALLOWED:
+ writeAcpURIs.append(AUTH_USERS_URI)
+ if bucket.AuthUsersFullControl == Permission.ALLOWED:
+ fullControlURIs.append(AUTH_USERS_URI)
+
+ if bucket.AllUsersRead == Permission.ALLOWED:
+ readURIs.append(ALL_USERS_URI)
+ if bucket.AllUsersWrite == Permission.ALLOWED:
+ writeURIs.append(ALL_USERS_URI)
+ if bucket.AllUsersReadACP == Permission.ALLOWED:
+ readAcpURIs.append(ALL_USERS_URI)
+ if bucket.AllUsersWriteACP == Permission.ALLOWED:
+ writeAcpURIs.append(ALL_USERS_URI)
+ if bucket.AllUsersFullControl == Permission.ALLOWED:
+ fullControlURIs.append(ALL_USERS_URI)
+
+ if self.aws_creds_configured: # Otherwise AWS will return "Request was missing a required header"
+ writeAcpURIs.append(AUTH_USERS_URI)
+ else:
+ writeAcpURIs.append(ALL_USERS_URI)
+ args = {'Bucket': bucket.name}
+ if len(readURIs) > 0:
+ args['GrantRead'] = ','.join(readURIs)
+ if len(writeURIs) > 0:
+ args['GrantWrite'] = ','.join(writeURIs)
+ if len(readAcpURIs) > 0:
+ args['GrantReadACP'] = ','.join(readAcpURIs)
+ if len(writeAcpURIs) > 0:
+ args['GrantWriteACP'] = ','.join(writeAcpURIs)
+ if len(fullControlURIs) > 0:
+ args['GrantFullControl'] = ','.join(fullControlURIs)
+ try:
+ self.s3_client.put_bucket_acl(**args)
+ if self.aws_creds_configured:
+ # Don't mark AuthUsersWriteACP as Allowed if it's due to implicit permission via AllUsersWriteACP
+ # Only mark it as allowed if the AuthUsers group is explicitly allowed
+ if bucket.AllUsersWriteACP != Permission.ALLOWED:
+ bucket.AuthUsersWriteACP = Permission.ALLOWED
+ else:
+ bucket.AuthUsersWriteACP = Permission.UNKNOWN
+ else:
+ bucket.AllUsersWriteACP = Permission.ALLOWED
+ except ClientError as e:
+ if e.response['Error']['Code'] == "AccessDenied" or e.response['Error']['Code'] == "AllAccessDisabled":
+ if self.aws_creds_configured:
+ bucket.AuthUsersWriteACP = Permission.DENIED
+ else:
+ bucket.AllUsersWriteACP = Permission.DENIED
+ else:
+ raise e
+
+ def dump_bucket_multithread(self, bucket, dest_directory, verbose=False, threads=4):
+ """
+ Takes a bucket and downloads all the objects to a local folder.
+ If the object exists locally and is the same size as the remote object, the object is skipped.
+ If the object exists locally and is a different size then the remote object, the local object is overwritten.
+
+ :param S3Bucket bucket: Bucket whose contents we want to dump
+ :param str dest_directory: Folder to save the objects to. Must include trailing slash
+ :param bool verbose: Output verbose messages to the user
+ :param int threads: Number of threads to use while dumping
+ :return: None
+ """
+ # TODO: Let the user choose whether or not to overwrite local files if different
+
+ print(f"{bucket.name} | Dumping contents using 4 threads...")
+ func = partial(self.download_file, dest_directory, bucket, verbose)
+
+ with ThreadPoolExecutor(max_workers=threads) as executor:
+ futures = {
+ executor.submit(func, obj): obj for obj in bucket.objects
+ }
+
+ for future in as_completed(futures):
+ if future.exception():
+ print(f"{bucket.name} | Download failed: {futures[future]}")
+
+ print(f"{bucket.name} | Dumping completed")
+
+ def download_file(self, dest_directory, bucket, verbose, obj):
+ """
+ Download `obj` from `bucket` into `dest_directory`
+
+ :param str dest_directory: Directory to store the object into
+ :param S3Bucket bucket: Bucket to download the object from
+ :param bool verbose: Output verbose messages to the user
+ :param S3BucketObject obj: Object to downlaod
+ :return: None
+ """
+ dest_file_path = pathlib.Path(normpath(dest_directory + obj.key))
+ if dest_file_path.exists():
+ if dest_file_path.stat().st_size == obj.size:
+ if verbose:
+ print(f"{bucket.name} | Skipping {obj.key} - already downloaded")
+ return
+ else:
+ if verbose:
+ print(f"{bucket.name} | Re-downloading {obj.key} - local size differs from remote")
+ else:
+ if verbose:
+ print(f"{bucket.name} | Downloading {obj.key}")
+ dest_file_path.parent.mkdir(parents=True, exist_ok=True) # Equivalent to `mkdir -p`
+ self.s3_client.download_file(bucket.name, obj.key, str(dest_file_path))
+
+ def enumerate_bucket_objects(self, bucket):
+ """
+ Enumerate all the objects in a bucket. Sets the `BucketSize`, `objects`, and `objects_enumerated` properties
+ of `bucket`.
+
+ :param S3Bucket bucket: Bucket to enumerate objects of
+ :raises Exception: If the bucket doesn't exist
+ :raises AccessDeniedException: If we are denied access to the bucket objects
+ :raises ClientError: If we encounter an unexpected ClientError from boto client
+ :return: None
+ """
+ if bucket.exists == BucketExists.UNKNOWN:
+ self.check_bucket_exists(bucket)
+ if bucket.exists == BucketExists.NO:
+ raise Exception("Bucket doesn't exist")
+
+ try:
+ for page in self.s3_client.get_paginator("list_objects_v2").paginate(Bucket=bucket.name):
+ if 'Contents' not in page: # No items in this bucket
+ bucket.objects_enumerated = True
+ return
+ for item in page['Contents']:
+ obj = S3BucketObject(key=item['Key'], last_modified=item['LastModified'], size=item['Size'])
+ bucket.add_object(obj)
+ except ClientError as e:
+ if e.response['Error']['Code'] == "AccessDenied" or e.response['Error']['Code'] == "AllAccessDisabled":
+ raise AccessDeniedException("AccessDenied while enumerating bucket objects")
+ bucket.objects_enumerated = True
+
+ def parse_found_acl(self, bucket):
+ """
+ Translate ACL grants into permission properties. If we were able to read the ACLs, we should be able to skip
+ manually checking most permissions
+
+ :param S3Bucket bucket: Bucket whose ACLs we want to parse
+ :return: None
+ """
+ if bucket.foundACL is None:
+ return
+
+ if 'Grants' in bucket.foundACL:
+ for grant in bucket.foundACL['Grants']:
+ if grant['Grantee']['Type'] == 'Group':
+ if 'URI' in grant['Grantee'] and grant['Grantee']['URI'] == 'http://acs.amazonaws.com/groups/global/AuthenticatedUsers':
+ # Permissions have been given to the AuthUsers group
+ if grant['Permission'] == 'FULL_CONTROL':
+ bucket.AuthUsersRead = Permission.ALLOWED
+ bucket.AuthUsersWrite = Permission.ALLOWED
+ bucket.AuthUsersReadACP = Permission.ALLOWED
+ bucket.AuthUsersWriteACP = Permission.ALLOWED
+ bucket.AuthUsersFullControl = Permission.ALLOWED
+ elif grant['Permission'] == 'READ':
+ bucket.AuthUsersRead = Permission.ALLOWED
+ elif grant['Permission'] == 'READ_ACP':
+ bucket.AuthUsersReadACP = Permission.ALLOWED
+ elif grant['Permission'] == 'WRITE':
+ bucket.AuthUsersWrite = Permission.ALLOWED
+ elif grant['Permission'] == 'WRITE_ACP':
+ bucket.AuthUsersWriteACP = Permission.ALLOWED
+
+ elif 'URI' in grant['Grantee'] and grant['Grantee']['URI'] == 'http://acs.amazonaws.com/groups/global/AllUsers':
+ # Permissions have been given to the AllUsers group
+ if grant['Permission'] == 'FULL_CONTROL':
+ bucket.AllUsersRead = Permission.ALLOWED
+ bucket.AllUsersWrite = Permission.ALLOWED
+ bucket.AllUsersReadACP = Permission.ALLOWED
+ bucket.AllUsersWriteACP = Permission.ALLOWED
+ bucket.AllUsersFullControl = Permission.ALLOWED
+ elif grant['Permission'] == 'READ':
+ bucket.AllUsersRead = Permission.ALLOWED
+ elif grant['Permission'] == 'READ_ACP':
+ bucket.AllUsersReadACP = Permission.ALLOWED
+ elif grant['Permission'] == 'WRITE':
+ bucket.AllUsersWrite = Permission.ALLOWED
+ elif grant['Permission'] == 'WRITE_ACP':
+ bucket.AllUsersWriteACP = Permission.ALLOWED
+
+ # All permissions not explicitly granted in the ACL are denied
+ # TODO: Simplify this
+ if bucket.AuthUsersRead == Permission.UNKNOWN:
+ bucket.AuthUsersRead = Permission.DENIED
+
+ if bucket.AuthUsersWrite == Permission.UNKNOWN:
+ bucket.AuthUsersWrite = Permission.DENIED
+
+ if bucket.AuthUsersReadACP == Permission.UNKNOWN:
+ bucket.AuthUsersReadACP = Permission.DENIED
+
+ if bucket.AuthUsersWriteACP == Permission.UNKNOWN:
+ bucket.AuthUsersWriteACP = Permission.DENIED
+
+ if bucket.AuthUsersFullControl == Permission.UNKNOWN:
+ bucket.AuthUsersFullControl = Permission.DENIED
+
+ if bucket.AllUsersRead == Permission.UNKNOWN:
+ bucket.AllUsersRead = Permission.DENIED
+
+ if bucket.AllUsersWrite == Permission.UNKNOWN:
+ bucket.AllUsersWrite = Permission.DENIED
+
+ if bucket.AllUsersReadACP == Permission.UNKNOWN:
+ bucket.AllUsersReadACP = Permission.DENIED
+
+ if bucket.AllUsersWriteACP == Permission.UNKNOWN:
+ bucket.AllUsersWriteACP = Permission.DENIED
+
+ if bucket.AllUsersFullControl == Permission.UNKNOWN:
+ bucket.AllUsersFullControl = Permission.DENIED
+
+ def validate_endpoint_url(self, use_ssl=True, verify_ssl=True, endpoint_address_style='path'):
+ """
+ Verify the user-supplied endpoint URL is S3-compliant by trying to list a maximum of 0 keys from a bucket which
+ is extremely unlikely to exist.
+
+ Note: Most S3-compliant services will return an error code of "NoSuchBucket". Some services which require auth
+ for most operations (like Minio) will return an error code of "AccessDenied" instead
+
+ :param bool use_ssl: Whether or not the endpoint serves HTTP over SSL
+ :param bool verify_ssl: Whether or not to verify the SSL connection.
+ :param str endpoint_address_style: Addressing style of endpoint. Must be either 'path' or 'vhost'
+ :return: bool: Whether or not the server responded in an S3-compliant way
+ """
+
+ # We always want to verify the endpoint using no creds
+ # so if the s3_client has creds configured, make a new anonymous client
+
+ addressing_style = 'virtual' if endpoint_address_style == 'vhost' else 'path'
+
+ validation_client = client('s3', config=Config(signature_version=UNSIGNED,
+ s3={'addressing_style': addressing_style}, connect_timeout=3,
+ retries={'max_attempts': 0}), endpoint_url=self.endpoint_url, use_ssl=use_ssl,
+ verify=verify_ssl)
+
+ non_existent_bucket = 's3scanner-' + str(datetime.datetime.now())[0:10]
+ 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']:
+ return True
+ return False
+ except botocore.exceptions.ConnectTimeoutError:
+ return False
+
+ # If we get here, the bucket either existed (unlikely) or the server returned a 200 for some reason
+ return False
diff --git a/S3Scanner/__init__.py b/S3Scanner/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/S3Scanner/__main__.py b/S3Scanner/__main__.py
new file mode 100644
index 0000000..afab1aa
--- /dev/null
+++ b/S3Scanner/__main__.py
@@ -0,0 +1,245 @@
+#########
+#
+# S3scanner - Audit unsecured S3 buckets
+#
+# Author: Dan Salmon (twitter.com/bltjetpack, github.com/sa7mon)
+# Created: 6/19/17
+# License: MIT
+#
+#########
+
+import argparse
+from os import path
+from sys import exit
+from .S3Bucket import S3Bucket, BucketExists, Permission
+from .S3Service import S3Service
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from .exceptions import InvalidEndpointException
+
+CURRENT_VERSION = '2.0.0'
+AWS_ENDPOINT = 'https://s3.amazonaws.com'
+
+
+# We want to use both formatter classes, so a custom class it is
+class CustomFormatter(argparse.RawTextHelpFormatter, argparse.RawDescriptionHelpFormatter):
+ pass
+
+
+def load_bucket_names_from_file(file_name):
+ """
+ Load in bucket names from a text file
+
+ :param str file_name: Path to text file
+ :return: set: All lines of text file
+ """
+ buckets = set()
+ if path.isfile(file_name):
+ with open(file_name, 'r') as f:
+ for line in f:
+ line = line.rstrip() # Remove any extra whitespace
+ buckets.add(line)
+ return buckets
+ else:
+ print("Error: '%s' is not a file" % file_name)
+ exit(1)
+
+
+def scan_single_bucket(s3service, anons3service, do_dangerous, bucket_name):
+ """
+ Scans a single bucket for permission issues. Exists on its own so we can do multi-threading
+
+ :param S3Service s3service: S3Service with credentials to use for scanning
+ :param S3Service anonS3Service: S3Service without credentials to use for scanning
+ :param bool do_dangerous: Whether or not to do dangerous checks
+ :param str bucket_name: Name of bucket to check
+ :return: None
+ """
+ try:
+ b = S3Bucket(bucket_name)
+ except ValueError as ve:
+ if str(ve) == "Invalid bucket name":
+ print(f"{bucket_name} | bucket_invalid_name")
+ return
+ else:
+ print(f"{bucket_name} | {str(ve)}")
+ return
+
+ # Check if bucket exists first
+ # Use credentials if configured and if we're hitting AWS. Otherwise, check anonymously
+ if s3service.endpoint_url == AWS_ENDPOINT:
+ s3service.check_bucket_exists(b)
+ else:
+ anons3service.check_bucket_exists(b)
+
+ if b.exists == BucketExists.NO:
+ print(f"{b.name} | bucket_not_exist")
+ return
+ checkAllUsersPerms = True
+ checkAuthUsersPerms = True
+
+ # 1. Check for ReadACP
+ anons3service.check_perm_read_acl(b) # Check for AllUsers
+ if s3service.aws_creds_configured and s3service.endpoint_url == AWS_ENDPOINT:
+ s3service.check_perm_read_acl(b) # Check for AuthUsers
+
+ # If FullControl is allowed for either AllUsers or AnonUsers, skip the remainder of those tests
+ if b.AuthUsersFullControl == Permission.ALLOWED:
+ checkAuthUsersPerms = False
+ if b.AllUsersFullControl == Permission.ALLOWED:
+ checkAllUsersPerms = False
+
+ # 2. Check for Read
+ if checkAllUsersPerms:
+ anons3service.check_perm_read(b)
+ if s3service.aws_creds_configured and checkAuthUsersPerms and s3service.endpoint_url == AWS_ENDPOINT:
+ s3service.check_perm_read(b)
+
+ # Do dangerous/destructive checks
+ if do_dangerous:
+ # 3. Check for Write
+ if checkAllUsersPerms:
+ anons3service.check_perm_write(b)
+ if s3service.aws_creds_configured and checkAuthUsersPerms:
+ s3service.check_perm_write(b)
+
+ # 4. Check for WriteACP
+ if checkAllUsersPerms:
+ anons3service.check_perm_write_acl(b)
+ if s3service.aws_creds_configured and checkAuthUsersPerms:
+ s3service.check_perm_write_acl(b)
+
+ print(f"{b.name} | bucket_exists | {b.get_human_readable_permissions()}")
+
+
+def main():
+ # Instantiate the parser
+ parser = argparse.ArgumentParser(description='s3scanner: Audit unsecured S3 buckets\n'
+ ' by Dan Salmon - github.com/sa7mon, @bltjetpack\n',
+ prog='s3scanner', allow_abbrev=False, formatter_class=CustomFormatter)
+ # Declare arguments
+ parser.add_argument('--version', action='version', version=CURRENT_VERSION,
+ help='Display the current version of this tool')
+ parser.add_argument('--threads', '-t', type=int, default=4, dest='threads', help='Number of threads to use. Default: 4',
+ metavar='n')
+ parser.add_argument('--endpoint-url', '-u', dest='endpoint_url',
+ help='URL of S3-compliant API. Default: https://s3.amazonaws.com',
+ default='https://s3.amazonaws.com')
+ parser.add_argument('--endpoint-address-style', '-s', dest='endpoint_address_style', choices=['path', 'vhost'],
+ default='path', help='Address style to use for the endpoint. Default: path')
+ parser.add_argument('--insecure', '-i', dest='verify_ssl', action='store_false', help='Do not verify SSL')
+ subparsers = parser.add_subparsers(title='mode', dest='mode', help='(Must choose one)')
+
+ # Scan mode
+ parser_scan = subparsers.add_parser('scan', help='Scan bucket permissions')
+ parser_scan.add_argument('--dangerous', action='store_true',
+ help='Include Write and WriteACP permissions checks (potentially destructive)')
+ parser_group = parser_scan.add_mutually_exclusive_group(required=True)
+ parser_group.add_argument('--buckets-file', '-f', dest='buckets_file',
+ help='Name of text file containing bucket names to check', metavar='file')
+ parser_group.add_argument('--bucket', '-b', dest='bucket', help='Name of bucket to check', metavar='bucket')
+ # TODO: Get help output to not repeat metavar names - i.e. `--bucket FILE, -f FILE`
+ # https://stackoverflow.com/a/9643162/2307994
+
+ # Dump mode
+ parser_dump = subparsers.add_parser('dump', help='Dump the contents of buckets')
+ parser_dump.add_argument('--dump-dir', '-d', required=True, dest='dump_dir', help='Directory to dump bucket into')
+ dump_parser_group = parser_dump.add_mutually_exclusive_group(required=True)
+ dump_parser_group.add_argument('--buckets-file', '-f', dest='dump_buckets_file',
+ help='Name of text file containing bucket names to check', metavar='file')
+ dump_parser_group.add_argument('--bucket', '-b', dest='dump_bucket', help='Name of bucket to check', metavar='bucket')
+ parser_dump.add_argument('--verbose', '-v', dest='dump_verbose', action='store_true',
+ help='Enable verbose output while dumping bucket(s)')
+
+ # Parse the args
+ args = parser.parse_args()
+
+ if 'http://' not in args.endpoint_url and 'https://' not in args.endpoint_url:
+ print("Error: endpoint_url must start with http:// or https:// scheme")
+ exit(1)
+
+ s3service = None
+ anons3service = None
+ try:
+ s3service = S3Service(endpoint_url=args.endpoint_url, verify_ssl=args.verify_ssl, endpoint_address_style=args.endpoint_address_style)
+ anons3service = S3Service(forceNoCreds=True, endpoint_url=args.endpoint_url, verify_ssl=args.verify_ssl, endpoint_address_style=args.endpoint_address_style)
+ except InvalidEndpointException as e:
+ print(f"Error: {e.message}")
+ exit(1)
+
+ if s3service.aws_creds_configured is False:
+ print("Warning: AWS credentials not configured - functionality will be limited. Run:"
+ " `aws configure` to fix this.\n")
+
+ bucketsIn = set()
+
+ if args.mode == 'scan':
+ if args.buckets_file is not None:
+ bucketsIn = load_bucket_names_from_file(args.buckets_file)
+ elif args.bucket is not None:
+ bucketsIn.add(args.bucket)
+
+ if args.dangerous:
+ print("INFO: Including dangerous checks. WARNING: This may change bucket ACL destructively")
+
+ with ThreadPoolExecutor(max_workers=args.threads) as executor:
+ futures = {
+ executor.submit(scan_single_bucket, s3service, anons3service, args.dangerous, bucketName): bucketName for bucketName in bucketsIn
+ }
+ for future in as_completed(futures):
+ if future.exception():
+ print(f"Bucket scan raised exception: {futures[future]} - {future.exception()}")
+
+ elif args.mode == 'dump':
+ if args.dump_dir is None or not path.isdir(args.dump_dir):
+ print("Error: Given --dump-dir does not exist or is not a directory")
+ exit(1)
+ if args.dump_buckets_file is not None:
+ bucketsIn = load_bucket_names_from_file(args.dump_buckets_file)
+ elif args.dump_bucket is not None:
+ bucketsIn.add(args.dump_bucket)
+
+ for bucketName in bucketsIn:
+ try:
+ b = S3Bucket(bucketName)
+ except ValueError as ve:
+ if str(ve) == "Invalid bucket name":
+ print(f"{bucketName} | bucket_name_invalid")
+ continue
+ else:
+ print(f"{bucketName} | {str(ve)}")
+ continue
+
+ # Check if bucket exists first
+ s3service.check_bucket_exists(b)
+
+ if b.exists == BucketExists.NO:
+ print(f"{b.name} | bucket_not_exist")
+ continue
+
+ s3service.check_perm_read(b)
+
+ if b.AuthUsersRead != Permission.ALLOWED:
+ anons3service.check_perm_read(b)
+ if b.AllUsersRead != Permission.ALLOWED:
+ print(f"{b.name} | Error: no read permissions")
+ else:
+ # Dump bucket without creds
+ print(f"{b.name} | Enumerating bucket objects...")
+ anons3service.enumerate_bucket_objects(b)
+ print(f"{b.name} | Total Objects: {str(len(b.objects))}, Total Size: {b.get_human_readable_size()}")
+ anons3service.dump_bucket_multithread(bucket=b, dest_directory=args.dump_dir,
+ verbose=args.dump_verbose, threads=args.threads)
+ else:
+ # Dump bucket with creds
+ print(f"{b.name} | Enumerating bucket objects...")
+ s3service.enumerate_bucket_objects(b)
+ print(f"{b.name} | Total Objects: {str(len(b.objects))}, Total Size: {b.get_human_readable_size()}")
+ s3service.dump_bucket_multithread(bucket=b, dest_directory=args.dump_dir,
+ verbose=args.dump_verbose, threads=args.threads)
+ else:
+ print("Invalid mode")
+ parser.print_help()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/S3Scanner/exceptions.py b/S3Scanner/exceptions.py
new file mode 100644
index 0000000..feb7227
--- /dev/null
+++ b/S3Scanner/exceptions.py
@@ -0,0 +1,18 @@
+class AccessDeniedException(Exception):
+ def __init__(self, message):
+ pass
+ # Call the base class constructor
+ # super().__init__(message, None)
+
+ # Now custom code
+ # self.errors = errors
+
+
+class InvalidEndpointException(Exception):
+ def __init__(self, message):
+ self.message = message
+
+
+class BucketMightNotExistException(Exception):
+ def __init__(self):
+ pass
diff --git a/buckets/.gitignore b/buckets/.gitignore
deleted file mode 100644
index 5e7d273..0000000
--- a/buckets/.gitignore
+++ /dev/null
@@ -1,4 +0,0 @@
-# Ignore everything in this directory
-*
-# Except this file
-!.gitignore
diff --git a/conftest.py b/conftest.py
new file mode 100644
index 0000000..33b565c
--- /dev/null
+++ b/conftest.py
@@ -0,0 +1,15 @@
+####
+# Pytest Configuration
+####
+
+
+def pytest_addoption(parser):
+ parser.addoption("--do-dangerous", action="store_true",
+ help="Run all tests, including ones where buckets are created.")
+
+
+def pytest_generate_tests(metafunc):
+ if "do_dangerous_test" in metafunc.fixturenames:
+ do_dangerous_test = True if metafunc.config.getoption("do_dangerous") else False
+ print("do_dangerous_test: " + str(do_dangerous_test))
+ metafunc.parametrize("do_dangerous_test", [do_dangerous_test])
\ No newline at end of file
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..f8d8975
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,6 @@
+[build-system]
+requires = [
+ "setuptools",
+ "wheel"
+]
+build-backend = "setuptools.build_meta"
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index f6e1969..ffce0ec 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,2 @@
-awscli
pytest-xdist
-coloredlogs
-boto3
-requests
\ No newline at end of file
+boto3
\ No newline at end of file
diff --git a/s3scanner.py b/s3scanner.py
deleted file mode 100644
index 8434054..0000000
--- a/s3scanner.py
+++ /dev/null
@@ -1,98 +0,0 @@
-#########
-#
-# AWS S3scanner - Scans domain names for S3 buckets
-#
-# Author: Dan Salmon (twitter.com/bltjetpack, github.com/sa7mon)
-# Created: 6/19/17
-# License: Creative Commons (CC BY-NC-SA 4.0))
-#
-#########
-
-import argparse
-import logging
-from os import path
-import sys
-
-import coloredlogs
-
-import s3utils as s3
-
-CURRENT_VERSION = '1.0.0'
-
-
-
-# We want to use both formatter classes, so a custom class it is
-class CustomFormatter(argparse.RawTextHelpFormatter, argparse.RawDescriptionHelpFormatter):
- pass
-
-
-# Instantiate the parser
-parser = argparse.ArgumentParser(description='# s3scanner - Find S3 buckets and dump!\n'
- '#\n'
- '# Author: Dan Salmon - @bltjetpack, github.com/sa7mon\n',
- prog='s3scanner', formatter_class=CustomFormatter)
-
-# Declare arguments
-parser.add_argument('-o', '--out-file', dest='outFile', default='./buckets.txt',
- help='Name of file to save the successfully checked buckets in (Default: buckets.txt)')
-# parser.add_argument('-c', '--include-closed', dest='includeClosed', action='store_true', default=False,
-# help='Include found but closed buckets in the out-file')
-parser.add_argument('-d', '--dump', dest='dump', action='store_true', default=False,
- help='Dump all found open buckets locally')
-parser.add_argument('-l', '--list', dest='list', action='store_true',
- help='Save bucket file listing to local file: ./list-buckets/${bucket}.txt')
-parser.add_argument('--version', action='version', version=CURRENT_VERSION,
- help='Display the current version of this tool')
-parser.add_argument('buckets', help='Name of text file containing buckets to check')
-
-
-# Parse the args
-args = parser.parse_args()
-
-# Create file logger
-flog = logging.getLogger('s3scanner-file')
-flog.setLevel(logging.DEBUG) # Set log level for logger object
-
-# Create file handler which logs even debug messages
-fh = logging.FileHandler(args.outFile)
-fh.setLevel(logging.DEBUG)
-
-# Add the handler to logger
-flog.addHandler(fh)
-
-# Create secondary logger for logging to screen
-slog = logging.getLogger('s3scanner-screen')
-slog.setLevel(logging.INFO)
-
-# Logging levels for the screen logger:
-# INFO = found
-# ERROR = not found
-# The levels serve no other purpose than to specify the output color
-
-levelStyles = {
- 'info': {'color': 'blue'},
- 'warning': {'color': 'yellow'},
- 'error': {'color': 'red'}
- }
-
-fieldStyles = {
- 'asctime': {'color': 'white'}
- }
-
-# Use coloredlogs to add color to screen logger. Define format and styles.
-coloredlogs.install(level='DEBUG', logger=slog, fmt='%(asctime)s %(message)s',
- level_styles=levelStyles, field_styles=fieldStyles)
-
-if not s3.checkAwsCreds():
- s3.AWS_CREDS_CONFIGURED = False
- slog.error("Warning: AWS credentials not configured. Open buckets will be shown as closed. Run:"
- " `aws configure` to fix this.\n")
-
-if path.isfile(args.buckets):
- with open(args.buckets, 'r') as f:
- for line in f:
- line = line.rstrip() # Remove any extra whitespace
- s3.checkBucket(line, slog, flog, args.dump, args.list)
-else:
- # It's a single bucket
- s3.checkBucket(args.buckets, slog, flog, args.dump, args.list)
diff --git a/s3utils.py b/s3utils.py
deleted file mode 100644
index ef5378b..0000000
--- a/s3utils.py
+++ /dev/null
@@ -1,279 +0,0 @@
-import os
-import re
-import signal
-from contextlib import contextmanager
-import datetime
-
-import boto3
-from botocore.exceptions import ClientError, NoCredentialsError, HTTPClientError
-from botocore.handlers import disable_signing
-from botocore import UNSIGNED
-from botocore.client import Config
-import requests
-
-
-SIZE_CHECK_TIMEOUT = 30 # How long to wait for getBucketSize to return
-AWS_CREDS_CONFIGURED = True
-ERROR_CODES = ['AccessDenied', 'AllAccessDisabled', '[Errno 21] Is a directory:']
-
-
-class TimeoutException(Exception): pass
-
-@contextmanager
-def time_limit(seconds):
- def signal_handler(signum, frame):
- raise TimeoutException("Timed out!")
- signal.signal(signal.SIGALRM, signal_handler)
- signal.alarm(seconds)
- try:
- yield
- finally:
- signal.alarm(0)
-
-
-
-def checkAcl(bucket):
- """
- Attempts to retrieve a bucket's ACL. This also functions as the main 'check if bucket exists' function.
- By trying to get the ACL, we combine 2 steps to minimize potentially slow network calls.
-
- :param bucket: Name of bucket to try to get the ACL of
- :return: A dictionary with 2 entries:
- found - Boolean. True/False whether or not the bucket was found
- acls - dictionary. If ACL was retrieved, contains 2 keys: 'allUsers' and 'authUsers'. If ACL was not
- retrieved,
- """
- allUsersGrants = []
- authUsersGrants = []
-
- s3 = boto3.resource('s3')
-
- try:
- bucket_acl = s3.BucketAcl(bucket)
- bucket_acl.load()
- except s3.meta.client.exceptions.NoSuchBucket:
- return {"found": False, "acls": {}}
-
- except ClientError as e:
- if e.response['Error']['Code'] == "AccessDenied":
- return {"found": True, "acls": "AccessDenied"}
- elif e.response['Error']['Code'] == "AllAccessDisabled":
- return {"found": True, "acls": "AllAccessDisabled"}
- else:
- raise e
-
- for grant in bucket_acl.grants:
- if 'URI' in grant['Grantee']:
- if grant['Grantee']['URI'] == "http://acs.amazonaws.com/groups/global/AllUsers":
- allUsersGrants.append(grant['Permission'])
- elif grant['Grantee']['URI'] == "http://acs.amazonaws.com/groups/global/AuthenticatedUsers":
- authUsersGrants.append(grant['Permission'])
-
- return {"found": True, "acls": {"allUsers": allUsersGrants, "authUsers": authUsersGrants}}
-
-
-def checkAwsCreds():
- """
- Checks to see if the user has credentials for AWS properly configured.
- This is essentially a requirement for getting accurate results.
-
- :return: True if AWS credentials are properly configured. False if not.
- """
-
- sts = boto3.client('sts')
- try:
- response = sts.get_caller_identity()
- except NoCredentialsError as e:
- return False
-
- return True
-
-
-def checkBucket(inBucket, slog, flog, argsDump, argsList):
- # Determine what kind of input we're given. Options:
- # bucket name i.e. mybucket
- # domain name i.e. flaws.cloud
- # full S3 url i.e. flaws.cloud.s3-us-west-2.amazonaws.com
- # bucket:region i.e. flaws.cloud:us-west-2
-
- if ".amazonaws.com" in inBucket: # We were given a full s3 url
- bucket = inBucket[:inBucket.rfind(".s3")]
- elif ":" in inBucket: # We were given a bucket in 'bucket:region' format
- bucket = inBucket.split(":")[0]
- else: # We were either given a bucket name or domain name
- bucket = inBucket
-
- valid = checkBucketName(bucket)
-
- if not valid:
- message = "{0:>11} : {1}".format("[invalid]", bucket)
- slog.error(message)
- # continue
- return
-
- if AWS_CREDS_CONFIGURED:
- b = checkAcl(bucket)
- else:
- a = checkBucketWithoutCreds(bucket)
- b = {"found": a, "acls": "unknown - no aws creds"}
-
- if b["found"]:
-
- size = getBucketSize(bucket) # Try to get the size of the bucket
-
- message = "{0:>11} : {1}".format("[found]", bucket + " | " + str(size) + " | ACLs: " + str(b["acls"]))
- slog.info(message)
- flog.debug(bucket)
-
- if argsDump:
- if size not in ["AccessDenied", "AllAccessDisabled"]:
- slog.info("{0:>11} : {1} - {2}".format("[found]", bucket, "Attempting to dump...this may take a while."))
- dumpBucket(bucket)
- if argsList:
- if str(b["acls"]) not in ["AccessDenied", "AllAccessDisabled"]:
- listBucket(bucket)
- else:
- message = "{0:>11} : {1}".format("[not found]", bucket)
- slog.error(message)
-
-
-def checkBucketName(bucket_name):
- """ Checks to make sure bucket names input are valid according to S3 naming conventions
- :param bucketName: Name of bucket to check
- :return: Boolean - whether or not the name is valid
- """
-
- # Bucket names can be 3-63 (inclusively) characters long.
- # Bucket names may only contain lowercase letters, numbers, periods, and hyphens
- pattern = r'(?=^.{3,63}$)(?!^(\d+\.)+\d+$)(^(([a-z0-9]|[a-z0-9][a-z0-9\-]*[a-z0-9])\.)*([a-z0-9]|[a-z0-9][a-z0-9\-]*[a-z0-9])$)'
-
-
- return bool(re.match(pattern, bucket_name))
-
-
-def checkBucketWithoutCreds(bucketName, triesLeft=2):
- """ Does a simple GET request with the Requests library and interprets the results.
- bucketName - A domain name without protocol (http[s]) """
-
- if triesLeft == 0:
- return False
-
- bucketUrl = 'http://' + bucketName + '.s3.amazonaws.com'
-
- r = requests.head(bucketUrl)
-
- if r.status_code == 200: # Successfully found a bucket!
- return True
- elif r.status_code == 403: # Bucket exists, but we're not allowed to LIST it.
- return True
- elif r.status_code == 404: # This is definitely not a valid bucket name.
- return False
- elif r.status_code == 503:
- return checkBucketWithoutCreds(bucketName, triesLeft - 1)
- else:
- raise ValueError("Got an unhandled status code back: " + str(r.status_code) + " for bucket: " + bucketName +
- ". Please open an issue at: https://github.com/sa7mon/s3scanner/issues and include this info.")
-
-
-def dumpBucket(bucketName):
- global dumped
- # Dump the bucket into bucket folder
- bucketDir = './buckets/' + bucketName
-
- if not os.path.exists(bucketDir):
- os.makedirs(bucketDir)
-
- dumped = True
-
- s3 = boto3.client('s3')
-
- try:
- if AWS_CREDS_CONFIGURED is False:
- s3 = boto3.client('s3', config=Config(signature_version=UNSIGNED))
-
- for page in s3.get_paginator("list_objects_v2").paginate(Bucket=bucketName):
- if 'Contents' in page:
- for item in page['Contents']:
- key = item['Key']
- s3.download_file(bucketName, key, bucketDir+"/"+key)
- dumped = True
- except ClientError as e:
- # global dumped
- if e.response['Error']['Code'] == 'AccessDenied':
- pass # TODO: Do something with the fact that we were denied
- dumped = False
- finally:
- # Check if folder is empty. If it is, delete it
- if not os.listdir(bucketDir):
- os.rmdir(bucketDir)
- return dumped
-
-
-def getBucketSize(bucketName):
- """
- Use awscli to 'ls' the bucket which will give us the total size of the bucket.
- NOTE:
- Function assumes the bucket exists and doesn't catch errors if it doesn't.
- """
- s3 = boto3.client('s3')
- try:
- if AWS_CREDS_CONFIGURED is False:
- s3 = boto3.client('s3', config=Config(signature_version=UNSIGNED))
- size_bytes = 0
- with time_limit(SIZE_CHECK_TIMEOUT):
- for page in s3.get_paginator("list_objects_v2").paginate(Bucket=bucketName):
- if 'Contents' in page:
- for item in page['Contents']:
- size_bytes += item['Size']
- return str(size_bytes) + " bytes"
-
- except HTTPClientError as e:
- if "Timed out!" in str(e):
- return "Unknown Size - timeout"
- else:
- raise e
- except ClientError as e:
- if e.response['Error']['Code'] == 'AccessDenied':
- return "AccessDenied"
- elif e.response['Error']['Code'] == 'AllAccessDisabled':
- return "AllAccessDisabled"
- elif e.response['Error']['Code'] == 'NoSuchBucket':
- return "NoSuchBucket"
- else:
- raise e
-
-
-def listBucket(bucketName):
- """
- If we find an open bucket, save the contents of the bucket listing to file.
- Returns:
- None if no errors were encountered
- """
-
- # Dump the bucket into bucket folder
- bucketDir = './list-buckets/' + bucketName + '.txt'
- if not os.path.exists('./list-buckets/'):
- os.makedirs('./list-buckets/')
-
- s3 = boto3.client('s3')
- objects = []
-
- try:
- if AWS_CREDS_CONFIGURED is False:
- s3 = boto3.client('s3', config=Config(signature_version=UNSIGNED))
-
- for page in s3.get_paginator("list_objects_v2").paginate(Bucket=bucketName):
- if 'Contents' in page:
- for item in page['Contents']:
- o = item['LastModified'].strftime('%Y-%m-%d %H:%M:%S') + " " + str(item['Size']) + " " + item['Key']
- objects.append(o)
-
- with open(bucketDir, 'w') as f:
- for o in objects:
- f.write(o + "\n")
-
- except ClientError as e:
- if e.response['Error']['Code'] == 'AccessDenied':
- return "AccessDenied"
- else:
- raise e
diff --git a/setup.cfg b/setup.cfg
new file mode 100644
index 0000000..7df77e8
--- /dev/null
+++ b/setup.cfg
@@ -0,0 +1,33 @@
+[metadata]
+name = S3Scanner
+version = 2.0.0
+author = Dan Salmon
+author_email = dan@salmon.cat
+description = Scan for open S3 buckets and dump the contents
+long_description = file: README.md
+long_description_content_type = text/markdown
+url = https://github.com/sa7mon/S3Scanner
+project_urls =
+ Bug Tracker = https://github.com/sa7mon/S3Scanner
+classifiers =
+ Programming Language :: Python :: 3.6
+ Programming Language :: Python :: 3.7
+ Programming Language :: Python :: 3.8
+ Programming Language :: Python :: 3.9
+ Topic :: Security
+ License :: OSI Approved :: MIT License
+ Operating System :: OS Independent
+
+[options]
+packages = S3Scanner
+install_requires =
+ boto3
+python_requires = >=3.6
+
+[options.entry_points]
+console_scripts =
+ s3scanner = S3Scanner.__main__:main
+
+[tool:pytest]
+python_files=test_*.py
+filterwarnings = ignore::pytest.PytestCollectionWarning
\ No newline at end of file
diff --git a/sites.txt b/sites.txt
deleted file mode 100644
index 635108c..0000000
--- a/sites.txt
+++ /dev/null
@@ -1,6 +0,0 @@
-flaws.cloud
-arstechnica.com
-lifehacker.com
-gizmodo.com
-reddit.com
-stackoverflow.com
\ No newline at end of file
diff --git a/test_scanner.py b/test_scanner.py
deleted file mode 100644
index f99940d..0000000
--- a/test_scanner.py
+++ /dev/null
@@ -1,408 +0,0 @@
-import s3utils as s3
-import os
-import sys
-import shutil
-import time
-import logging
-import subprocess
-
-
-pyVersion = sys.version_info # pyVersion[0] can be 2 or 3
-
-
-s3scannerLocation = "./"
-testingFolder = "./test/"
-
-setupRan = False
-
-
-def test_setup():
- """ Setup code to run before we run all tests. """
- global setupRan
-
- if setupRan: # We only need to run this once per test-run
- return
-
- # Check if AWS creds are configured
- s3.AWS_CREDS_CONFIGURED = s3.checkAwsCreds()
-
- print("--> AWS credentials configured: " + str(s3.AWS_CREDS_CONFIGURED))
-
- # Create testingFolder if it doesn't exist
- if not os.path.exists(testingFolder) or not os.path.isdir(testingFolder):
- os.makedirs(testingFolder)
-
- setupRan = True
-
-
-def test_arguments():
- """
- Scenario mainargs.1: No args supplied
- Scenario mainargs.2: --out-file
- Scenario mainargs.3: --list
- Scenario mainargs.4: --dump
- """
-
- test_setup()
-
- # mainargs.1
- a = subprocess.run(['python3', s3scannerLocation + 's3scanner.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- assert a.stderr == b'usage: s3scanner [-h] [-o OUTFILE] [-d] [-l] [--version] buckets\ns3scanner: error: the following arguments are required: buckets\n'
-
- # mainargs.2
-
- # Put one bucket into a new file
- with open(testingFolder + "mainargs.2_input.txt", "w") as f:
- f.write('s3scanner-bucketsize\n')
-
- try:
- a = subprocess.run(['python3', s3scannerLocation + 's3scanner.py', '--out-file', testingFolder + 'mainargs.2_output.txt',
- testingFolder + 'mainargs.2_input.txt'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-
- with open(testingFolder + "mainargs.2_output.txt") as f:
- line = f.readline().strip()
-
- assert line == 's3scanner-bucketsize'
-
- finally:
- # No matter what happens with the test, clean up the test files at the end
- try:
- os.remove(testingFolder + 'mainargs.2_output.txt')
- os.remove(testingFolder + 'mainargs.2_input.txt')
- except OSError:
- pass
-
- # mainargs.3
- # mainargs.4
-
-
-def test_checkAcl():
- """
- Scenario checkAcl.1 - ACL listing enabled
- Expected:
- found = True
- acls = {'allUsers': ['READ', 'READ_ACP'], 'authUsers': ['READ', 'READ_ACP']}
- Scenario checkAcl.2 - AccessDenied for ACL listing
- Expected:
- found = True
- acls = 'AccessDenied'
- Scenario checkAcl.3 - Bucket access is disabled
- Expected:
- found = True
- acls = "AllAccessDisabled"
- Scenario checkAcl.4 - Bucket doesn't exist
- Expected:
- found = False
- acls = {}
- """
- test_setup()
-
- if not s3.AWS_CREDS_CONFIGURED: # Don't run tests if AWS creds aren't configured
- return
-
- # checkAcl.1
- r1 = s3.checkAcl('aneta')
- assert r1["found"] is True
- assert r1["acls"] == {'allUsers': ['READ', 'READ_ACP'], 'authUsers': ['READ', 'READ_ACP']}
-
- # checkAcl.2
- result = s3.checkAcl('flaws.cloud')
- assert result["found"] is True
- assert result["acls"] == "AccessDenied"
-
- # checkAcl.3
- result = s3.checkAcl('amazon.com')
- assert result["found"] is True
- assert result["acls"] == "AllAccessDisabled"
-
- # checkAcl.4
- result = s3.checkAcl('hopethisdoesntexist1234asdf')
- assert result["found"] is False
- assert result["acls"] == {}
-
-
-def test_checkAwsCreds():
- """
- Scenario checkAwsCreds.1 - Output of checkAwsCreds() matches a more intense check for creds
- """
- test_setup()
-
- # Check more thoroughly for creds being set.
- vars = os.environ
-
- keyid = vars.get("AWS_ACCESS_KEY_ID")
- key = vars.get("AWS_SECRET_ACCESS_KEY")
- credsFile = os.path.expanduser("~") + "/.aws/credentials"
-
- if keyid is not None and len(keyid) == 20:
- if key is not None and len(key) == 40:
- credsActuallyConfigured = True
- else:
- credsActuallyConfigured = False
- else:
- credsActuallyConfigured = False
-
- if os.path.exists(credsFile):
- print("credsFile path exists")
- if not credsActuallyConfigured:
- keyIdSet = None
- keySet = None
-
- # Check the ~/.aws/credentials file
- with open(credsFile, "r") as f:
- for line in f:
- line = line.strip()
- if line[0:17].lower() == 'aws_access_key_id':
- if len(line) >= 38: # key + value = length of at least 38 if no spaces around equals
- keyIdSet = True
- else:
- keyIdSet = False
-
- if line[0:21].lower() == 'aws_secret_access_key':
- if len(line) >= 62:
- keySet = True
- else:
- keySet = False
-
- if keyIdSet and keySet:
- credsActuallyConfigured = True
-
- # checkAwsCreds.1
- assert s3.checkAwsCreds() == credsActuallyConfigured
-
-def test_checkBucket():
- """
- checkBucket.1 - Bucket name
- checkBucket.2 - Domain name
- checkBucket.3 - Full s3 url
- checkBucket.4 - bucket:region
- """
-
- test_setup()
-
- testFile = './test/test_checkBucket.txt'
-
- # Create file logger
- flog = logging.getLogger('s3scanner-file')
- flog.setLevel(logging.DEBUG) # Set log level for logger object
-
- # Create file handler which logs even debug messages
- fh = logging.FileHandler(testFile)
- fh.setLevel(logging.DEBUG)
-
- # Add the handler to logger
- flog.addHandler(fh)
-
- # Create secondary logger for logging to screen
- slog = logging.getLogger('s3scanner-screen')
- slog.setLevel(logging.CRITICAL)
-
- try:
- # checkBucket.1
- s3.checkBucket("flaws.cloud", slog, flog, False, False)
-
- # checkBucket.2
- s3.checkBucket("flaws.cloud.s3-us-west-2.amazonaws.com", slog, flog, False, False)
-
- # checkBucket.3
- s3.checkBucket("flaws.cloud:us-west-2", slog, flog, False, False)
-
- # Read in test loggin file and assert
- f = open(testFile, 'r')
- results = f.readlines()
- f.close()
-
- assert results[0].rstrip() == "flaws.cloud"
- assert results[1].rstrip() == "flaws.cloud"
- assert results[2].rstrip() == "flaws.cloud"
-
- finally:
- # Delete test file
- os.remove(testFile)
-
-def test_checkBucketName():
- """
- Scenario checkBucketName.1 - Under length requirements
- Expected: False
- Scenario checkBucketName.2 - Over length requirements
- Expected: False
- Scenario checkBucketName.3 - Contains forbidden characters
- Expected: False
- Scenario checkBucketName.4 - Blank name
- Expected: False
- Scenario checkBucketName.5 - Good name
- Expected: True
- """
- test_setup()
-
- # checkBucketName.1
- result = s3.checkBucketName('ab')
- assert result is False
-
- # checkBucketName.2
- tooLong = "asdfasdf12834092834nMSdfnasjdfhu23y49u2y4jsdkfjbasdfbasdmn4asfasdf23423423423423" # 80 characters
- result = s3.checkBucketName(tooLong)
- assert result is False
-
- # checkBucketName.3
- badBucket = "mycoolbucket:dev"
- assert s3.checkBucketName(badBucket) is False
-
- # checkBucketName.4
- assert s3.checkBucketName('') is False
-
- # checkBucketName.5
- assert s3.checkBucketName('arathergoodname') is True
-
-
-def test_checkBucketWithoutCreds():
- """
- Scenario checkBucketwc.1 - Non-existent bucket
- Scenario checkBucketwc.2 - Good bucket
- Scenario checkBucketwc.3 - No public read perm
- """
- test_setup()
-
- if s3.AWS_CREDS_CONFIGURED:
- return
-
- # checkBucketwc.1
- assert s3.checkBucketWithoutCreds('ireallyhopethisbucketdoesntexist') is False
-
- # checkBucketwc.2
- assert s3.checkBucketWithoutCreds('flaws.cloud') is True
-
- # checkBucketwc.3
- assert s3.checkBucketWithoutCreds('blog') is True
-
-
-def test_dumpBucket():
- """
- Scenario dumpBucket.1 - Public read permission enabled
- Expected: Dumping the bucket "flaws.cloud" should result in 6 files being downloaded into the buckets folder.
- The expected file sizes of each file are listed in the 'expectedFiles' dictionary.
- Scenario dumpBucket.2 - Public read objects disabled
- Expected: The function returns false and the bucket directory doesn't exist
- Scenario dumpBucket.3 - Authenticated users read enabled, public users read disabled
- Expected: The function returns true and the bucket directory exists. Opposite for if no aws creds are set
- """
- test_setup()
-
- # dumpBucket.1
-
- s3.dumpBucket("flaws.cloud")
-
- dumpDir = './buckets/flaws.cloud/' # Folder to look for the files in
-
- # Expected sizes of each file
- expectedFiles = {'hint1.html': 2575, 'hint2.html': 1707, 'hint3.html': 1101, 'index.html': 3082,
- 'robots.txt': 46, 'secret-dd02c7c.html': 1051, 'logo.png': 15979}
-
- try:
- # Assert number of files in the folder
- assert len(os.listdir(dumpDir)) == len(expectedFiles)
-
- # For each file, assert the size
- for file, size in expectedFiles.items():
- assert os.path.getsize(dumpDir + file) == size
- finally:
- # No matter what happens with the asserts, cleanup after the test by deleting the flaws.cloud directory
- shutil.rmtree(dumpDir)
-
- # dumpBucket.2
- assert s3.dumpBucket('s3scanner-private') is s3.AWS_CREDS_CONFIGURED
- assert os.path.exists('./buckets/s3scanner-private') is False
-
- # dumpBucket.3
- assert s3.dumpBucket('s3scanner-auth') is s3.AWS_CREDS_CONFIGURED # Asserts should both follow whether or not creds are set
- assert os.path.exists('./buckets/s3scanner-auth') is False
-
-
-def test_getBucketSize():
- """
- Scenario getBucketSize.1 - Public read enabled
- Expected: The s3scanner-bucketsize bucket returns size: 43 bytes
- Scenario getBucketSize.2 - Public read disabled
- Expected: app-dev bucket has public read permissions disabled
- Scenario getBucketSize.3 - Bucket doesn't exist
- Expected: We should get back "NoSuchBucket"
- Scenario getBucketSize.4 - Public read enabled, more than 1,000 objects
- Expected: The s3scanner-long bucket returns size: 3900 bytes
- """
- test_setup()
-
- # getBucketSize.1
- assert s3.getBucketSize('s3scanner-bucketsize') == "43 bytes"
-
- # getBucketSize.2
- assert s3.getBucketSize('app-dev') == "AccessDenied"
-
- # getBucketSize.3
- assert s3.getBucketSize('thiswillprobablynotexistihope') == "NoSuchBucket"
-
- # getBucketSize.4
- assert s3.getBucketSize('s3scanner-long') == "4000 bytes"
-
-
-def test_getBucketSizeTimeout():
- """
- Scenario getBucketSize.1 - Too many files to list so it times out
- Expected: The function returns a timeout error after the specified wait time
- Note: Verify that getBucketSize returns an unknown size and doesn't take longer
- than sizeCheckTimeout set in s3utils
- """
- test_setup()
-
- # s3.AWS_CREDS_CONFIGURED = False
- # s3.SIZE_CHECK_TIMEOUT = 2 # In case we have a fast connection
-
- # output = s3.getBucketSize("s3scanner-long")
-
- # # Assert that the size check timed out
- # assert output == "Unknown Size - timeout"
-
- print("!! Notes: test_getBucketSizeTimeout temporarily disabled.")
-
-
-def test_listBucket():
- """
- Scenario listBucket.1 - Public read enabled
- Expected: Listing bucket flaws.cloud will create the directory, create flaws.cloud.txt, and write the listing to file
- Scenario listBucket.2 - Public read disabled
- Scenario listBucket.3 - Public read enabled, long listing
-
- """
- test_setup()
-
- # listBucket.1
-
- listFile = './list-buckets/s3scanner-bucketsize.txt'
-
- s3.listBucket('s3scanner-bucketsize')
-
- assert os.path.exists(listFile) # Assert file was created in the correct location
-
- lines = []
- with open(listFile, 'r') as g:
- for line in g:
- lines.append(line)
-
- assert lines[0].rstrip().endswith('test-file.txt') # Assert the first line is correct
- assert len(lines) == 1 # Assert number of lines in the file is correct
-
- # listBucket.2
- if s3.AWS_CREDS_CONFIGURED:
- assert s3.listBucket('s3scanner-private') == None
- else:
- assert s3.listBucket('s3scanner-private') == "AccessDenied"
-
- # listBucket.3
- longFile = './list-buckets/s3scanner-long.txt'
- s3.listBucket('s3scanner-long')
- assert os.path.exists(longFile)
-
- lines = []
- with open(longFile, 'r') as f:
- for line in f:
- lines.append(f)
- assert len(lines) == 3501
\ No newline at end of file
diff --git a/tests/TestUtils.py b/tests/TestUtils.py
new file mode 100644
index 0000000..efa541a
--- /dev/null
+++ b/tests/TestUtils.py
@@ -0,0 +1,49 @@
+import random
+import string
+import boto3
+
+
+class TestBucketService:
+ def __init__(self):
+ self.session = boto3.Session(profile_name='privileged')
+ self.s3_client = self.session.client('s3')
+
+ @staticmethod
+ def generate_random_bucket_name(length=40):
+ candidates = string.ascii_lowercase + string.digits
+ return 's3scanner-' + ''.join(random.choice(candidates) for i in range(length))
+
+ def delete_bucket(self, bucket_name):
+ self.s3_client.delete_bucket(Bucket=bucket_name)
+
+ def create_bucket(self, danger_bucket):
+ bucket_name = self.generate_random_bucket_name()
+
+ # For type descriptions, refer to: https://github.com/sa7mon/S3Scanner/wiki/Test-Buckets
+ if danger_bucket == 1:
+ self.s3_client.create_bucket(Bucket=bucket_name,
+ GrantWrite='uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers')
+ self.s3_client.put_bucket_acl(Bucket=bucket_name,
+ GrantWrite='uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers',
+ GrantWriteACP='uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers')
+ elif danger_bucket == 2:
+ self.s3_client.create_bucket(Bucket=bucket_name,
+ GrantWrite='uri=http://acs.amazonaws.com/groups/global/AllUsers',
+ GrantWriteACP='uri=http://acs.amazonaws.com/groups/global/AllUsers')
+ elif danger_bucket == 3:
+ self.s3_client.create_bucket(Bucket=bucket_name,
+ GrantRead='uri=http://acs.amazonaws.com/groups/global/AllUsers',
+ GrantWrite='uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers',
+ GrantWriteACP='uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers')
+ elif danger_bucket == 4:
+ self.s3_client.create_bucket(Bucket=bucket_name,
+ GrantWrite='uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers,'
+ 'uri=http://acs.amazonaws.com/groups/global/AllUsers')
+ elif danger_bucket == 5:
+ self.s3_client.create_bucket(Bucket=bucket_name,
+ GrantWriteACP='uri=http://acs.amazonaws.com/groups/global/AuthenticatedUsers,'
+ 'uri=http://acs.amazonaws.com/groups/global/AllUsers')
+ else:
+ raise Exception("Unknown danger bucket type")
+
+ return bucket_name
diff --git a/tests/test_bucket.py b/tests/test_bucket.py
new file mode 100644
index 0000000..c975f92
--- /dev/null
+++ b/tests/test_bucket.py
@@ -0,0 +1,48 @@
+from S3Scanner.S3Bucket import S3Bucket, S3BucketObject, Permission
+
+"""
+Tests for S3Bucket class go here
+"""
+
+
+def test_invalid_bucket_name():
+ try:
+ S3Bucket(name="asdf,;0()")
+ except ValueError as ve:
+ if str(ve) != "Invalid bucket name":
+ raise ve
+
+
+def test_s3_bucket_object():
+ o1 = S3BucketObject(key='index.html', size=8096, last_modified='2018-03-02T08:10:25.000Z')
+ o2 = S3BucketObject(key='home.html', size=2, last_modified='2018-03-02T08:10:25.000Z')
+
+ assert o1 != o2
+ assert o2 < o1 # test __lt__ method which compares keys
+ assert str(o1) == "Key: index.html, Size: 8096, LastModified: 2018-03-02T08:10:25.000Z"
+ assert o1.get_human_readable_size() == "7.9KB"
+
+
+def test_check_bucket_name():
+ S3Bucket(name="asdfasdf.s3.amazonaws.com")
+ S3Bucket(name="asdf:us-west-1")
+
+
+def test_get_human_readable_permissions():
+ b = S3Bucket(name='asdf')
+ b.AllUsersRead = Permission.ALLOWED
+ b.AllUsersWrite = Permission.ALLOWED
+ b.AllUsersReadACP = Permission.ALLOWED
+ b.AllUsersWriteACP = Permission.ALLOWED
+ b.AuthUsersRead = Permission.ALLOWED
+ b.AuthUsersWrite = Permission.ALLOWED
+ b.AuthUsersReadACP = Permission.ALLOWED
+ b.AuthUsersWriteACP = Permission.ALLOWED
+
+ b.get_human_readable_permissions()
+
+ b.AllUsersFullControl = Permission.ALLOWED
+ b.AuthUsersFullControl = Permission.ALLOWED
+
+ b.get_human_readable_permissions()
+
diff --git a/tests/test_scanner.py b/tests/test_scanner.py
new file mode 100644
index 0000000..c6c1dc9
--- /dev/null
+++ b/tests/test_scanner.py
@@ -0,0 +1,139 @@
+import sys
+import subprocess
+import os
+import time
+import shutil
+
+from S3Scanner.S3Service import S3Service
+
+
+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'
+
+ 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())
+
+ c = subprocess.run([sys.executable, '-m', 'S3Scanner', 'scan', '--bucket', 'asdfasdf---,'], stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
+ assert_scanner_output(s, 'asdfasdf---, | bucket_invalid_name', c.stdout.decode('utf-8').strip())
+
+ d = subprocess.run([sys.executable, '-m', 'S3Scanner', 'scan', '--bucket', 'isurehopethisbucketdoesntexistasdfasdf'],
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert_scanner_output(s, 'isurehopethisbucketdoesntexistasdfasdf | bucket_not_exist', d.stdout.decode('utf-8').strip())
+
+ e = subprocess.run([sys.executable, '-m', 'S3Scanner', 'scan', '--bucket', 'flaws.cloud', '--dangerous'],
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert_scanner_output(s, f"INFO: Including dangerous checks. WARNING: This may change bucket ACL destructively{os.linesep}flaws.cloud | bucket_exists | AuthUsers: [], AllUsers: [Read]", e.stdout.decode('utf-8').strip())
+
+ f = subprocess.run([sys.executable, '-m', 'S3Scanner', 'dump', '--bucket', 'flaws.cloud', '--dump-dir', './asfasdf'],
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert_scanner_output(s, "Error: Given --dump-dir does not exist or is not a directory", f.stdout.decode('utf-8').strip())
+
+ # Create temp folder to dump into
+ test_folder = os.path.join(os.getcwd(), 'testing_' + str(time.time())[0:10], '')
+ os.mkdir(test_folder)
+
+ try:
+ f = subprocess.run([sys.executable, '-m', 'S3Scanner', 'dump', '--bucket', 'flaws.cloud', '--dump-dir', test_folder],
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert_scanner_output(s, f"flaws.cloud | Enumerating bucket objects...{os.linesep}flaws.cloud | Total Objects: 7, Total Size: 25.0KB{os.linesep}flaws.cloud | Dumping contents using 4 threads...{os.linesep}flaws.cloud | Dumping completed", f.stdout.decode('utf-8').strip())
+
+ g = subprocess.run([sys.executable, '-m', 'S3Scanner', 'dump', '--bucket', 'asdfasdf,asdfasd,', '--dump-dir', test_folder],
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert_scanner_output(s, "asdfasdf,asdfasd, | bucket_name_invalid", g.stdout.decode('utf-8').strip())
+
+ h = subprocess.run([sys.executable, '-m', 'S3Scanner', 'dump', '--bucket', 'isurehopethisbucketdoesntexistasdfasdf', '--dump-dir', test_folder],
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert_scanner_output(s, 'isurehopethisbucketdoesntexistasdfasdf | bucket_not_exist', h.stdout.decode('utf-8').strip())
+ finally:
+ shutil.rmtree(test_folder) # Cleanup the testing folder
+
+
+def test_endpoints():
+ """
+ Test the handling of non-AWS endpoints
+ :return:
+ """
+ s = S3Service()
+ b = subprocess.run([sys.executable, '-m', 'S3Scanner', '--endpoint-url', 'https://sfo2.digitaloceanspaces.com',
+ 'scan', '--bucket', 's3scanner'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert_scanner_output(s, 's3scanner | bucket_not_exist',
+ b.stdout.decode('utf-8').strip())
+
+ c = subprocess.run([sys.executable, '-m', 'S3Scanner', '--endpoint-url', 'http://example.com', 'scan', '--bucket',
+ 's3scanner'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ assert c.stdout.decode('utf-8').strip() == "Error: Endpoint 'http://example.com' does not appear to be S3-compliant"
+
+
+def assert_scanner_output(service, expected_output, found_output):
+ """
+ If the tests are run without AWS creds configured, all the output from scanner.py will have a warning banner.
+ This is a convenience method to simplify comparing the expected output to the found output
+
+ :param service: s3service
+ :param expected_output: string
+ :param found_output: string
+ :return: boolean
+ """
+ creds_warning = "Warning: AWS credentials not configured - functionality will be limited. Run: `aws configure` to fix this."
+
+ if service.aws_creds_configured:
+ assert expected_output == found_output
+ else:
+ assert f"{creds_warning}{os.linesep}{os.linesep}{expected_output}" == found_output
+
+
+def test_check_aws_creds():
+ """
+ Scenario checkAwsCreds.1 - Output of checkAwsCreds() matches a more intense check for creds
+ """
+ print("test_checkAwsCreds temporarily disabled.")
+
+ # test_setup()
+ #
+ # # Check more thoroughly for creds being set.
+ # vars = os.environ
+ #
+ # keyid = vars.get("AWS_ACCESS_KEY_ID")
+ # key = vars.get("AWS_SECRET_ACCESS_KEY")
+ # credsFile = os.path.expanduser("~") + "/.aws/credentials"
+ #
+ # if keyid is not None and len(keyid) == 20:
+ # if key is not None and len(key) == 40:
+ # credsActuallyConfigured = True
+ # else:
+ # credsActuallyConfigured = False
+ # else:
+ # credsActuallyConfigured = False
+ #
+ # if os.path.exists(credsFile):
+ # print("credsFile path exists")
+ # if not credsActuallyConfigured:
+ # keyIdSet = None
+ # keySet = None
+ #
+ # # Check the ~/.aws/credentials file
+ # with open(credsFile, "r") as f:
+ # for line in f:
+ # line = line.strip()
+ # if line[0:17].lower() == 'aws_access_key_id':
+ # if len(line) >= 38: # key + value = length of at least 38 if no spaces around equals
+ # keyIdSet = True
+ # else:
+ # keyIdSet = False
+ #
+ # if line[0:21].lower() == 'aws_secret_access_key':
+ # if len(line) >= 62:
+ # keySet = True
+ # else:
+ # keySet = False
+ #
+ # if keyIdSet and keySet:
+ # credsActuallyConfigured = True
+ #
+ # # checkAwsCreds.1
+ # assert s3.checkAwsCreds() == credsActuallyConfigured
+
diff --git a/tests/test_service.py b/tests/test_service.py
new file mode 100644
index 0000000..377ff84
--- /dev/null
+++ b/tests/test_service.py
@@ -0,0 +1,545 @@
+import os
+
+import pytest
+
+from S3Scanner.S3Service import S3Service
+from S3Scanner.S3Bucket import BucketExists, Permission, S3BucketObject, S3Bucket
+from TestUtils import TestBucketService
+from S3Scanner.exceptions import AccessDeniedException, BucketMightNotExistException
+from pathlib import Path
+
+testingFolder = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), 'test/')
+setupRan = False
+
+
+"""
+S3Service.py methods to test:
+
+- init()
+ - โ๏ธ Test service.aws_creds_configured is false when forceNoCreds = False
+- check_bucket_exists()
+ - โ๏ธ Test against that exists
+ - โ๏ธ Test against one that doesn't
+- check_perm_read_acl()
+ - โ๏ธ Test against bucket with AllUsers allowed
+ - โ๏ธ Test against bucket with AuthUsers allowed
+ - โ๏ธ Test against bucket with all denied
+- check_perm_read()
+ - โ๏ธ Test against bucket with AuthUsers read permission
+ - โ๏ธ Test against bucket with AllUsers read permission
+ - โ๏ธ Test against bucket with no read permission
+- check_perm_write()
+ - โ๏ธ Test against bucket with no write permissions
+ - โ๏ธ Test against bucket with AuthUsers write permission
+ - โ๏ธ Test against bucket with AllUsers write permission
+ - โ๏ธ Test against bucket with AllUsers and AuthUsers write permission
+- check_perm_write_acl()
+ - โ๏ธ Test against bucket with AllUsers allowed
+ - โ๏ธ Test against bucket with AuthUsers allowed
+ - โ๏ธ Test against bucket with both AllUsers allowed
+ - โ๏ธ Test against bucket with no groups allowed
+- enumerate_bucket_objects()
+ - โ๏ธ Test against empty bucket
+ - โ๏ธ Test against not empty bucket with read permission
+ - โ๏ธ Test against bucket without read permission
+- parse_found_acl()
+ - โ๏ธ Test against JSON with FULL_CONTROL for AllUsers
+ - โ๏ธ Test against JSON with FULL_CONTROL for AuthUsers
+ - โ๏ธ Test against empty JSON
+ - โ๏ธ Test against JSON with ReadACP for AuthUsers and Write for AllUsers
+"""
+
+
+def test_setup_new():
+ global setupRan
+ if setupRan: # We only need to run this once per test-run
+ return
+
+ # Create testingFolder if it doesn't exist
+ if not os.path.exists(testingFolder) or not os.path.isdir(testingFolder):
+ os.makedirs(testingFolder)
+ setupRan = True
+
+
+def test_init():
+ test_setup_new()
+
+ s = S3Service(forceNoCreds=True)
+ assert s.aws_creds_configured is False
+
+
+def test_bucket_exists():
+ test_setup_new()
+
+ s = S3Service()
+
+ # Bucket that does exist
+ b1 = S3Bucket('s3scanner-private')
+ s.check_bucket_exists(b1)
+ assert b1.exists is BucketExists.YES
+
+ # Bucket that doesn't exist (hopefully)
+ b2 = S3Bucket('asfasfasdfasdfasdf')
+ s.check_bucket_exists(b2)
+ assert b2.exists is BucketExists.NO
+
+ # Pass a thing that's not a bucket
+ with pytest.raises(ValueError):
+ s.check_bucket_exists("asdfasdf")
+
+
+def test_check_perm_read():
+ test_setup_new()
+
+ s = S3Service()
+
+ # Bucket that no one can list
+ b1 = S3Bucket('s3scanner-private')
+ b1.exists = BucketExists.YES
+ s.check_perm_read(b1)
+ if s.aws_creds_configured:
+ assert b1.AuthUsersRead == Permission.DENIED
+ else:
+ assert b1.AllUsersRead == Permission.DENIED
+
+ # Bucket that only AuthenticatedUsers can list
+ b2 = S3Bucket('s3scanner-auth-read')
+ b2.exists = BucketExists.YES
+ s.check_perm_read(b2)
+ if s.aws_creds_configured:
+ assert b2.AuthUsersRead == Permission.ALLOWED
+ else:
+ assert b2.AllUsersRead == Permission.DENIED
+
+ # Bucket that Everyone can list
+ b3 = S3Bucket('s3scanner-long')
+ b3.exists = BucketExists.YES
+ s.check_perm_read(b3)
+ if s.aws_creds_configured:
+ assert b3.AuthUsersRead == Permission.ALLOWED
+ else:
+ assert b3.AllUsersRead == Permission.ALLOWED
+
+
+def test_enumerate_bucket_objects():
+ test_setup_new()
+
+ s = S3Service()
+
+ # Empty bucket
+ b1 = S3Bucket('s3scanner-empty')
+ b1.exists = BucketExists.YES
+ s.check_perm_read(b1)
+ if s.aws_creds_configured:
+ assert b1.AuthUsersRead == Permission.ALLOWED
+ else:
+ assert b1.AllUsersRead == Permission.ALLOWED
+ s.enumerate_bucket_objects(b1)
+ assert b1.objects_enumerated is True
+ assert b1.bucketSize == 0
+
+ # Bucket with > 1000 items
+ if s.aws_creds_configured:
+ b2 = S3Bucket('s3scanner-auth-read')
+ b2.exists = BucketExists.YES
+ s.check_perm_read(b2)
+ assert b2.AuthUsersRead == Permission.ALLOWED
+ s.enumerate_bucket_objects(b2)
+ assert b2.objects_enumerated is True
+ assert b2.bucketSize == 4143
+ assert b2.get_human_readable_size() == "4.0KB"
+ else:
+ print("[test_enumerate_bucket_objects] Skipping test due to no AWS creds")
+
+ # Bucket without read permission
+ b3 = S3Bucket('s3scanner-private')
+ b3.exists = BucketExists.YES
+ s.check_perm_read(b3)
+ if s.aws_creds_configured:
+ assert b3.AuthUsersRead == Permission.DENIED
+ else:
+ assert b3.AllUsersRead == Permission.DENIED
+ try:
+ s.enumerate_bucket_objects(b3)
+ except AccessDeniedException:
+ pass
+
+ # Try to enumerate before checking if bucket exists
+ b4 = S3Bucket('s3scanner-enumerate-bucket')
+ with pytest.raises(Exception):
+ s.enumerate_bucket_objects(b4)
+
+
+def test_check_perm_read_acl():
+ test_setup_new()
+ s = S3Service()
+
+ # Bucket with no read ACL perms
+ b1 = S3Bucket('s3scanner-private')
+ b1.exists = BucketExists.YES
+ s.check_perm_read_acl(b1)
+ if s.aws_creds_configured:
+ assert b1.AuthUsersReadACP == Permission.DENIED
+ else:
+ assert b1.AllUsersReadACP == Permission.DENIED
+
+ # Bucket that allows AuthenticatedUsers to read ACL
+ if s.aws_creds_configured:
+ b2 = S3Bucket('s3scanner-auth-read-acl')
+ b2.exists = BucketExists.YES
+ s.check_perm_read_acl(b2)
+ if s.aws_creds_configured:
+ assert b2.AuthUsersReadACP == Permission.ALLOWED
+ else:
+ assert b2.AllUsersReadACP == Permission.DENIED
+
+ # Bucket that allows AllUsers to read ACL
+ b3 = S3Bucket('s3scanner-all-readacp')
+ b3.exists = BucketExists.YES
+ s.check_perm_read_acl(b3)
+ assert b3.AllUsersReadACP == Permission.ALLOWED
+ assert b3.AllUsersWrite == Permission.DENIED
+ assert b3.AllUsersWriteACP == Permission.DENIED
+ assert b3.AuthUsersReadACP == Permission.DENIED
+ assert b3.AuthUsersWriteACP == Permission.DENIED
+ assert b3.AuthUsersWrite == Permission.DENIED
+
+
+def test_check_perm_write(do_dangerous_test):
+ test_setup_new()
+ s = S3Service()
+ sAnon = S3Service(forceNoCreds=True)
+
+ # Bucket with no write perms
+ b1 = S3Bucket('flaws.cloud')
+ b1.exists = BucketExists.YES
+ s.check_perm_write(b1)
+
+ if s.aws_creds_configured:
+ assert b1.AuthUsersWrite == Permission.DENIED
+ else:
+ assert b1.AllUsersWrite == Permission.DENIED
+
+ if do_dangerous_test:
+ print("[test_check_perm_write] Doing dangerous test")
+ ts = TestBucketService()
+
+ danger_bucket_1 = ts.create_bucket(1) # Bucket with AuthUser Write, WriteACP permissions
+ try:
+ b2 = S3Bucket(danger_bucket_1)
+ b2.exists = BucketExists.YES
+ sAnon.check_perm_write(b2)
+ s.check_perm_write(b2)
+ assert b2.AuthUsersWrite == Permission.ALLOWED
+ assert b2.AllUsersWrite == Permission.DENIED
+ finally:
+ ts.delete_bucket(danger_bucket_1)
+
+ danger_bucket_2 = ts.create_bucket(2) # Bucket with AllUser Write, WriteACP permissions
+ try:
+ b3 = S3Bucket(danger_bucket_2)
+ b3.exists = BucketExists.YES
+ sAnon.check_perm_write(b3)
+ s.check_perm_write(b3)
+ assert b3.AllUsersWrite == Permission.ALLOWED
+ assert b3.AuthUsersWrite == Permission.UNKNOWN
+ finally:
+ ts.delete_bucket(danger_bucket_2)
+
+ # Bucket with AllUsers and AuthUser Write permissions
+ danger_bucket_4 = ts.create_bucket(4)
+ try:
+ b4 = S3Bucket(danger_bucket_4)
+ b4.exists = BucketExists.YES
+ sAnon.check_perm_write(b4)
+ s.check_perm_write(b4)
+ assert b4.AllUsersWrite == Permission.ALLOWED
+ assert b4.AuthUsersWrite == Permission.UNKNOWN
+ finally:
+ ts.delete_bucket(danger_bucket_4)
+ else:
+ print("[test_check_perm_write] Skipping dangerous test")
+
+
+def test_check_perm_write_acl(do_dangerous_test):
+ test_setup_new()
+ s = S3Service()
+ sNoCreds = S3Service(forceNoCreds=True)
+
+ # Bucket with no permissions
+ b1 = S3Bucket('s3scanner-private')
+ b1.exists = BucketExists.YES
+ s.check_perm_write_acl(b1)
+ if s.aws_creds_configured:
+ assert b1.AuthUsersWriteACP == Permission.DENIED
+ assert b1.AllUsersWriteACP == Permission.UNKNOWN
+ else:
+ assert b1.AllUsersWriteACP == Permission.DENIED
+ assert b1.AuthUsersWriteACP == Permission.UNKNOWN
+
+ if do_dangerous_test:
+ print("[test_check_perm_write_acl] Doing dangerous tests...")
+ ts = TestBucketService()
+
+ # Bucket with WRITE_ACP enabled for AuthUsers
+ danger_bucket_3 = ts.create_bucket(3)
+ try:
+ b2 = S3Bucket(danger_bucket_3)
+ b2.exists = BucketExists.YES
+
+ # Check for read/write permissions so when we check for write_acl we
+ # send the same perms that it had originally
+ sNoCreds.check_perm_read(b2)
+ s.check_perm_read(b2)
+ sNoCreds.check_perm_write(b2)
+ s.check_perm_write(b2)
+
+ # Check for WriteACP
+ sNoCreds.check_perm_write_acl(b2)
+ s.check_perm_write_acl(b2)
+
+ # Grab permissions after our check so we can compare to original
+ sNoCreds.check_perm_write(b2)
+ s.check_perm_write(b2)
+ sNoCreds.check_perm_read(b2)
+ s.check_perm_read(b2)
+ if s.aws_creds_configured:
+ assert b2.AuthUsersWriteACP == Permission.ALLOWED
+
+ # Make sure we didn't change the original permissions
+ assert b2.AuthUsersWrite == Permission.ALLOWED
+ assert b2.AllUsersWrite == Permission.DENIED
+ assert b2.AllUsersRead == Permission.ALLOWED
+ assert b2.AuthUsersRead == Permission.UNKNOWN
+ else:
+ assert b2.AllUsersRead == Permission.ALLOWED
+ assert b2.AuthUsersWriteACP == Permission.UNKNOWN
+ except Exception as e:
+ raise e
+ finally:
+ ts.delete_bucket(danger_bucket_3)
+
+ # Bucket with WRITE_ACP enabled for AllUsers
+ danger_bucket_2 = ts.create_bucket(2)
+ try:
+ b3 = S3Bucket(danger_bucket_2)
+ b3.exists = BucketExists.YES
+ sNoCreds.check_perm_read(b3)
+ s.check_perm_read(b3)
+ sNoCreds.check_perm_write(b3)
+ s.check_perm_write(b3)
+ sNoCreds.check_perm_write_acl(b3)
+ s.check_perm_write_acl(b3)
+ sNoCreds.check_perm_write(b3)
+ s.check_perm_write(b3)
+ sNoCreds.check_perm_read(b3)
+ s.check_perm_read(b3)
+ if s.aws_creds_configured:
+ assert b3.AllUsersWriteACP == Permission.ALLOWED
+ assert b3.AuthUsersWriteACP == Permission.UNKNOWN
+ assert b3.AllUsersWrite == Permission.ALLOWED
+ else:
+ assert b3.AllUsersRead == Permission.ALLOWED
+ assert b3.AuthUsersWriteACP == Permission.UNKNOWN
+ except Exception as e:
+ raise e
+ finally:
+ ts.delete_bucket(danger_bucket_2)
+
+ # Bucket with WRITE_ACP enabled for both AllUsers and AuthUsers
+ danger_bucket_5 = ts.create_bucket(5)
+ try:
+ b5 = S3Bucket(danger_bucket_5)
+ b5.exists = BucketExists.YES
+ sNoCreds.check_perm_read(b5)
+ s.check_perm_read(b5)
+ sNoCreds.check_perm_write(b5)
+ s.check_perm_write(b5)
+ sNoCreds.check_perm_write_acl(b5)
+ s.check_perm_write_acl(b5)
+ sNoCreds.check_perm_write(b5)
+ s.check_perm_write(b5)
+ sNoCreds.check_perm_read(b5)
+ s.check_perm_read(b5)
+ assert b5.AllUsersWriteACP == Permission.ALLOWED
+ assert b5.AuthUsersWriteACP == Permission.UNKNOWN
+ assert b5.AllUsersWrite == Permission.DENIED
+ assert b5.AuthUsersWrite == Permission.DENIED
+ except Exception as e:
+ raise e
+ finally:
+ ts.delete_bucket(danger_bucket_5)
+ else:
+ print("[test_check_perm_write_acl] Skipping dangerous test...")
+
+
+def test_parse_found_acl():
+ test_setup_new()
+ sAnon = S3Service(forceNoCreds=True)
+
+ b1 = S3Bucket('s3scanner-all-read-readacl')
+ b1.exists = BucketExists.YES
+ sAnon.check_perm_read_acl(b1)
+
+ assert b1.foundACL is not None
+ assert b1.AllUsersRead == Permission.ALLOWED
+ assert b1.AllUsersReadACP == Permission.ALLOWED
+ assert b1.AllUsersWrite == Permission.DENIED
+ assert b1.AllUsersWriteACP == Permission.DENIED
+ assert b1.AllUsersFullControl == Permission.DENIED
+
+ assert b1.AuthUsersReadACP == Permission.DENIED
+ assert b1.AuthUsersRead == Permission.DENIED
+ assert b1.AuthUsersWrite == Permission.DENIED
+ assert b1.AuthUsersWriteACP == Permission.DENIED
+ assert b1.AuthUsersFullControl == Permission.DENIED
+
+ test_acls_1 = {
+ 'Grants': [
+ {
+ 'Grantee': {
+ 'Type': 'Group',
+ 'URI': 'http://acs.amazonaws.com/groups/global/AllUsers'
+ },
+ 'Permission': 'FULL_CONTROL'
+ }
+ ]
+ }
+
+ b2 = S3Bucket('test-acl-doesnt-exist')
+ b2.exists = BucketExists.YES
+ b2.foundACL = test_acls_1
+ sAnon.parse_found_acl(b2)
+ assert b2.AllUsersRead == Permission.ALLOWED
+ assert b2.AllUsersReadACP == Permission.ALLOWED
+ assert b2.AllUsersWrite == Permission.ALLOWED
+ assert b2.AllUsersWriteACP == Permission.ALLOWED
+ assert b2.AllUsersFullControl == Permission.ALLOWED
+ assert b2.AuthUsersRead == Permission.DENIED
+ assert b2.AuthUsersReadACP == Permission.DENIED
+ assert b2.AuthUsersWrite == Permission.DENIED
+ assert b2.AuthUsersWriteACP == Permission.DENIED
+ assert b2.AuthUsersFullControl == Permission.DENIED
+
+ test_acls_2 = {
+ 'Grants': [
+ {
+ 'Grantee': {
+ 'Type': 'Group',
+ 'URI': 'http://acs.amazonaws.com/groups/global/AuthenticatedUsers'
+ },
+ 'Permission': 'FULL_CONTROL'
+ }
+ ]
+ }
+
+ b3 = S3Bucket('test-acl2-doesnt-exist')
+ b3.exists = BucketExists.YES
+ b3.foundACL = test_acls_2
+ sAnon.parse_found_acl(b3)
+ assert b3.AllUsersRead == Permission.DENIED
+ assert b3.AllUsersReadACP == Permission.DENIED
+ assert b3.AllUsersWrite == Permission.DENIED
+ assert b3.AllUsersWriteACP == Permission.DENIED
+ assert b3.AllUsersFullControl == Permission.DENIED
+ assert b3.AuthUsersRead == Permission.ALLOWED
+ assert b3.AuthUsersReadACP == Permission.ALLOWED
+ assert b3.AuthUsersWrite == Permission.ALLOWED
+ assert b3.AuthUsersWriteACP == Permission.ALLOWED
+ assert b3.AuthUsersFullControl == Permission.ALLOWED
+
+ test_acls_3 = {
+ 'Grants': [
+ {
+ 'Grantee': {
+ 'Type': 'Group',
+ 'URI': 'asdfasdf'
+ },
+ 'Permission': 'READ'
+ }
+ ]
+ }
+
+ b4 = S3Bucket('test-acl3-doesnt-exist')
+ b4.exists = BucketExists.YES
+ b4.foundACL = test_acls_3
+ sAnon.parse_found_acl(b4)
+
+ all_permissions = [b4.AllUsersRead, b4.AllUsersReadACP, b4.AllUsersWrite, b4.AllUsersWriteACP,
+ b4.AllUsersFullControl, b4.AuthUsersRead, b4.AuthUsersReadACP, b4.AuthUsersWrite,
+ b4.AuthUsersWriteACP, b4.AuthUsersFullControl]
+
+ for p in all_permissions:
+ assert p == Permission.DENIED
+
+ test_acls_4 = {
+ 'Grants': [
+ {
+ 'Grantee': {
+ 'Type': 'Group',
+ 'URI': 'http://acs.amazonaws.com/groups/global/AuthenticatedUsers'
+ },
+ 'Permission': 'READ_ACP'
+ },
+ {
+ 'Grantee': {
+ 'Type': 'Group',
+ 'URI': 'http://acs.amazonaws.com/groups/global/AllUsers'
+ },
+ 'Permission': 'READ_ACP'
+ }
+ ]
+ }
+
+ b5 = S3Bucket('test-acl4-doesnt-exist')
+ b5.exists = BucketExists.YES
+ b5.foundACL = test_acls_4
+ sAnon.parse_found_acl(b5)
+ assert b5.AllUsersRead == Permission.DENIED
+ assert b5.AllUsersReadACP == Permission.ALLOWED
+ assert b5.AllUsersWrite == Permission.DENIED
+ assert b5.AllUsersWriteACP == Permission.DENIED
+ assert b5.AllUsersFullControl == Permission.DENIED
+ assert b5.AuthUsersRead == Permission.DENIED
+ assert b5.AuthUsersReadACP == Permission.ALLOWED
+ assert b5.AuthUsersWrite == Permission.DENIED
+ assert b5.AuthUsersWriteACP == Permission.DENIED
+ assert b5.AuthUsersFullControl == Permission.DENIED
+
+
+def test_check_perms_without_checking_bucket_exists():
+ test_setup_new()
+ sAnon = S3Service(forceNoCreds=True)
+
+ b1 = S3Bucket('blahblah')
+ with pytest.raises(BucketMightNotExistException):
+ sAnon.check_perm_read_acl(b1)
+
+ with pytest.raises(BucketMightNotExistException):
+ sAnon.check_perm_read(b1)
+
+ with pytest.raises(BucketMightNotExistException):
+ sAnon.check_perm_write(b1)
+
+ with pytest.raises(BucketMightNotExistException):
+ sAnon.check_perm_write_acl(b1)
+
+
+def test_no_ssl():
+ test_setup_new()
+ S3Service(verify_ssl=False)
+
+
+def test_download_file():
+ test_setup_new()
+ s = S3Service()
+
+ # Try to download a file that already exists
+ dest_folder = os.path.realpath(testingFolder)
+ Path(os.path.join(dest_folder, 'test_download_file.txt')).touch()
+ size = Path(os.path.join(dest_folder, 'test_download_file.txt')).stat().st_size
+
+ o = S3BucketObject(size=size, last_modified="2020-12-31_03-02-11z", key="test_download_file.txt")
+
+ b = S3Bucket("bucket-no-existo")
+ s.download_file(os.path.join(dest_folder, ''), b, True, o)