modified verification emails to use itsdangerous tokens
This commit is contained in:
parent
a90b350f71
commit
342f06f7bd
@ -58,9 +58,6 @@ class ChangePassForm(wtforms.Form):
|
||||
'Password',
|
||||
[wtforms.validators.Required(),
|
||||
wtforms.validators.Length(min=5, max=1024)])
|
||||
userid = wtforms.HiddenField(
|
||||
'',
|
||||
[wtforms.validators.Required()])
|
||||
token = wtforms.HiddenField(
|
||||
'',
|
||||
[wtforms.validators.Required()])
|
||||
|
@ -20,6 +20,7 @@ import bcrypt
|
||||
|
||||
from mediagoblin.tools.mail import send_email
|
||||
from mediagoblin.tools.template import render_template
|
||||
from mediagoblin.tools.crypto import get_timed_signer_url
|
||||
from mediagoblin import mg_globals
|
||||
|
||||
|
||||
@ -91,8 +92,8 @@ def fake_login_attempt():
|
||||
|
||||
|
||||
EMAIL_FP_VERIFICATION_TEMPLATE = (
|
||||
u"http://{host}{uri}?"
|
||||
u"userid={userid}&token={fp_verification_key}")
|
||||
u"{uri}?"
|
||||
u"token={fp_verification_key}")
|
||||
|
||||
|
||||
def send_fp_verification_email(user, request):
|
||||
@ -103,14 +104,16 @@ def send_fp_verification_email(user, request):
|
||||
- user: a user object
|
||||
- request: the request
|
||||
"""
|
||||
fp_verification_key = get_timed_signer_url('mail_verification_token') \
|
||||
.dumps(user.id)
|
||||
|
||||
rendered_email = render_template(
|
||||
request, 'mediagoblin/auth/fp_verification_email.txt',
|
||||
{'username': user.username,
|
||||
'verification_url': EMAIL_FP_VERIFICATION_TEMPLATE.format(
|
||||
host=request.host,
|
||||
uri=request.urlgen('mediagoblin.auth.verify_forgot_password'),
|
||||
userid=unicode(user.id),
|
||||
fp_verification_key=user.fp_verification_key)})
|
||||
uri=request.urlgen('mediagoblin.auth.verify_forgot_password',
|
||||
qualified=True),
|
||||
fp_verification_key=fp_verification_key)})
|
||||
|
||||
# TODO: There is no error handling in place
|
||||
send_email(
|
||||
|
@ -62,8 +62,8 @@ def normalize_user_or_email_field(allow_email=True, allow_user=True):
|
||||
|
||||
|
||||
EMAIL_VERIFICATION_TEMPLATE = (
|
||||
u"http://{host}{uri}?"
|
||||
u"userid={userid}&token={verification_key}")
|
||||
u"{uri}?"
|
||||
u"token={verification_key}")
|
||||
|
||||
|
||||
def send_verification_email(user, request, email=None,
|
||||
@ -79,14 +79,15 @@ def send_verification_email(user, request, email=None,
|
||||
email = user.email
|
||||
|
||||
if not rendered_email:
|
||||
verification_key = get_timed_signer_url('mail_verification_token') \
|
||||
.dumps(user.id)
|
||||
rendered_email = render_template(
|
||||
request, 'mediagoblin/auth/verification_email.txt',
|
||||
{'username': user.username,
|
||||
'verification_url': EMAIL_VERIFICATION_TEMPLATE.format(
|
||||
host=request.host,
|
||||
uri=request.urlgen('mediagoblin.auth.verify_email'),
|
||||
userid=unicode(user.id),
|
||||
verification_key=user.verification_key)})
|
||||
uri=request.urlgen('mediagoblin.auth.verify_email',
|
||||
qualified=True),
|
||||
verification_key=verification_key)})
|
||||
|
||||
# TODO: There is no error handling in place
|
||||
send_email(
|
||||
|
@ -15,10 +15,11 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import uuid
|
||||
import datetime
|
||||
from itsdangerous import BadSignature
|
||||
|
||||
from mediagoblin import messages, mg_globals
|
||||
from mediagoblin.db.models import User
|
||||
from mediagoblin.tools.crypto import get_timed_signer_url
|
||||
from mediagoblin.tools.response import render_to_response, redirect, render_404
|
||||
from mediagoblin.tools.translate import pass_to_ugettext as _
|
||||
from mediagoblin.tools.mail import email_debug_message
|
||||
@ -115,16 +116,28 @@ def verify_email(request):
|
||||
you are lucky :)
|
||||
"""
|
||||
# If we don't have userid and token parameters, we can't do anything; 404
|
||||
if not 'userid' in request.GET or not 'token' in request.GET:
|
||||
if not 'token' in request.GET:
|
||||
return render_404(request)
|
||||
|
||||
user = User.query.filter_by(id=request.args['userid']).first()
|
||||
# Catch error if token is faked or expired
|
||||
try:
|
||||
token = get_timed_signer_url("mail_verification_token") \
|
||||
.loads(request.GET['token'], max_age=10*24*3600)
|
||||
except BadSignature:
|
||||
messages.add_message(
|
||||
request,
|
||||
messages.ERROR,
|
||||
_('The verification key or user id is incorrect.'))
|
||||
|
||||
if user and user.verification_key == unicode(request.GET['token']):
|
||||
return redirect(
|
||||
request,
|
||||
'index')
|
||||
|
||||
user = User.query.filter_by(id=int(token)).first()
|
||||
|
||||
if user and user.email_verified is False:
|
||||
user.status = u'active'
|
||||
user.email_verified = True
|
||||
user.verification_key = None
|
||||
|
||||
user.save()
|
||||
|
||||
messages.add_message(
|
||||
@ -166,9 +179,6 @@ def resend_activation(request):
|
||||
|
||||
return redirect(request, "mediagoblin.user_pages.user_home", user=request.user['username'])
|
||||
|
||||
request.user.verification_key = unicode(uuid.uuid4())
|
||||
request.user.save()
|
||||
|
||||
email_debug_message(request)
|
||||
send_verification_email(request.user, request)
|
||||
|
||||
@ -235,11 +245,6 @@ def forgot_password(request):
|
||||
|
||||
# SUCCESS. Send reminder and return to login page
|
||||
if user:
|
||||
user.fp_verification_key = unicode(uuid.uuid4())
|
||||
user.fp_token_expire = datetime.datetime.now() + \
|
||||
datetime.timedelta(days=10)
|
||||
user.save()
|
||||
|
||||
email_debug_message(request)
|
||||
send_fp_verification_email(user, request)
|
||||
|
||||
@ -254,31 +259,44 @@ def verify_forgot_password(request):
|
||||
"""
|
||||
# get form data variables, and specifically check for presence of token
|
||||
formdata = _process_for_token(request)
|
||||
if not formdata['has_userid_and_token']:
|
||||
if not formdata['has_token']:
|
||||
return render_404(request)
|
||||
|
||||
formdata_token = formdata['vars']['token']
|
||||
formdata_userid = formdata['vars']['userid']
|
||||
formdata_vars = formdata['vars']
|
||||
|
||||
# check if it's a valid user id
|
||||
user = User.query.filter_by(id=formdata_userid).first()
|
||||
if not user:
|
||||
return render_404(request)
|
||||
# Catch error if token is faked or expired
|
||||
try:
|
||||
token = get_timed_signer_url("mail_verification_token") \
|
||||
.loads(formdata_vars['token'], max_age=10*24*3600)
|
||||
except BadSignature:
|
||||
messages.add_message(
|
||||
request,
|
||||
messages.ERROR,
|
||||
_('The verification key or user id is incorrect.'))
|
||||
|
||||
# check if we have a real user and correct token
|
||||
if ((user and user.fp_verification_key and
|
||||
user.fp_verification_key == unicode(formdata_token) and
|
||||
datetime.datetime.now() < user.fp_token_expire
|
||||
and user.email_verified and user.status == 'active')):
|
||||
return redirect(
|
||||
request,
|
||||
'index')
|
||||
|
||||
# check if it's a valid user id
|
||||
user = User.query.filter_by(id=int(token)).first()
|
||||
|
||||
# no user in db
|
||||
if not user:
|
||||
messages.add_message(
|
||||
request, messages.ERROR,
|
||||
_('The user id is incorrect.'))
|
||||
return redirect(
|
||||
request, 'index')
|
||||
|
||||
# check if user active and has email verified
|
||||
if user.email_verified and user.status == 'active':
|
||||
|
||||
cp_form = auth_forms.ChangePassForm(formdata_vars)
|
||||
|
||||
if request.method == 'POST' and cp_form.validate():
|
||||
user.pw_hash = auth_lib.bcrypt_gen_password_hash(
|
||||
cp_form.password.data)
|
||||
user.fp_verification_key = None
|
||||
user.fp_token_expire = None
|
||||
user.save()
|
||||
|
||||
messages.add_message(
|
||||
@ -292,10 +310,20 @@ def verify_forgot_password(request):
|
||||
'mediagoblin/auth/change_fp.html',
|
||||
{'cp_form': cp_form})
|
||||
|
||||
# in case there is a valid id but no user with that id in the db
|
||||
# or the token expired
|
||||
else:
|
||||
return render_404(request)
|
||||
if not user.email_verified:
|
||||
messages.add_message(
|
||||
request, messages.ERROR,
|
||||
_('You need to verify your email before you can reset your'
|
||||
' password.'))
|
||||
|
||||
if not user.status == 'active':
|
||||
messages.add_message(
|
||||
request, messages.ERROR,
|
||||
_('You are no longer an active user. Please contact the system'
|
||||
' admin to reactivate your accoutn.'))
|
||||
|
||||
return redirect(
|
||||
request, 'index')
|
||||
|
||||
|
||||
def _process_for_token(request):
|
||||
@ -313,7 +341,6 @@ def _process_for_token(request):
|
||||
|
||||
formdata = {
|
||||
'vars': formdata_vars,
|
||||
'has_userid_and_token':
|
||||
'userid' in formdata_vars and 'token' in formdata_vars}
|
||||
'has_token': 'token' in formdata_vars}
|
||||
|
||||
return formdata
|
||||
|
@ -287,3 +287,23 @@ def unique_collections_slug(db):
|
||||
constraint.create()
|
||||
|
||||
db.commit()
|
||||
|
||||
|
||||
@RegisterMigration(11, MIGRATIONS)
|
||||
def drop_token_related_User_columns(db):
|
||||
"""
|
||||
Drop unneeded columns from the User table after switching to using
|
||||
itsdangerous tokens for email and forgot password verification.
|
||||
"""
|
||||
metadata = MetaData(bind=db.bind)
|
||||
user_table = inspect_table(metadata, 'core__users')
|
||||
|
||||
verification_key = user_table.columns['verification_key']
|
||||
fp_verification_key = user_table.columns['fp_verification_key']
|
||||
fp_token_expire = user_table.columns['fp_token_expire']
|
||||
|
||||
verification_key.drop()
|
||||
fp_verification_key.drop()
|
||||
fp_token_expire.drop()
|
||||
|
||||
db.commit()
|
||||
|
@ -68,12 +68,9 @@ class User(Base, UserMixin):
|
||||
# set to nullable=True implicitly.
|
||||
wants_comment_notification = Column(Boolean, default=True)
|
||||
license_preference = Column(Unicode)
|
||||
verification_key = Column(Unicode)
|
||||
is_admin = Column(Boolean, default=False, nullable=False)
|
||||
url = Column(Unicode)
|
||||
bio = Column(UnicodeText) # ??
|
||||
fp_verification_key = Column(Unicode)
|
||||
fp_token_expire = Column(DateTime)
|
||||
|
||||
## TODO
|
||||
# plugin data would be in a separate model
|
||||
|
@ -156,20 +156,15 @@ def test_register_views(test_app):
|
||||
assert path == u'/auth/verify_email/'
|
||||
parsed_get_params = urlparse.parse_qs(get_params)
|
||||
|
||||
### user should have these same parameters
|
||||
assert parsed_get_params['userid'] == [
|
||||
unicode(new_user.id)]
|
||||
assert parsed_get_params['token'] == [
|
||||
new_user.verification_key]
|
||||
|
||||
## Try verifying with bs verification key, shouldn't work
|
||||
template.clear_test_template_context()
|
||||
response = test_app.get(
|
||||
"/auth/verify_email/?userid=%s&token=total_bs" % unicode(
|
||||
new_user.id))
|
||||
"/auth/verify_email/?token=total_bs")
|
||||
response.follow()
|
||||
context = template.TEMPLATE_TEST_CONTEXT[
|
||||
'mediagoblin/user_pages/user.html']
|
||||
|
||||
# Correct redirect?
|
||||
assert urlparse.urlsplit(response.location)[2] == '/'
|
||||
|
||||
# assert context['verification_successful'] == True
|
||||
# TODO: Would be good to test messages here when we can do so...
|
||||
new_user = mg_globals.database.User.find_one(
|
||||
@ -233,35 +228,17 @@ def test_register_views(test_app):
|
||||
|
||||
path = urlparse.urlsplit(email_context['verification_url'])[2]
|
||||
get_params = urlparse.urlsplit(email_context['verification_url'])[3]
|
||||
assert path == u'/auth/forgot_password/verify/'
|
||||
parsed_get_params = urlparse.parse_qs(get_params)
|
||||
|
||||
# user should have matching parameters
|
||||
new_user = mg_globals.database.User.find_one({'username': u'happygirl'})
|
||||
assert parsed_get_params['userid'] == [unicode(new_user.id)]
|
||||
assert parsed_get_params['token'] == [new_user.fp_verification_key]
|
||||
|
||||
### The forgotten password token should be set to expire in ~ 10 days
|
||||
# A few ticks have expired so there are only 9 full days left...
|
||||
assert (new_user.fp_token_expire - datetime.datetime.now()).days == 9
|
||||
assert path == u'/auth/forgot_password/verify/'
|
||||
|
||||
## Try using a bs password-changing verification key, shouldn't work
|
||||
template.clear_test_template_context()
|
||||
response = test_app.get(
|
||||
"/auth/forgot_password/verify/?userid=%s&token=total_bs" % unicode(
|
||||
new_user.id), status=404)
|
||||
assert response.status.split()[0] == u'404' # status="404 NOT FOUND"
|
||||
"/auth/forgot_password/verify/?token=total_bs")
|
||||
response.follow()
|
||||
|
||||
## Try using an expired token to change password, shouldn't work
|
||||
template.clear_test_template_context()
|
||||
new_user = mg_globals.database.User.find_one({'username': u'happygirl'})
|
||||
real_token_expiration = new_user.fp_token_expire
|
||||
new_user.fp_token_expire = datetime.datetime.now()
|
||||
new_user.save()
|
||||
response = test_app.get("%s?%s" % (path, get_params), status=404)
|
||||
assert response.status.split()[0] == u'404' # status="404 NOT FOUND"
|
||||
new_user.fp_token_expire = real_token_expiration
|
||||
new_user.save()
|
||||
# Correct redirect?
|
||||
assert urlparse.urlsplit(response.location)[2] == '/'
|
||||
|
||||
## Verify step 1 of password-change works -- can see form to change password
|
||||
template.clear_test_template_context()
|
||||
@ -272,7 +249,6 @@ def test_register_views(test_app):
|
||||
template.clear_test_template_context()
|
||||
response = test_app.post(
|
||||
'/auth/forgot_password/verify/', {
|
||||
'userid': parsed_get_params['userid'],
|
||||
'password': 'iamveryveryhappy',
|
||||
'token': parsed_get_params['token']})
|
||||
response.follow()
|
||||
|
Loading…
x
Reference in New Issue
Block a user