login.py

import os
from datetime import (
    datetime,
    timedelta,
    )
from urllib.parse import urlparse
from pyramid.response import Response
from pyramid.view import view_config
from pyramid.httpexceptions import (
    HTTPFound,
    HTTPForbidden,
    )
from pyramid.security import (
    remember,
    forget,
    authenticated_userid,
    )
from pyramid.i18n import TranslationStringFactory
import colander
from deform import (
    Form,
    ValidationFailure,
    widget,
    Button,
    )
from ziggurat_foundations.models.services.user import UserService
from pyramid_mailer.message import Message
from ..tools.waktu import create_now
from ..tools.this_framework import get_settings
from ..models import DBSession
from ..models.ziggurat import User


_ = TranslationStringFactory('login')


class Login(colander.Schema):
    username = colander.SchemaNode(colander.String(), title=_('Username'))
    password = colander.SchemaNode(
            colander.String(), widget=widget.PasswordWidget(),
            title=_('Password'),
            description=_('Forgot password'))


# http://deformdemo.repoze.org/interfield/
def login_validator(form, value):
    user = form.user
    if not user or \
            not user.status or \
            not user.user_password or \
            not UserService.check_password(user, value['password']):
        raise colander.Invalid(form, _('Login failed'))


def login_ok(request, user, route='home'):
    headers = remember(request, user.id)
    user.last_login_date = create_now()
    DBSession.add(user)
    return HTTPFound(location=request.route_url(route), headers=headers)


def get_user_by_identity(request):
    identity = request.POST.get('username')
    if identity.find('@') > -1:
        q = DBSession.query(User).filter_by(email=identity)
    else:
        q = DBSession.query(User).filter_by(user_name=identity)
    return q.first()


one_hour = timedelta(1/24)
two_minutes = timedelta(1/24/60)


def login_by_code(request):
    q = DBSession.query(User).filter_by(security_code=request.GET['code'])
    user = q.first()
    if not user or \
            create_now() - user.security_code_date > one_hour:
        return HTTPFound(location=request.route_url('login-by-code-failed'))
    user.security_code = None
    DBSession.add(user)
    DBSession.flush()
    return login_ok(request, user, 'change-password')


def login_default_response():
    return dict(title=_('Login'), label_forgot_password=_('Forgot password'))


@view_config(route_name='login', renderer='templates/login.pt')
def view_login(request):
    if authenticated_userid(request):
        return HTTPFound(location=request.route_url('home'))
    if '_LOCALE_' in request.GET:
        resp = Response()
        resp.set_cookie('_LOCALE_', request.GET['_LOCALE_'], 31536000)
        return HTTPFound(
            location=request.route_url('login'), headers=resp.headers)
    if 'code' in request.GET:
        return login_by_code(request)
    resp = login_default_response()
    schema = Login(validator=login_validator)
    btn_submit = Button('submit', _('Submit'))
    form = Form(schema, buttons=(btn_submit,))
    if 'submit' not in request.POST:
        resp['form'] = form.render()
        return resp
    controls = request.POST.items()
    schema.user = user = get_user_by_identity(request)
    try:
        c = form.validate(controls)
    except ValidationFailure:
        resp['form'] = form.render()
        return resp
    return login_ok(request, user)


@view_config(route_name='logout')
def view_logout(request):
    headers = forget(request)
    return HTTPFound(
            location=request.route_url('home'), headers=headers)


@view_config(
    route_name='login-by-code-failed',
    renderer='templates/login-by-code-failed.pt')
def view_login_by_code_failed(request):
    return dict(title='Login by code failed')


###################
# Change password #
###################
class ChangePassword(colander.Schema):
    new_password = colander.SchemaNode(
            colander.String(), title=_('New password'),
            widget=widget.PasswordWidget())
    retype_password = colander.SchemaNode(
            colander.String(), title=_('Retype new password'),
            widget=widget.PasswordWidget())


