Skip to content

Instantly share code, notes, and snippets.

@massenz
Last active December 24, 2017 15:15
Show Gist options
  • Save massenz/1c218b549039e50deaf9 to your computer and use it in GitHub Desktop.
Save massenz/1c218b549039e50deaf9 to your computer and use it in GitHub Desktop.
Simple Flask server to demonstrate how to use LaunchKey (http://www.launchkey.com) to authenticate users
# Copyright AlertAvert.com (c) 2013. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import json
from flask import Flask, make_response, jsonify
import time
import launchkey
import logging
FORMAT = '%(asctime)-15s [%(levelname)s] %(message)s'
DATE_FMT = '%m/%d/%Y %H:%M:%S'
# TODO: read from config.yaml instead
app_key = 1234567890
secret_key = "1d0n7c4r3wh4u7h1nk0fm3"
private_key = None
def parse_args():
""" Parse command line arguments and returns a configuration object
"""
parser = argparse.ArgumentParser()
parser.add_argument('-f', '--key-file', required=True,
help="The file containing the Private Key")
parser.add_argument('-p', '--port', help="The port for the server to listen on",
default=5050)
parser.add_argument('-v', '--verbose', action='store_true', help='Enables debug logging')
return parser.parse_args()
app = Flask(__name__)
def _launchkey_private_key(config):
# TODO: globals are evil! don't do this in real code
global private_key
if not private_key:
if not config:
raise RuntimeError("Must seed the private key first, passing in the config object")
with open(config.key_file) as key_file:
private_key = key_file.read()
return private_key
def get_api_key(username):
# TODO: this would of course look this up in a backing DB
return '_'.join(['deadfad', username, 'feedbeef'])
class NotAuthorized(Exception):
status_code = 401
def __init__(self, message, status_code=None, payload=None):
Exception.__init__(self)
self.message = message
if status_code is not None:
self.status_code = status_code
self.payload = payload
def to_dict(self):
rv = dict(self.payload or ())
rv['message'] = self.message
return rv
def auth(username):
api = launchkey.API(app_key, secret_key, private_key)
auth_request = api.authorize(username, True)
auth_response = {}
# TODO: make this configurable (config.yaml)
max_retries = 10
delay_sec = 15
while not 'auth' in auth_response and max_retries > 0:
auth_response = api.poll_request(auth_request)
if 'message' in auth_response:
logging.info('Waiting for {username} to authenticate: {msg}'.format(
username=username, msg=auth_response.get('message')
))
if max_retries > 0:
max_retries -= 1
time.sleep(delay_sec)
auth_hash = auth_response.get('auth')
if not auth_hash or not api.is_authorized(auth_request, auth_hash):
raise NotAuthorized('Could not authenticate {username}'.format(username=username))
@app.route('/login/<username>')
def login(username):
logging.info('Logging in {username}'.format(username=username))
auth(username)
logging.info('{username} successfully authenticated'.format(username=username))
response = {
"username": username,
"api_key": get_api_key(username)
}
return make_response(json.dumps(response))
@app.errorhandler(NotAuthorized)
def handle_invalid_usage(error):
logging.error(error.message)
response = jsonify(error.to_dict())
response.status_code = error.status_code
return response
def run_server():
config = parse_args()
loglevel = logging.DEBUG if config.verbose else logging.INFO
logging.basicConfig(format=FORMAT, datefmt=DATE_FMT, level=loglevel)
# Initialize the private key global
_launchkey_private_key(config)
app.run(port=config.port, debug=config.verbose)
if __name__ == '__main__':
run_server()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment