# Licensed to the Apache Software Foundation (ASF) under one
#       or more contributor license agreements.  See the NOTICE file
#       distributed with this work for additional information
#       regarding copyright ownership.  The ASF licenses this file
#       to you under the Apache License, Version 2.0 (the
#       "License"); you may not use this file except in compliance
#       with the License.  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#       Unless required by applicable law or agreed to in writing,
#       software distributed under the License is distributed on an
#       "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#       KIND, either express or implied.  See the License for the
#       specific language governing permissions and limitations
#       under the License.
from __future__ import annotations

import calendar
from base64 import b32encode
from datetime import datetime, time, timedelta
from time import time as time_time
import json

from six.moves.urllib.parse import urlparse, parse_qs
from six.moves.urllib.parse import urlencode

from bson import ObjectId
import re

from testfixtures import LogCapture

from ming.orm.ormsession import ThreadLocalORMSession, session
from tg import config, expose
from mock import patch, Mock
import mock
from alluratest.tools import (
    assert_equal,
    assert_not_equal,
    assert_is_none,
    assert_is_not_none,
    assert_in,
    assert_not_in,
    assert_true,
    assert_false,
)
from tg import tmpl_context as c, app_globals as g

from allura.tests import TestController
from allura.tests import decorators as td
from allura.tests.decorators import audits, out_audits, assert_logmsg
from alluratest.controller import setup_trove_categories, TestRestApiBase, oauth1_webtest
from allura import model as M
from allura.lib import plugin
from allura.lib import helpers as h
from allura.lib.multifactor import TotpService, RecoveryCodeService


def unentity(s):
    return s.replace('&quot;', '"').replace('&#34;', '"')


class TestAuth(TestController):
    def test_login(self):
        self.app.get('/auth/')
        r = self.app.post('/auth/send_verification_link', params=dict(a='test@example.com',
                                                                      _session_id=self.app.cookies['_session_id']))
        email = M.User.query.get(username='test-admin').email_addresses[0]
        r = self.app.post('/auth/send_verification_link', params=dict(a=email,
                                                                      _session_id=self.app.cookies['_session_id']))
        ThreadLocalORMSession.flush_all()
        r = self.app.get('/auth/verify_addr', params=dict(a='foo'))
        assert json.loads(self.webflash(r))['status'] == 'error', self.webflash(r)
        ea = M.EmailAddress.find({'email': email}).first()
        r = self.app.get('/auth/verify_addr', params=dict(a=ea.nonce))
        assert json.loads(self.webflash(r))['status'] == 'ok', self.webflash(r)
        r = self.app.get('/auth/logout')

        with audits('Successful login', user=True):
            r = self.app.post('/auth/do_login', params=dict(
                username='test-user', password='foo',
                _session_id=self.app.cookies['_session_id']),
                antispam=True).follow()
            assert_equal(r.headers['Location'], 'http://localhost/dashboard')

        r = self.app.post('/auth/do_login', antispam=True, params=dict(
            username='test-user', password='foo', honey1='robot',  # bad honeypot value
            _session_id=self.app.cookies['_session_id']),
                          extra_environ={'regular_antispam_err_handling_even_when_tests': 'true'},
                          status=302)
        wf = json.loads(self.webflash(r))
        assert_equal(wf['status'], 'error')
        assert_equal(wf['message'], 'Spambot protection engaged')

        with audits('Failed login', user=True):
            r = self.app.post('/auth/do_login', antispam=True, params=dict(
                username='test-user', password='food',
                _session_id=self.app.cookies['_session_id']))
            assert 'Invalid login' in str(r), r.showbrowser()

        r = self.app.post('/auth/do_login', antispam=True, params=dict(
            username='test-usera', password='foo',
            _session_id=self.app.cookies['_session_id']))
        assert 'Invalid login' in str(r), r.showbrowser()

    def test_login_invalid_username(self):
        extra = {'username': '*anonymous'}
        r = self.app.get('/auth/', extra_environ=extra)
        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test@user.com'
        f[encoded['password']] = 'foo'
        r = f.submit(extra_environ={'username': '*anonymous'})
        r.mustcontain('Usernames only include small letters, ')

    def test_login_diff_ips_ok(self):
        # exercises AntiSpam.validate methods
        extra = {'username': '*anonymous', 'REMOTE_ADDR': '11.22.33.44'}
        r = self.app.get('/auth/', extra_environ=extra)

        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test-user'
        f[encoded['password']] = 'foo'
        with audits('Successful login', user=True):
            r = f.submit(extra_environ={'username': '*anonymous', 'REMOTE_ADDR': '11.22.33.99'})

    def test_login_diff_ips_bad(self):
        # exercises AntiSpam.validate methods
        extra = {'username': '*anonymous', 'REMOTE_ADDR': '24.52.32.123'}
        r = self.app.get('/auth/', extra_environ=extra)

        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test-user'
        f[encoded['password']] = 'foo'
        r = f.submit(extra_environ={'username': '*anonymous', 'REMOTE_ADDR': '11.22.33.99',
                                    'regular_antispam_err_handling_even_when_tests': 'true'},
                     status=302)
        wf = json.loads(self.webflash(r))
        assert_equal(wf['status'], 'error')
        assert_equal(wf['message'], 'Spambot protection engaged')

    @patch('allura.lib.plugin.AuthenticationProvider.hibp_password_check_enabled', Mock(return_value=True))
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    def test_login_hibp_compromised_password_untrusted_client(self, sendsimplemail):
        # first & only login by this user, so won't have any trusted previous logins
        self.app.extra_environ = {'disable_auth_magic': 'True'}
        r = self.app.get('/auth/')
        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test-user'
        f[encoded['password']] = 'foo'

        with audits('Attempted login from untrusted location with password in HIBP breach database', user=True):
            r = f.submit(status=200)

        r.mustcontain('reset your password via email.')
        r.mustcontain('reset your password via email.<br>\nPlease check your email')

        args, kwargs = sendsimplemail.post.call_args
        assert_equal(sendsimplemail.post.call_count, 1)
        assert_equal(kwargs['subject'], 'Update your %s password' % config['site_name'])
        assert_in('/auth/forgotten_password/', kwargs['text'])

        assert_equal([], M.UserLoginDetails.query.find().all())  # no records created

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    def test_login_hibp_compromised_password_trusted_client(self, sendsimplemail):
        self.app.extra_environ = {'disable_auth_magic': 'True'}

        # regular login first, so IP address will be recorded and then trusted
        r = self.app.get('/auth/')
        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test-user'
        f[encoded['password']] = 'foo'
        with audits('Successful login', user=True):
            f.submit(status=302)
        self.app.get('/auth/logout')

        # this login will get caught by HIBP check, but trusted due to IP address being same
        with patch('allura.lib.plugin.AuthenticationProvider.hibp_password_check_enabled', Mock(return_value=True)):
            r = self.app.get('/auth/')
            f = r.forms[0]
            encoded = self.app.antispam_field_names(f)
            f[encoded['username']] = 'test-user'
            f[encoded['password']] = 'foo'

            with audits(r'Successful login with password in HIBP breach database, from trusted source '
                        r'\(reason: exact ip\)', user=True):
                r = f.submit(status=302)

            assert r.session.get('pwd-expired')
            assert_equal(r.session.get('expired-reason'), 'hibp')
            assert_equal(r.location, 'http://localhost/auth/pwd_expired')

            r = r.follow()
            r.mustcontain('must be updated to be more secure')

            # changing password covered in TestPasswordExpire

    def test_login_disabled(self):
        u = M.User.query.get(username='test-user')
        u.disabled = True
        r = self.app.get('/auth/', extra_environ={'username': '*anonymous'})
        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test-user'
        f[encoded['password']] = 'foo'
        with audits('Failed login', user=True):
            r = f.submit(extra_environ={'username': '*anonymous'})

    def test_login_pending(self):
        u = M.User.query.get(username='test-user')
        u.pending = True
        r = self.app.get('/auth/', extra_environ={'username': '*anonymous'})
        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test-user'
        f[encoded['password']] = 'foo'
        with audits('Failed login', user=True):
            r = f.submit(extra_environ={'username': '*anonymous'})

    def test_login_overlay(self):
        r = self.app.get('/auth/login_fragment/', extra_environ={'username': '*anonymous'})
        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = 'test-user'
        f[encoded['password']] = 'foo'
        with audits('Successful login', user=True):
            r = f.submit(extra_environ={'username': '*anonymous'})

    def test_logout(self):
        self.app.extra_environ = {'disable_auth_magic': 'True'}
        nav_pattern = ('nav', {'class': 'nav-main'})
        r = self.app.get('/auth/')

        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            _session_id=self.app.cookies['_session_id']),
            extra_environ={'REMOTE_ADDR': '127.0.0.1'},
            antispam=True).follow().follow()

        logged_in_session = r.session['_id']
        links = r.html.find(*nav_pattern).findAll('a')
        assert_equal(links[-1].string, "Log Out")

        r = self.app.get('/auth/logout').follow().follow()
        logged_out_session = r.session['_id']
        assert logged_in_session is not logged_out_session
        links = r.html.find(*nav_pattern).findAll('a')
        assert_equal(links[-1].string, 'Log In')

    def test_track_login(self):
        user = M.User.by_username('test-user')
        assert_equal(user.last_access['login_date'], None)
        assert_equal(user.last_access['login_ip'], None)
        assert_equal(user.last_access['login_ua'], None)

        self.app.get('/').follow()  # establish session
        self.app.post('/auth/do_login',
                      headers={'User-Agent': 'browser'},
                      extra_environ={'REMOTE_ADDR': '127.0.0.1'},
                      params=dict(
                          username='test-user',
                          password='foo',
                          _session_id=self.app.cookies['_session_id'],
                      ),
                      antispam=True,
                      )
        user = M.User.by_username('test-user')
        assert_not_equal(user.last_access['login_date'], None)
        assert_equal(user.last_access['login_ip'], '127.0.0.1')
        assert_equal(user.last_access['login_ua'], 'browser')

    def test_rememberme(self):
        username = M.User.query.get(username='test-user').username

        r = self.app.get('/').follow()  # establish session

        # Login as test-user with remember me checkbox off
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            _session_id=self.app.cookies['_session_id'],
        ), antispam=True)
        assert_equal(r.session['username'], username)
        assert_equal(r.session['login_expires'], True)

        for header, contents in r.headerlist:
            if header == 'Set-cookie':
                assert_not_in('expires', contents)

        # Login as test-user with remember me checkbox on
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo', rememberme='on',
            _session_id=self.app.cookies['_session_id'],
        ), antispam=True)
        assert_equal(r.session['username'], username)
        assert_not_equal(r.session['login_expires'], True)

        for header, contents in r.headerlist:
            if header == 'Set-cookie':
                assert_in('expires', contents)

    @td.with_user_project('test-admin')
    def test_user_can_not_claim_duplicate_emails(self):
        email_address = 'test_abcd_123@domain.net'
        user = M.User.query.get(username='test-admin')
        addresses_number = len(user.email_addresses)
        self.app.get('/').follow()  # establish session
        self.app.post('/auth/preferences/update_emails',
                      params={
                          'new_addr.addr': email_address,
                          'new_addr.claim': 'Claim Address',
                          'primary_addr': 'test-admin@users.localhost',
                          'preferences.email_format': 'plain',
                          'password': 'foo',
                          '_session_id': self.app.cookies['_session_id'],
                      },
                      extra_environ=dict(username='test-admin'))

        assert M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).count() == 1
        r = self.app.post('/auth/preferences/update_emails',
                          params={
                              'new_addr.addr': email_address,
                              'new_addr.claim': 'Claim Address',
                              'primary_addr': 'test-admin@users.localhost',
                              'preferences.email_format': 'plain',
                              'password': 'foo',
                              '_session_id': self.app.cookies['_session_id'],
                          },
                          extra_environ=dict(username='test-admin'))

        assert json.loads(self.webflash(r))['status'] == 'error', self.webflash(r)
        assert M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).count() == 1
        assert len(M.User.query.get(username='test-admin').email_addresses) == addresses_number + 1

    @td.with_user_project('test-admin')
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_added_claimed_address_by_other_user_confirmed(self, gen_message_id, sendsimplemail):
        self.app.get('/').follow()  # establish session
        email_address = 'test_abcd_123@domain.net'

        # test-user claimed & confirmed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.find(dict(email=email_address)).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()

        # Claiming the same email address by test-admin
        # the email should be added to the email_addresses list but notifications should not be sent

        admin = M.User.query.get(username='test-admin')
        addresses_number = len(admin.email_addresses)
        r = self.app.post('/auth/preferences/update_emails',
                          params={
                              'new_addr.addr': email_address,
                              'new_addr.claim': 'Claim Address',
                              'primary_addr': 'test-admin@users.localhost',
                              'preferences.email_format': 'plain',
                              'password': 'foo',
                              '_session_id': self.app.cookies['_session_id'],
                          },
                          extra_environ=dict(username='test-admin'))

        assert json.loads(self.webflash(r))['status'] == 'ok'
        assert json.loads(self.webflash(r))['message'] == 'A verification email has been sent.  ' \
                                                          'Please check your email and click to confirm.'

        args, kwargs = sendsimplemail.post.call_args
        assert sendsimplemail.post.call_count == 1
        assert kwargs['toaddr'] == email_address
        assert kwargs['subject'] == '%s - Email address claim attempt' % config['site_name']
        assert "You tried to add %s to your Allura account, " \
               "but it is already claimed by your %s account." % (email_address, user.username) in kwargs['text']

        assert len(M.User.query.get(username='test-admin').email_addresses) == addresses_number + 1
        assert len(M.EmailAddress.find(dict(email=email_address)).all()) == 2

    @td.with_user_project('test-admin')
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_added_claimed_address_by_other_user_not_confirmed(self, gen_message_id, sendsimplemail):
        email_address = 'test_abcd_1235@domain.net'

        # test-user claimed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.find(dict(email=email_address)).first()
        email.confirmed = False
        ThreadLocalORMSession.flush_all()
        # Claiming the same email address by test-admin
        # the email should be added to the email_addresses list but notifications should not be sent

        user1 = M.User.query.get(username='test-user-1')
        addresses_number = len(user1.email_addresses)
        self.app.get('/').follow()  # establish session
        r = self.app.post('/auth/preferences/update_emails',
                          params={
                              'new_addr.addr': email_address,
                              'new_addr.claim': 'Claim Address',
                              'primary_addr': 'test-user-1@users.localhost',
                              'preferences.email_format': 'plain',
                              'password': 'foo',
                              '_session_id': self.app.cookies['_session_id'],
                          },
                          extra_environ=dict(username='test-user-1'))

        assert json.loads(self.webflash(r))['status'] == 'ok'
        assert json.loads(self.webflash(r))['message'] == 'A verification email has been sent.  ' \
                                                          'Please check your email and click to confirm.'
        assert sendsimplemail.post.called
        assert len(M.User.query.get(username='test-user-1').email_addresses) == addresses_number + 1
        assert len(M.EmailAddress.find(dict(email=email_address)).all()) == 2

    @td.with_user_project('test-admin')
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_cannot_claim_more_than_max_limit(self, gen_message_id, sendsimplemail):
        with h.push_config(config, **{'user_prefs.maximum_claimed_emails': '2'}):
            self.app.get('/').follow()  # establish session
            r = self.app.post('/auth/preferences/update_emails',
                              params={
                                  'new_addr.addr': 'test_abcd_1@domain.net',
                                  'new_addr.claim': 'Claim Address',
                                  'primary_addr': 'test-user-1@users.localhost',
                                  'preferences.email_format': 'plain',
                                  'password': 'foo',
                                  '_session_id': self.app.cookies['_session_id'],
                              },
                              extra_environ=dict(username='test-user-1'))
            assert json.loads(self.webflash(r))['status'] == 'ok'

            r = self.app.post('/auth/preferences/update_emails',
                              params={
                                  'new_addr.addr': 'test_abcd_2@domain.net',
                                  'new_addr.claim': 'Claim Address',
                                  'primary_addr': 'test-user-1@users.localhost',
                                  'preferences.email_format': 'plain',
                                  'password': 'foo',
                                  '_session_id': self.app.cookies['_session_id'],
                              },
                              extra_environ=dict(username='test-user-1'))

            assert json.loads(self.webflash(r))['status'] == 'error'
            assert json.loads(self.webflash(r))['message'] == 'You cannot claim more than 2 email addresses.'

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_verification_link_for_confirmed_email(self, gen_message_id, sendsimplemail):
        self.app.get('/').follow()  # establish session
        email_address = 'test_abcd@domain.net'

        # test-user claimed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first()
        email.confirmed = True

        user1 = M.User.query.get(username='test-user-1')
        user1.claim_address(email_address)
        email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user1._id)).first()
        email.confirmed = False

        ThreadLocalORMSession.flush_all()

        r = self.app.post('/auth/send_verification_link',
                          params=dict(a=email_address, _session_id=self.app.cookies['_session_id']),
                          extra_environ=dict(username='test-user-1', _session_id=self.app.cookies['_session_id']))

        assert json.loads(self.webflash(r))['status'] == 'ok'
        assert json.loads(self.webflash(r))['message'] == 'Verification link sent'

        args, kwargs = sendsimplemail.post.call_args
        assert sendsimplemail.post.call_count == 1
        assert kwargs['toaddr'] == email_address
        assert kwargs['subject'] == '%s - Email address claim attempt' % config['site_name']
        assert "You tried to add %s to your Allura account, " \
               "but it is already claimed by your %s account." % (email_address, user.username) in kwargs['text']

    def test_invalidate_verification_link_if_email_was_confirmed(self):
        self.app.get('/').follow()  # establish session
        email_address = 'test_abcd@domain.net'

        # test-user claimed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first()
        email.confirmed = False
        ThreadLocalORMSession.flush_all()

        self.app.post('/auth/send_verification_link',
                      params=dict(a=email_address,
                                  _session_id=self.app.cookies['_session_id']),
                      extra_environ=dict(username='test-user'))

        user1 = M.User.query.get(username='test-user-1')
        user1.claim_address(email_address)
        email1 = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user1._id)).first()
        email1.confirmed = True
        ThreadLocalORMSession.flush_all()
        # Verify first email with the verification link
        r = self.app.get('/auth/verify_addr', params=dict(a=email.nonce),
                         extra_environ=dict(username='test-user'))

        assert json.loads(self.webflash(r))['status'] == 'error'
        email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first()
        assert not email.confirmed

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_verify_addr_correct_session(self, gen_message_id, sendsimplemail):
        self.app.get('/').follow()  # establish session
        email_address = 'test_abcd@domain.net'

        # test-user claimed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.find(dict(email=email_address, claimed_by_user_id=user._id)).first()
        email.confirmed = False
        ThreadLocalORMSession.flush_all()

        self.app.post('/auth/send_verification_link',
                      params=dict(a=email_address,
                                  _session_id=self.app.cookies['_session_id']),
                      extra_environ=dict(username='test-user'))

        # logged out, gets redirected to login page
        r = self.app.get('/auth/verify_addr', params=dict(a=email.nonce),
                         extra_environ=dict(username='*anonymous'))
        assert_in('/auth/?return_to=%2Fauth%2Fverify_addr', r.location)

        # logged in as someone else
        r = self.app.get('/auth/verify_addr', params=dict(a=email.nonce),
                         extra_environ=dict(username='test-admin'))
        assert_in('/auth/?return_to=%2Fauth%2Fverify_addr', r.location)
        assert_equal('You must be logged in to the correct account', json.loads(self.webflash(r))['message'])
        assert_equal('warning', json.loads(self.webflash(r))['status'])

        # logged in as correct user
        r = self.app.get('/auth/verify_addr', params=dict(a=email.nonce),
                         extra_environ=dict(username='test-user'))
        assert_in('confirmed', json.loads(self.webflash(r))['message'])
        assert_equal('ok', json.loads(self.webflash(r))['status'])

        # assert 'email added' notification email sent
        args, kwargs = sendsimplemail.post.call_args
        assert_equal(kwargs['toaddr'], user._id)
        assert_equal(kwargs['subject'], 'New Email Address Added')

    @staticmethod
    def _create_password_reset_hash():
        """ Generates a password reset token for a given user.

        :return: User object
        :rtype: User
        """
        # test-user claimed email address
        user = M.User.by_username('test-admin')
        user.set_tool_data('AuthPasswordReset',
                           hash="generated_hash_value",
                           hash_expiry="04-08-2020")
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        session(user).flush(user)

        hash_expiry = user.get_tool_data('AuthPasswordReset', 'hash_expiry')
        assert_equal(hash, 'generated_hash_value')
        assert_equal(hash_expiry, '04-08-2020')
        return user

    def test_token_generator(self):
        """ Generates new token invalidation tests.

        The tests cover: changing, claiming, updating, removing email addresses.
        :returns: email_change_invalidates_token
        """
        _params = [{'new_addr.addr': 'test_abcd@domain.net',  # Change primary address
                    'primary_addr': 'test@example.com', },
                   {'new_addr.addr': 'test@example.com',  # Claim new address
                    'new_addr.claim': 'Claim Address',
                    'primary_addr': 'test-admin@users.localhost',
                    'password': 'foo',
                    'preferences.email_format': 'plain'},
                   {'addr-1.ord': '1',  # remove test-admin@users.localhost
                    'addr-1.delete': 'on',
                    'addr-2.ord': '2',
                    'new_addr.addr': '',
                    'primary_addr': 'test-admin@users.localhost',
                    'password': 'foo',
                    'preferences.email_format': 'plain'},
                   {'addr-1.ord': '1',  # Remove email
                    'addr-2.ord': '2',
                    'addr-2.delete': 'on',
                    'new_addr.addr': '',
                    'primary_addr': 'test-admin@users.localhost'}]

        for param in _params:
            yield self.email_change_invalidates_token, param

    def email_change_invalidates_token(self, change_params):
        user = self._create_password_reset_hash()
        session(user).flush(user)

        self.app.get('/').follow()  # establish session
        change_params['_session_id'] = self.app.cookies['_session_id']
        self.app.post('/auth/preferences/update_emails',
                      extra_environ=dict(username='test-admin'),
                      params=change_params)

        u = M.User.by_username('test-admin')
        print(u.get_tool_data('AuthPasswordReset', 'hash'))
        assert_equal(u.get_tool_data('AuthPasswordReset', 'hash'), '')
        assert_equal(u.get_tool_data('AuthPasswordReset', 'hash_expiry'), '')

    @td.with_user_project('test-admin')
    def test_change_password(self):
        self.app.get('/').follow()  # establish session
        # Get and assert user with password reset token.
        user = self._create_password_reset_hash()
        old_pass = user.get_pref('password')

        # Change password
        with audits('Password changed', user=True):
            self.app.post('/auth/preferences/change_password',
                          extra_environ=dict(username='test-admin'),
                          params={
                              'oldpw': 'foo',
                              'pw': 'asdfasdf',
                              'pw2': 'asdfasdf',
                              '_session_id': self.app.cookies['_session_id'],
                          })

        # Confirm password was changed.
        assert_not_equal(old_pass, user.get_pref('password'))

        # Confirm any existing tokens were reset.
        assert_equal(user.get_tool_data('AuthPasswordReset', 'hash'), '')
        assert_equal(user.get_tool_data('AuthPasswordReset', 'hash_expiry'), '')

        # Confirm an email was sent
        tasks = M.MonQTask.query.find(dict(task_name='allura.tasks.mail_tasks.sendsimplemail')).all()
        assert_equal(len(tasks), 1)
        assert_equal(tasks[0].kwargs['subject'], 'Password Changed')
        assert_in('The password for your', tasks[0].kwargs['text'])

    @patch('allura.lib.plugin.AuthenticationProvider.hibp_password_check_enabled', Mock(return_value=True))
    @td.with_user_project('test-admin')
    def test_change_password_hibp(self):
        self.app.get('/').follow()  # establish session
        # Get and assert user with password reset token.
        user = self._create_password_reset_hash()
        old_pass = user.get_pref('password')

        # Attempt change password with weak pwd
        r = self.app.post('/auth/preferences/change_password',
                          extra_environ=dict(username='test-admin'),
                          params={
                              'oldpw': 'foo',
                              'pw': 'password',
                              'pw2': 'password',
                              '_session_id': self.app.cookies['_session_id'],
                          })

        assert 'Unsafe' in str(r.headers)

        r = self.app.post('/auth/preferences/change_password',
                          extra_environ=dict(username='test-admin'),
                          params={
                              'oldpw': 'foo',
                              'pw': '3j84rhoirwnoiwrnoiw',
                              'pw2': '3j84rhoirwnoiwrnoiw',
                              '_session_id': self.app.cookies['_session_id'],
                          })
        assert 'Unsafe' not in str(r.headers)

        # Confirm password was changed.
        user = M.User.by_username('test-admin')
        assert_not_equal(old_pass, user.get_pref('password'))

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    @td.with_user_project('test-admin')
    def test_prefs(self, gen_message_id, sendsimplemail):
        r = self.app.get('/auth/preferences/',
                         extra_environ=dict(username='test-admin'))
        # check preconditions of test data
        assert 'test@example.com' not in r
        assert 'test-admin@users.localhost' in r
        assert_equal(M.User.query.get(username='test-admin').get_pref('email_address'),
                     'test-admin@users.localhost')

        # add test@example
        with td.audits('New email address: test@example.com', user=True):
            r = self.app.post('/auth/preferences/update_emails',
                              extra_environ=dict(username='test-admin'),
                              params={
                                  'new_addr.addr': 'test@example.com',
                                  'new_addr.claim': 'Claim Address',
                                  'primary_addr': 'test-admin@users.localhost',
                                  'password': 'foo',
                                  'preferences.email_format': 'plain',
                                  '_session_id': self.app.cookies['_session_id'],
                              })
        r = self.app.get('/auth/preferences/')
        assert 'test@example.com' in r
        user = M.User.query.get(username='test-admin')
        assert_equal(user.get_pref('email_address'), 'test-admin@users.localhost')

        # remove test-admin@users.localhost
        with td.audits('Email address deleted: test-admin@users.localhost', user=True):
            r = self.app.post('/auth/preferences/update_emails',
                              extra_environ=dict(username='test-admin'),
                              params={
                                  'addr-1.ord': '1',
                                  'addr-1.delete': 'on',
                                  'addr-2.ord': '2',
                                  'new_addr.addr': '',
                                  'primary_addr': 'test-admin@users.localhost',
                                  'password': 'foo',
                                  'preferences.email_format': 'plain',
                                  '_session_id': self.app.cookies['_session_id'],
                              })

        # assert 'email_removed' notification email sent
        args, kwargs = sendsimplemail.post.call_args
        assert_equal(kwargs['toaddr'], user._id)
        assert_equal(kwargs['subject'], 'Email Address Removed')

        r = self.app.get('/auth/preferences/')
        assert 'test-admin@users.localhost' not in r
        # preferred address has not changed if email is not verified
        user = M.User.query.get(username='test-admin')
        assert_equal(user.get_pref('email_address'), None)

        with td.audits('Display Name changed Test Admin => Admin', user=True):
            r = self.app.post('/auth/preferences/update',
                              params={'preferences.display_name': 'Admin',
                                      '_session_id': self.app.cookies['_session_id'],
                                      },
                              extra_environ=dict(username='test-admin'))

    @td.with_user_project('test-admin')
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_email_prefs_change_requires_password(self, gen_message_id, sendsimplemail):
        self.app.get('/').follow()  # establish session
        # Claim new email
        new_email_params = {
            'new_addr.addr': 'test@example.com',
            'new_addr.claim': 'Claim Address',
            'primary_addr': 'test-admin@users.localhost',
            '_session_id': self.app.cookies['_session_id'],
        }
        r = self.app.post('/auth/preferences/update_emails',
                          params=new_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to claim new email', self.webflash(r))
        assert_not_in('test@example.com', r.follow())
        new_email_params['password'] = 'bad pass'

        r = self.app.post('/auth/preferences/update_emails',
                          params=new_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to claim new email', self.webflash(r))
        assert_not_in('test@example.com', r.follow())
        new_email_params['password'] = 'foo'  # valid password

        r = self.app.post('/auth/preferences/update_emails',
                          params=new_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_not_in('You must provide your current password to claim new email', self.webflash(r))
        assert_in('test@example.com', r.follow())

        # Change primary address
        change_primary_params = {
            'new_addr.addr': '',
            'primary_addr': 'test@example.com',
            '_session_id': self.app.cookies['_session_id'],
        }
        r = self.app.post('/auth/preferences/update_emails',
                          params=change_primary_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to change primary address', self.webflash(r))
        assert_equal(M.User.by_username('test-admin').get_pref('email_address'), 'test-admin@users.localhost')
        change_primary_params['password'] = 'bad pass'

        r = self.app.post('/auth/preferences/update_emails',
                          params=change_primary_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to change primary address', self.webflash(r))
        assert_equal(M.User.by_username('test-admin').get_pref('email_address'), 'test-admin@users.localhost')
        change_primary_params['password'] = 'foo'  # valid password

        self.app.get('/auth/preferences/')  # let previous 'flash' message cookie get used up
        r = self.app.post('/auth/preferences/update_emails',
                          params=change_primary_params,
                          extra_environ=dict(username='test-admin'))
        assert_not_in('You must provide your current password to change primary address', self.webflash(r))
        assert_equal(M.User.by_username('test-admin').get_pref('email_address'), 'test@example.com')

        # assert 'email added' notification email sent using original primary addr
        args, kwargs = sendsimplemail.post.call_args
        assert_equal(kwargs['toaddr'], 'test-admin@users.localhost')
        assert_equal(kwargs['subject'], 'Primary Email Address Changed')

        # Remove email
        remove_email_params = {
            'addr-1.ord': '1',
            'addr-2.ord': '2',
            'addr-2.delete': 'on',
            'new_addr.addr': '',
            'primary_addr': 'test-admin@users.localhost',
            '_session_id': self.app.cookies['_session_id'],
        }
        r = self.app.post('/auth/preferences/update_emails',
                          params=remove_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to delete an email', self.webflash(r))
        assert_in('test@example.com', r.follow())
        remove_email_params['password'] = 'bad pass'
        r = self.app.post('/auth/preferences/update_emails',
                          params=remove_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to delete an email', self.webflash(r))
        assert_in('test@example.com', r.follow())
        remove_email_params['password'] = 'foo'  # vallid password
        r = self.app.post('/auth/preferences/update_emails',
                          params=remove_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_not_in('You must provide your current password to delete an email', self.webflash(r))
        assert_not_in('test@example.com', r.follow())

    @td.with_user_project('test-admin')
    def test_prefs_subscriptions(self):
        r = self.app.get('/auth/subscriptions/',
                         extra_environ=dict(username='test-admin'))
        subscriptions = M.Mailbox.query.find(dict(
            user_id=c.user._id, is_flash=False)).all()
        # make sure page actually lists all the user's subscriptions
        assert len(subscriptions) > 0, 'Test user has no subscriptions, cannot verify that they are shown'
        for m in subscriptions:
            assert str(m._id) in r, "Page doesn't list subscription for Mailbox._id = %s" % m._id

        # make sure page lists all tools which user can subscribe
        user = M.User.query.get(username='test-admin')
        for p in user.my_projects():
            for ac in p.app_configs:
                if not M.Mailbox.subscribed(project_id=p._id, app_config_id=ac._id):
                    if ac.tool_name in ('activity', 'admin', 'search', 'userstats', 'profile'):
                        # these have has_notifications=False
                        assert str(ac._id) not in r, "Page lists tool %s but it should not" % ac.tool_name
                    else:
                        assert str(ac._id) in r, "Page doesn't list tool %s" % ac.tool_name

    @td.with_user_project('test-admin')
    def test_update_user_notifications(self):
        self.app.get('/').follow()  # establish session
        assert not M.User.query.get(username='test-admin').get_pref('mention_notifications')
        self.app.post('/auth/subscriptions/update_user_notifications',
                      params={'_session_id': self.app.cookies['_session_id'],
                              })
        assert not M.User.query.get(username='test-admin').get_pref('mention_notifications')
        self.app.post('/auth/subscriptions/update_user_notifications',
                      params={'allow_umnotif': 'on',
                              '_session_id': self.app.cookies['_session_id'],
                              })
        assert M.User.query.get(username='test-admin').get_pref('mention_notifications')

    def _find_subscriptions_form(self, r):
        form = None
        for f in r.forms.values():
            if f.action == 'update_subscriptions':
                form = f
                break
        assert form is not None, "Can't find subscriptions form"
        return form

    def _find_subscriptions_field(self, form, subscribed=False):
        field_name = None
        for k, v in form.fields.items():
            if subscribed:
                check = v and v[0].value == 'on'
            else:
                check = v and v[0].value != 'on'
            if k and k.endswith('.subscribed') and check:
                field_name = k.replace('.subscribed', '')
        assert field_name, "Can't find unsubscribed tool for user"
        return field_name

    @td.with_user_project('test-admin')
    def test_prefs_subscriptions_subscribe(self):
        resp = self.app.get('/auth/subscriptions/',
                            extra_environ=dict(username='test-admin'))
        form = self._find_subscriptions_form(resp)
        # find not subscribed tool, subscribe and verify
        field_name = self._find_subscriptions_field(form, subscribed=False)
        t_id = ObjectId(form.fields[field_name + '.tool_id'][0].value)
        p_id = ObjectId(form.fields[field_name + '.project_id'][0].value)
        subscribed = M.Mailbox.subscribed(project_id=p_id, app_config_id=t_id)
        assert not subscribed, "User already subscribed for tool %s" % t_id
        form.fields[field_name + '.subscribed'][0].value = 'on'
        form.submit()
        subscribed = M.Mailbox.subscribed(project_id=p_id, app_config_id=t_id)
        assert subscribed, "User is not subscribed for tool %s" % t_id

    @td.with_user_project('test-admin')
    def test_prefs_subscriptions_unsubscribe(self):
        resp = self.app.get('/auth/subscriptions/',
                            extra_environ=dict(username='test-admin'))
        form = self._find_subscriptions_form(resp)
        field_name = self._find_subscriptions_field(form, subscribed=True)
        s_id = ObjectId(form.fields[field_name + '.subscription_id'][0].value)
        s = M.Mailbox.query.get(_id=s_id)
        assert s, "User has not subscription with Mailbox._id = %s" % s_id
        form.fields[field_name + '.subscribed'][0].value = None
        form.submit()
        s = M.Mailbox.query.get(_id=s_id)
        assert not s, "User still has subscription with Mailbox._id %s" % s_id

    def test_format_email(self):
        self.app.get('/').follow()  # establish session
        self.app.post('/auth/subscriptions/update_subscriptions',
                      params={'email_format': 'plain', 'subscriptions': '',
                              '_session_id': self.app.cookies['_session_id']})
        r = self.app.get('/auth/subscriptions/')
        assert '<option selected value="plain">Plain Text</option>' in r
        self.app.post('/auth/subscriptions/update_subscriptions',
                      params={'email_format': 'both', 'subscriptions': '',
                              '_session_id': self.app.cookies['_session_id']})
        r = self.app.get('/auth/subscriptions/')
        assert '<option selected value="both">HTML</option>' in r

    def test_create_account(self):
        r = self.app.get('/auth/create_account')
        assert 'Create an Account' in r
        r = self.app.post('/auth/save_new',
                          params=dict(username='AAA', pw='123',
                                      _session_id=self.app.cookies['_session_id']))
        assert_in('Enter a value 6 characters long or more', r)
        assert_in('Usernames must include only small letters, numbers, '
                  'and dashes. They must also start with a letter and be '
                  'at least 3 characters long.', r)
        r = self.app.post(
            '/auth/save_new',
            params=dict(
                username='aaa',
                pw='12345678',
                pw2='12345678',
                display_name='Test Me',
                _session_id=self.app.cookies['_session_id'],
            ))
        r = r.follow().follow()
        assert 'User "aaa" registered' in unentity(r.text)
        r = self.app.post(
            '/auth/save_new',
            params=dict(
                username='aaa',
                pw='12345678',
                pw2='12345678',
                display_name='Test Me',
                _session_id=self.app.cookies['_session_id'],
            ))
        assert 'That username is already taken. Please choose another.' in r
        r = self.app.get('/auth/logout')
        r = self.app.post(
            '/auth/do_login',
            params=dict(username='aaa', password='12345678',
                        _session_id=self.app.cookies['_session_id']), antispam=True,
            status=302)

    def test_create_account_require_email(self):
        self.app.get('/').follow()  # establish session
        with h.push_config(config, **{'auth.require_email_addr': 'false'}):
            self.app.post(
                '/auth/save_new',
                params=dict(
                    username='aaa',
                    pw='12345678',
                    pw2='12345678',
                    display_name='Test Me',
                    email='test@example.com',
                    _session_id=self.app.cookies['_session_id'],
                ))
            user = M.User.query.get(username='aaa')
            assert not user.pending
            assert_equal(M.Project.query.find({'name': 'u/aaa'}).count(), 1)
        with h.push_config(config, **{'auth.require_email_addr': 'true'}):
            self.app.post(
                '/auth/save_new',
                params=dict(
                    username='bbb',
                    pw='12345678',
                    pw2='12345678',
                    display_name='Test Me',
                    email='test@example.com',
                    _session_id=self.app.cookies['_session_id']
                ))
            user = M.User.query.get(username='bbb')
            assert user.pending
            assert_equal(M.Project.query.find({'name': 'u/bbb'}).count(), 0)

    def test_verify_email(self):
        with h.push_config(config, **{'auth.require_email_addr': 'true'}):
            self.app.get('/').follow()  # establish session
            r = self.app.post(
                '/auth/save_new',
                params=dict(
                    username='aaa',
                    pw='12345678',
                    pw2='12345678',
                    display_name='Test Me',
                    email='test@example.com',
                    _session_id=self.app.cookies['_session_id']
                ))
            r = r.follow()
            user = M.User.query.get(username='aaa')
            em = M.EmailAddress.get(email='test@example.com')
            assert user._id == em.claimed_by_user_id
            r = self.app.get('/auth/verify_addr', params=dict(a=em.nonce))
            user = M.User.query.get(username='aaa')
            em = M.EmailAddress.get(email='test@example.com')
            assert not user.pending
            assert em.confirmed
            assert user.get_pref('email_address')
            assert_equal(M.Project.query.find({'name': 'u/aaa'}).count(), 1)

    def test_create_account_disabled_header_link(self):
        with h.push_config(config, **{'auth.allow_user_registration': 'false'}):
            r = self.app.get('/')
            assert 'Register' not in r

    def test_create_account_disabled_form_gone(self):
        with h.push_config(config, **{'auth.allow_user_registration': 'false'}):
            r = self.app.get('/auth/create_account', status=404)
            assert 'Create an Account' not in r

    def test_create_account_disabled_submit_fails(self):
        with h.push_config(config, **{'auth.allow_user_registration': 'false'}):
            self.app.get('/').follow()  # establish session
            self.app.post('/auth/save_new',
                          params=dict(
                              username='aaa',
                              pw='12345678',
                              pw2='12345678',
                              display_name='Test Me',
                              _session_id=self.app.cookies['_session_id']
                          ),
                          status=404)

    def test_one_project_role(self):
        """Make sure when a user goes to a new project only one project role is created.
           There was an issue with extra project roles getting created if a user went directly to
           an admin page."""
        p_nbhd = M.Neighborhood.query.get(name='Projects')
        p = M.Project.query.get(shortname='test', neighborhood_id=p_nbhd._id)
        self.app.get('/').follow()  # establish session
        self.app.post('/auth/save_new', params=dict(
            username='aaa',
            pw='12345678',
            pw2='12345678',
            display_name='Test Me',
            email='test@example.com',
            _session_id=self.app.cookies['_session_id'],
        )).follow()
        user = M.User.query.get(username='aaa')
        user.pending = False
        session(user).flush(user)
        assert M.ProjectRole.query.find(
            dict(user_id=user._id, project_id=p._id)).count() == 0

        self.app.get('/p/test/admin/permissions',
                     extra_environ=dict(username='aaa'), status=403)
        assert M.ProjectRole.query.find(
            dict(user_id=user._id, project_id=p._id)).count() <= 1

    def test_default_lookup(self):
        # Make sure that default _lookup() throws 404
        self.app.get('/auth/foobar', status=404)

    def test_disabled_user(self):
        user = M.User.query.get(username='test-admin')
        sess = session(user)
        assert not user.disabled
        r = self.app.get('/p/test/admin/',
                         extra_environ={'username': 'test-admin'})
        assert_equal(r.status_int, 200, 'Redirect to %s' % r.location)
        user.disabled = True
        sess.save(user)
        sess.flush()
        user = M.User.query.get(username='test-admin')
        assert user.disabled
        r = self.app.get('/p/test/admin/',
                         extra_environ={'username': 'test-admin'})
        assert_equal(r.status_int, 302)
        assert_equal(r.location, 'http://localhost/auth/?return_to=%2Fp%2Ftest%2Fadmin%2F')

    def test_no_open_return_to(self):
        r = self.app.get('/auth/logout').follow().follow()
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            return_to='/foo',
            _session_id=self.app.cookies['_session_id']),
            antispam=True
        )
        assert_equal(r.location, 'http://localhost/foo')

        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', antispam=True, params=dict(
            username='test-user', password='foo',
            return_to='http://localhost/foo',
            _session_id=self.app.cookies['_session_id']))
        assert_equal(r.location, 'http://localhost/foo')

        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', antispam=True, params=dict(
            username='test-user', password='foo',
            return_to='http://example.com/foo',
            _session_id=self.app.cookies['_session_id'])).follow()
        assert_equal(r.location, 'http://localhost/dashboard')

        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', antispam=True, params=dict(
            username='test-user', password='foo',
            return_to='//example.com/foo',
            _session_id=self.app.cookies['_session_id'])).follow()
        assert_equal(r.location, 'http://localhost/dashboard')

    def test_no_injected_headers_in_return_to(self):
        r = self.app.get('/auth/logout').follow().follow()
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            return_to='/foo\nContent-Length: 777',
            # WebTest actually will raise an error if there's an invalid header (webob itself does not)
            _session_id=self.app.cookies['_session_id']),
            antispam=True
        )
        assert_equal(r.location, 'http://localhost/')
        assert_not_equal(r.content_length, 777)