def password_validator(form, value):
    if value['new_password'] != value['retype_password']:
        raise colander.Invalid(form, _('Retype mismatch'))


@view_config(
    route_name='change-password', renderer='templates/change-password.pt',
    permission='view')
def view_change_password(request):
    schema = ChangePassword(validator=password_validator)
    btn_submit = Button('save', _('Save'))
    btn_cancel = Button('cancel', _('Cancel'))
    form = Form(schema, buttons=(btn_submit, btn_cancel))
    resp = dict(title=_('Change password'))
    if not request.POST:
        resp['form'] = form.render()
        return resp
    if 'save' not in request.POST:
        return HTTPFound(location=request.route_url('home'))
    schema.request = request
    controls = request.POST.items()
    try:
        c = form.validate(controls)
    except ValidationFailure as e:
        resp['form'] = form.render()
        return resp
    UserService.set_password(request.user, c['new_password'])
    DBSession.add(request.user)
    return HTTPFound(location=request.route_url('change-password-done'))


@view_config(
    route_name='change-password-done',
    renderer='templates/change-password-done.pt', permission='view')
def view_change_password_done(request):
    return dict(title=_('Change password'))


##################
# Reset password #
##################
class ResetPassword(colander.Schema):
    email = colander.SchemaNode(
                colander.String(), title=_('Email'),
                description=_(
                    'email-reset-password',
                    default='Enter your email address and we will send you '
                            'a link to reset your password.')
            )


def reset_password_validator(form, value):
    user = form.user
    if not user or not user.status:
        raise colander.Invalid(form, _('Invalid email'))


def security_code_age(user):
    return create_now() - user.security_code_date


def send_email_security_code(
        request, user, time_remain, subject, body_msg_id, body_default_file):
    settings = get_settings()
    up = urlparse(request.url)
    url = '{}://{}/login?code={}'.format(
            up.scheme, up.netloc, user.security_code)
    minutes = int(time_remain.seconds / 60)
    data = dict(url=url, minutes=minutes)
    here = os.path.abspath(os.path.dirname(__file__))
    body_file = os.path.join(here, body_default_file)
    with open(body_file) as f:
        body_tpl = f.read()
    body = _(body_msg_id, default=body_tpl, mapping=data)
    body = request.localizer.translate(body)
    sender = '{} <{}>'.format(
            settings['mail.sender_name'], settings['mail.username'])
    subject = request.localizer.translate(_(subject))
    message = Message(
        subject=subject, sender=sender, recipients=[user.email], body=body)
    mailer = request.registry['mailer']
    mailer.send(message)


def regenerate_security_code(user):
    age = security_code_age(user)
    remain = one_hour - age
    if user.security_code and age < one_hour and remain > two_minutes:
        return remain
    UserService.regenerate_security_code(user)
    user.security_code_date = create_now()
    DBSession.add(user)
    return one_hour


@view_config(
    route_name='reset-password', renderer='templates/reset-password.pt')
def view_reset_password(request):
    if authenticated_userid(request):
        return HTTPFound(location=request.route_url('home'))
    resp = dict(title=_('Reset password'))
    schema = ResetPassword(validator=reset_password_validator)
    btn_submit = Button('submit', _('Send password reset email'))
    form = Form(schema, buttons=(btn_submit,))
    if 'submit' in request.POST:
        controls = request.POST.items()
        identity = request.POST.get('email')
        q = DBSession.query(User).filter_by(email=identity)
        schema.user = user = q.first()
        try:
            c = form.validate(controls)
        except ValidationFailure:
            resp['form'] = form.render()
            return resp
        remain = regenerate_security_code(user)
        send_email_security_code(
            request, user, remain, 'Reset password', 'reset-password-body',
            'reset-password-body.tpl')
        return HTTPFound(location=request.route_url('reset-password-sent'))
    resp['form'] = form.render()
    return resp


@view_config(
    route_name='reset-password-sent',
    renderer='templates/reset-password-sent.pt')
def view_reset_password_sent(request):
    return dict(title=_('Reset password'))