test_rest_framework.py

import unittest
from datetime import timedelta

from django.conf.urls import include, url
from django.contrib.auth import get_user_model
from django.http import HttpResponse
from django.test import TestCase
from django.test.utils import override_settings
from django.utils import timezone

from oauth2_provider.models import get_access_token_model, get_application_model
from oauth2_provider.settings import oauth2_settings


try:
    from unittest import mock
except ImportError:
    import mock


Application = get_application_model()
AccessToken = get_access_token_model()
UserModel = get_user_model()


try:
    from rest_framework import permissions
    from rest_framework.views import APIView
    from rest_framework.test import force_authenticate, APIRequestFactory
    from oauth2_provider.contrib.rest_framework import (
        IsAuthenticatedOrTokenHasScope, OAuth2Authentication, TokenHasScope,
        TokenHasReadWriteScope, TokenHasMethodScope, RequiredMethodScopes,
        TokenHasMethodPathScope, TokenHasResourceScope
    )

    class MockView(APIView):
        permission_classes = (permissions.IsAuthenticated,)

        def get(self, request):
            return HttpResponse({"a": 1, "b": 2, "c": 3})

        def post(self, request):
            return HttpResponse({"a": 1, "b": 2, "c": 3})

    class OAuth2View(MockView):
        authentication_classes = [OAuth2Authentication]

    class ScopedView(OAuth2View):
        permission_classes = [permissions.IsAuthenticated, TokenHasScope]
        required_scopes = ["scope1"]

    class AuthenticatedOrScopedView(OAuth2View):
        permission_classes = [IsAuthenticatedOrTokenHasScope]
        required_scopes = ["scope1"]

    class ReadWriteScopedView(OAuth2View):
        permission_classes = [permissions.IsAuthenticated, TokenHasReadWriteScope]

    class MethodScopeView(OAuth2View):
        permission_classes = [TokenHasMethodScope]
        required_scopes_map = {
            "GET": ["read"],
            "POST": ["create"],
            "PUT": ["update", "put"],
            "DELETE": ["delete"],
        }

    class MethodPathScopeView(OAuth2View):
        permission_classes = [TokenHasMethodPathScope]
        required_scopes_map_list = [
            RequiredMethodScopes("GET", r"^/oauth2-method-path-scope-test/$", ["read", "get scope2"]),
            RequiredMethodScopes("POST", r"^/oauth2-method-path-scope-test/$", ["create", "post scope2"]),
            RequiredMethodScopes("PUT", r"^/oauth2-method-path-scope-test/$", ["update", "put scope2"]),
            RequiredMethodScopes("DELETE", r"^/oauth2-method-path-scope-test/$", ["delete", "scope2 scope3"]),
        ]

    class ResourceScopedView(OAuth2View):
        permission_classes = [permissions.IsAuthenticated, TokenHasResourceScope]
        required_scopes = ["resource1"]

    urlpatterns = [
        url(r"^oauth2/", include("oauth2_provider.urls")),
        url(r"^oauth2-test/$", OAuth2View.as_view()),
        url(r"^oauth2-scoped-test/$", ScopedView.as_view()),
        url(r"^oauth2-read-write-test/$", ReadWriteScopedView.as_view()),
        url(r"^oauth2-resource-scoped-test/$", ResourceScopedView.as_view()),
        url(r"^oauth2-authenticated-or-scoped-test/$", AuthenticatedOrScopedView.as_view()),
        url(r"^oauth2-method-scope-test/$", MethodScopeView.as_view()),
        url(r"^oauth2-method-path-scope-test/$", MethodPathScopeView.as_view()),
    ]

    rest_framework_installed = True
except ImportError:
    rest_framework_installed = False


