Skip to content

Commit

Permalink
Delete unused S3AccessPolicy class from api.aws module
Browse files Browse the repository at this point in the history
  • Loading branch information
Josh Rowe authored and Josh Rowe committed Oct 1, 2019
1 parent 59a4396 commit 5e3bb3a
Show file tree
Hide file tree
Showing 2 changed files with 0 additions and 300 deletions.
95 changes: 0 additions & 95 deletions controlpanel/api/aws.py
Original file line number Diff line number Diff line change
Expand Up @@ -269,98 +269,3 @@ def list_role_names(self, prefix="/"):


aws = AWSClient()


class S3AccessPolicy(object):
def __init__(self, document=None):
self._readonly_arns = set()
self._readwrite_arns = set()

if document:
self._load(document)

def revoke_access(self, bucket_arn):
self._readonly_arns.discard(bucket_arn)
self._readwrite_arns.discard(bucket_arn)

def grant_access(self, bucket_arn, readwrite=False):
self.revoke_access(bucket_arn)

if readwrite:
self._readwrite_arns.add(bucket_arn)
else:
self._readonly_arns.add(bucket_arn)

@property
def document(self):
statements = [
{
"Sid": "console",
"Effect": "Allow",
"Action": ["s3:GetBucketLocation", "s3:ListAllMyBuckets"],
"Resource": "arn:aws:s3:::*",
}
]

all_buckets_arns = self._readonly_arns | self._readwrite_arns
if all_buckets_arns:
statements.append(self._list_statement(all_buckets_arns))

if self._readonly_arns:
statements.append(self._readonly_statement)

if self._readwrite_arns:
statements.append(self._readwrite_statement)

return {"Version": "2012-10-17", "Statement": statements}

def _list_statement(self, all_buckets_arns):
return {
"Sid": "list",
"Action": ["s3:ListBucket"],
"Effect": "Allow",
"Resource": list(all_buckets_arns),
}

@property
def _readonly_statement(self):
return {
"Sid": "readonly",
"Action": ["s3:GetObject", "s3:GetObjectAcl", "s3:GetObjectVersion"],
"Effect": "Allow",
"Resource": self._s3_objects_arns(self._readonly_arns),
}

@property
def _readwrite_statement(self):
return {
"Sid": "readwrite",
"Action": [
"s3:GetObject",
"s3:GetObjectAcl",
"s3:GetObjectVersion",
"s3:DeleteObject",
"s3:DeleteObjectVersion",
"s3:PutObject",
"s3:PutObjectAcl",
"s3:RestoreObject",
],
"Effect": "Allow",
"Resource": self._s3_objects_arns(self._readwrite_arns),
}

def _s3_objects_arns(self, arns):
return [f"{arn}/*" for arn in arns]

def _s3_buckets_arns(self, arns):
return [arn.rsplit("/*", 1)[0] for arn in arns]

def _load(self, document):
for statement in document["Statement"]:
sid = statement["Sid"]
if sid in ("readonly", "readwrite"):
arns = set(self._s3_buckets_arns(statement["Resource"]))
if sid == "readwrite":
self._readwrite_arns = arns
else:
self._readonly_arns = arns
205 changes: 0 additions & 205 deletions tests/api/test_aws_s3_access_policy.py

This file was deleted.

0 comments on commit 5e3bb3a

Please sign in to comment.