controllers.py 文件源码

def presign(auth_token, url, ownerid=None):
    """Generates S3 presigned URLs if necessary
    :param auth_token: authentication token from auth
    :param ownerid: ownerid for dataset
    :param url: url to check for sigend URL
    """
    s3 = get_s3_client()
    try:
        needs_signed_url = requests.head(url)
        if needs_signed_url.status_code != 403:
            return json.dumps({'url': url})
        # Verify client, deny access if not verified
        if ownerid is None:
            return Response(status=401)
        if not services.verify(auth_token, ownerid):
            return Response(status=403)
        parsed_url = urllib.parse.urlparse(url)
        bucket = parsed_url.netloc
        key = parsed_url.path.lstrip('/')
        # Handle s3 path-style URLs
        if bucket.endswith('amazonaws.com'):
            bucket, key = key.split('/', 1)

        # Make sure file belongs to user (only in case of pkgstore)
        if (config['STORAGE_BUCKET_NAME'] != bucket) and (ownerid not in url):
            return Response(status=403)

        signed_url = s3.generate_presigned_url(
            ClientMethod='get_object',
            Params={
                'Bucket': bucket,
                'Key': key
            },
            ExpiresIn=3600*24)
        return json.dumps({'url': signed_url})
    except Exception as exception:
        logging.exception('Bad request')
        return Response(status=400)