@override_settings(ROOT_URLCONF=__name__)
class TestOAuth2Authentication(TestCase):
    def setUp(self):
        oauth2_settings._SCOPES = ["read", "write", "scope1", "scope2", "resource1"]

        self.test_user = UserModel.objects.create_user("test_user", "test@example.com", "123456")
        self.dev_user = UserModel.objects.create_user("dev_user", "dev@example.com", "123456")

        self.application = Application.objects.create(
            name="Test Application",
            redirect_uris="http://localhost http://example.com http://example.org",
            user=self.dev_user,
            client_type=Application.CLIENT_CONFIDENTIAL,
            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
        )

        self.access_token = AccessToken.objects.create(
            user=self.test_user,
            scope="read write",
            expires=timezone.now() + timedelta(seconds=300),
            token="secret-access-token-key",
            application=self.application
        )

    def tearDown(self):
        oauth2_settings._SCOPES = ["read", "write"]

    def _create_authorization_header(self, token):
        return "Bearer {0}".format(token)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_authentication_allow(self):
        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_authentication_denied(self):
        auth = self._create_authorization_header("fake-token")
        response = self.client.get("/oauth2-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 401)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_authentication_or_scope_denied(self):
        # user is not authenticated
        # not a correct token
        auth = self._create_authorization_header("fake-token")
        response = self.client.get("/oauth2-authenticated-or-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 401)
        # token doesn"t have correct scope
        auth = self._create_authorization_header(self.access_token.token)

        factory = APIRequestFactory()
        request = factory.get("/oauth2-authenticated-or-scoped-test/")
        request.auth = auth
        force_authenticate(request, token=self.access_token)
        response = AuthenticatedOrScopedView.as_view()(request)
        # authenticated but wrong scope, this is 403, not 401
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_scoped_permission_allow(self):
        self.access_token.scope = "scope1"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_authenticated_or_scoped_permission_allow(self):
        self.access_token.scope = "scope1"
        self.access_token.save()
        # correct token and correct scope
        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-authenticated-or-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

        auth = self._create_authorization_header("fake-token")
        # incorrect token  but authenticated
        factory = APIRequestFactory()
        request = factory.get("/oauth2-authenticated-or-scoped-test/")
        request.auth = auth
        force_authenticate(request, self.test_user)
        response = AuthenticatedOrScopedView.as_view()(request)
        self.assertEqual(response.status_code, 200)

        # correct token  but not authenticated
        request = factory.get("/oauth2-authenticated-or-scoped-test/")
        request.auth = auth
        self.access_token.scope = "scope1"
        self.access_token.save()
        force_authenticate(request, token=self.access_token)
        response = AuthenticatedOrScopedView.as_view()(request)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_scoped_permission_deny(self):
        self.access_token.scope = "scope2"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_read_write_permission_get_allow(self):
        self.access_token.scope = "read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-read-write-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_read_write_permission_post_allow(self):
        self.access_token.scope = "write"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-read-write-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_read_write_permission_get_deny(self):
        self.access_token.scope = "write"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-read-write-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_read_write_permission_post_deny(self):
        self.access_token.scope = "read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-read-write-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_resource_scoped_permission_get_allow(self):
        self.access_token.scope = "resource1:read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-resource-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_resource_scoped_permission_post_allow(self):
        self.access_token.scope = "resource1:write"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-resource-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_resource_scoped_permission_get_denied(self):
        self.access_token.scope = "resource1:write"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-resource-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_resource_scoped_permission_post_denied(self):
        self.access_token.scope = "resource1:read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-resource-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    @mock.patch.object(oauth2_settings, "ERROR_RESPONSE_WITH_SCOPES", new=True)
    def test_required_scope_in_response(self):
        self.access_token.scope = "scope2"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(response.data["required_scopes"], ["scope1"])

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_required_scope_not_in_response_by_default(self):
        self.access_token.scope = "scope2"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-scoped-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)
        self.assertNotIn("required_scopes", response.data)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_scope_permission_get_allow(self):
        self.access_token.scope = "read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_scope_permission_post_allow(self):
        self.access_token.scope = "create"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_scope_permission_get_deny(self):
        self.access_token.scope = "write"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_scope_permission_post_deny(self):
        self.access_token.scope = "read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-method-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_path_scope_permission_get_allow_1(self):
        self.access_token.scope = "read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-method-path-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_path_scope_permission_get_allow_2(self):
        self.access_token.scope = "get scope2"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-method-path-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_path_scope_permission_post_allow_1(self):
        self.access_token.scope = "create"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-method-path-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_path_scope_permission_post_allow_2(self):
        self.access_token.scope = "post scope2"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-method-path-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 200)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_path_scope_permission_get_deny(self):
        self.access_token.scope = "write"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.get("/oauth2-method-path-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)

    @unittest.skipUnless(rest_framework_installed, "djangorestframework not installed")
    def test_method_path_scope_permission_post_deny(self):
        self.access_token.scope = "read"
        self.access_token.save()

        auth = self._create_authorization_header(self.access_token.token)
        response = self.client.post("/oauth2-method-path-scope-test/", HTTP_AUTHORIZATION=auth)
        self.assertEqual(response.status_code, 403)