vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Change cloudfront storage to report non-cloudfront (i.e. S3) URLs for all requests missing IP information

Browse source 
This ensures the security scanner gets S3 URLs

Fixes https://jira.coreos.com/browse/QUAY-954
This commit is contained in:
Joseph Schorr 2018-07-05 16:16:20 +03:00
parent 100d7eae81
commit 7426f9e93a
2 changed files with 11 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
storage/cloud.py
View file

@ -615,22 +615,22 @@ class CloudFrontedS3Storage(S3Storage):
def get_direct_download_url(self, path, request_ip=None, expires_in=60, requires_cors=False,
head=False):
# If CloudFront could not be loaded, fall back to normal S3.
if self.cloudfront_privatekey is None:
if self.cloudfront_privatekey is None or request_ip is None:
return super(CloudFrontedS3Storage, self).get_direct_download_url(path, request_ip,
expires_in, requires_cors,
head)
resolved_ip_info = None
logger.debug('Got direct download request for path "%s" with IP "%s"', path, request_ip)
if request_ip is not None:
# Lookup the IP address in our resolution table and determine whether it is under AWS.
# If it is, then return an S3 signed URL, since we are in-network.
resolved_ip_info = self._context.ip_resolver.resolve_ip(request_ip)
logger.debug('Resolved IP information for IP %s: %s', request_ip, resolved_ip_info)
if resolved_ip_info and resolved_ip_info.provider == 'aws':
return super(CloudFrontedS3Storage, self).get_direct_download_url(path, request_ip,
expires_in, requires_cors,
head)
# Lookup the IP address in our resolution table and determine whether it is under AWS.
# If it is, then return an S3 signed URL, since we are in-network.
resolved_ip_info = self._context.ip_resolver.resolve_ip(request_ip)
logger.debug('Resolved IP information for IP %s: %s', request_ip, resolved_ip_info)
if resolved_ip_info and resolved_ip_info.provider == 'aws':
return super(CloudFrontedS3Storage, self).get_direct_download_url(path, request_ip,
expires_in, requires_cors,
head)
url = 'https://%s/%s' % (self.cloudfront_distribution_domain, path)
expire_date = datetime.now() + timedelta(seconds=expires_in)

3
storage/test/test_cloudfront.py
View file

@ -64,5 +64,4 @@ def test_direct_download_no_ip(test_aws_ip, aws_ip_range_data, ipranges_populate
_TEST_BUCKET, _TEST_USER, _TEST_PASSWORD)
engine.put_content(_TEST_PATH, _TEST_CONTENT)
assert engine.exists(_TEST_PATH)
assert 'cloudfrontdomain' in engine.get_direct_download_url(_TEST_PATH)
assert 's3.amazonaws.com' in engine.get_direct_download_url(_TEST_PATH)