Boto3 Check Framework

checks.py

from utils import Checker, CheckError
from botocore.exceptions import ClientError
import requests


class AnonChecks(Checker):

    def check_anonymous_http_get(self):
        """ 1 """
        r = requests.get("https://{}.s3.amazonaws.com/".format(self.bucket.name))
        if r.status_code not in (403, 404):
            raise CheckError("Anonymous HTTP GET succeeded, expected 403")

    def check_list_bucket(self):
        """ 4 """
        try:
            self.client.list_objects(Bucket=self.bucket.name)
        except ClientError as e:
            response = e.response["Error"]["Code"]
            if response in ("AccessDenied",):
                return
            raise CheckError("Response {} was not expected".format(response))
        raise CheckError("Did not cause ClientError")

    def check_put(self):
        """ 3, 5 """
        try:
            self.client.put_object(Bucket=self.bucket.name, Key="s3_ACL_test", Body=b"RandomData")
        except ClientError as e:
            response = e.response["Error"]["Code"]
            if response in ("AccessDenied",):
                return
            raise CheckError("Response {} was not expected".format(response))
        raise CheckError("Did not cause ClientError")


class AuthChecks(Checker):
    def check_authn_write(self):
        """ 2 """
        try:
            self.client.put_object(Bucket=self.bucket.name, Key="s3_ACL_test", Body=b"RandomData")
        except ClientError as e:
            response = e.response["Error"]["Code"]
            if response in ("AccessDenied",):
                return
            raise CheckError("Response {} was not expected".format(response))
        raise CheckError("Did not cause ClientError")

utils.py

import logging
import boto3
from collections import namedtuple


class Bucket(namedtuple("Bucket", ("description", "vendor", "name"))):
    def __str__(self):
        return "{};{};{}".format(self.description, self.vendor, self.name)


class CheckError(Exception):
    """ Custom Exception for catching test failures """
    pass


class Checker(object):
    """ Abstract class for collections of checks """
    def __init__(self, bucket: Bucket, client: boto3.client):
        self.log = logging.getLogger(str(bucket))
        self.bucket = bucket
        self.client = client

    def __call__(self, *args, **kwargs):
        passes = 0
        errors = 0
        for method in dir(self):
            if method.startswith("check_") and callable(getattr(self, method)):
                try:
                    getattr(self, method)()
                    self.log.info("{}: Ok".format(method))
                    passes += 1
                except CheckError as e:
                    self.log.error("{}: {}".format(method, e))
                    errors += 1

        if errors == 0:
            self.log.info("Checks passed ({0}/{0})".format(passes))
        else:
            self.log.error("Checks failed ({0}/{1})".format(errors, errors + passes))