path: root/tests/contrib/django_util/test_views.py

# Copyright 2016 Google Inc. All Rights Reserved.
#
# Licensed 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.

"""Unit test for django_util views"""

import copy
import json

import django
from django import http
import django.conf
from django.contrib.auth import models as django_models
import mock
from six.moves import reload_module

from oauth2client import client
import oauth2client.contrib.django_util
from oauth2client.contrib.django_util import views
from tests.contrib import django_util as tests_django_util
from tests.contrib.django_util import models as tests_models


class OAuth2AuthorizeTest(tests_django_util.TestWithDjangoEnvironment):

    def setUp(self):
        super(OAuth2AuthorizeTest, self).setUp()
        self.save_settings = copy.deepcopy(django.conf.settings)
        reload_module(oauth2client.contrib.django_util)
        self.user = django_models.User.objects.create_user(
            username='bill', email='bill@example.com', password='hunter2')

    def tearDown(self):
        django.conf.settings = copy.deepcopy(self.save_settings)

    def test_authorize_works(self):
        request = self.factory.get('oauth2/oauth2authorize')
        request.session = self.session
        request.user = self.user
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)

    def test_authorize_anonymous_user(self):
        request = self.factory.get('oauth2/oauth2authorize')
        request.session = self.session
        request.user = django_models.AnonymousUser()
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)

    def test_authorize_works_explicit_return_url(self):
        request = self.factory.get('oauth2/oauth2authorize',
                                   data={'return_url': '/return_endpoint'})
        request.session = self.session
        request.user = self.user
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)


class Oauth2AuthorizeStorageModelTest(
        tests_django_util.TestWithDjangoEnvironment):

    def setUp(self):
        super(Oauth2AuthorizeStorageModelTest, self).setUp()
        self.save_settings = copy.deepcopy(django.conf.settings)

        STORAGE_MODEL = {
            'model': 'tests.contrib.django_util.models.CredentialsModel',
            'user_property': 'user_id',
            'credentials_property': 'credentials'
        }
        django.conf.settings.GOOGLE_OAUTH2_STORAGE_MODEL = STORAGE_MODEL

        # OAuth2 Settings gets configured based on Django settings
        # at import time, so in order for us to reload the settings
        # we need to reload the module
        reload_module(oauth2client.contrib.django_util)
        self.user = django_models.User.objects.create_user(
            username='bill', email='bill@example.com', password='hunter2')

    def tearDown(self):
        django.conf.settings = copy.deepcopy(self.save_settings)

    def test_authorize_works(self):
        request = self.factory.get('oauth2/oauth2authorize')
        request.session = self.session
        request.user = self.user
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)
        # redirects to Google oauth
        self.assertIn('accounts.google.com', response.url)

    def test_authorize_anonymous_user_redirects_login(self):
        request = self.factory.get('oauth2/oauth2authorize')
        request.session = self.session
        request.user = django_models.AnonymousUser()
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)
        # redirects to Django login
        self.assertIn(django.conf.settings.LOGIN_URL, response.url)

    def test_authorize_works_explicit_return_url(self):
        request = self.factory.get('oauth2/oauth2authorize',
                                   data={'return_url': '/return_endpoint'})
        request.session = self.session
        request.user = self.user
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)

    def test_authorized_user_no_credentials_redirects(self):
        request = self.factory.get('oauth2/oauth2authorize',
                                   data={'return_url': '/return_endpoint'})
        request.session = self.session

        authorized_user = django_models.User.objects.create_user(
            username='bill2', email='bill@example.com', password='hunter2')

        tests_models.CredentialsModel.objects.create(
            user_id=authorized_user,
            credentials=None)

        request.user = authorized_user
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)

    def test_already_authorized(self):
        request = self.factory.get('oauth2/oauth2authorize',
                                   data={'return_url': '/return_endpoint'})
        request.session = self.session

        authorized_user = django_models.User.objects.create_user(
            username='bill2', email='bill@example.com', password='hunter2')

        credentials = _Credentials()
        tests_models.CredentialsModel.objects.create(
            user_id=authorized_user,
            credentials=credentials)

        request.user = authorized_user
        response = views.oauth2_authorize(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)
        self.assertEqual(response.url, '/return_endpoint')


class _Credentials(object):
    # Can't use mock when testing Django models
    # https://code.djangoproject.com/ticket/25493
    def __init__(self):
        self.invalid = False
        self.scopes = set()

    def has_scopes(self, _):
        return True


class Oauth2CallbackTest(tests_django_util.TestWithDjangoEnvironment):

    def setUp(self):
        super(Oauth2CallbackTest, self).setUp()
        self.save_settings = copy.deepcopy(django.conf.settings)
        reload_module(oauth2client.contrib.django_util)

        self.CSRF_TOKEN = 'token'
        self.RETURN_URL = 'http://return-url.com'
        self.fake_state = {
            'csrf_token': self.CSRF_TOKEN,
            'return_url': self.RETURN_URL,
            'scopes': django.conf.settings.GOOGLE_OAUTH2_SCOPES
        }
        self.user = django_models.User.objects.create_user(
            username='bill', email='bill@example.com', password='hunter2')

    @mock.patch('oauth2client.contrib.django_util.views.jsonpickle')
    def test_callback_works(self, jsonpickle_mock):
        request = self.factory.get('oauth2/oauth2callback', data={
            'state': json.dumps(self.fake_state),
            'code': 123
        })

        self.session['google_oauth2_csrf_token'] = self.CSRF_TOKEN

        flow = client.OAuth2WebServerFlow(
            client_id='clientid',
            client_secret='clientsecret',
            scope=['email'],
            state=json.dumps(self.fake_state),
            redirect_uri=request.build_absolute_uri("oauth2/oauth2callback"))

        name = 'google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)
        pickled_flow = object()
        self.session[name] = pickled_flow
        flow.step2_exchange = mock.Mock()
        jsonpickle_mock.decode.return_value = flow

        request.session = self.session
        request.user = self.user
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseRedirect)
        self.assertEqual(
            response.status_code, django.http.HttpResponseRedirect.status_code)
        self.assertEqual(response['Location'], self.RETURN_URL)
        jsonpickle_mock.decode.assert_called_once_with(pickled_flow)

    @mock.patch('oauth2client.contrib.django_util.views.jsonpickle')
    def test_callback_handles_bad_flow_exchange(self, jsonpickle_mock):
        request = self.factory.get('oauth2/oauth2callback', data={
            "state": json.dumps(self.fake_state),
            "code": 123
        })

        self.session['google_oauth2_csrf_token'] = self.CSRF_TOKEN

        flow = client.OAuth2WebServerFlow(
            client_id='clientid',
            client_secret='clientsecret',
            scope=['email'],
            state=json.dumps(self.fake_state),
            redirect_uri=request.build_absolute_uri('oauth2/oauth2callback'))

        session_key = 'google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)
        pickled_flow = object()
        self.session[session_key] = pickled_flow

        def local_throws(code):
            raise client.FlowExchangeError('test')

        flow.step2_exchange = local_throws
        jsonpickle_mock.decode.return_value = flow

        request.session = self.session
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)
        jsonpickle_mock.decode.assert_called_once_with(pickled_flow)

    def test_error_returns_bad_request(self):
        request = self.factory.get('oauth2/oauth2callback', data={
            'error': 'There was an error in your authorization.',
        })
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)
        self.assertIn(b'Authorization failed', response.content)

    def test_error_escapes_html(self):
        request = self.factory.get('oauth2/oauth2callback', data={
            'error': '<script>bad</script>',
        })
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)
        self.assertNotIn(b'<script>', response.content)
        self.assertIn(b'&lt;script&gt;', response.content)

    def test_no_session(self):
        request = self.factory.get('oauth2/oauth2callback', data={
            'code': 123,
            'state': json.dumps(self.fake_state)
        })

        request.session = self.session
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)
        self.assertEqual(
            response.content, b'No existing session for this flow.')

    def test_missing_state_returns_bad_request(self):
        request = self.factory.get('oauth2/oauth2callback', data={
            'code': 123
        })
        self.session['google_oauth2_csrf_token'] = "token"
        request.session = self.session
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)

    def test_bad_state(self):
        request = self.factory.get('oauth2/oauth2callback', data={
            'code': 123,
            'state': json.dumps({'wrong': 'state'})
        })
        self.session['google_oauth2_csrf_token'] = 'token'
        request.session = self.session
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)
        self.assertEqual(response.content, b'Invalid state parameter.')

    def test_bad_csrf(self):
        request = self.factory.get('oauth2/oauth2callback', data={
            "state": json.dumps(self.fake_state),
            "code": 123
        })
        self.session['google_oauth2_csrf_token'] = 'WRONG TOKEN'
        request.session = self.session
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)
        self.assertEqual(response.content, b'Invalid CSRF token.')

    def test_no_saved_flow(self):
        request = self.factory.get('oauth2/oauth2callback', data={
            'state': json.dumps(self.fake_state),
            'code': 123
        })
        self.session['google_oauth2_csrf_token'] = self.CSRF_TOKEN
        self.session['google_oauth2_flow_{0}'.format(self.CSRF_TOKEN)] = None
        request.session = self.session
        response = views.oauth2_callback(request)
        self.assertIsInstance(response, http.HttpResponseBadRequest)
        self.assertEqual(response.content, b'Missing Oauth2 flow.')