#!/usr/bin/python # -*- coding: UTF-8 -*- from cStringIO import StringIO from datetime import datetime from datetime import timedelta from email.mime.text import MIMEText from threading import Thread from uuid import uuid4 import base64 import functools import hashlib import json import os import smtplib from flask import Flask from flask import jsonify from flask import render_template, send_from_directory from flask import request from flask import send_file from flask import session from flask import url_for from flask_compress import Compress from flask_session import Session from pyotp import random_base32 from werkzeug import redirect import pyotp import pytz import qrcode import webauthn from functions import pbkdf2 from functions import random_data from functions import render_jinja_html import config ################################################################################ app = Flask( __name__ ) app.config.from_pyfile( 'config.py' ) Compress( app ) Session( app ) debug = os.environ.get( "DEBUG", False ) baseurl = os.environ.get( "BASEURL", "" ) ################################################################################ # Loggin decorator def session_field_required( field, value ): def decorator( func ): @functools.wraps( func ) def wrapper_login_required( *args, **kwargs ): if not field in session: return redirect( url_for( "login" ) ) elif not session.get( field ) == value: return redirect( url_for( "login" ) ) return func( *args, **kwargs ) return wrapper_login_required return decorator def login_required( func ): @functools.wraps( func ) def wrapper_login_required( *args, **kwargs ): if not session.get( 'logged', False ) : return redirect( url_for( "login" ) ) return func( *args, **kwargs ) return wrapper_login_required def admin_required( func ): @functools.wraps( func ) def wrapper_login_required( *args, **kwargs ): if not session.get( 'logged', False ) or not session.get( 'account_type', None ) == 1: return redirect( url_for( "login" ) ) return func( *args, **kwargs ) return wrapper_login_required ################################################################################ # Generic routing @app.route( baseurl + '/ping' ) def ping(): return "pong" ################################################################################ # App serving @app.route( baseurl + '/app/' ) def send_app_files( path ): return send_from_directory( 'app', path ) @app.route( baseurl + '/static/' ) def send_static_files( path ): return send_from_directory( 'static', path ) ################################################################################ # Sessions @app.before_request def renew_session(): session.permanent = True app.permanent_session_lifetime = timedelta( seconds = config.session_timeout ) @app.route( baseurl + '/logout' ) def logout(): session.clear() return redirect( url_for( 'home' ) ) @app.route( baseurl + '/login' ) def login(): session.clear() session[ 'process' ] = "login" session[ 'need_to_check' ] = [ 'password' ] session[ 'logged' ] = False return render_template( "login.html", baseurl = baseurl, js = config.cdnjs, css = config.cdncss ) @app.route( baseurl + '/do_login', methods = [ 'POST' ] ) @session_field_required( "process", "login" ) def do_login(): need_to_check = session.get( "need_to_check", [ 'password' ] ) try: current_check = need_to_check[ 0 ] except: current_check = None session[ 'need_to_check' ] = need_to_check ############################################################################ if current_check == "password": q = config.db.query( 'SELECT * FROM users WHERE username = %s', ( request.form.get( "username" ), ) ) user = q.fetchone() if user == None: session[ 'logged' ] = False return jsonify( { 'error': False, 'logged': False } ) form_password = request.form.get( "password", None ) if form_password == None or not pbkdf2( form_password, user[ 'password' ] ): session[ 'logged' ] = False return jsonify( { 'error': False, 'logged': False, } ) elif not user[ 'active' ]: session[ 'logged' ] = False return jsonify( { 'error': False, 'logged': False, 'message': 'Your account is not activated. Please contact an administrator.' } ) else: session[ 'session_id' ] = str( uuid4() ) session[ 'username' ] = user[ 'username' ] session[ 'user_id' ] = user[ 'id' ] session[ 'password_check' ] = True session[ 'need_to_check' ].remove( current_check ) if user[ 'must_use_totp' ]: session[ 'need_to_check' ].append( 'totp' ) if user[ 'must_use_securitykey' ]: session[ 'need_to_check' ].append( 'securitykey' ) elif current_check == 'totp': q = config.db.query( 'SELECT username, totp FROM users WHERE username = %s', ( session[ 'username' ], ) ) user = q.fetchone() if not pyotp.TOTP( user[ 'totp' ] ).verify( request.form[ "totp" ], valid_window = 1 ): session[ 'logged' ] = False return jsonify( { 'error': False, 'logged': False, 'message': 'Wrong TOTP' } ) else: session[ 'need_to_check' ].remove( current_check ) if len( session[ 'need_to_check' ] ) == 0 and session.get( "password_check", False ): for key in [ 'process', 'need_to_check', 'password_check' ]: session.pop( key ) session[ 'logged' ] = True q = config.db.query( 'SELECT type FROM users WHERE username = %s', ( session[ 'username' ], ) ) user = q.fetchone() session[ 'account_type' ] = user[ 'type' ] return jsonify( { 'error': False, 'logged': True, } ) else: return jsonify( { 'error': False, 'next_step': session[ 'need_to_check' ][ 0 ] } ) ################################################################################ # Reset @app.route( baseurl + '/reset_password' ) def password_reset(): session.clear() session[ 'process' ] = "request_password_reset" return render_template( "password_reset.html", baseurl = baseurl, js = config.cdnjs, css = config.cdncss ) @app.route( baseurl + '/do_reset_password', methods = [ 'POST' ] ) def do_password_reset(): email = request.form.get( "email", None ) Thread( target = do_password_reset_thread, args = ( email, ) ).start() return jsonify( { 'error': False, 'message': 'OK' } ) def do_password_reset_thread( email ): q = config.db.query( 'SELECT id, username, email FROM users' ) users = q.fetchall() for user in users: if not user[ 'email' ].startswith( "pbkdf2$" ): continue elif pbkdf2( email, user[ 'email' ] ): id = hashlib.sha512( random_data( 100 ) ).hexdigest() #################################################################### data = { 'process': 'password_reset', 'process_id': id, 'user_id': user[ 'id' ] } data = json.dumps( data ) data = base64.b64encode( data ) config.redis_shared.set( "reset_" + id, data, ex = 24 * 3600 ) #################################################################### email_content = render_jinja_html( "templates/email", "reset.html", id = id, url = config.domain + baseurl + "/reset_password_stage2" ) msg = MIMEText( email_content, "html" ) msg[ 'Subject' ] = 'ICNML - User password reset' msg[ 'From' ] = config.sender msg[ 'To' ] = email s = smtplib.SMTP( config.smtpserver ) s.sendmail( config.sender, [ email ], msg.as_string() ) s.quit() break else: print email + " not found" @app.route( baseurl + '/reset_password_stage2/', methods = [ 'GET', 'POST' ] ) def password_reset_stage2( id ): id = str( id ) data = config.redis_shared.get( "reset_" + id ) if data != None: data = base64.b64decode( data ) data = json.loads( data ) password = request.form.get( "password", None ) userid = data.get( "user_id", None ) if password != None: password = pbkdf2( password, random_data( 50 ), 50000 ) config.db.query( "UPDATE users SET password = %s WHERE id = %s", ( password, userid ) ) config.db.commit() config.redis_shared.delete( "reset_" + id ) return jsonify( { 'error': False, 'password_updated': True } ) else: return render_template( "password_reset_stage2.html", baseurl = baseurl, id = id, js = config.cdnjs, css = config.cdncss ) else: return jsonify( { 'error': True, 'message': 'Reset procedure not found/expired' } ) ################################################################################ # Webauthn @app.route( baseurl + '/webauthn_key_management' ) @login_required def webauthn_key_management(): return render_template( "webauthn_key_management.html", baseurl = baseurl, js = config.cdnjs, css = config.cdncss, keys = do_webauthn_get_list_of_keys( all = True ) ) @app.route( baseurl + '/webauthn_begin_activate', methods = [ 'POST' ] ) @login_required def webauthn_begin_activate(): session[ 'key_name' ] = request.form.get( "key_name", None ) username = session.get( "username" ) challenge = random_base32( 64 ) ukey = random_base32( 64 ) session[ 'challenge' ] = challenge session[ 'register_ukey' ] = ukey make_credential_options = webauthn.WebAuthnMakeCredentialOptions( challenge, config.rp_name, config.RP_ID, ukey, username, username, None ) return jsonify( make_credential_options.registration_dict ) @app.route( baseurl + '/verify_credential_info', methods = [ 'POST' ] ) @login_required def verify_credential_info(): challenge = session[ 'challenge' ] user_id = session[ 'user_id' ] key_name = session.get( "key_name", None ) ukey = session[ 'register_ukey' ] response = request.form webauthn_registration_response = webauthn.WebAuthnRegistrationResponse( config.RP_ID, config.ORIGIN, response, challenge ) try: webauthn_credential = webauthn_registration_response.verify() except Exception as e: return jsonify( { 'error': True, 'message': 'Registration failed. Error: {}'.format( e ) } ) config.db.query( """ INSERT INTO webauthn ( user_id, key_name, ukey, credential_id, pub_key, sign_count ) VALUES ( %s, %s, %s, %s, %s, %s ) """, ( user_id, key_name, ukey, webauthn_credential.credential_id, webauthn_credential.public_key, webauthn_credential.sign_count, ) ) config.db.commit() return jsonify( { 'success': 'User successfully registered.' } ) ################################################################################ def do_webauthn_get_list_of_keys( uid = None, all = False ): user_id = session.get( "user_id", uid ) sql = "SELECT id, key_name as name, created_on, last_usage, usage_counter, active FROM webauthn WHERE user_id = %s" if not all: sql += " AND active = true" sql += " ORDER BY sign_count DESC" q = config.db.query( sql, ( user_id, ) ) keys = q.fetchall() data = [] for key in keys: data.append( dict( key ) ) return data @app.route( baseurl + '/webauthn_get_list_of_keys', methods = [ 'POST' ] ) def webauthn_get_list_of_keys(): return jsonify( { 'error': False, 'data': do_webauthn_get_list_of_keys() } ) @app.route( baseurl + '/webauthn_delete_key', methods = [ 'POST' ] ) @login_required def webauthn_delete_key(): key_id = request.form.get( "key_id", False ) key_name = request.form.get( "key_name", False ) userid = session[ 'user_id' ] try: config.db.query( "DELETE FROM webauthn WHERE id = %s AND key_name = %s AND user_id = %s", ( key_id, key_name, userid, ) ) config.db.commit() return jsonify( { 'error': False } ) except Exception as e: return jsonify( { 'error': True } ); @app.route( baseurl + '/webauthn_disable_key', methods = [ 'POST' ] ) @login_required def webauthn_disable_key(): key_id = request.form.get( "key_id", False ) key_name = request.form.get( "key_name", False ) userid = session[ 'user_id' ] try: config.db.query( "UPDATE webauthn SET active = False WHERE id = %s AND key_name = %s AND user_id = %s", ( key_id, key_name, userid, ) ) config.db.commit() return jsonify( { 'error': False } ) except Exception as e: return jsonify( { 'error': True } ); @app.route( baseurl + '/webauthn_enable_key', methods = [ 'POST' ] ) @login_required def webauthn_enable_key(): key_id = request.form.get( "key_id", False ) key_name = request.form.get( "key_name", False ) userid = session[ 'user_id' ] try: config.db.query( "UPDATE webauthn SET active = True WHERE id = %s AND key_name = %s AND user_id = %s", ( key_id, key_name, userid, ) ) config.db.commit() return jsonify( { 'error': False } ) except Exception as e: return jsonify( { 'error': True } ); @app.route( baseurl + '/webauthn_begin_assertion', methods = [ 'POST' ] ) def webauthn_begin_assertion(): key_name = request.form.get( "key_name", None ) user_id = session.get( "user_id" ) if 'challenge' in session: del session[ 'challenge' ] challenge = random_base32( 64 ) session[ 'challenge' ] = challenge q = config.db.query( "SELECT * FROM webauthn WHERE user_id = %s AND key_name = %s", ( user_id, key_name ) ) key = q.fetchone() webauthn_user = webauthn.WebAuthnUser( key[ 'ukey' ], session[ 'username' ], session[ 'username' ], None, key[ 'credential_id' ], key[ 'pub_key' ], key[ 'sign_count' ], config.RP_ID ) webauthn_assertion_options = webauthn.WebAuthnAssertionOptions( webauthn_user, challenge ) return jsonify( { 'error': False, 'data': webauthn_assertion_options.assertion_dict } ) @app.route( baseurl + '/verify_assertion', methods = [ 'POST' ] ) def verify_assertion(): challenge = session.get( 'challenge' ) assertion_response = request.form credential_id = assertion_response.get( 'id' ) q = config.db.query( "SELECT * FROM webauthn WHERE credential_id = %s", ( credential_id, ) ) user = q.fetchone() webauthn_user = webauthn.WebAuthnUser( user[ 'ukey' ], session[ 'username' ], session[ 'username' ], None, user[ 'credential_id' ], user[ 'pub_key' ], user[ 'sign_count' ], "icnml.unil.ch" ) webauthn_assertion_response = webauthn.WebAuthnAssertionResponse( webauthn_user, assertion_response, challenge, config.ORIGIN, uv_required = False ) try: sign_count = webauthn_assertion_response.verify() except Exception as e: return jsonify( { 'error': True, 'message': 'Assertion failed. Error: {}'.format( e ) } ) dt = datetime.now( pytz.timezone( 'Europe/Zurich' ) ) q = config.db.query( "UPDATE webauthn SET sign_count = %s, last_usage = %s, usage_counter = usage_counter + 1 WHERE credential_id = %s", ( sign_count, dt, credential_id, ) ) config.db.commit() session[ 'need_to_check' ].remove( "securitykey" ) do_login() return jsonify( { 'error': False } ) ################################################################################ # New user @app.route( baseurl + '/signin' ) def new_user(): q = config.db.query( "SELECT id, name FROM account_type WHERE can_singin = true" ) r = q.fetchall() account_type = [] for rr in r: account_type.append( dict( rr ) ) return render_template( "signin.html", baseurl = baseurl, list_account_type = account_type, js = config.cdnjs, css = config.cdncss ) @app.route( baseurl + '/do_signin', methods = [ 'POST' ] ) def add_account_request_to_db(): try: first_name = request.form[ 'first_name' ] last_name = request.form[ 'last_name' ] email = request.form[ 'email' ] account_type = request.form[ 'account_type' ] uuid = str( uuid4() ) config.db.query( """ INSERT INTO signin_requests ( first_name, last_name, email, account_type, uuid ) VALUES ( %s, %s, %s, %s, %s ) """, ( first_name, last_name, email, account_type, uuid ) ) config.db.commit() return jsonify( { 'error': False, 'uuid': uuid } ) except: return jsonify( { 'error': True } ) @app.route( baseurl + '/validate_signin' ) @admin_required def validate_signin(): q = config.db.query( """ SELECT signin_requests.*, account_type.name as account_type FROM signin_requests LEFT JOIN account_type ON signin_requests.account_type = account_type.id WHERE signin_requests.validation_time IS NULL """ ) r = q.fetchall() users = [] for rr in r: users.append( dict( rr ) ) return render_template( "validate_signin.html", baseurl = baseurl, users = users, js = config.cdnjs, css = config.cdncss ) @app.route( baseurl + '/do_validate_signin', methods = [ 'POST' ] ) @admin_required def do_validate_signin(): request_id = request.form.get( "id" ) q = config.db.query( 'SELECT * FROM signin_requests WHERE id = %s', ( request_id, ) ) s = q.fetchone() s = dict( s ) r = {} r[ 'user' ] = s r[ 'user' ][ 'request_time' ] = str( r[ 'user' ][ 'request_time' ] ) r[ 'user' ][ 'validation_time' ] = str( r[ 'user' ][ 'validation_time' ] ) r[ 'acceptance' ] = {} r[ 'acceptance' ][ 'username' ] = session[ 'username' ] r[ 'acceptance' ][ 'time' ] = str( datetime.now() ) j = json.dumps( r ) challenge = base64.b64encode( j ) challenge = challenge.replace( "=", "" ) session[ 'validation_user_challenge' ] = challenge ############################################################################ user_id = session[ 'user_id' ] q = config.db.query( "SELECT * FROM webauthn WHERE user_id = %s ORDER BY last_usage DESC LIMIT 1", ( user_id, ) ) key = q.fetchone() webauthn_user = webauthn.WebAuthnUser( key[ 'ukey' ], session[ 'username' ], session[ 'username' ], None, key[ 'credential_id' ], key[ 'pub_key' ], key[ 'sign_count' ], config.RP_ID ) webauthn_assertion_options = webauthn.WebAuthnAssertionOptions( webauthn_user, challenge ) return jsonify( { 'error': False, 'data': webauthn_assertion_options.assertion_dict } ) @app.route( baseurl + '/do_validate_signin_2', methods = [ 'POST' ] ) @admin_required def do_validate_signin_2(): challenge = session.get( 'validation_user_challenge' ) assertion_response = request.form assertion_response_s = base64.b64encode( json.dumps( assertion_response ) ) credential_id = assertion_response.get( 'id' ) q = config.db.query( "SELECT * FROM webauthn WHERE credential_id = %s", ( credential_id, ) ) user = q.fetchone() webauthn_user = webauthn.WebAuthnUser( user[ 'ukey' ], session[ 'username' ], session[ 'username' ], None, user[ 'credential_id' ], user[ 'pub_key' ], user[ 'sign_count' ], "icnml.unil.ch" ) webauthn_assertion_response = webauthn.WebAuthnAssertionResponse( webauthn_user, assertion_response, challenge, config.ORIGIN, uv_required = False ) try: sign_count = webauthn_assertion_response.verify() except Exception as e: return jsonify( { 'error': True, 'message': 'Assertion failed. Error: {}'.format( e ) } ) ############################################################################ if len( challenge ) % 4 != 0: challenge += "=" * ( 4 - ( len( challenge ) % 4 ) ) newuser = base64.b64decode( challenge ) newuser = json.loads( newuser ) user_type = newuser[ 'user' ][ 'account_type' ] email = newuser[ 'user' ][ 'email' ] email_hash = pbkdf2( email, random_data( 100 ), 50000 ) request_uuid = newuser[ 'user' ][ 'uuid' ] request_id = newuser[ 'user' ][ 'id' ] username_id = newuser[ 'user' ][ 'username_id' ] n = config.db.query( "SELECT name FROM account_type WHERE id = %s", ( user_type, ) ) n = n.fetchone() n = n[ 0 ] username = n + "_" + str( username_id ) username = username.lower() try: config.db.query( "UPDATE signin_requests SET validation_time = now(), assertion_response = %s", ( assertion_response_s, ) ) config.db.query( "INSERT INTO users ( username, email, type ) VALUES ( %s, %s, %s )", ( username, email_hash, user_type ) ) config.db.commit() except: return jsonify( { 'error': True, 'message': 'Can not insert into database.' } ) ############################################################################ email_content = render_jinja_html( "templates/email", "signin.html", username = username, url = "https://icnml.unil.ch/config/" + newuser[ 'user' ][ 'uuid' ] ) msg = MIMEText( email_content, "html" ) msg[ 'Subject' ] = 'ICNML - Login information' msg[ 'From' ] = config.sender msg[ 'To' ] = email s = smtplib.SMTP( config.smtpserver ) s.sendmail( config.sender, [ email ], msg.as_string() ) s.quit() ############################################################################ return jsonify( { 'error': False } ) @app.route( baseurl + '/config/' ) def config_new_user( uuid ): session.clear() q = config.db.query( "SELECT email FROM signin_requests WHERE uuid = %s", ( uuid, ) ) r = q.fetchone() try: email = r[ 'email' ] session[ 'signin_user_validation_email' ] = email session[ 'signin_user_validation_uuid' ] = uuid return render_template( "user_config.html", baseurl = baseurl, js = config.cdnjs, css = config.cdncss, session_timeout = config.session_timeout ) except: return redirect( url_for( 'home' ) ) @app.route( baseurl + '/do_config', methods = [ 'POST' ] ) def do_config_new_user(): email = session[ 'signin_user_validation_email' ] uuid = session[ 'signin_user_validation_uuid' ] username = request.form.get( "username" ) password = request.form.get( "password" ) session[ 'username' ] = username ############################################################################ q = config.db.query( "SELECT count(*) FROM signin_requests WHERE uuid = %s AND email = %s", ( uuid, email, ) ) r = q.fetchone() r = r[ 0 ] if r == 0: return jsonify( { 'error': True, 'message': "no signin request" } ) q = config.db.query( "SELECT * FROM users WHERE username = %s", ( username, ) ) user = q.fetchone() if user == None: return jsonify( { 'error': True, 'message': "no user" } ) elif not password.startswith( "pbkdf2" ): return jsonify( { 'error': True, 'message': "password not in the correct format" } ) elif not pbkdf2( email, user[ 'email' ] ): return jsonify( { 'error': True, 'message': "email not corresponding to the request form" } ) elif user.get( 'password', None ) != None: return jsonify( { 'error': True, 'message': "password already set" } ) ############################################################################ password = pbkdf2( password, random_data( 65 ), 50000 ) q = config.db.query( "UPDATE users SET password = %s WHERE username = %s", ( password, username, ) ) config.db.commit() session[ 'username' ] = username ############################################################################ return jsonify( { 'error': False } ) @app.route( baseurl + '/totp_help' ) def totp_help(): return render_template( "totp_help.html", baseurl = baseurl, js = config.cdnjs, css = config.cdncss, ) ################################################################################ # QR Code generation def renew_secret(): secret = random_base32( 40 ) session[ 'secret' ] = secret return secret def get_secret(): secret = session.get( "secret", None ) if secret == None: secret = renew_secret() return secret @app.route( baseurl + '/set_secret' ) def set_secret(): config.db.query( "UPDATE users SET totp = %s WHERE username = %s", ( session[ 'secret' ], session[ 'username' ], ) ) config.db.commit() return jsonify( { 'error': False } ) @app.route( baseurl + '/secret' ) def request_secret(): get_secret() return jsonify( { 'error': False, 'secret': session[ 'secret' ] } ) @app.route( baseurl + '/new_secret' ) @login_required def request_renew_secret(): renew_secret() return jsonify( { 'error': False, 'secret': session[ 'secret' ] } ) @app.route( baseurl + '/qrcode' ) def send_qrcode(): if 'username' in session: qrcode_value = 'otpauth://totp/ICNML%20' + session[ 'username' ] + '?secret=' + get_secret() else: qrcode_value = 'otpauth://totp/ICNML?secret=' + get_secret() img = qrcode.make( qrcode_value ) temp = StringIO() img.save( temp, format = "png" ) temp.seek( 0 ) return send_file( temp, mimetype = 'image/png' ) @app.route( baseurl + '/user_qrcode' ) @login_required def serve_qrcode(): return render_template( "qrcode.html", baseurl = baseurl, secret = get_secret(), js = config.cdnjs, css = config.cdncss ) ################################################################################ # Home page @app.route( baseurl + '/' ) @login_required def home(): return render_template( "index.html", baseurl = baseurl, admin = int( session[ 'account_type' ] ) == 1, js = config.cdnjs, css = config.cdncss, session_timeout = config.session_timeout ) ################################################################################ # Main startup if __name__ == '__main__': app.run( debug = debug, host = "0.0.0.0", threaded = True )