CTFd/tests/users/test_auth.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from CTFd.models import Users
from CTFd.utils import set_config, get_config
from CTFd.utils.security.signing import serialize
from CTFd.utils.security.passwords import hash_password, check_password
from freezegun import freeze_time
from tests.helpers import *
from mock import patch


def test_register_user():
    """Can a user be registered"""
    app = create_ctfd()
    with app.app_context():
        register_user(app)
        user_count = Users.query.count()
        assert user_count == 2  # There's the admin user and the created user
    destroy_ctfd(app)


def test_register_unicode_user():
    """Can a user with a unicode name be registered"""
    app = create_ctfd()
    with app.app_context():
        register_user(app, name="你好")
        user_count = Users.query.count()
        assert user_count == 2  # There's the admin user and the created user
    destroy_ctfd(app)


def test_register_duplicate_username():
    """A user shouldn't be able to use an already registered team name"""
    app = create_ctfd()
    with app.app_context():
        register_user(app, name="user1", email="user1@ctfd.io", password="password", raise_for_error=False)
        register_user(app, name="user1", email="user2@ctfd.io", password="password", raise_for_error=False)
        user_count = Users.query.count()
        assert user_count == 2  # There's the admin user and the first created user
    destroy_ctfd(app)


def test_register_duplicate_email():
    """A user shouldn't be able to use an already registered email address"""
    app = create_ctfd()
    with app.app_context():
        register_user(app, name="user1", email="user1@ctfd.io", password="password", raise_for_error=False)
        register_user(app, name="user2", email="user1@ctfd.io", password="password", raise_for_error=False)
        user_count = Users.query.count()
        assert user_count == 2  # There's the admin user and the first created user
    destroy_ctfd(app)


def test_register_whitelisted_email():
    """A user shouldn't be able to register with an email that isn't on the whitelist"""
    app = create_ctfd()
    with app.app_context():
        set_config('domain_whitelist', 'whitelisted.com, whitelisted.org, whitelisted.net')
        register_user(app, name="not_whitelisted", email='user@nope.com', raise_for_error=False)
        assert Users.query.count() == 1

        register_user(app, name="user1", email='user@whitelisted.com')
        assert Users.query.count() == 2

        register_user(app, name="user2", email='user@whitelisted.org')
        assert Users.query.count() == 3

        register_user(app, name="user3", email='user@whitelisted.net')
        assert Users.query.count() == 4
    destroy_ctfd(app)


def test_user_bad_login():
    """A user should not be able to login with an incorrect password"""
    app = create_ctfd()
    with app.app_context():
        register_user(app)
        client = login_as_user(app, name="user", password="wrong_password", raise_for_error=False)
        with client.session_transaction() as sess:
            assert sess.get('id') is None
        r = client.get('/profile')
        assert r.location.startswith("http://localhost/login")  # We got redirected to login
    destroy_ctfd(app)


def test_user_login():
    """Can a registered user can login"""
    app = create_ctfd()
    with app.app_context():
        register_user(app)
        client = login_as_user(app)
        r = client.get('/profile')
        assert r.location != "http://localhost/login"  # We didn't get redirected to login
        assert r.status_code == 200
    destroy_ctfd(app)


def test_user_login_with_email():
    """Can a registered user can login with an email address instead of a team name"""
    app = create_ctfd()
    with app.app_context():
        register_user(app)
        client = login_as_user(app, name="user@ctfd.io", password="password")
        r = client.get('/profile')
        assert r.location != "http://localhost/login"  # We didn't get redirected to login
        assert r.status_code == 200
    destroy_ctfd(app)


def test_user_get_logout():
    """Can a registered user load /logout"""
    app = create_ctfd()
    with app.app_context():
        register_user(app)
        client = login_as_user(app)
        client.get('/logout', follow_redirects=True)
        r = client.get('/challenges')
        assert r.location == "http://localhost/login?next=%2Fchallenges"
        assert r.status_code == 302
    destroy_ctfd(app)


def test_user_isnt_admin():
    """A registered user cannot access admin pages"""
    app = create_ctfd()
    with app.app_context():
        register_user(app)
        client = login_as_user(app)
        for page in ['pages', 'users', 'teams', 'scoreboard', 'challenges', 'statistics', 'config']:
            r = client.get('/admin/{}'.format(page))
            assert r.location.startswith("http://localhost/login?next=")
            assert r.status_code == 302
    destroy_ctfd(app)


def test_expired_confirmation_links():
    """Test that expired confirmation links are reported to the user"""
    app = create_ctfd()
    with app.app_context(), freeze_time("2019-02-24 03:21:34"):
        set_config('verify_emails', True)

        register_user(app, email="user@user.com")
        client = login_as_user(app, name="user", password="password")

        # user@user.com "2012-01-14 03:21:34"
        confirm_link = 'http://localhost/confirm/InVzZXJAdXNlci5jb20i.TxD0vg.cAGwAy8cK1T0saEEbrDEBVF2plI'
        r = client.get(confirm_link)

        assert "Your confirmation link has expired" in r.get_data(as_text=True)
        user = Users.query.filter_by(email='user@user.com').first()
        assert user.verified is not True
    destroy_ctfd(app)


def test_invalid_confirmation_links():
    """Test that invalid confirmation links are reported to the user"""
    app = create_ctfd()
    with app.app_context():
        set_config('verify_emails', True)

        register_user(app, email="user@user.com")
        client = login_as_user(app, name="user", password="password")

        # user@user.com "2012-01-14 03:21:34"
        confirm_link = 'http://localhost/confirm/a8375iyu<script>alert(1)<script>hn3048wueorighkgnsfg'
        r = client.get(confirm_link)

        assert "Your confirmation token is invalid" in r.get_data(as_text=True)
        user = Users.query.filter_by(email='user@user.com').first()
        assert user.verified is not True
    destroy_ctfd(app)


def test_expired_reset_password_link():
    """Test that expired reset password links are reported to the user"""
    app = create_ctfd()
    with app.app_context():
        set_config('mail_server', 'localhost')
        set_config('mail_port', 25)
        set_config('mail_useauth', True)
        set_config('mail_username', 'username')
        set_config('mail_password', 'password')

        register_user(app, name="user1", email="user@user.com")

        with app.test_client() as client, freeze_time("2019-02-24 03:21:34"):
            # user@user.com "2012-01-14 03:21:34"
            forgot_link = 'http://localhost/reset_password/InVzZXJAdXNlci5jb20i.TxD0vg.cAGwAy8cK1T0saEEbrDEBVF2plI'
            r = client.get(forgot_link)

            assert "Your link has expired" in r.get_data(as_text=True)
    destroy_ctfd(app)


def test_invalid_reset_password_link():
    """Test that invalid reset password links are reported to the user"""
    app = create_ctfd()
    with app.app_context():
        set_config('mail_server', 'localhost')
        set_config('mail_port', 25)
        set_config('mail_useauth', True)
        set_config('mail_username', 'username')
        set_config('mail_password', 'password')

        register_user(app, name="user1", email="user@user.com")

        with app.test_client() as client:
            # user@user.com "2012-01-14 03:21:34"
            forgot_link = 'http://localhost/reset_password/5678ytfghjiu876tyfg<INVALID DATA>hvbnmkoi9u87y6trdf'
            r = client.get(forgot_link)

            assert "Your reset token is invalid" in r.get_data(as_text=True)
    destroy_ctfd(app)


def test_contact_for_password_reset():
    """Test that if there is no mailserver configured, users should contact admins"""
    app = create_ctfd()
    with app.app_context():
        register_user(app, name="user1", email="user@user.com")

        with app.test_client() as client:
            forgot_link = 'http://localhost/reset_password'
            r = client.get(forgot_link)

            assert "Contact a CTF organizer" in r.get_data(as_text=True)
    destroy_ctfd(app)


@patch('smtplib.SMTP')
def test_user_can_confirm_email(mock_smtp):
    """Test that a user is capable of confirming their email address"""
    app = create_ctfd()
    with app.app_context(), freeze_time("2012-01-14 03:21:34"):
        # Set CTFd to only allow confirmed users and send emails
        set_config('verify_emails', True)
        set_config('mail_server', 'localhost')
        set_config('mail_port', 25)
        set_config('mail_useauth', True)
        set_config('mail_username', 'username')
        set_config('mail_password', 'password')

        register_user(app, name="user1", email="user@user.com")

        # Teams are not verified by default
        user = Users.query.filter_by(email='user@user.com').first()
        assert user.verified is False

        client = login_as_user(app, name="user1", password="password")

        r = client.get('http://localhost/confirm')
        assert "Need to resend the confirmation email?" in r.get_data(as_text=True)

        # smtp.sendmail was called
        mock_smtp.return_value.sendmail.assert_called()

        with client.session_transaction() as sess:
            data = {
                "nonce": sess.get('nonce')
            }
            r = client.post('http://localhost/confirm', data=data)
            assert "confirmation email has been resent" in r.get_data(as_text=True)

            r = client.get('/challenges')
            assert r.location == "http://localhost/confirm"  # We got redirected to /confirm

            r = client.get(
                'http://localhost/confirm/' + serialize('user@user.com')
            )
            assert r.location == 'http://localhost/challenges'

            # The team is now verified
            user = Users.query.filter_by(email='user@user.com').first()
            assert user.verified is True

            r = client.get('http://localhost/confirm')
            assert r.location == "http://localhost/settings"
    destroy_ctfd(app)


@patch('smtplib.SMTP')
def test_user_can_reset_password(mock_smtp):
    """Test that a user is capable of resetting their password"""
    from email.mime.text import MIMEText
    app = create_ctfd()
    with app.app_context(), freeze_time("2012-01-14 03:21:34"):
        # Set CTFd to send emails
        set_config('mail_server', 'localhost')
        set_config('mail_port', 25)
        set_config('mail_useauth', True)
        set_config('mail_username', 'username')
        set_config('mail_password', 'password')

        # Create a user
        register_user(app, name="user1", email="user@user.com")

        with app.test_client() as client:
            r = client.get('/reset_password')

            # Build reset password data
            with client.session_transaction() as sess:
                data = {
                    'nonce': sess.get('nonce'),
                    'email': 'user@user.com'
                }

            # Issue the password reset request
            r = client.post('/reset_password', data=data)

            from_addr = get_config('mailfrom_addr') or app.config.get('MAILFROM_ADDR')
            to_addr = 'user@user.com'

            # Build the email
            msg = ("""Did you initiate a password reset? Click the following link to reset """
                   """your password:\n\nhttp://localhost/reset_password/InVzZXIxIg.TxD0vg.-gvVg-KVy0RWkiclAE6JViv1I0M\n\n""")
            email_msg = MIMEText(msg)
            email_msg['Subject'] = "Message from CTFd"
            email_msg['From'] = from_addr
            email_msg['To'] = to_addr

            # Make sure that the reset password email is sent
            mock_smtp.return_value.sendmail.assert_called_with(from_addr, [to_addr], email_msg.as_string())

            # Get user's original password
            user = Users.query.filter_by(email="user@user.com").first()
            user_password_saved = user.password

            # Build the POST data
            with client.session_transaction() as sess:
                data = {
                    'nonce': sess.get('nonce'),
                    'password': 'passwordtwo'
                }

            # Do the password reset
            r = client.get('/reset_password/InVzZXIxIg.TxD0vg.-gvVg-KVy0RWkiclAE6JViv1I0M')
            r = client.post('/reset_password/InVzZXIxIg.TxD0vg.-gvVg-KVy0RWkiclAE6JViv1I0M', data=data)

            # Make sure that the user's password changed
            user = Users.query.filter_by(email="user@user.com").first()
            assert check_password('passwordtwo', user.password)
    destroy_ctfd(app)


def test_banned_user():
    app = create_ctfd()
    with app.app_context():
        register_user(app)
        client = login_as_user(app)
        user = Users.query.filter_by(id=2).first()
        user.banned = True
        db.session.commit()

        routes = ['/', '/challenges', '/api/v1/challenges']
        for route in routes:
            r = client.get(route)
            assert r.status_code == 403
    destroy_ctfd(app)