class TestAuthRest(TestRestApiBase):

    def test_tools_list_anon(self):
        resp = self.api_get('/rest/auth/tools/wiki', user='*anonymous')
        assert_equal(resp.json, {
            'tools': []
        })

    def test_tools_list_invalid_tool(self):
        resp = self.api_get('/rest/auth/tools/af732q9547235')
        assert_equal(resp.json, {
            'tools': []
        })

    @td.with_tool('test', 'Wiki', mount_point='docs', mount_label='Documentation')
    def test_tools_list_wiki(self):
        resp = self.api_get('/rest/auth/tools/wiki')
        assert_equal(resp.json, {
            'tools': [
                {
                    'mount_label': 'Wiki',
                    'mount_point': 'wiki',
                    'name': 'wiki',
                    'project_name': 'Home Project for Adobe',
                    'url': 'http://localhost/adobe/wiki/',
                    'api_url': 'http://localhost/rest/adobe/wiki/',
                },
                {
                    'mount_label': 'Documentation',
                    'mount_point': 'docs',
                    'name': 'wiki',
                    'project_name': 'Test Project',
                    'url': 'http://localhost/p/test/docs/',
                    'api_url': 'http://localhost/rest/p/test/docs/',
                },
            ]
        })


class TestPreferences(TestController):
    @td.with_user_project('test-admin')
    def test_personal_data(self):
        from pytz import country_names

        setsex, setbirthdate, setcountry, setcity, settimezone = \
            ('Male', '19/08/1988', 'IT', 'Milan', 'Europe/Rome')
        self.app.get('/auth/user_info/')

        # Check if personal data is properly set
        r = self.app.post('/auth/user_info/change_personal_data',
                          params=dict(
                              sex=setsex,
                              birthdate=setbirthdate,
                              country=setcountry,
                              city=setcity,
                              timezone=settimezone,
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        sex = user.sex
        assert sex == setsex
        birthdate = user.birthdate.strftime('%d/%m/%Y')
        assert birthdate == setbirthdate
        country = user.localization.country
        assert country_names.get(setcountry) == country
        city = user.localization.city
        assert city == setcity
        timezone = user.timezone
        assert timezone == settimezone

        # Check if setting a wrong date everything works correctly
        r = self.app.post('/auth/user_info/change_personal_data',
                          params=dict(birthdate='30/02/1998', _session_id=self.app.cookies['_session_id']))
        assert 'Please enter a valid date' in r.text
        user = M.User.query.get(username='test-admin')
        sex = user.sex
        assert sex == setsex
        birthdate = user.birthdate.strftime('%d/%m/%Y')
        assert birthdate == setbirthdate
        country = user.localization.country
        assert country_names.get(setcountry) == country
        city = user.localization.city
        assert city == setcity
        timezone = user.timezone
        assert timezone == settimezone

        # Check deleting birthdate
        r = self.app.post('/auth/user_info/change_personal_data',
                          params=dict(
                              sex=setsex,
                              birthdate='',
                              country=setcountry,
                              city=setcity,
                              timezone=settimezone,
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        assert user.birthdate is None

    @td.with_user_project('test-admin')
    def test_contacts(self):
        # Add skype account
        testvalue = 'testaccount'
        self.app.get('/auth/user_info/contacts/')
        self.app.post('/auth/user_info/contacts/skype_account',
                      params=dict(skypeaccount=testvalue, _session_id=self.app.cookies['_session_id']))
        user = M.User.query.get(username='test-admin')
        assert user.skypeaccount == testvalue

        # Add social network account
        socialnetwork = 'Facebook'
        accounturl = 'http://www.facebook.com/test'
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(socialnetwork=socialnetwork,
                                  accounturl=accounturl,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1
        assert_equal(user.socialnetworks[0].socialnetwork, socialnetwork)
        assert_equal(user.socialnetworks[0].accounturl, accounturl)

        # Add second social network account
        socialnetwork2 = 'Twitter'
        accounturl2 = 'http://twitter.com/test'
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(socialnetwork=socialnetwork2,
                                  accounturl='@test',
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 2
        assert_in({'socialnetwork': socialnetwork, 'accounturl': accounturl}, user.socialnetworks)
        assert_in({'socialnetwork': socialnetwork2, 'accounturl': accounturl2}, user.socialnetworks)

        # Remove first social network account
        self.app.post('/auth/user_info/contacts/remove_social_network',
                      params=dict(socialnetwork=socialnetwork,
                                  account=accounturl,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1
        assert_in({'socialnetwork': socialnetwork2, 'accounturl': accounturl2}, user.socialnetworks)

        # Add empty social network account
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(accounturl=accounturl, socialnetwork='',
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1
        assert_in({'socialnetwork': socialnetwork2, 'accounturl': accounturl2}, user.socialnetworks)

        # Add invalid social network account
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(accounturl=accounturl, socialnetwork='invalid',
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1
        assert_in({'socialnetwork': socialnetwork2, 'accounturl': accounturl2}, user.socialnetworks)

        # Add telephone number
        telnumber = '+3902123456'
        self.app.post('/auth/user_info/contacts/add_telnumber',
                      params=dict(newnumber=telnumber,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert (len(user.telnumbers) == 1 and (user.telnumbers[0] == telnumber))

        # Add second telephone number
        telnumber2 = '+3902654321'
        self.app.post('/auth/user_info/contacts/add_telnumber',
                      params=dict(newnumber=telnumber2,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert (len(user.telnumbers) == 2 and telnumber in user.telnumbers and telnumber2 in user.telnumbers)

        # Remove first telephone number
        self.app.post('/auth/user_info/contacts/remove_telnumber',
                      params=dict(oldvalue=telnumber,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert (len(user.telnumbers) == 1 and telnumber2 in user.telnumbers)

        # Add website
        website = 'http://www.testurl.com'
        self.app.post('/auth/user_info/contacts/add_webpage',
                      params=dict(newwebsite=website,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert (len(user.webpages) == 1 and (website in user.webpages))

        # Add second website
        website2 = 'http://www.testurl2.com'
        self.app.post('/auth/user_info/contacts/add_webpage',
                      params=dict(newwebsite=website2,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert (len(user.webpages) == 2 and website in user.webpages and website2 in user.webpages)

        # Remove first website
        self.app.post('/auth/user_info/contacts/remove_webpage',
                      params=dict(oldvalue=website,
                                  _session_id=self.app.cookies['_session_id'],
                                  ))
        user = M.User.query.get(username='test-admin')
        assert (len(user.webpages) == 1 and website2 in user.webpages)

    @td.with_user_project('test-admin')
    def test_availability(self):
        # Add availability timeslot
        weekday = 'Monday'
        starttime = time(9, 0, 0)
        endtime = time(12, 0, 0)

        self.app.get('/auth/user_info/availability/')
        r = self.app.post('/auth/user_info/availability/add_timeslot',
                          params=dict(
                              weekday=weekday,
                              starttime=starttime.strftime('%H:%M'),
                              endtime=endtime.strftime('%H:%M'),
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        timeslot1dict = dict(week_day=weekday, start_time=starttime, end_time=endtime)
        assert len(user.availability) == 1 and timeslot1dict in user.get_availability_timeslots()

        weekday2 = 'Tuesday'
        starttime2 = time(14, 0, 0)
        endtime2 = time(16, 0, 0)

        # Add second availability timeslot
        r = self.app.post('/auth/user_info/availability/add_timeslot',
                          params=dict(
                              weekday=weekday2,
                              starttime=starttime2.strftime('%H:%M'),
                              endtime=endtime2.strftime('%H:%M'),
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        timeslot2dict = dict(week_day=weekday2, start_time=starttime2, end_time=endtime2)
        assert len(user.availability) == 2
        assert_in(timeslot1dict, user.get_availability_timeslots())
        assert_in(timeslot2dict, user.get_availability_timeslots())

        # Remove availability timeslot
        r = self.app.post('/auth/user_info/availability/remove_timeslot',
                          params=dict(
                              weekday=weekday,
                              starttime=starttime.strftime('%H:%M'),
                              endtime=endtime.strftime('%H:%M'),
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        assert len(user.availability) == 1 and timeslot2dict in user.get_availability_timeslots()

        # Add invalid availability timeslot
        r = self.app.post('/auth/user_info/availability/add_timeslot',
                          params=dict(
                              weekday=weekday2,
                              starttime=endtime2.strftime('%H:%M'),
                              endtime=starttime2.strftime('%H:%M'),
                              _session_id=self.app.cookies['_session_id'],
                          ))
        assert 'Invalid period:' in str(r)
        user = M.User.query.get(username='test-admin')
        timeslot2dict = dict(week_day=weekday2, start_time=starttime2, end_time=endtime2)
        assert len(user.availability) == 1 and timeslot2dict in user.get_availability_timeslots()

    @td.with_user_project('test-admin')
    def test_inactivity(self):
        # Add inactivity period
        now = datetime.utcnow().date()
        now = datetime(now.year, now.month, now.day)
        startdate = now + timedelta(days=1)
        enddate = now + timedelta(days=7)
        self.app.get('/auth/user_info/availability/')
        r = self.app.post('/auth/user_info/availability/add_inactive_period',
                          params=dict(
                              startdate=startdate.strftime('%d/%m/%Y'),
                              enddate=enddate.strftime('%d/%m/%Y'),
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        period1dict = dict(start_date=startdate, end_date=enddate)
        assert len(user.inactiveperiod) == 1 and period1dict in user.get_inactive_periods()

        # Add second inactivity period
        startdate2 = now + timedelta(days=24)
        enddate2 = now + timedelta(days=28)
        r = self.app.post('/auth/user_info/availability/add_inactive_period',
                          params=dict(
                              startdate=startdate2.strftime('%d/%m/%Y'),
                              enddate=enddate2.strftime('%d/%m/%Y'),
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        period2dict = dict(start_date=startdate2, end_date=enddate2)
        assert len(user.inactiveperiod) == 2
        assert_in(period1dict, user.get_inactive_periods())
        assert_in(period2dict, user.get_inactive_periods())

        # Remove first inactivity period
        r = self.app.post(
            '/auth/user_info/availability/remove_inactive_period',
            params=dict(
                startdate=startdate.strftime('%d/%m/%Y'),
                enddate=enddate.strftime('%d/%m/%Y'),
                _session_id=self.app.cookies['_session_id'],
            ))
        user = M.User.query.get(username='test-admin')
        assert len(user.inactiveperiod) == 1 and period2dict in user.get_inactive_periods()

        # Add invalid inactivity period
        r = self.app.post('/auth/user_info/availability/add_inactive_period',
                          params=dict(
                              startdate='NOT/A/DATE',
                              enddate=enddate2.strftime('%d/%m/%Y'),
                              _session_id=self.app.cookies['_session_id'],
                          ))
        user = M.User.query.get(username='test-admin')
        assert 'Please enter a valid date' in str(r)
        assert len(user.inactiveperiod) == 1 and period2dict in user.get_inactive_periods()

    @td.with_user_project('test-admin')
    def test_skills(self):
        setup_trove_categories()
        # Add a skill
        skill_cat = M.TroveCategory.query.get(show_as_skill=True)
        level = 'low'
        comment = 'test comment'
        self.app.get('/auth/user_info/skills/')
        self.app.post('/auth/user_info/skills/save_skill',
                      params=dict(
                          level=level,
                          comment=comment,
                          selected_skill=str(skill_cat.trove_cat_id),
                          _session_id=self.app.cookies['_session_id'],
                      ))
        user = M.User.query.get(username='test-admin')
        skilldict = dict(category_id=skill_cat._id,
                         comment=comment, level=level)
        assert len(user.skills) == 1 and skilldict in user.skills

        # Add again the same skill
        level = 'medium'
        comment = 'test comment 2'
        self.app.get('/auth/user_info/skills/')
        self.app.post('/auth/user_info/skills/save_skill',
                      params=dict(
                          level=level,
                          comment=comment,
                          selected_skill=str(skill_cat.trove_cat_id),
                          _session_id=self.app.cookies['_session_id'],
                      ))
        user = M.User.query.get(username='test-admin')
        skilldict = dict(category_id=skill_cat._id,
                         comment=comment, level=level)
        assert len(user.skills) == 1 and skilldict in user.skills

        # Add an invalid skill
        level2 = 'not a level'
        comment2 = 'test comment 2'
        self.app.post('/auth/user_info/skills/save_skill',
                      params=dict(
                          level=level2,
                          comment=comment2,
                          selected_skill=str(skill_cat.trove_cat_id),
                          _session_id=self.app.cookies['_session_id'],
                      ))
        user = M.User.query.get(username='test-admin')
        # Check that everything is as it was before
        assert len(user.skills) == 1 and skilldict in user.skills

        # Remove a skill
        self.app.get('/auth/user_info/skills/')
        self.app.post('/auth/user_info/skills/remove_skill',
                      params=dict(
                          categoryid=str(skill_cat.trove_cat_id),
                          _session_id=self.app.cookies['_session_id'],
                      ))
        user = M.User.query.get(username='test-admin')
        assert len(user.skills) == 0

    @td.with_user_project('test-admin')
    def test_user_message(self):
        self.app.get('/').follow()  # establish session
        assert not M.User.query.get(username='test-admin').get_pref('disable_user_messages')
        self.app.post('/auth/preferences/user_message',
                      params={'_session_id': self.app.cookies['_session_id'],
                              })
        assert M.User.query.get(username='test-admin').get_pref('disable_user_messages')
        self.app.post('/auth/preferences/user_message',
                      params={'allow_user_messages': 'on',
                              '_session_id': self.app.cookies['_session_id'],
                              })
        assert not M.User.query.get(username='test-admin').get_pref('disable_user_messages')

    @td.with_user_project('test-admin')
    def test_additional_page(self):
        class MyPP(plugin.UserPreferencesProvider):
            def not_page(self):
                return 'not page'

            @expose()
            def new_page(self):
                return 'new page'

        with mock.patch.object(plugin.UserPreferencesProvider, 'get') as upp_get:
            upp_get.return_value = MyPP()
            r = self.app.get('/auth/new_page')
            assert_equal(r.text, 'new page')
            self.app.get('/auth/not_page', status=404)


class TestPasswordReset(TestController):
    test_primary_email = 'testprimaryaddr@mail.com'

    def setUp(self):
        super().setUp()
        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True'}

    @patch('allura.model.User.send_password_reset_email')
    @patch('allura.lib.plugin.LocalAuthenticationProvider.resend_verification_link')
    @patch('allura.tasks.mail_tasks.sendmail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_email_unconfirmed(self, gen_message_id, sendmail, p_sendlink, p_sendpwd):
        user = M.User.query.get(username='test-admin')
        user.pending = True
        email = M.EmailAddress.find(
            {'claimed_by_user_id': user._id}).first()
        email.confirmed = False
        ThreadLocalORMSession.flush_all()
        self.app.get('/').follow()  # establish session
        self.app.post('/auth/password_recovery_hash', {'email': email.email,
                                                       '_session_id': self.app.cookies['_session_id'],
                                                       })
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        assert hash is None
        p_sendlink.assert_called_once()
        p_sendpwd.assert_not_called()

    @patch('allura.tasks.mail_tasks.sendmail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_disabled(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.find(
            {'claimed_by_user_id': user._id}).first()
        user.disabled = True
        ThreadLocalORMSession.flush_all()
        self.app.get('/').follow()  # establish session
        self.app.post('/auth/password_recovery_hash', {'email': email.email,
                                                       '_session_id': self.app.cookies['_session_id'],
                                                       })
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        assert hash is None

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_only_primary_email_reset_allowed(self, gen_message_id, sendmail):
        self.app.get('/').follow()  # establish session
        user = M.User.query.get(username='test-admin')
        user.claim_address(self.test_primary_email)
        user.set_pref('email_address', self.test_primary_email)

        email = M.EmailAddress.find({'email': self.test_primary_email}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()

        with h.push_config(config, **{'auth.allow_non_primary_email_password_reset': 'false'}):
            self.app.post('/auth/password_recovery_hash', {'email': self.test_primary_email,
                                                           '_session_id': self.app.cookies['_session_id'],
                                                           })
            hash = user.get_tool_data('AuthPasswordReset', 'hash')
            assert hash is not None
            args, kwargs = sendmail.post.call_args
            assert_equal(kwargs['toaddr'], self.test_primary_email)

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_non_primary_email_reset_allowed(self, gen_message_id, sendmail):
        self.app.get('/').follow()  # establish session
        user = M.User.query.get(username='test-admin')
        email1 = M.EmailAddress.find({'claimed_by_user_id': user._id}).first()
        user.claim_address(self.test_primary_email)
        user.set_pref('email_address', self.test_primary_email)
        email = M.EmailAddress.find({'email': self.test_primary_email}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()
        with h.push_config(config, **{'auth.allow_non_primary_email_password_reset': 'true'}):
            self.app.post('/auth/password_recovery_hash', {'email': email1.email,
                                                           '_session_id': self.app.cookies['_session_id'],
                                                           })
            hash = user.get_tool_data('AuthPasswordReset', 'hash')
            assert hash is not None
            args, kwargs = sendmail.post.call_args
            assert_equal(kwargs['toaddr'], email1.email)

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_password_reset(self, gen_message_id, sendsimplemail):
        self.app.get('/').follow()  # establish session
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.find({'claimed_by_user_id': user._id}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()
        old_pw_hash = user.password

        # request a reset
        with td.audits('Password recovery link sent to: ' + email.email, user=True):
            r = self.app.post('/auth/password_recovery_hash', {'email': email.email,
                                                               '_session_id': self.app.cookies['_session_id'],
                                                               })
        # confirm some fields
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        hash_expiry = user.get_tool_data('AuthPasswordReset', 'hash_expiry')
        assert hash is not None
        assert hash_expiry is not None

        # confirm email sent
        text = '''Your username is test-admin

To update your password on %s, please visit the following URL:

%s/auth/forgotten_password/%s''' % (config['site_name'], config['base_url'], hash)
        sendsimplemail.post.assert_called_once_with(
            sender='noreply@localhost',
            toaddr=email.email,
            fromaddr='"{}" <{}>'.format(config['site_name'], config['forgemail.return_path']),
            reply_to=config['forgemail.return_path'],
            subject='Allura Password recovery',
            message_id=gen_message_id(),
            text=text)

        # load reset form and fill it out
        r = self.app.get('/auth/forgotten_password/%s' % hash)
        assert_in('Enter a new password for: test-admin', r)
        assert_in('New Password:', r)
        assert_in('New Password (again):', r)
        form = r.forms[0]
        form['pw'] = form['pw2'] = new_password = '154321'
        with td.audits(r'Password changed \(through recovery process\)', user=True):
            # escape parentheses, so they would not be treated as regex group
            r = form.submit()

        # verify 'Password Changed' email sent
        args, kwargs = sendsimplemail.post.call_args
        assert_equal(kwargs['toaddr'], user._id)
        assert_equal(kwargs['subject'], 'Password Changed')

        # confirm password changed and works
        user = M.User.query.get(username='test-admin')
        assert_not_equal(old_pw_hash, user.password)
        provider = plugin.LocalAuthenticationProvider(None)
        assert_true(provider._validate_password(user, new_password))

        # confirm reset fields cleared
        user = M.User.query.get(username='test-admin')
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        hash_expiry = user.get_tool_data('AuthPasswordReset', 'hash_expiry')
        assert_equal(hash, '')
        assert_equal(hash_expiry, '')

        # confirm can log in now in same session
        r = r.follow()
        assert 'Log Out' not in r, r
        form = r.forms[0]
        encoded = self.app.antispam_field_names(r.form)
        form[encoded['username']] = 'test-admin'
        form[encoded['password']] = new_password
        r = form.submit(status=302)
        r = r.follow().follow()
        assert 'Log Out' in r, r

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_hash_expired(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.find(
            {'claimed_by_user_id': user._id}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()
        self.app.get('/').follow()  # establish session
        r = self.app.post('/auth/password_recovery_hash', {'email': email.email,
                                                           '_session_id': self.app.cookies['_session_id'],
                                                           })
        user = M.User.by_username('test-admin')
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        user.set_tool_data('AuthPasswordReset',
                           hash_expiry=datetime(2000, 10, 10))
        r = self.app.get('/auth/forgotten_password/%s' % hash.encode('utf-8'))
        assert_in('Unable to process reset, please try again', r.follow().text)
        r = self.app.post('/auth/set_new_password/%s' %
                          hash.encode('utf-8'), {'pw': '154321', 'pw2': '154321',
                                                 '_session_id': self.app.cookies['_session_id'],
                                                 })
        assert_in('Unable to process reset, please try again', r.follow().text)

    def test_hash_invalid(self):
        r = self.app.get('/auth/forgotten_password/123412341234', status=302)
        assert_in('Unable to process reset, please try again', r.follow().text)

    @patch('allura.lib.plugin.AuthenticationProvider')
    def test_provider_disabled(self, AP):
        user = M.User.query.get(username='test-admin')
        ap = AP.get()
        ap.forgotten_password_process = False
        ap.authenticate_request()._id = user._id
        ap.by_username().username = user.username
        self.app.get('/auth/forgotten_password', status=404)
        self.app.get('/').follow()  # establish session
        self.app.post('/auth/set_new_password',
                      {'pw': 'foo', 'pw2': 'foo', '_session_id': self.app.cookies['_session_id']},
                      status=404)
        self.app.post('/auth/password_recovery_hash',
                      {'email': 'foo', '_session_id': self.app.cookies['_session_id']},
                      status=404)

    @patch('allura.lib.plugin.AuthenticationProvider.hibp_password_check_enabled', Mock(return_value=True))
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_pwd_reset_hibp_check(self, gen_message_id, sendmail):
        self.app.get('/').follow()  # establish session
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.find({'claimed_by_user_id': user._id}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()

        # request a reset
        r = self.app.post('/auth/password_recovery_hash', {'email': email.email,
                                                           '_session_id': self.app.cookies['_session_id'],
                                                           })
        hash = user.get_tool_data('AuthPasswordReset', 'hash')

        # load reset form and fill it out with weak password
        r = self.app.get('/auth/forgotten_password/%s' % hash)
        form = r.forms[0]
        form['pw'] = form['pw2'] = new_password = 'password'
        r = form.submit()
        assert 'Unsafe' in str(r.headers)

        # fill it out again, with a stronger password
        r = r.follow()
        form = r.forms[0]
        form['pw'] = form['pw2'] = new_password = 'oj35h9u34280j924hnuiw'  # something unlikely to trip at hibp
        r = form.submit()
        assert 'Unsafe' not in str(r.headers)

        # confirm password changed and works
        user = M.User.query.get(username='test-admin')
        provider = plugin.LocalAuthenticationProvider(None)
        assert_true(provider._validate_password(user, new_password))

        # confirm can log in now in same session
        r = r.follow()
        assert 'Log Out' not in r, r
        form = r.forms[0]
        encoded = self.app.antispam_field_names(r.form)
        form[encoded['username']] = 'test-admin'
        form[encoded['password']] = new_password
        r = form.submit(status=302)
        r = r.follow().follow()
        assert 'Log Out' in r, r


class TestOAuth(TestController):
    def test_register_deregister_app(self):
        # register
        r = self.app.get('/auth/oauth/')
        r = self.app.post('/auth/oauth/register',
                          params={'application_name': 'oautstapp', 'application_description': 'Oauth rulez',
                                  '_session_id': self.app.cookies['_session_id'],
                                  }).follow()
        assert 'oautstapp' in r
        # deregister
        assert_equal(r.forms[0].action, 'deregister')
        r.forms[0].submit()
        r = self.app.get('/auth/oauth/')
        assert 'oautstapp' not in r

    def test_generate_revoke_access_token(self):
        # generate
        self.app.get('/').follow()  # establish session
        r = self.app.post('/auth/oauth/register',
                          params={'application_name': 'oautstapp', 'application_description': 'Oauth rulez',
                                  '_session_id': self.app.cookies['_session_id'],
                                  }, status=302)
        r = self.app.get('/auth/oauth/')
        assert_equal(r.forms[1].action, 'generate_access_token')
        r = r.forms[1].submit(extra_environ={'username': 'test-user'})  # not the right user
        assert_in("Invalid app ID", self.webflash(r))                   # gets an error
        r = self.app.get('/auth/oauth/')                                # do it again
        r = r.forms[1].submit()                                         # as correct user
        assert_equal('', self.webflash(r))

        r = self.app.get('/auth/oauth/')
        assert 'Bearer Token:' in r
        assert_not_equal(
            M.OAuthAccessToken.for_user(M.User.by_username('test-admin')), [])
        # revoke
        assert_equal(r.forms[0].action, 'revoke_access_token')
        r.forms[0].submit()
        r = self.app.get('/auth/oauth/')
        assert_not_equal(r.forms[0].action, 'revoke_access_token')
        assert_equal(
            M.OAuthAccessToken.for_user(M.User.by_username('test-admin')), [])

    def test_interactive(self):
        user = M.User.by_username('test-admin')
        M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            secret_key='test-client-secret',
            user_id=user._id,
            description='ctok_desc',
        )
        ThreadLocalORMSession.flush_all()
        oauth_params = dict(
            client_key='api_key_api_key_12345',
            client_secret='test-client-secret',
            callback_uri='http://my.domain.com/callback',
        )
        r = self.app.post(*oauth1_webtest('/rest/oauth/request_token', oauth_params, method='POST'))
        rtok = parse_qs(r.text)['oauth_token'][0]
        rsecr = parse_qs(r.text)['oauth_token_secret'][0]
        assert rtok
        assert rsecr
        r = self.app.post('/rest/oauth/authorize',
                          params={'oauth_token': rtok})
        r = r.forms[0].submit('yes')
        assert r.location.startswith('http://my.domain.com/callback')
        pin = parse_qs(urlparse(r.location).query)['oauth_verifier'][0]
        assert pin

        oauth_params = dict(
            client_key='api_key_api_key_12345',
            client_secret='test-client-secret',
            resource_owner_key=rtok,
            resource_owner_secret=rsecr,
            verifier=pin,
        )
        r = self.app.get(*oauth1_webtest('/rest/oauth/access_token', oauth_params))
        atok = parse_qs(r.text)
        assert_equal(len(atok['oauth_token']), 1)
        assert_equal(len(atok['oauth_token_secret']), 1)

        # now use the tokens & secrets to make a full OAuth request:
        oauth_token = atok['oauth_token'][0]
        oauth_secret = atok['oauth_token_secret'][0]
        oaurl, oaparams, oahdrs = oauth1_webtest('/rest/p/test/', dict(
            client_key='api_key_api_key_12345',
            client_secret='test-client-secret',
            resource_owner_key=oauth_token,
            resource_owner_secret=oauth_secret,
            signature_type='query'
        ))
        self.app.get(oaurl, oaparams, oahdrs, status=200)
        self.app.get(oaurl.replace('oauth_signature=', 'removed='), oaparams, oahdrs, status=401)

    def test_authorize_ok(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            consumer_token_id=ctok._id,
            callback='oob',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/authorize', params={'oauth_token': 'api_key_reqtok_12345'})
        assert_in('ctok_desc', r.text)
        assert_in('api_key_reqtok_12345', r.text)

    def test_authorize_invalid(self):
        self.app.post('/rest/oauth/authorize', params={'oauth_token': 'api_key_reqtok_12345'}, status=401)

    def test_do_authorize_no(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            consumer_token_id=ctok._id,
            callback='oob',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        self.app.post('/rest/oauth/do_authorize',
                      params={'no': '1', 'oauth_token': 'api_key_reqtok_12345'})
        assert_is_none(M.OAuthRequestToken.query.get(api_key='api_key_reqtok_12345'))

    def test_do_authorize_oob(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            consumer_token_id=ctok._id,
            callback='oob',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/do_authorize', params={'yes': '1', 'oauth_token': 'api_key_reqtok_12345'})
        assert_is_not_none(r.html.find(text=re.compile('^PIN: ')))

    def test_do_authorize_cb(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/do_authorize', params={'yes': '1', 'oauth_token': 'api_key_reqtok_12345'})
        assert r.location.startswith('http://my.domain.com/callback?oauth_token=api_key_reqtok_12345&oauth_verifier=')

    def test_do_authorize_cb_params(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/do_authorize', params={'yes': '1', 'oauth_token': 'api_key_reqtok_12345'})
        assert r.location.startswith('http://my.domain.com/callback?myparam=foo&oauth_token=api_key_reqtok_12345&oauth_verifier=')


class TestOAuthRequestToken(TestController):

    oauth_params = dict(
        client_key='api_key_api_key_12345',
        client_secret='test-client-secret',
    )

    def test_request_token_valid(self):
        user = M.User.by_username('test-user')
        consumer_token = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            secret_key='test-client-secret',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post(*oauth1_webtest('/rest/oauth/request_token', self.oauth_params, method='POST'))

        request_token = M.OAuthRequestToken.query.get(consumer_token_id=consumer_token._id)
        assert_is_not_none(request_token)
        assert_equal(r.text, request_token.to_string())

    def test_request_token_no_consumer_token_matching(self):
        with LogCapture() as logs:
            self.app.post(*oauth1_webtest('/rest/oauth/request_token', self.oauth_params), status=401)
        assert_logmsg(logs, 'Invalid consumer token')

    def test_request_token_no_consumer_token_given(self):
        oauth_params = self.oauth_params.copy()
        oauth_params['signature_type'] = 'query'  # so we can more easily remove a param next
        url, params, hdrs = oauth1_webtest('/rest/oauth/request_token', oauth_params)
        url = url.replace('oauth_consumer_key', 'gone')
        with LogCapture() as logs:
            self.app.post(url, params, hdrs, status=401)
        assert_logmsg(logs, 'Invalid consumer token')

    def test_request_token_invalid(self):
        user = M.User.by_username('test-user')
        M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            secret_key='test-client-secret--INVALID',
        )
        ThreadLocalORMSession.flush_all()
        with LogCapture() as logs:
            self.app.post(*oauth1_webtest('/rest/oauth/request_token', self.oauth_params, method='POST'),
                          status=401)
        assert_logmsg(logs, "Invalid signature <class 'oauth2.Error'> Invalid signature.")


class TestOAuthAccessToken(TestController):

    oauth_params = dict(
        client_key='api_key_api_key_12345',
        client_secret='test-client-secret',
        resource_owner_key='api_key_reqtok_12345',
        resource_owner_secret='test-token-secret',
        verifier='good_verifier_123456',
    )

    def test_access_token_no_consumer(self):
        with LogCapture() as logs:
            self.app.get(*oauth1_webtest('/rest/oauth/access_token', self.oauth_params), status=401)
        assert_logmsg(logs, 'Invalid consumer token')

    def test_access_token_no_request(self):
        user = M.User.by_username('test-admin')
        M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
        )
        ThreadLocalORMSession.flush_all()
        with LogCapture() as logs:
            self.app.get(*oauth1_webtest('/rest/oauth/access_token', self.oauth_params), status=401)
        assert_logmsg(logs, 'Invalid request token')

    def test_access_token_bad_pin(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
            validation_pin='good_verifier_123456',
        )
        ThreadLocalORMSession.flush_all()
        with LogCapture() as logs:
            oauth_params = self.oauth_params.copy()
            oauth_params['verifier'] = 'bad_verifier_1234567'
            self.app.get(*oauth1_webtest('/rest/oauth/access_token', oauth_params),
                         status=401)
        assert_logmsg(logs, 'Invalid verifier')

    def test_access_token_bad_sig(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            user_id=user._id,
            description='ctok_desc',
            secret_key='test-client-secret',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
            validation_pin='good_verifier_123456',
            secret_key='test-token-secret--INVALID',
        )
        ThreadLocalORMSession.flush_all()
        with LogCapture() as logs:
            self.app.get(*oauth1_webtest('/rest/oauth/access_token', self.oauth_params), status=401)
        assert_logmsg(logs, "Invalid signature <class 'oauth2.Error'> Invalid signature.")

    def test_access_token_ok(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key_api_key_12345',
            secret_key='test-client-secret',
            user_id=user._id,
            description='ctok_desc',
        )
        M.OAuthRequestToken(
            api_key='api_key_reqtok_12345',
            secret_key='test-token-secret',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
            validation_pin='good_verifier_123456',
        )
        ThreadLocalORMSession.flush_all()

        r = self.app.get(*oauth1_webtest('/rest/oauth/access_token', self.oauth_params))
        atok = parse_qs(r.text)
        assert_equal(len(atok['oauth_token']), 1)
        assert_equal(len(atok['oauth_token_secret']), 1)

        oauth_params = dict(self.oauth_params, signature_type='query')
        r = self.app.get(*oauth1_webtest('/rest/oauth/access_token', oauth_params))
        atok = parse_qs(r.text)
        assert_equal(len(atok['oauth_token']), 1)
        assert_equal(len(atok['oauth_token_secret']), 1)


class TestDisableAccount(TestController):
    def test_not_authenticated(self):
        r = self.app.get(
            '/auth/disable/',
            extra_environ={'username': '*anonymous'})
        assert_equal(r.status_int, 302)
        assert_equal(r.location,
                     'http://localhost/auth/?return_to=%2Fauth%2Fdisable%2F')

    def test_lists_user_projects(self):
        r = self.app.get('/auth/disable/')
        user = M.User.by_username('test-admin')
        for p in user.my_projects_by_role_name('Admin'):
            if p.name == 'u/test-admin':
                continue
            assert_in(p.name, r)
            assert_in(p.url(), r)

    def test_has_asks_password(self):
        r = self.app.get('/auth/disable/')
        form = r.html.find('form', {'action': 'do_disable'})
        assert form is not None

    def test_bad_password(self):
        self.app.get('/').follow()  # establish session
        r = self.app.post('/auth/disable/do_disable', {'password': 'bad',
                                                       '_session_id': self.app.cookies['_session_id'], })
        assert_in('Invalid password', r)
        user = M.User.by_username('test-admin')
        assert_equal(user.disabled, False)

    def test_disable(self):
        self.app.get('/').follow()  # establish session
        r = self.app.post('/auth/disable/do_disable', {'password': 'foo',
                                                       '_session_id': self.app.cookies['_session_id'], })
        assert_equal(r.status_int, 302)
        assert_equal(r.location, 'http://localhost/')
        flash = json.loads(self.webflash(r))
        assert_equal(flash['status'], 'ok')
        assert_equal(flash['message'], 'Your account was successfully disabled!')
        user = M.User.by_username('test-admin')
        assert_equal(user.disabled, True)


class TestPasswordExpire(TestController):
    def login(self, username='test-user', pwd='foo', query_string=''):
        extra = {'username': '*anonymous', 'REMOTE_ADDR': '127.0.0.1'}
        r = self.app.get('/auth/' + query_string, extra_environ=extra)

        f = r.forms[0]
        encoded = self.app.antispam_field_names(f)
        f[encoded['username']] = username
        f[encoded['password']] = pwd
        return f.submit(extra_environ={'username': '*anonymous'})

    def assert_redirects(self, where='/'):
        resp = self.app.get(where, extra_environ={'username': 'test-user'}, status=302)
        assert_equal(resp.location, 'http://localhost/auth/pwd_expired?' + urlencode({'return_to': where}))

    def assert_not_redirects(self, where='/neighborhood'):
        self.app.get(where, extra_environ={'username': 'test-user'}, status=200)

    def test_disabled(self):
        r = self.login()
        assert_false(r.session.get('pwd-expired'))
        self.assert_not_redirects()

    def expired(self, r):
        return r.session.get('pwd-expired')

    def set_expire_for_user(self, username='test-user', days=100):
        user = M.User.by_username(username)
        user.last_password_updated = datetime.utcnow() - timedelta(days=days)
        session(user).flush(user)
        return user

    def test_days(self):
        self.set_expire_for_user()

        with h.push_config(config, **{'auth.pwdexpire.days': 180}):
            r = self.login()
            assert_false(self.expired(r))
            self.assert_not_redirects()

        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

    def test_before(self):
        self.set_expire_for_user()

        before = datetime.utcnow() - timedelta(days=180)
        before = calendar.timegm(before.timetuple())
        with h.push_config(config, **{'auth.pwdexpire.before': before}):
            r = self.login()
            assert_false(self.expired(r))
            self.assert_not_redirects()

        before = datetime.utcnow() - timedelta(days=90)
        before = calendar.timegm(before.timetuple())
        with h.push_config(config, **{'auth.pwdexpire.before': before}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

    def test_logout(self):
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()
            r = self.app.get('/auth/logout', extra_environ={'username': 'test-user'})
            assert_false(self.expired(r))
            self.assert_not_redirects()

    def test_change_pwd(self):
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

            user = M.User.by_username('test-user')
            old_update_time = user.last_password_updated
            old_password = user.password
            r = self.app.get('/auth/pwd_expired', extra_environ={'username': 'test-user'})
            f = r.forms[0]
            f['oldpw'] = 'foo'
            f['pw'] = 'qwerty'
            f['pw2'] = 'qwerty'
            r = f.submit(extra_environ={'username': 'test-user'}, status=302)
            assert_equal(r.location, 'http://localhost/')
            assert_false(self.expired(r))
            user = M.User.by_username('test-user')
            assert_true(user.last_password_updated > old_update_time)
            assert_not_equal(user.password, old_password)

            # Can log in with new password and change isn't required anymore
            r = self.login(pwd='qwerty').follow()
            assert_equal(r.location, 'http://localhost/dashboard')
            assert_not_in('Invalid login', r)
            assert_false(self.expired(r))
            self.assert_not_redirects()

            # and can't log in with old password
            r = self.login(pwd='foo')
            assert_in('Invalid login', r)

    def test_expired_pwd_change_invalidates_token(self):
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()
            user = M.User.by_username('test-user')
            user.set_tool_data('AuthPasswordReset',
                               hash="generated_hash_value",
                               hash_expiry="04-08-2020")
            hash = user.get_tool_data('AuthPasswordReset', 'hash')
            hash_expiry = user.get_tool_data('AuthPasswordReset', 'hash_expiry')
            assert_equal(hash, 'generated_hash_value')
            assert_equal(hash_expiry, '04-08-2020')
            session(user).flush(user)

            # Change expired password
            r = self.app.get('/auth/pwd_expired', extra_environ={'username': 'test-user'})
            f = r.forms[0]
            f['oldpw'] = 'foo'
            f['pw'] = 'qwerty'
            f['pw2'] = 'qwerty'
            r = f.submit(extra_environ={'username': 'test-user'}, status=302)
            assert_equal(r.location, 'http://localhost/')

            user = M.User.by_username('test-user')
            hash = user.get_tool_data('AuthPasswordReset', 'hash')
            hash_expiry = user.get_tool_data('AuthPasswordReset', 'hash_expiry')

            assert_equal(hash, '')
            assert_equal(hash_expiry, '')

    def check_validation(self, oldpw, pw, pw2):
        user = M.User.by_username('test-user')
        old_update_time = user.last_password_updated
        old_password = user.password
        r = self.app.get('/auth/pwd_expired', extra_environ={'username': 'test-user'})
        f = r.forms[0]
        f['oldpw'] = oldpw
        f['pw'] = pw
        f['pw2'] = pw2
        r = f.submit(extra_environ={'username': 'test-user'})
        assert_true(self.expired(r))
        user = M.User.by_username('test-user')
        assert_equal(user.last_password_updated, old_update_time)
        assert_equal(user.password, old_password)
        return r

    def test_change_pwd_validation(self):
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

            r = self.check_validation('', '', '')
            assert_in('Please enter a value', r)
            r = self.check_validation('', 'qwe', 'qwerty')
            assert_in('Enter a value 6 characters long or more', r)
            r = self.check_validation('bad', 'qwerty1', 'qwerty')
            assert_in('Passwords must match', r)
            r = self.check_validation('bad', 'qwerty', 'qwerty')
            assert_in('Incorrect password', self.webflash(r))
            assert_equal(r.location, 'http://localhost/auth/pwd_expired?return_to=')

            with h.push_config(config, **{'auth.min_password_len': 3}):
                r = self.check_validation('foo', 'foo', 'foo')
                assert_in('Your old and new password should not be the same', r)

    def test_return_to(self):
        return_to = '/p/test/tickets/?milestone=1.0&page=2'
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login(query_string='?' + urlencode({'return_to': return_to}))
            # don't go to the return_to yet
            assert_equal(r.location, 'http://localhost/auth/pwd_expired?' + urlencode({'return_to': return_to}))

            # but if user tries to go directly there anyway, intercept and redirect back
            self.assert_redirects(where=return_to)

            r = self.app.get('/auth/pwd_expired', extra_environ={'username': 'test-user'})
            f = r.forms[0]
            f['oldpw'] = 'foo'
            f['pw'] = 'qwerty'
            f['pw2'] = 'qwerty'
            f['return_to'] = return_to
            r = f.submit(extra_environ={'username': 'test-user'}, status=302)
            assert_equal(r.location, 'http://localhost/p/test/tickets/?milestone=1.0&page=2')


class TestCSRFProtection(TestController):
    def test_blocks_invalid(self):
        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True', 'REMOTE_ADDR': '127.0.0.1'}

        # regular login
        r = self.app.get('/auth/')

        r = self.app.post('/auth/do_login', params=dict(
            username='test-admin', password='foo',
            _session_id=self.app.cookies['_session_id']),
            antispam=True)

        # regular form submit
        r = self.app.get('/admin/overview')
        r = r.form.submit()
        assert_equal(r.location, 'http://localhost/admin/overview')

        # invalid form submit
        r = self.app.get('/admin/overview')
        r.form['_session_id'] = 'bogus'
        r = r.form.submit()
        assert_equal(r.location, 'http://localhost/auth/')

    def test_blocks_invalid_on_login(self):
        r = self.app.get('/auth/')
        r.form['_session_id'] = 'bogus'
        r.form.submit(status=403)

    def test_token_present_on_first_request(self):
        r = self.app.get('/auth/')
        assert_true(r.form['_session_id'].value)


class TestTwoFactor(TestController):

    sample_key = b'\x00K\xda\xbfv\xc2B\xaa\x1a\xbe\xa5\x96b\xb2\xa0Z:\xc9\xcf\x8a'
    sample_b32 = 'ABF5VP3WYJBKUGV6UWLGFMVALI5MTT4K'

    def _init_totp(self, username='test-admin'):
        user = M.User.query.get(username=username)
        totp_srv = TotpService().get()
        totp_srv.set_secret_key(user, self.sample_key)
        user.set_pref('multifactor', True)

    def test_settings_on(self):
        r = self.app.get('/auth/preferences/')
        assert r.html.find(attrs={'class': 'preferences multifactor'})

    def test_settings_off(self):
        with h.push_config(config, **{'auth.multifactor.totp': 'false'}):
            r = self.app.get('/auth/preferences/')
            assert not r.html.find(attrs={'class': 'preferences multifactor'})

            for url in ['/auth/preferences/totp_new',
                        '/auth/preferences/totp_view',
                        '/auth/preferences/totp_set',
                        '/auth/preferences/totp_send_link',
                        '/auth/preferences/multifactor_disable',
                        '/auth/preferences/multifactor_recovery',
                        '/auth/preferences/multifactor_recovery_regen',
                        '/auth/multifactor',
                        '/auth/do_multifactor',
                        ]:
                self.app.post(url,
                              {'password': 'foo', '_session_id': self.app.cookies['_session_id']},
                              status=404)

    def test_user_disabled(self):
        r = self.app.get('/auth/preferences/')
        info_html = str(r.html.find(attrs={'class': 'preferences multifactor'}))
        assert_in('disabled', info_html)

    def test_user_enabled(self):
        self._init_totp()
        r = self.app.get('/auth/preferences/')
        info_html = str(r.html.find(attrs={'class': 'preferences multifactor'}))
        assert_in('enabled', info_html)

    def test_reconfirm_auth(self):
        from datetime import datetime as real_datetime
        with patch('allura.lib.decorators.datetime') as datetime:
            datetime.min = real_datetime.min

            # reconfirm required at first
            datetime.utcnow.return_value = real_datetime(2016, 1, 1, 0, 0, 0)
            r = self.app.get('/auth/preferences/totp_new')
            assert_in('Password Confirmation', r)

            # submit form, and its not required
            r.form['password'] = 'foo'
            r = r.form.submit()
            assert_not_in('Password Confirmation', r)

            # still not required
            datetime.utcnow.return_value = real_datetime(2016, 1, 1, 0, 1, 45)
            r = self.app.get('/auth/preferences/totp_new')
            assert_not_in('Password Confirmation', r)

            # required later
            datetime.utcnow.return_value = real_datetime(2016, 1, 1, 0, 2, 3)
            r = self.app.get('/auth/preferences/totp_new')
            assert_in('Password Confirmation', r)

    def test_enable_totp(self):
        # create a separate session, for later use in the test
        other_session = TestController()
        other_session.setUp()
        other_session.app.get('/auth/preferences/')

        with out_audits(user=True):
            r = self.app.get('/auth/preferences/totp_new')
            assert_in('Password Confirmation', r)

        with audits('Visited multifactor new TOTP page', user=True):
            r.form['password'] = 'foo'
            r = r.form.submit()
            assert_in('Scan this', r)
            assert_in('Or enter setup key: ', r)

        first_key_shown = r.session['totp_new_key']

        with audits(r'Failed to set up multifactor TOTP \(wrong code\)', user=True):
            form = r.forms['totp_set']
            form['code'] = ''
            r = form.submit()
            assert_in('Invalid', r)
            assert_in(f'Or enter setup key: {b32encode(first_key_shown).decode()}', r)
            assert_equal(first_key_shown, r.session['totp_new_key'])  # different keys on each pageload would be bad!

        new_totp = TotpService().Totp(r.session['totp_new_key'])
        code = new_totp.generate(time_time())
        form = r.forms['totp_set']
        form['code'] = code
        with audits('Set up multifactor TOTP', user=True):
            r = form.submit()
            assert_equal('Two factor authentication has now been set up.', json.loads(self.webflash(r))['message'],
                         self.webflash(r))

        tasks = M.MonQTask.query.find(dict(task_name='allura.tasks.mail_tasks.sendsimplemail')).all()
        assert_equal(len(tasks), 1)
        assert_equal(tasks[0].kwargs['subject'], 'Two-Factor Authentication Enabled')
        assert_in('new two-factor authentication', tasks[0].kwargs['text'])

        r = r.follow()
        assert_in('Recovery Codes', r)

        # Confirm any pre-existing sessions have to re-authenticate
        r = other_session.app.get('/auth/preferences/')
        assert_in('/auth/?return_to', r.headers['Location'])
        other_session.tearDown()

    def test_reset_totp(self):
        self._init_totp()

        # access page
        r = self.app.get('/auth/preferences/totp_new')
        assert_in('Password Confirmation', r)

        # reconfirm password to get to it
        r.form['password'] = 'foo'
        r = r.form.submit()

        # confirm warning message, and key is not changed yet
        assert_in('Scan this', r)
        assert_in('Or enter setup key: ', r)
        assert_in('this will invalidate your previous', r)
        current_key = TotpService.get().get_secret_key(M.User.query.get(username='test-admin'))
        assert_equal(self.sample_key, current_key)

        # incorrect submission
        form = r.forms['totp_set']
        form['code'] = ''
        r = form.submit()
        assert_in('Invalid', r)

        # still unchanged key
        current_key = TotpService.get().get_secret_key(M.User.query.get(username='test-admin'))
        assert_equal(self.sample_key, current_key)

        # valid submission
        new_key = r.session['totp_new_key']
        new_totp = TotpService().Totp(new_key)
        code = new_totp.generate(time_time())
        form = r.forms['totp_set']
        form['code'] = code
        r = form.submit()
        assert_equal('Two factor authentication has now been set up.', json.loads(self.webflash(r))['message'],
                     self.webflash(r))

        # new key in place
        current_key = TotpService.get().get_secret_key(M.User.query.get(username='test-admin'))
        assert_equal(new_key, current_key)
        assert_not_equal(self.sample_key, current_key)

    def test_disable(self):
        self._init_totp()

        self.app.get('/auth/preferences/multifactor_disable', status=405)  # GET not allowed

        # get form and submit
        r = self.app.get('/auth/preferences/')
        form = r.forms['multifactor_disable']
        r = form.submit()

        # confirm first, no change
        assert_in('Password Confirmation', r)
        user = M.User.query.get(username='test-admin')
        assert_equal(user.get_pref('multifactor'), True)

        # confirm submit, everything goes off
        r.form['password'] = 'foo'
        with audits('Disabled multifactor TOTP', user=True):
            r = r.form.submit()
            assert_equal('Multifactor authentication has now been disabled.', json.loads(self.webflash(r))['message'],
                         self.webflash(r))
        user = M.User.query.get(username='test-admin')
        assert_equal(user.get_pref('multifactor'), False)
        assert_equal(TotpService().get().get_secret_key(user), None)
        assert_equal(RecoveryCodeService().get().get_codes(user), [])

        # email confirmation
        tasks = M.MonQTask.query.find(dict(task_name='allura.tasks.mail_tasks.sendsimplemail')).all()
        assert_equal(len(tasks), 1)
        assert_equal(tasks[0].kwargs['subject'], 'Two-Factor Authentication Disabled')
        assert_in('disabled two-factor authentication', tasks[0].kwargs['text'])

    def test_login_totp(self):
        self._init_totp()

        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True'}

        # regular login
        r = self.app.get('/auth/?return_to=/p/foo')
        encoded = self.app.antispam_field_names(r.form)
        r.form[encoded['username']] = 'test-admin'
        r.form[encoded['password']] = 'foo'
        with audits('Multifactor login - password ok, code not entered yet', user=True):
            r = r.form.submit()

        # check results
        assert r.location.endswith('/auth/multifactor?return_to=%2Fp%2Ffoo'), r
        r = r.follow()
        assert not r.session.get('username')

        # try an invalid code
        r.form['code'] = 'invalid-code'
        with audits('Multifactor login - invalid code', user=True):
            r = r.form.submit()
        assert_in('Invalid code', r)
        assert not r.session.get('username')

        # use a valid code
        totp = TotpService().Totp(self.sample_key)
        code = totp.generate(time_time())
        r.form['code'] = code
        with audits('Successful login', user=True):
            r = r.form.submit()

        # confirm login and final page
        assert_equal(r.session['username'], 'test-admin')
        assert r.location.endswith('/p/foo'), r

    def test_login_rate_limit(self):
        self._init_totp()

        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True'}

        # regular login
        r = self.app.get('/auth/?return_to=/p/foo')
        encoded = self.app.antispam_field_names(r.form)

        r.form[encoded['username']] = 'test-admin'
        r.form[encoded['password']] = 'foo'
        r = r.form.submit()
        r = r.follow()

        # try some invalid codes
        for i in range(3):
            r.form['code'] = 'invalid-code'
            r = r.form.submit()
            assert_in('Invalid code', r)

        # use a valid code, but it'll hit rate limit
        totp = TotpService().Totp(self.sample_key)
        code = totp.generate(time_time())
        r.form['code'] = code
        with audits('Multifactor login - rate limit', user=True):
            r = r.form.submit()

        assert_in('rate limit exceeded', r)
        assert not r.session.get('username')

    def test_login_totp_disrupted(self):
        self._init_totp()

        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True'}

        # regular login
        r = self.app.get('/auth/')
        encoded = self.app.antispam_field_names(r.form)
        r.form[encoded['username']] = 'test-admin'
        r.form[encoded['password']] = 'foo'
        r = r.form.submit()
        r = r.follow()

        # go to some other page instead of filling out the 2FA code
        other_r = self.app.get('/')

        # then try to complete the 2FA form
        totp = TotpService().Totp(self.sample_key)
        code = totp.generate(time_time())
        r.form['code'] = code
        r = r.form.submit()

        # sent back to regular login
        assert_equal('Your multifactor login was disrupted, please start over.',
                     json.loads(self.webflash(r))['message'],
                     self.webflash(r))
        r = r.follow()
        assert_in('Password Login', r)

    def test_login_recovery_code(self):
        self._init_totp()

        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True'}

        # regular login
        r = self.app.get('/auth/?return_to=/p/foo')
        encoded = self.app.antispam_field_names(r.form)
        r.form[encoded['username']] = 'test-admin'
        r.form[encoded['password']] = 'foo'
        r = r.form.submit()

        # check results
        assert r.location.endswith('/auth/multifactor?return_to=%2Fp%2Ffoo'), r
        r = r.follow()
        assert not r.session.get('username')

        # change login mode
        r.form['mode'] = 'recovery'

        # try an invalid code
        r.form['code'] = 'invalid-code'
        r = r.form.submit()
        assert_in('Invalid code', r)
        assert not r.session.get('username')

        # use a valid code
        user = M.User.by_username('test-admin')
        recovery = RecoveryCodeService().get()
        recovery.regenerate_codes(user)
        recovery_code = recovery.get_codes(user)[0]
        r.form['code'] = recovery_code
        with audits('Logged in using a multifactor recovery code', user=True):
            r = r.form.submit()

        # confirm login and final page
        assert_equal(r.session['username'], 'test-admin')
        assert r.location.endswith('/p/foo'), r

        # confirm code used up
        assert_not_in(recovery_code, RecoveryCodeService().get().get_codes(user))

    @patch('allura.lib.plugin.AuthenticationProvider.hibp_password_check_enabled', Mock(return_value=True))
    def test_login_totp_with_hibp(self):
        # this is essentially the same as regular TOTP test, just making sure that HIBP doesn't get in the way
        # or cause any problems.  It shouldn't even run since a password isn't present when the final login happens

        self._init_totp()

        # so test-admin isn't automatically logged in for all requests
        self.app.extra_environ = {'disable_auth_magic': 'True'}

        # regular login
        r = self.app.get('/auth/?return_to=/p/foo')
        encoded = self.app.antispam_field_names(r.form)
        r.form[encoded['username']] = 'test-admin'
        r.form[encoded['password']] = 'foo'
        with audits('Multifactor login - password ok, code not entered yet', user=True):
            r = r.form.submit()

        # check results
        assert r.location.endswith('/auth/multifactor?return_to=%2Fp%2Ffoo'), r
        r = r.follow()
        assert not r.session.get('username')

        # use a valid code
        totp = TotpService().Totp(self.sample_key)
        code = totp.generate(time_time())
        r.form['code'] = code
        with audits('Successful login', user=True):
            r = r.form.submit()

        # confirm login and final page
        assert_equal(r.session['username'], 'test-admin')
        assert r.location.endswith('/p/foo'), r

    def test_view_key(self):
        self._init_totp()

        with out_audits(user=True):
            r = self.app.get('/auth/preferences/totp_view')
            assert_in('Password Confirmation', r)

        with audits('Viewed multifactor TOTP config page', user=True):
            r.form['password'] = 'foo'
            r = r.form.submit()
            assert_in('Scan this', r)
            assert_in(f'Or enter setup key: {self.sample_b32}', r)

    def test_view_recovery_codes_and_regen(self):
        self._init_totp()

        # reconfirm password
        with out_audits(user=True):
            r = self.app.get('/auth/preferences/multifactor_recovery')
            assert_in('Password Confirmation', r)

        # actual visit
        with audits('Viewed multifactor recovery codes', user=True):
            r.form['password'] = 'foo'
            r = r.form.submit()
            assert_in('Download', r)
            assert_in('Print', r)

        # regenerate codes
        with audits('Regenerated multifactor recovery codes', user=True):
            r = r.forms['multifactor_recovery_regen'].submit()

        # email confirmation
        tasks = M.MonQTask.query.find(dict(task_name='allura.tasks.mail_tasks.sendsimplemail')).all()
        assert_equal(len(tasks), 1)
        assert_equal(tasks[0].kwargs['subject'], 'Two-Factor Recovery Codes Regenerated')
        assert_in('regenerated', tasks[0].kwargs['text'])

    def test_send_links(self):
        r = self.app.get('/auth/preferences/totp_new')
        r.form['password'] = 'foo'
        r = r.form.submit()

        r = r.forms['totp_send_link'].submit()

        tasks = M.MonQTask.query.find(dict(task_name='allura.tasks.mail_tasks.sendsimplemail')).all()
        assert_equal(len(tasks), 1)
        assert_equal(tasks[0].kwargs['subject'], 'Two-Factor Authentication Apps')
        assert_in('itunes.apple.com', tasks[0].kwargs['text'])
        assert_in('play.google.com', tasks[0].kwargs['text'])