twitch.py

from __future__ import print_function
import base64
import json
import logging
import urllib
import boto3
from flask import Flask, redirect, session, url_for, jsonify
from flask_oauthlib.client import OAuth
import requests
ssm = boto3.client('ssm')

logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

client_params = ssm.get_parameters(
    Names=['TwitchAccess', 'TwitchSecret']
)['Parameters']

app = Flask(__name__)
oauth = OAuth(app)
twitch = oauth.remote_app(
    'twitch',
    base_url='https://api.twitch.tv/kraken/',
    authorize_url='https://api.twitch.tv/kraken/oauth2/authorize',
    access_token_url='https://api.twitch.tv/kraken/oauth2/token',
    access_token_method='POST',
    request_token_params={'scope': 'user_read'},
    consumer_key=client_params[0]['Value'],
    consumer_secret=client_params[1]['Value']
)
app.config['SECRET_KEY'] = client_params[1]['Value']

#ddb = boto3.resource('dynamodb').Table('twitch_auth')
sts = boto3.client('sts')


def include_consumer_key(uri, headers, body):
    auth = headers.get('Authorization')
    if auth and 'oauth2' not in uri:
        headers['Authorization'] = auth.replace('Bearer', 'OAuth')
        headers['Client-ID'] = client_params[0]['Value']
    return uri, headers, body

twitch.pre_request = include_consumer_key


@twitch.tokengetter
def get_twitch_oauth_token():
    return session.get('twitch_token')


@app.route('/twitch_login')
def login():
    return twitch.authorize(callback=url_for('authorized', _external=True))


@app.route('/twitch_login/authorized')
def authorized():
    try:
        resp = twitch.authorized_response()
    except Exception as ex:
        logger.info(ex)
        return jsonify(ex.data), 401
    if resp is None or resp.get('access_token') is None:
        return 'Access Denied', 403
    session['twitch_token'] = (resp['access_token'], '')
    user = twitch.get('user').data
    assume_role_resp = sts.assume_role(
        RoleArn='arn:aws:iam::309575541351:role/twitch-plays',
        RoleSessionName=user['display_name'],
        ExternalId=str(user['_id']),
        DurationSeconds=900  # minimum range
    )
    creds = {
        'sessionId': assume_role_resp['Credentials']['AccessKeyId'],
        'sessionKey': assume_role_resp['Credentials']['SecretAccessKey'],
        'sessionToken': assume_role_resp['Credentials']['SessionToken']
    }
    params = {
        "Action": "getSigninToken",
        "Session": json.dumps(creds)
    }
    auth_url = "https://signin.aws.amazon.com/federation"
    resp = requests.get(auth_url, params=params)
    signin_token = resp.json().get('SigninToken')
    params = {
        'Action': 'login',
        'Issuer': 'twitch.tv',
        'Destination': 'https://console.aws.amazon.com/',
        'SigninToken': signin_token
    }
    url = requests.Request('GET', auth_url, params=params).prepare().url
    return redirect(url)


if __name__ == '__main__':
    app.run(debug=True)