'Password',
[wtforms.validators.Required(),
wtforms.validators.Length(min=5, max=1024)])
- userid = wtforms.HiddenField(
- '',
- [wtforms.validators.Required()])
token = wtforms.HiddenField(
'',
[wtforms.validators.Required()])
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
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):
- 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(
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,
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(
# 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
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(
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)
# 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)
"""
# 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']
+ # 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.'))
+
+ return redirect(
+ request,
+ 'index')
+
# check if it's a valid user id
- user = User.query.filter_by(id=formdata_userid).first()
+ user = User.query.filter_by(id=int(token)).first()
+
+ # no user in db
if not user:
- return render_404(request)
+ messages.add_message(
+ request, messages.ERROR,
+ _('The user id is incorrect.'))
+ return redirect(
+ request, 'index')
- # 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')):
+ # 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(
'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):
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
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()
# 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
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(
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()
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()
projects / mediagoblin.git / commitdiff