"""Module to implement two-factor autentication using
rfc4226 HOTP or rfc6238 TOTP on your tracker.
Templating functions including QR code generator and two
actions:
* login using TOTP/HOTP
* Manage TOTP/HOTP - enroll using HOTP or TOTP, unenroll 2FA
Configuration is done using the extensions/config.ini file.
Settings with default values:
[HOTP]
# Presence of the key disables this mechanism. User will
# not be able to enroll using the mechanism. They will be
# able to login using the method unless the value is
# 'login_deny' (no quotes). This will tell the user to see
# an administrator.
disable =
# If the counter on the token generator and the tracker get
# out of sync, increment the counter up to window times to
# try to match the token the user entered.
window = 10
# Number of digits in the OTP token. 6 or 8 are usually supported by
# authenticators.
length = 6
# Name for this tracker used for issuer in the QR code.
# If empty, the string before the last / in the TRACKER_WEB
# config.ini setting.
issuer =
[TOTP]
# Same as HOTP.
disable =
# If the time sync on the token generator and this server
# are out of sync calculate the token for window 30 second
# periods before and after now. 0 means the user must
# generate and the server must process the token within the
# current 30 second window.
window = 0
# Same as HOTP.
length = 6
# Same as HOTP.
issuer =
"""
from roundup.anypy.strings import b2s
from roundup.anypy import urllib_
from roundup.i18n import _
import logging
# import for generator
from roundup.cgi.actions import EditItemAction
from roundup.configuration import InvalidOptionError
import random
# imports for LoginAction replacement
from roundup.cgi.actions import LoginAction
from roundup.cgi.exceptions import Redirect, LoginError
import cgi
import math
import re
import time
# If this fails OTP doesn't work at all.
# So fail on import rather than at runtime.
from onetimepass import get_hotp, valid_hotp, valid_totp
try:
# Hopefully this works for python2
from StringIO import cStringIO as IOBuff
except ImportError:
# python 3
from io import StringIO as IOBuff
try:
# qrcode only supports python 3
import qrcode
import qrcode.image.svg
except ImportError:
qrcode = None
logger = logging.getLogger('extension')
def chunk_text(text_or_password, chunksize=4, errormsg=None):
""" Template function to chunk text by breaking one of
three types of text source:
plain text
HTML form field with _value
Password hyperdb class with _value.password
"""
if not errormsg:
errormsg = _("No value present.")
if not isinstance(text_or_password, str):
try:
text = text_or_password._value.password
except AttributeError:
text = text_or_password._value
else:
text = text_or_password
if text == '-':
return errormsg
return ' '.join([text[x*chunksize:x*chunksize+chunksize] for x in
range(0, math.ceil(len(text)/chunksize))])
def format_current_count(count):
if count == -1:
return _("2FA disabled. No counter.")
if count == -2:
return _("Using time based 2FA. Counter unused.")
return count
def hotpIntegrityValue(user_context):
'''Return the OTP for counter value 0. Google Authenticator
uses this as the Integrity Check value for the key.
The first OTP that will be accepted starts at counter
value 1.
'''
if user_context.secret_counter._value == -2:
return _("Using time based 2FA. Check value not available.")
try:
key = user_context.secret_2fa._value.password
except AttributeError:
key = user_context.secret_2fa._value
if key == '-':
return _("2FA disabled. Check value not available.")
value = str(get_hotp(key, 0))
return chunk_text(value.rjust(6, '0'), chunksize=3)
def qrify(user_context, output="svg"):
'''Generate a QR code to enroll the user using Google
Authenticator and other client apps.
The QR code is a special url to enroll the user.
We set the following info in the QR:
identity: issuer:username@TRACKER_NAME
issuer: last element of TRACKER_WEB, HOTP_ISSUER or
TOTP_ISSUER from extensions/config.ini
All values are url encoded.
Generates an svg or ascii qr code. Default output=svg.
Ascii is useful for text browsers.
'''
if qrcode is None:
return _("QR code is not available.")
if user_context.secret_2fa._value == '-':
return _("No secret key is set.")
def get_ext_setting(setting, default=None):
try:
return user_context._db.config.ext[setting]
except InvalidOptionError:
return default
counter = int(user_context.secret_counter._value)
# can be a Password or web text item.
try:
secret = urllib_.quote(user_context.secret_2fa._value.password)
except AttributeError:
secret = urllib_.quote(user_context.secret_2fa._value)
user = urllib_.quote(user_context.username._value)
web_url = urllib_.quote(user_context._db.config['TRACKER_WEB'])
if user_context.secret_counter == -2:
# TOTP
issuer = urllib_.quote(get_ext_setting('TOTP_ISSUER', ''))
else:
issuer = urllib_.quote(get_ext_setting('HOTP_ISSUER', ''))
if not issuer:
# take last component of / terminated web url
issuer = urllib_.quote(web_url.split('/')[-2])
tracker = urllib_.quote(user_context._db.config['TRACKER_NAME'])
# neither sissuer nor user can have a ':'
issuer.replace(':', '_')
user.replace(':', '_')
# https://github.com/google/google-authenticator/wiki/Key-Uri-Format
if counter == 0: # HOTP
url = (("otpauth://hotp/%(issuer)s%%3A%(user)s@%(tracker)s?"
"secret=%(secret)s&issuer=%(issuer)s&counter=0") %
locals())
else: # counter = -2 TOTP
url = (("otpauth://totp/%(issuer)s%%3A%(user)s@%(tracker)s?"
"secret=%(secret)s&issuer=%(issuer)s") % locals())
if output == "svg":
factory = qrcode.image.svg.SvgPathImage
qr_svg = qrcode.make(url, image_factory=factory)
return b2s(qr_svg.to_string())
elif output == "ascii":
qr = qrcode.QRCode()
qr.add_data(url)
file_mock = IOBuff()
qr.print_ascii(out=file_mock)
file_mock.seek(0)
qr_text = file_mock.read()
# wrap it in pre tags to keep formatting
return '%s
' % qr_text
# The following css is required to make it work on
# browsers that implement css. If the line height is
# not 1, blank space is created between the lines.
# The space makes QR recognition fail. It works fine
# without it on text browsers.
# pre.ascii_otp {line-height: 1em;}
else:
return "Error generating QR code."
def set_form_wins(client, value):
'''set client's form_wins attribute to the boolean value
passed in. If value is False, the templating code displays
values from the database rather than from the current form.
By default, the value is True so that a rejected form
displays the values the user entered rather than overwriting
them with the values from the database.
However we want to display the database values on the
user.2fa.html template as there is no user entered data
on the form that changes the secret key.
'''
client.form_wins = value
class Generate2faSecret(EditItemAction):
'''Change users HOTP secret or disable 2FA.
Values for schema properties:
secret_2fa: 32 characters from base32 alphabet or - if disabled
secret_counter:
integer >= 0 2fa HOTP enabled
integer = -1 2fa disabled
integer = -2 reserved for 2fa TOTP enabled (not yet implmented)
To prevent a user from manually setting the key by submitting
a form, we add a property to the db handle that the detector
looks for. If a web request comes in without that property on
the db, it raises an error.
'''
def handle(self):
''' Generate new 2FA HOTP secret or remove 2FA enrollment.
'''
if __debug__:
logger.debug("Generate2faSecret: enter")
def get_ext_setting(setting, default=None):
try:
return self.db.config.ext[setting]
except InvalidOptionError:
return default
# add property on db object to be used by detector
# so it knows request came from here.
self.db.Generate2faSecret = "Generate2faSecret"
# Process unenrolling request.
# 'remove_2fa' is the name of the un-enrolling button
# in web interface.
if 'remove_2fa' in self.form:
# zero out 2FA creds and counter
self.form.list.append(
cgi.MiniFieldStorage("secret_2fa", "-")
)
# all password changes need confirmation element
self.form.list.append(
cgi.MiniFieldStorage("@confirm@secret_2fa", "-")
)
# empty reset counter when removing 2fa
self.form.list.append(
cgi.MiniFieldStorage("secret_counter", "-1")
)
# call the core EditItemAction to process the edit.
EditItemAction.handle(self)
return
# Now all we have is to enroll the user.
# define base32 alphabet for random string generation
alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"
# get available otp methods
otp_methods = []
if get_ext_setting("HOTP_DISABLE") is None:
otp_methods.append('hotp')
if get_ext_setting("TOTP_DISABLE") is None:
otp_methods.append('totp')
if not otp_methods:
raise ValueError(_("All two factor methods are disabled."))
if '__enable_TOTP' in self.form:
if 'totp' in otp_methods:
use_totp = True
else:
raise ValueError(_("Time based method disabled"))
else:
if 'hotp' in otp_methods:
use_totp = False
else:
raise ValueError(_("Counter based method disabled"))
# Is this the best we can do?? How random is it?
# 160 bits (32 characters with 5 bits) of entropy.
new_secret = ''.join(
[random.choice(alphabet) for x in range(0, 32)])
self.form.list.append(
cgi.MiniFieldStorage("secret_2fa", new_secret)
)
self.form.list.append(
cgi.MiniFieldStorage("@confirm@secret_2fa", new_secret)
)
if use_totp is False:
# always reset counter to 0 when changing HOTP secret
self.form.list.append(
cgi.MiniFieldStorage("secret_counter", "0")
)
else:
self.form.list.append(
cgi.MiniFieldStorage("secret_counter", "-2")
)
# call the core EditItemAction to process the edit.
EditItemAction.handle(self)
def make_redirect_url(urlparse_dict):
"""As part of 2FA login redirect to new url composed from source
URL plus new query parameters.
"""
return urllib_.urlunparse((urlparse_dict['scheme'],
urlparse_dict['netloc'],
urlparse_dict['path'],
urlparse_dict['params'],
urllib_.urlencode(
list(
sorted(
urlparse_dict['query'].items()
)
),
doseq=True
),
urlparse_dict['fragment'])
)
def clean_url_dict(self):
"""Return a dict with cleaned up components of source URL.
As part of login code, we need to redirect back to source page
but without @ok_message or @error_message query params.
"""
if '__came_from' in self.form:
# On valid or invalid login, redirect the user back to the page
# they started on. Searches, issue, and other pages
# are all preserved in __came_from. Clean out any old feedback
# @error_message, @ok_message from the __came_from url.
#
# 1. Split the url into components.
# 2. Split the query string into parts.
# 3. Delete @error_message and @ok_message if present.
# 4. Define a new redirect_url missing the @...message entries.
# This will be redefined if there is a login error to include
# a new error message
clean_url = self.examine_url(self.form['__came_from'].value)
redirect_url_tuple = urllib_.urlparse(clean_url)
# now I have a tuple form for the __came_from url
try:
query = urllib_.parse_qs(redirect_url_tuple.query)
if "@error_message" in query:
del query["@error_message"]
if "@ok_message" in query:
del query["@ok_message"]
if "@action" in query:
# also remove the action from the redirect
# there is only ever one @action value.
if query['@action'] == ["logout"]:
del query["@action"]
except AttributeError:
# no query param so nothing to remove. Just define.
query = {}
pass
urlparse_dict = {"scheme": redirect_url_tuple.scheme,
"netloc": redirect_url_tuple.netloc,
"path": redirect_url_tuple.path,
"params": redirect_url_tuple.params,
"query": query,
"fragment": redirect_url_tuple.fragment
}
return urlparse_dict
else:
return None
class OtpLoginAction(LoginAction):
'''Login Action requiring one time password (HOTP/TOTP)
to successfully log in.
'''
# Calculate TOTP for an interval of this many seconds
# Changing this INVALIDATES all TOTP's.
# Some authenticators won't allow a value other than 30.
# Read the rfc before changing this.
totp_interval_in_sec = 30
def handle(self):
if __debug__:
logger.debug("OtpLoginAction: enter")
redirect_url_dict = clean_url_dict(self)
def get_ext_setting(setting, default=None):
try:
return self.db.config.ext[setting]
except InvalidOptionError:
return default
# store exceptions thrown by verifyLogin to expose
# error message later
verifyLoginException = None
# Find enabled methods.
otp_methods = []
if get_ext_setting("HOTP_DISABLE") is None:
otp_methods.append('hotp')
if get_ext_setting("TOTP_DISABLE") is None:
otp_methods.append('totp')
if '__login_name' in self.form:
username = self.form['__login_name'].value
try:
# get the info for the user
userid = self.db.user.lookup(username)
try:
secret = self.db.user.get(userid, 'secret_2fa',
"").password
except AttributeError:
# thrown if secret is empty
secret = '-'
last = self.db.user.get(userid, 'secret_counter')
if __debug__:
logger.debug("Login username %s is valid", username)
except KeyError:
# No such user so login will fail.
# Set variables as though OTP is disabled.
# Don't return here as a timing attack can find
# valid usernames.
secret = "-" # user's TOTP/HOTP secret
last = -1 # sequence number of last used HOTP -1 invalid
try:
last = int(last)
except (ValueError, TypeError):
# string but not an int or non-string
last = -1 # sentinal value - indicates OTP disabled
# sanity check
if (secret == "-" and last != -1) or \
(secret != "-" and last == -1):
# somebody is playing games. This should never happen.
# abort immediately.
errormsg = _("Internal error. Secret key and counter mismatch.")
# keep user on original page and report error.
if redirect_url_dict is None:
self.client.add_error_message(errormsg)
return
else:
redirect_url_dict['query']['@error_message'] = errormsg
raise Redirect(make_redirect_url(redirect_url_dict))
if '__login_password' in self.form:
password = self.form['__login_password'].value
else:
password = ""
if '__otp' in self.form:
token = self.form['__otp'].value
# remove spaces/tabs. Some auth generators
# chunk the token, copy/paste includes the spaces
token = re.sub(r'\s', '', token)
else:
if secret == '-':
# 2fa not enabled, do a normal login
LoginAction.handle(self)
# LoginAction "returns" an exception
# so the return below should not be reached
return
else:
token = ""
if last == -2:
# TOTP - window is the number of 30 second
# (interval) periods before/after the current 30
# second period. 0 only matches the code for
# this (30) second period.
windows = int(get_ext_setting('TOTP_WINDOW', 0))
else:
# HOTP - window we will search for a match so
# up to 10 keys past the last used one can be valid.
windows = int(get_ext_setting('HOTP_WINDOW', 10))
if last == -2:
# TOTP - token length number of numbers in token
# 6 or 8 are usually supported
token_length = int(get_ext_setting('TOTP_LENGTH', 6))
else:
token_length = int(get_ext_setting('HOTP_LENGTH', 6))
# Verify TOTP/HOTP and prevent replay
if last >= 0:
# uppercase because it's used for retreiving a config setting.
otp_method = "HOTP"
# valid_hotp returns: integer > 0 (match sequence number) or False
otp_ok = valid_hotp(token,
secret,
token_length=token_length,
trials=windows,
last=last)
# FIXME: should the TOTP stuff be done here to prevent
# leaking that the user is using HOTP and not TOTP. TOTP
# validation will take longer. Try sleeping
# random.random()/10 seconds.
time.sleep(random.random()/10)
# invalidate an exposed HOTP to make it single
# use even if username/password fails.
if otp_ok:
self.db.user.set(userid, secret_counter=otp_ok)
self.db.commit()
elif last == -2:
# uppercase because it's used for retreiving a config setting.
otp_method = "TOTP"
interval = self.totp_interval_in_sec
# FIXME: consider setting clock argument for
# valid_otp to (now - interval) and then None.
# Allow one interval look back to account for
# entry delay and network transmission between
# synchronized clocks. This is a look back
# window only. Note setting windows=1 looks
# forward and back one interval.
# valid_totp returns: True or False
otp_ok = valid_totp(token,
secret,
token_length=token_length,
window=windows,
interval_length=interval)
# verify that token has not been used
# check against last TOTP to prevent replay
# Use session db so multiple tokens
# (TOTP_used--) with lifetime of
# now + (window * 30 seconds or (interval))
# can be saved and expired by usual method?
otk = self.client.db.Otk
totp_key = "TOTP_used-%s-%s" % (username, token)
try:
_discard = otk.getall(totp_key)
# token has been seen before, login must fail.
otp_ok = False
except KeyError:
# ignore if key not found
pass
# Calculate a timestamp that will make OTK.clean()
# expire the entry 1.1 x interval plus windows * interval
# after the current time. 1.1 is just to make sure we record
# the TOTP through it's entire interval.
# (FYI expiration is 1 week, so we subtract a week first)
ts = time.time() - (60 * 60 * 24 * 7) \
+ (interval * windows) + 1.1*interval
if otp_ok:
# this is the only code that depends on a valid otp
otk.set(totp_key, token=token, __timestamp=ts)
# The time spent cleaning the OTK db depends on
# on number of records in database. Attempts to
# determine if the TOTP was correct via timing attack
# may be made more difficult by doing housekeeping.
otk.clean()
otk.commit()
else:
# process the token so we take the same amount of
# time as when 2fa is enabled but the token is wrong.
_discard = valid_hotp(token, 'A'*32, last=0, trials=windows)
time.sleep(random.random()/10) # combat timing attack
# Since 2fa not enabled: set otp_ok False
otp_ok = False
login_ok = False
try:
LoginAction.verifyLogin(self, username, password)
login_ok = True
except LoginError as e:
# store for later use
verifyLoginException = e
# If user supplied the next OTP in the HOTP sequence
# or if it passes with a TOTP, call core LoginAction.
if otp_ok and ((otp_ok == last + 1) or (last == -2)):
# if otp method is not supported and logins disabled
# using it, reject with a suitable error.
if get_ext_setting(otp_method + "_DISABLE") == 'login_deny':
# implement part of workflow from core LoginAction.handle
self.client.make_user_anonymous()
self.client.session_api.destroy()
errormsg = _('Login disabled please see Administrator')
if redirect_url_dict is None:
self.client.add_error_message(errormsg)
return
else:
redirect_url_dict['query']['@error_message'] = errormsg
raise Redirect(make_redirect_url(redirect_url_dict))
# perform normal user login
LoginAction.handle(self)
return # just in case control returns here
if (last != -2) and otp_ok and login_ok:
# User has skipped one or more HOTP's save the sequence
# number and reject the login. Request they enter the next
# number in the sequence.
#
# FIXME how to detect hotp and not login_ok to force two
# hotp keys in a row. Maybe a new field secret_missing
# to store state between login attempts?
# implement part of workflow from core LoginAction.handle
self.client.make_user_anonymous()
self.client.session_api.destroy()
errormsg = _(
'Multiple One Time Passwords were skipped. '
'Please log in again with your next One Time Password.')
if redirect_url_dict is None:
self.client.add_error_message(errormsg)
return
else:
redirect_url_dict['query']['@error_message'] = errormsg
raise Redirect(make_redirect_url(redirect_url_dict))
if not login_ok:
# implement part of workflow from core LoginAction.handle
self.client.make_user_anonymous()
# keep user on same page, report error from verifyLogin.
if redirect_url_dict is None:
for arg in verifyLoginException.args:
self.client.add_error_message(arg)
return
else:
redirect_url_dict['query']['@error_message'] = \
verifyLoginException.args
raise Redirect(make_redirect_url(redirect_url_dict))
# If hotp/totp was not matched, fail login.
if otp_ok is False:
# implement part of workflow from core LoginAction.handle
self.client.make_user_anonymous()
self.client.session_api.destroy()
# Match LoginAction.handle error so we don't give away info
errormsg = _('Invalid login')
# keep user on original page and report error.
if redirect_url_dict is None:
self.client.add_error_message(errormsg)
return
else:
redirect_url_dict['query']['@error_message'] = errormsg
raise Redirect(make_redirect_url(redirect_url_dict))
def init(instance):
instance.registerUtil('chunk_text', chunk_text)
instance.registerUtil('format_current_count', format_current_count)
instance.registerUtil('hotpIntegrityValue', hotpIntegrityValue)
instance.registerUtil('qrify', qrify)
if 'set_form_wins' not in instance.templating_utils:
instance.registerUtil('set_form_wins', set_form_wins)
instance.registerAction('gen2fasecret', Generate2faSecret)
instance.registerAction('login', OtpLoginAction)