').attr({'id':'qrcode_div','style':'text-align:center;padding-bottom:15px;'});
+
+ if (setup){
+ direction = $('
').attr('id','qr_info').text('Enter Code displayed in OTP App.');
+ qrcode_div.append(direction);
+ $('#otp_div').prepend(qrcode_div);
+ } else {
+ direction = $('
').attr('id','qr_info').text('OTP setup using OTP App was not completed. Please contact Administrator.');
+ qrcode_div.append(direction);
+ $('#otp_div').prepend(qrcode_div);
+ }
+}
+
+var continue_sms = function(setup, prompt){
+ request_otp();
+ var sms_div = $('
').attr({'id':'sms_div','style':'padding-bottom:15px;text-align:center;'});
+
+ if (setup){
+ sms_div.append(prompt)
+ $('#otp_div').prepend(sms_div);
+ } else {
+ direction = $('
').attr('id','qr_info').text(prompt || 'SMS was not sent. Please contact Administrator.');
+ sms_div.append(direction);
+ $('#otp_div').prepend(sms_div)
+ }
+}
+
+var continue_email = function(setup, prompt){
+ request_otp();
+ var email_div = $('
').attr({'id':'email_div','style':'padding-bottom:15px;text-align:center;'});
+
+ if (setup){
+ email_div.append(prompt)
+ $('#otp_div').prepend(email_div);
+ } else {
+ var direction = $('
').attr('id','qr_info').text(prompt || 'Verification code email not sent. Please contact Administrator.');
+ email_div.append(direction);
+ $('#otp_div').prepend(email_div);
+ }
+}
\ No newline at end of file
diff --git a/frappe/tests/test_twofactor.py b/frappe/tests/test_twofactor.py
new file mode 100644
index 0000000000..900e617360
--- /dev/null
+++ b/frappe/tests/test_twofactor.py
@@ -0,0 +1,128 @@
+# Copyright (c) 2017, Frappe Technologies Pvt. Ltd. and Contributors
+# MIT License. See license.txt
+from __future__ import unicode_literals
+
+import unittest, frappe, pyotp
+from werkzeug.wrappers import Request
+from werkzeug.test import EnvironBuilder
+from frappe.auth import LoginManager, HTTPRequest
+from frappe.twofactor import *
+import time
+
+
+
+class TestTwoFactor(unittest.TestCase):
+
+
+ def setUp(self):
+ self.http_requests = create_http_request()
+ self.login_manager = frappe.local.login_manager
+ self.user = self.login_manager.user
+
+ def tearDown(self):
+ tmp_id = frappe.local.response['tmp_id']
+ frappe.local.response['verification'] = None
+ frappe.local.response['tmp_id'] = None
+ frappe.clear_cache(user=self.user)
+
+
+ def test_should_run_2fa(self):
+ '''Should return true if enabled.'''
+ toggle_2fa_all_role(state=True)
+ self.assertTrue(should_run_2fa(self.user))
+ toggle_2fa_all_role(state=False)
+ self.assertFalse(should_run_2fa(self.user))
+
+ def test_get_cached_user_pass(self):
+ '''Cached data should not contain user and pass before 2fa.'''
+ user,pwd = get_cached_user_pass()
+ self.assertTrue(all([not user, not pwd]))
+
+ def test_authenticate_for_2factor(self):
+ '''Verification obj and tmp_id should be set in frappe.local.'''
+ authenticate_for_2factor(self.user)
+ verification_obj = frappe.local.response['verification']
+ tmp_id = frappe.local.response['tmp_id']
+ self.assertTrue(verification_obj)
+ self.assertTrue(tmp_id)
+ for k in ['_usr','_pwd','_otp_secret']:
+ self.assertTrue(frappe.cache().get('{0}{1}'.format(tmp_id,k)),
+ '{} not available'.format(k))
+
+ def test_two_factor_is_enabled_for_user(self):
+ '''Should return true if enabled for user.'''
+ toggle_2fa_all_role(state=True)
+ self.assertTrue(two_factor_is_enabled_for_(self.user))
+ toggle_2fa_all_role(state=False)
+ self.assertFalse(two_factor_is_enabled_for_(self.user))
+
+ def test_get_otpsecret_for_user(self):
+ '''OTP secret should be set for user.'''
+ self.assertTrue(get_otpsecret_for_(self.user))
+ self.assertTrue(frappe.db.get_default(self.user + '_otpsecret'))
+
+ def test_confirm_otp_token(self):
+ '''Ensure otp is confirmed'''
+ authenticate_for_2factor(self.user)
+ tmp_id = frappe.local.response['tmp_id']
+ otp = 'wrongotp'
+ with self.assertRaises(frappe.AuthenticationError):
+ confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id)
+ otp = get_otp(self.user)
+ self.assertTrue(confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id))
+ if frappe.flags.tests_verbose:
+ print('Sleeping for 30secs to confirm token expires..')
+ time.sleep(30)
+ with self.assertRaises(frappe.AuthenticationError):
+ confirm_otp_token(self.login_manager,otp=otp,tmp_id=tmp_id)
+
+ def test_get_verification_obj(self):
+ '''Confirm verification object is returned.'''
+ otp_secret = get_otpsecret_for_(self.user)
+ token = int(pyotp.TOTP(otp_secret).now())
+ self.assertTrue(get_verification_obj(self.user,token,otp_secret))
+
+ def test_render_string_template(self):
+ '''String template renders as expected with variables.'''
+ args = {'issuer_name':'Frappe Technologies'}
+ _str = 'Verification Code from {{issuer_name}}'
+ _str = render_string_template(_str,args)
+ self.assertEqual(_str,'Verification Code from Frappe Technologies')
+
+
+def set_request(**kwargs):
+ builder = EnvironBuilder(**kwargs)
+ frappe.local.request = Request(builder.get_environ())
+
+def create_http_request():
+ '''Get http request object.'''
+ set_request(method='POST', path='login')
+ enable_2fa()
+ frappe.form_dict['usr'] = 'test@erpnext.com'
+ frappe.form_dict['pwd'] = 'test'
+ frappe.local.form_dict['cmd'] = 'login'
+ http_requests = HTTPRequest()
+ return http_requests
+
+def enable_2fa():
+ '''Enable Two factor in system settings.'''
+ toggle_two_factor_auth(True)
+ system_settings = frappe.get_doc('System Settings')
+ system_settings.two_factor_method = 'OTP App'
+ system_settings.save(ignore_permissions=True)
+ frappe.db.commit()
+
+def toggle_2fa_all_role(state=None):
+ '''Enable or disable 2fa for 'all' role on the system.'''
+ all_role = frappe.get_doc('Role','All')
+ if state == None:
+ state = False if all_role.two_factor_auth == True else False
+ if state not in [True,False]:return
+ all_role.two_factor_auth = state
+ all_role.save(ignore_permissions=True)
+ frappe.db.commit()
+
+def get_otp(user):
+ otp_secret = get_otpsecret_for_(user)
+ otp = pyotp.TOTP(otp_secret)
+ return otp.now()
\ No newline at end of file
diff --git a/frappe/twofactor.py b/frappe/twofactor.py
new file mode 100644
index 0000000000..44e3fb24df
--- /dev/null
+++ b/frappe/twofactor.py
@@ -0,0 +1,358 @@
+# Copyright (c) 2017, Frappe Technologies Pvt. Ltd. and Contributors
+# MIT License. See license.txt
+
+from __future__ import unicode_literals
+
+import frappe
+from frappe import _
+import pyotp,base64,os
+from frappe.utils.background_jobs import enqueue
+from jinja2 import Template
+from pyqrcode import create as qrcreate
+from StringIO import StringIO
+from base64 import b64encode,b32encode
+from frappe.utils import get_url, get_datetime, time_diff_in_seconds
+from frappe.installer import update_site_config
+
+
+class ExpiredLoginException(Exception):pass
+
+
+def toggle_two_factor_auth(state,roles=[]):
+ '''Enable or disable 2FA in site_config and roles'''
+ update_site_config('enable_two_factor_auth',state)
+ for role in roles:
+ role = frappe.get_doc('Role',{'role_name':role})
+ role.two_factor_auth = state
+ role.save(ignore_permissions=True)
+
+
+def two_factor_is_enabled(user=None):
+ '''Returns True if 2FA is enabled.'''
+ enabled = frappe.local.conf.get('enable_two_factor_auth',False)
+ if not user or not enabled:
+ return enabled
+ return two_factor_is_enabled_for_(user)
+
+def should_run_2fa(user):
+ '''Check if 2fa should run.'''
+ return two_factor_is_enabled(user=user)
+
+
+def get_cached_user_pass():
+ '''Get user and password if set.'''
+ user = pwd = None
+ tmp_id = frappe.form_dict.get('tmp_id')
+ if tmp_id:
+ user = frappe.cache().get(tmp_id+'_usr')
+ pwd = frappe.cache().get(tmp_id+'_pwd')
+ return (user,pwd)
+
+
+def authenticate_for_2factor(user):
+ '''Authenticate two factor for enabled user before login.'''
+ if frappe.form_dict.get('otp'):return
+ otp_secret = get_otpsecret_for_(user)
+ verification_method = frappe.db.get_value('System Settings', None, 'two_factor_method')
+ token = int(pyotp.TOTP(otp_secret).now())
+ tmp_id = frappe.generate_hash(length=8)
+ cache_2fa_data(user,token,otp_secret,tmp_id)
+ verification_obj = get_verification_obj(user,token,otp_secret)
+ # Save data in local
+ frappe.local.response['verification'] = verification_obj
+ frappe.local.response['tmp_id'] = tmp_id
+
+def cache_2fa_data(user,token,otp_secret,tmp_id):
+ '''Cache and set expiry for data.'''
+ pwd = frappe.form_dict.get('pwd')
+ verification_method = get_verification_method()
+
+ # set increased expiry time for SMS and Email
+ if verification_method in ['SMS', 'Email']:
+ expiry_time = 300
+ frappe.cache().set(tmp_id + '_token', token)
+ frappe.cache().expire(tmp_id + '_token', expiry_time)
+ else:
+ expiry_time = 180
+ for k,v in {'_usr':user,'_pwd':pwd,'_otp_secret':otp_secret}.iteritems():
+ frappe.cache().set("{0}{1}".format(tmp_id,k),v)
+ frappe.cache().expire("{0}{1}".format(tmp_id,k),expiry_time)
+
+def two_factor_is_enabled_for_(user):
+ '''Check if 2factor is enabled for user.'''
+ if isinstance(user,basestring):
+ user = frappe.get_doc('User',user)
+ if user.roles:
+ query = """select name from `tabRole` where two_factor_auth=1
+ and name in ("All",{0});""".format(', '.join('\"{}\"'.format(i.role) for \
+ i in user.roles))
+ if len(frappe.db.sql(query)) > 0:
+ return True
+ return False
+
+def get_otpsecret_for_(user):
+ '''Set OTP Secret for user even if not set.'''
+ otp_secret = frappe.db.get_default(user + '_otpsecret')
+ if not otp_secret:
+ otp_secret = b32encode(os.urandom(10)).decode('utf-8')
+ frappe.db.set_default(user + '_otpsecret', otp_secret)
+ frappe.db.commit()
+ return otp_secret
+
+def get_verification_method():
+ return frappe.db.get_value('System Settings', None, 'two_factor_method')
+
+
+
+def confirm_otp_token(login_manager,otp=None,tmp_id=None):
+ '''Confirm otp matches.'''
+ if not otp:
+ otp = frappe.form_dict.get('otp')
+ if not otp:
+ if two_factor_is_enabled_for_(login_manager.user):
+ return False
+ return True
+ if not tmp_id:
+ tmp_id = frappe.form_dict.get('tmp_id')
+ hotp_token = frappe.cache().get(tmp_id + '_token')
+ otp_secret = frappe.cache().get(tmp_id + '_otp_secret')
+ if not otp_secret:
+ raise ExpiredLoginException(_('Login session expired, refresh page to retry'))
+ hotp = pyotp.HOTP(otp_secret)
+ if hotp_token:
+ if hotp.verify(otp, int(hotp_token)):
+ frappe.cache().delete(tmp_id + '_token')
+ return True
+ else:
+ login_manager.fail(_('Incorrect Verification code'), login_manager.user)
+
+ totp = pyotp.TOTP(otp_secret)
+ if totp.verify(otp):
+ # show qr code only once
+ if not frappe.db.get_default(login_manager.user + '_otplogin'):
+ frappe.db.set_default(login_manager.user + '_otplogin', 1)
+ delete_qrimage(login_manager.user)
+ return True
+ else:
+ login_manager.fail(_('Incorrect Verification code'), login_manager.user)
+
+
+def get_verification_obj(user,token,otp_secret):
+ otp_issuer = frappe.db.get_value('System Settings', 'System Settings', 'otp_issuer_name')
+ verification_method = get_verification_method()
+ verification_obj = None
+ if verification_method == 'SMS':
+ verification_obj = process_2fa_for_sms(user,token,otp_secret)
+ elif verification_method == 'OTP App':
+ #check if this if the first time that the user is trying to login. If so, send an email
+ if not frappe.db.get_default(user + '_otplogin'):
+ verification_obj = process_2fa_for_email(user,token,otp_secret,otp_issuer,method='OTP App')
+ else:
+ verification_obj = process_2fa_for_otp_app(user,otp_secret,otp_issuer)
+ elif verification_method == 'Email':
+ verification_obj = process_2fa_for_email(user,token,otp_secret,otp_issuer)
+ return verification_obj
+
+
+def process_2fa_for_sms(user,token,otp_secret):
+ '''Process sms method for 2fa.'''
+ phone = frappe.db.get_value('User', user, ['phone','mobile_no'], as_dict=1)
+ phone = phone.mobile_no or phone.phone
+ status = send_token_via_sms(otp_secret,token=token, phone_no=phone)
+ verification_obj = {'token_delivery': status,
+ 'prompt': status and 'Enter verification code sent to {}'.format(phone[:4] + '******' + phone[-3:]),
+ 'method': 'SMS',
+ 'setup': status}
+ return verification_obj
+
+def process_2fa_for_otp_app(user,otp_secret,otp_issuer):
+ '''Process OTP App method for 2fa.'''
+ totp_uri = pyotp.TOTP(otp_secret).provisioning_uri(user, issuer_name=otp_issuer)
+ if frappe.db.get_default(user + '_otplogin'):
+ otp_setup_completed = True
+ else:
+ otp_setup_completed = False
+
+ verification_obj = {'totp_uri': totp_uri,
+ 'method': 'OTP App',
+ 'qrcode': get_qr_svg_code(totp_uri),
+ 'setup': otp_setup_completed }
+ return verification_obj
+
+def process_2fa_for_email(user,token,otp_secret,otp_issuer,method='Email'):
+ '''Process Email method for 2fa.'''
+ subject = None
+ message = None
+ status = True
+ prompt = ''
+ if method == 'OTP App' and not frappe.db.get_default(user + '_otplogin'):
+ '''Sending one-time email for OTP App'''
+ totp_uri = pyotp.TOTP(otp_secret).provisioning_uri(user, issuer_name=otp_issuer)
+ qrcode_link = get_link_for_qrcode(user,totp_uri)
+ message = get_email_body_for_qr_code({'qrcode_link':qrcode_link})
+ subject = get_email_subject_for_qr_code({'qrcode_link':qrcode_link})
+ prompt = 'Please check your registered email address for instructions on how to proceed. Do not close this window as you will have to return to it!!'
+ else:
+ '''Sending email verification'''
+ prompt = 'Verification code has been sent to your registered email address.'
+ status = send_token_via_email(user,token,otp_secret,otp_issuer,subject=subject,message=message)
+ verification_obj = {'token_delivery': status,
+ 'prompt': status and prompt,
+ 'method': 'Email',
+ 'setup': status}
+ return verification_obj
+
+def get_email_subject_for_2fa(kwargs_dict):
+ '''Get email subject for 2fa.'''
+ subject_template = 'Verification Code from {}'.format(frappe.db.get_value('System Settings', 'System Settings', 'otp_issuer_name'))
+ subject = render_string_template(subject_template,kwargs_dict)
+ return subject
+
+def get_email_body_for_2fa(kwargs_dict):
+ '''Get email body for 2fa.'''
+ body_template = 'Use this token to login
{{otp}}' + body = render_string_template(body_template,kwargs_dict) + return body + +def get_email_subject_for_qr_code(kwargs_dict): + '''Get QRCode email subject.''' + subject_template = 'OTP Registration Code from {}'.format(frappe.db.get_value('System Settings', 'System Settings', 'otp_issuer_name')) + subject = render_string_template(subject_template,kwargs_dict) + return subject + +def get_email_body_for_qr_code(kwargs_dict): + '''Get QRCode email body.''' + body_template = 'Please click on the following link and follow the instructions on the page.
{{qrcode_link}}' + body = render_string_template(body_template,kwargs_dict) + return body + +def render_string_template(_str,kwargs_dict): + '''Render string with jinja.''' + s = Template(_str) + s = s.render(**kwargs_dict) + return s + +def get_link_for_qrcode(user,totp_uri): + '''Get link to temporary page showing QRCode.''' + key = frappe.generate_hash(length=20) + key_user = "{}_user".format(key) + key_uri = "{}_uri".format(key) + lifespan = int(frappe.db.get_value('System Settings', 'System Settings', 'lifespan_qrcode_image')) + if lifespan<=0: + lifespan = 240 + frappe.cache().set_value(key_uri,totp_uri,expires_in_sec=lifespan) + frappe.cache().set_value(key_user,user,expires_in_sec=lifespan) + return get_url('/qrcode?k={}'.format(key)) + +def send_token_via_sms(otpsecret, token=None, phone_no=None): + '''Send token as sms to user.''' + otp_issuer = frappe.db.get_value('System Settings', 'System Settings', 'otp_issuer_name') + try: + from frappe.core.doctype.sms_settings.sms_settings import send_request + except: + return False + + if not phone_no: + return False + + ss = frappe.get_doc('SMS Settings', 'SMS Settings') + if not ss.sms_gateway_url: + return False + + hotp = pyotp.HOTP(otpsecret) + args = {ss.message_parameter: 'Your verification code is {}'.format(hotp.at(int(token))), ss.sms_sender_name: otp_issuer} + for d in ss.get("parameters"): + args[d.parameter] = d.value + + args[ss.receiver_parameter] = phone_no + + sms_args = {'gateway_url':ss.sms_gateway_url,'params':args} + enqueue(method=send_request, queue='short', timeout=300, event=None, async=True, job_name=None, now=False, **sms_args) + return True + +def send_token_via_email(user, token, otp_secret, otp_issuer,subject=None,message=None): + '''Send token to user as email.''' + user_email = frappe.db.get_value('User', user, 'email') + if not user_email: + return False + hotp = pyotp.HOTP(otp_secret) + otp = hotp.at(int(token)) + template_args = {'otp':otp,'otp_issuer':otp_issuer} + if not subject: + subject = get_email_subject_for_2fa(template_args) + if not message: + message = get_email_body_for_2fa(template_args) + email_args = { + 'recipients':user_email, 'sender':None, 'subject':subject, + 'message':message, + 'delayed':False, 'retry':3 } + + enqueue(method=frappe.sendmail, queue='short', timeout=300, event=None, async=True, job_name=None, now=False, **email_args) + return True + +def get_qr_svg_code(totp_uri): + '''Get SVG code to display Qrcode for OTP.''' + url = qrcreate(totp_uri) + svg = '' + stream = StringIO() + try: + url.svg(stream, scale=4, background="#eee", module_color="#222") + svg = stream.getvalue().replace('\n','') + svg = b64encode(bytes(svg)) + finally: + stream.close() + return svg + +def qrcode_as_png(user,totp_uri): + '''Save temporary Qrcode to server.''' + from frappe.utils.file_manager import save_file + folder = create_barcode_folder() + png_file_name = '{}.png'.format(frappe.generate_hash(length=20)) + file_obj = save_file(png_file_name,png_file_name,'User',user,folder=folder) + frappe.db.commit() + file_url = get_url(file_obj.file_url) + file_path = os.path.join(frappe.get_site_path('public', 'files'),file_obj.file_name) + url = qrcreate(totp_uri) + with open(file_path,'w') as png_file: + url.png(png_file,scale=8, module_color=[0, 0, 0, 180], background=[0xff, 0xff, 0xcc]) + return file_url + +def create_barcode_folder(): + '''Get Barcodes folder.''' + folder_name = 'Barcodes' + folder = frappe.db.exists('File',{'file_name':folder_name}) + if folder: + return folder + folder = frappe.get_doc({ + 'doctype':'File', + 'file_name':folder_name, + 'is_folder':1, + 'folder':'Home' + }) + folder.insert(ignore_permissions=True) + return folder.name + +def delete_qrimage(user,check_expiry=False): + '''Delete Qrimage when user logs in.''' + user_barcodes = frappe.get_all('File',{'attached_to_doctype':'User', + 'attached_to_name':user,'folder':'Home/Barcodes'}) + for barcode in user_barcodes: + if check_expiry and not should_remove_barcode_image(barcode):continue + barcode = frappe.get_doc('File',barcode.name) + frappe.delete_doc('File',barcode.name,ignore_permissions=True) + +def delete_all_barcodes_for_users(): + '''Task to delete all barcodes for user.''' + users = frappe.get_all('User',{'enabled':1}) + for user in users: + delete_qrimage(user.name,check_expiry=True) + +def should_remove_barcode_image(barcode): + '''Check if it's time to delete barcode image from server. ''' + if isinstance(barcode, basestring): + barcode = frappe.get_doc('File',barcode) + lifespan = frappe.db.get_value('System Settings', 'System Settings', 'lifespan_qrcode_image') + if time_diff_in_seconds(get_datetime(),barcode.creation) > int(lifespan): + return True + return False + diff --git a/frappe/www/login.py b/frappe/www/login.py index cc149abbec..5002a44b35 100644 --- a/frappe/www/login.py +++ b/frappe/www/login.py @@ -68,4 +68,3 @@ def login_via_token(login_token): frappe.local.login_manager = LoginManager() redirect_post_login(desk_user = frappe.db.get_value("User", frappe.session.user, "user_type")=="System User") - diff --git a/frappe/www/qrcode.html b/frappe/www/qrcode.html new file mode 100644 index 0000000000..3b9c81c531 --- /dev/null +++ b/frappe/www/qrcode.html @@ -0,0 +1,31 @@ +{% extends "templates/web.html" %} + +{% block title %}Register OTP Secret{% endblock %} + +{% block page_content %} +
{{otp}}' + body = render_string_template(body_template,kwargs_dict) + return body + +def get_email_subject_for_qr_code(kwargs_dict): + '''Get QRCode email subject.''' + subject_template = 'OTP Registration Code from {}'.format(frappe.db.get_value('System Settings', 'System Settings', 'otp_issuer_name')) + subject = render_string_template(subject_template,kwargs_dict) + return subject + +def get_email_body_for_qr_code(kwargs_dict): + '''Get QRCode email body.''' + body_template = 'Please click on the following link and follow the instructions on the page.
{{qrcode_link}}' + body = render_string_template(body_template,kwargs_dict) + return body + +def render_string_template(_str,kwargs_dict): + '''Render string with jinja.''' + s = Template(_str) + s = s.render(**kwargs_dict) + return s + +def get_link_for_qrcode(user,totp_uri): + '''Get link to temporary page showing QRCode.''' + key = frappe.generate_hash(length=20) + key_user = "{}_user".format(key) + key_uri = "{}_uri".format(key) + lifespan = int(frappe.db.get_value('System Settings', 'System Settings', 'lifespan_qrcode_image')) + if lifespan<=0: + lifespan = 240 + frappe.cache().set_value(key_uri,totp_uri,expires_in_sec=lifespan) + frappe.cache().set_value(key_user,user,expires_in_sec=lifespan) + return get_url('/qrcode?k={}'.format(key)) + +def send_token_via_sms(otpsecret, token=None, phone_no=None): + '''Send token as sms to user.''' + otp_issuer = frappe.db.get_value('System Settings', 'System Settings', 'otp_issuer_name') + try: + from frappe.core.doctype.sms_settings.sms_settings import send_request + except: + return False + + if not phone_no: + return False + + ss = frappe.get_doc('SMS Settings', 'SMS Settings') + if not ss.sms_gateway_url: + return False + + hotp = pyotp.HOTP(otpsecret) + args = {ss.message_parameter: 'Your verification code is {}'.format(hotp.at(int(token))), ss.sms_sender_name: otp_issuer} + for d in ss.get("parameters"): + args[d.parameter] = d.value + + args[ss.receiver_parameter] = phone_no + + sms_args = {'gateway_url':ss.sms_gateway_url,'params':args} + enqueue(method=send_request, queue='short', timeout=300, event=None, async=True, job_name=None, now=False, **sms_args) + return True + +def send_token_via_email(user, token, otp_secret, otp_issuer,subject=None,message=None): + '''Send token to user as email.''' + user_email = frappe.db.get_value('User', user, 'email') + if not user_email: + return False + hotp = pyotp.HOTP(otp_secret) + otp = hotp.at(int(token)) + template_args = {'otp':otp,'otp_issuer':otp_issuer} + if not subject: + subject = get_email_subject_for_2fa(template_args) + if not message: + message = get_email_body_for_2fa(template_args) + email_args = { + 'recipients':user_email, 'sender':None, 'subject':subject, + 'message':message, + 'delayed':False, 'retry':3 } + + enqueue(method=frappe.sendmail, queue='short', timeout=300, event=None, async=True, job_name=None, now=False, **email_args) + return True + +def get_qr_svg_code(totp_uri): + '''Get SVG code to display Qrcode for OTP.''' + url = qrcreate(totp_uri) + svg = '' + stream = StringIO() + try: + url.svg(stream, scale=4, background="#eee", module_color="#222") + svg = stream.getvalue().replace('\n','') + svg = b64encode(bytes(svg)) + finally: + stream.close() + return svg + +def qrcode_as_png(user,totp_uri): + '''Save temporary Qrcode to server.''' + from frappe.utils.file_manager import save_file + folder = create_barcode_folder() + png_file_name = '{}.png'.format(frappe.generate_hash(length=20)) + file_obj = save_file(png_file_name,png_file_name,'User',user,folder=folder) + frappe.db.commit() + file_url = get_url(file_obj.file_url) + file_path = os.path.join(frappe.get_site_path('public', 'files'),file_obj.file_name) + url = qrcreate(totp_uri) + with open(file_path,'w') as png_file: + url.png(png_file,scale=8, module_color=[0, 0, 0, 180], background=[0xff, 0xff, 0xcc]) + return file_url + +def create_barcode_folder(): + '''Get Barcodes folder.''' + folder_name = 'Barcodes' + folder = frappe.db.exists('File',{'file_name':folder_name}) + if folder: + return folder + folder = frappe.get_doc({ + 'doctype':'File', + 'file_name':folder_name, + 'is_folder':1, + 'folder':'Home' + }) + folder.insert(ignore_permissions=True) + return folder.name + +def delete_qrimage(user,check_expiry=False): + '''Delete Qrimage when user logs in.''' + user_barcodes = frappe.get_all('File',{'attached_to_doctype':'User', + 'attached_to_name':user,'folder':'Home/Barcodes'}) + for barcode in user_barcodes: + if check_expiry and not should_remove_barcode_image(barcode):continue + barcode = frappe.get_doc('File',barcode.name) + frappe.delete_doc('File',barcode.name,ignore_permissions=True) + +def delete_all_barcodes_for_users(): + '''Task to delete all barcodes for user.''' + users = frappe.get_all('User',{'enabled':1}) + for user in users: + delete_qrimage(user.name,check_expiry=True) + +def should_remove_barcode_image(barcode): + '''Check if it's time to delete barcode image from server. ''' + if isinstance(barcode, basestring): + barcode = frappe.get_doc('File',barcode) + lifespan = frappe.db.get_value('System Settings', 'System Settings', 'lifespan_qrcode_image') + if time_diff_in_seconds(get_datetime(),barcode.creation) > int(lifespan): + return True + return False + diff --git a/frappe/www/login.py b/frappe/www/login.py index cc149abbec..5002a44b35 100644 --- a/frappe/www/login.py +++ b/frappe/www/login.py @@ -68,4 +68,3 @@ def login_via_token(login_token): frappe.local.login_manager = LoginManager() redirect_post_login(desk_user = frappe.db.get_value("User", frappe.session.user, "user_type")=="System User") - diff --git a/frappe/www/qrcode.html b/frappe/www/qrcode.html new file mode 100644 index 0000000000..3b9c81c531 --- /dev/null +++ b/frappe/www/qrcode.html @@ -0,0 +1,31 @@ +{% extends "templates/web.html" %} + +{% block title %}Register OTP Secret{% endblock %} + +{% block page_content %} +
+
+{% endblock %}
\ No newline at end of file
diff --git a/frappe/www/qrcode.py b/frappe/www/qrcode.py
new file mode 100644
index 0000000000..636eabac35
--- /dev/null
+++ b/frappe/www/qrcode.py
@@ -0,0 +1,39 @@
+# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
+# MIT License. See license.txt
+
+from __future__ import unicode_literals
+
+import frappe
+from frappe import _
+from urlparse import parse_qs
+from frappe.twofactor import get_qr_svg_code
+
+no_cache = 1
+
+
+def get_context(context):
+ context.qr_code_user,context.qrcode_svg = get_user_svg_from_cache()
+
+def get_query_key():
+ '''Return query string arg.'''
+ query_string = frappe.local.request.query_string
+ query = parse_qs(query_string)
+ if not 'k' in query.keys():
+ frappe.throw(_('Not Permitted'),frappe.PermissionError)
+ query = (query['k'][0]).strip()
+ if False in [i.isalpha() or i.isdigit() for i in query]:
+ frappe.throw(_('Not Permitted'),frappe.PermissionError)
+ return query
+
+def get_user_svg_from_cache():
+ '''Get User and SVG code from cache.'''
+ key = get_query_key()
+ totp_uri = frappe.cache().get_value("{}_uri".format(key))
+ user = frappe.cache().get_value("{}_user".format(key))
+ if not totp_uri or not user:
+ frappe.throw(_('Page has expired!'),frappe.PermissionError)
+ if not frappe.db.exists('User',user):
+ frappe.throw(_('Not Permitted'), frappe.PermissionError)
+ user = frappe.get_doc('User',user)
+ svg = get_qr_svg_code(totp_uri)
+ return (user,svg)
diff --git a/requirements.txt b/requirements.txt
index 5beb2ecc3c..0f6a4ef421 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -41,4 +41,8 @@ oauthlib
PyJWT
pypdf
openpyxl
+pyotp
+pyqrcode
+pypng
premailer
+
+
+
+
+|
+
+
+ + Hi {{qr_code_user.first_name}}, please perform the following actions: + Examples of Authentication Apps you can use are Google Authenticator, Lastpass Authenticator, Authy and Duo Mobile. + + |
+
+
+
+ 
+ |
+