Authentication API

curl -H "Authorization: Bearer {PAT}" -X GET https://api.optimizely.com/v2/export/credentials\?duration\=1h

{
 "credentials": {
   "accessKeyId": "access_key_id",
   "secretAccessKey": "access_secret_key",
   "sessionToken": "session_token",
   "expiration": 1591026407394
 },
 "s3Path": "s3://optimizely-events-data/v1/account_id=123"
}

#!/usr/bin/env python3

from botocore.credentials import RefreshableCredentials
from botocore.session import get_session
from datetime import datetime
import requests as r
import boto3
import pytz
import sys


class BearerAuth(r.auth.AuthBase):
    """
    Bearer authentication class that integrates with requests'
    native authentication mechanism.
    """

    def __init__(self, token):
        """
        Args:
            token (str): Bearer token used to authenticate requests to an API
        """
        self.token = token

    def __call__(self, request):
        """Sets the Authorization header on the current HTTP request.
        Args:
            request (request): The current HTTP request object.
        Returns:
            request: The current HTTP request with additional
                     Authorization header
        """
        request.headers['Authorization'] = f"Bearer {self.token}"
        return request


class OptimizelyS3Client(object):
    """
    AWS S3 client using credentials from Optimizely
    """

    EXPORT_CREDENTIALS_URL = 'https://api.optimizely.com/v2/export/credentials'
    CREDENTIALS_IDENTIFIER = 'Optimizely'
    EVENT_EXPORT_BUCKET_REGION = 'us-east-1'

    def __init__(self, token: str):
        """
        Args:
            token (str): Optimizely personal access token
        """
        self.token = token

    def get_bucket_prefix(self, account_id: str) -> str:
        """Get the bucket prefix for a given Optimizely account ID
        Args:
            account_id (str): Optimizely account ID
        Returns:
            str: The S3 bucket prefix to access events data
        """
        return f"v1/account_id={account_id}/"

    def get_creds(self) -> dict:
        """Get AWS credentials from the Optimizely public API
        Returns:
            dict: The AWS credentials for a given Optimizely account
        """
        response = r.get(
            OptimizelyS3Client.EXPORT_CREDENTIALS_URL,
            auth=BearerAuth(self.token)
        )

        node = response.json()
        return node['credentials']

    def as_boto3_s3_client(self):
        """
        Convert the Optimizely S3 client to a standard boto s3
        client with refreshable credentials
        Returns:
            botocore.client.S3: Standard boto3 S3 client
        """
        creds = self.get_creds()

        # The API response is in milliseconds
        expiry_time = int(creds['expiration'] / 1000)

        # boto expects the expiry time to be a UTC datetime
        expiry_time = datetime.fromtimestamp(expiry_time, pytz.utc)

        opz_refreshable_credentials = RefreshableCredentials(
            creds['accessKeyId'],
            creds['secretAccessKey'],
            creds['sessionToken'],
            expiry_time,
            self.get_creds,
            OptimizelyS3Client.CREDENTIALS_IDENTIFIER
        )

        session = get_session()
        session._credentials = opz_refreshable_credentials
        session.set_config_variable('region', OptimizelyS3Client.EVENT_EXPORT_BUCKET_REGION)
        opz_session = boto3.Session(botocore_session=session)
        s3_client = opz_session.client('s3')

        return s3_client


def main():
    EVENT_EXPORT_BUCKET_NAME = 'optimizely-events-data'
    OPTIMIZELY_ACCOUNT_ID = '8506653257'
    OPTIMIZELY_PERSONAL_ACCESS_TOKEN = '2:4gLnqLABma8j2B4hyJ1ssvRJNvDR_YX1LfNjwS742PNKgOs0GNcU'

    optimizely_s3_client = OptimizelyS3Client(OPTIMIZELY_PERSONAL_ACCESS_TOKEN)
    s3_client = optimizely_s3_client.as_boto3_s3_client()

    all_objects = s3_client.list_objects(
        Bucket=EVENT_EXPORT_BUCKET_NAME,
        Prefix=optimizely_s3_client.get_bucket_prefix(OPTIMIZELY_ACCOUNT_ID)
    )
    print(all_objects)


if __name__ == '__main__':
    main()