mediagoblin/tests/test_oauth.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 import json
  18 import logging
  19
  20 import pytest
  21 from urlparse import parse_qs, urlparse
  22
  23 from mediagoblin import mg_globals
  24 from mediagoblin.tools import template, pluginapi
  25 from mediagoblin.tests.tools import fixture_add_user
  26 from mediagoblin.tools.testing import _activate_testing
  27
  28 _activate_testing()
  29
  30
  31 _log = logging.getLogger(__name__)
  32
  33
  34 class TestOAuth(object):
  35     @pytest.fixture(autouse=True)
  36     def setup(self, test_app):
  37         self.test_app = test_app
  38
  39         self.db = mg_globals.database
  40
  41         self.pman = pluginapi.PluginManager()
  42
  43         self.user_password = u'4cc355_70k3N'
  44         self.user = fixture_add_user(u'joauth', self.user_password)
  45
  46         self.login()
  47
  48     def login(self):
  49         self.test_app.post(
  50             '/auth/login/', {
  51                 'username': self.user.username,
  52                 'password': self.user_password})
  53
  54     def register_client(self, name, client_type, description=None,
  55                         redirect_uri=''):
  56         return self.test_app.post(
  57                 '/oauth/client/register', {
  58                     'name': name,
  59                     'description': description,
  60                     'type': client_type,
  61                     'redirect_uri': redirect_uri})
  62
  63     def get_context(self, template_name):
  64         return template.TEMPLATE_TEST_CONTEXT[template_name]
  65
  66     def test_1_public_client_registration_without_redirect_uri(self):
  67         ''' Test 'public' OAuth client registration without any redirect uri '''
  68         response = self.register_client(
  69             u'OMGOMGOMG', 'public', 'OMGOMG Apache License v2')
  70
  71         ctx = self.get_context('oauth/client/register.html')
  72
  73         client = self.db.OAuthClient.query.filter(
  74                 self.db.OAuthClient.name == u'OMGOMGOMG').first()
  75
  76         assert response.status_int == 200
  77
  78         # Should display an error
  79         assert len(ctx['form'].redirect_uri.errors)
  80
  81         # Should not pass through
  82         assert not client
  83
  84     def test_2_successful_public_client_registration(self):
  85         ''' Successfully register a public client '''
  86         uri = 'http://foo.example'
  87         self.register_client(
  88             u'OMGOMG', 'public', 'OMG!', uri)
  89
  90         client = self.db.OAuthClient.query.filter(
  91                 self.db.OAuthClient.name == u'OMGOMG').first()
  92
  93         # redirect_uri should be set
  94         assert client.redirect_uri == uri
  95
  96         # Client should have been registered
  97         assert client
  98
  99     def test_3_successful_confidential_client_reg(self):
 100         ''' Register a confidential OAuth client '''
 101         response = self.register_client(
 102             u'GMOGMO', 'confidential', 'NO GMO!')
 103
 104         assert response.status_int == 302
 105
 106         client = self.db.OAuthClient.query.filter(
 107                 self.db.OAuthClient.name == u'GMOGMO').first()
 108
 109         # Client should have been registered
 110         assert client
 111
 112         return client
 113
 114     def test_4_authorize_confidential_client(self):
 115         ''' Authorize a confidential client as a logged in user '''
 116         client = self.test_3_successful_confidential_client_reg()
 117
 118         client_identifier = client.identifier
 119
 120         redirect_uri = 'https://foo.example'
 121         response = self.test_app.get('/oauth/authorize', {
 122                 'client_id': client.identifier,
 123                 'scope': 'all',
 124                 'redirect_uri': redirect_uri})
 125
 126         # User-agent should NOT be redirected
 127         assert response.status_int == 200
 128
 129         ctx = self.get_context('oauth/authorize.html')
 130
 131         form = ctx['form']
 132
 133         # Short for client authorization post reponse
 134         capr = self.test_app.post(
 135                 '/oauth/client/authorize', {
 136                     'client_id': form.client_id.data,
 137                     'allow': 'Allow',
 138                     'next': form.next.data})
 139
 140         assert capr.status_int == 302
 141
 142         authorization_response = capr.follow()
 143
 144         assert authorization_response.location.startswith(redirect_uri)
 145
 146         return authorization_response, client_identifier
 147
 148     def get_code_from_redirect_uri(self, uri):
 149         ''' Get the value of ?code= from an URI '''
 150         return parse_qs(urlparse(uri).query)['code'][0]
 151
 152     def test_token_endpoint_successful_confidential_request(self):
 153         ''' Successful request against token endpoint '''
 154         code_redirect, client_id = self.test_4_authorize_confidential_client()
 155
 156         code = self.get_code_from_redirect_uri(code_redirect.location)
 157
 158         client = self.db.OAuthClient.query.filter(
 159                 self.db.OAuthClient.identifier == unicode(client_id)).first()
 160
 161         token_res = self.test_app.get('/oauth/access_token?client_id={0}&\
 162 code={1}&client_secret={2}'.format(client_id, code, client.secret))
 163
 164         assert token_res.status_int == 200
 165
 166         token_data = json.loads(token_res.body)
 167
 168         assert not 'error' in token_data
 169         assert 'access_token' in token_data
 170         assert 'token_type' in token_data
 171         assert 'expires_in' in token_data
 172         assert type(token_data['expires_in']) == int
 173         assert token_data['expires_in'] > 0
 174
 175         # There should be a refresh token provided in the token data
 176         assert len(token_data['refresh_token'])
 177
 178         return client_id, token_data
 179
 180     def test_token_endpont_missing_id_confidential_request(self):
 181         ''' Unsuccessful request against token endpoint, missing client_id '''
 182         code_redirect, client_id = self.test_4_authorize_confidential_client()
 183
 184         code = self.get_code_from_redirect_uri(code_redirect.location)
 185
 186         client = self.db.OAuthClient.query.filter(
 187                 self.db.OAuthClient.identifier == unicode(client_id)).first()
 188
 189         token_res = self.test_app.get('/oauth/access_token?\
 190 code={0}&client_secret={1}'.format(code, client.secret))
 191
 192         assert token_res.status_int == 200
 193
 194         token_data = json.loads(token_res.body)
 195
 196         assert 'error' in token_data
 197         assert not 'access_token' in token_data
 198         assert token_data['error'] == 'invalid_request'
 199         assert len(token_data['error_description'])
 200
 201     def test_refresh_token(self):
 202         ''' Try to get a new access token using the refresh token '''
 203         # Get an access token and a refresh token
 204         client_id, token_data =\
 205             self.test_token_endpoint_successful_confidential_request()
 206
 207         client = self.db.OAuthClient.query.filter(
 208             self.db.OAuthClient.identifier == client_id).first()
 209
 210         token_res = self.test_app.get('/oauth/access_token',
 211                      {'refresh_token': token_data['refresh_token'],
 212                       'client_id': client_id,
 213                       'client_secret': client.secret
 214                       })
 215
 216         assert token_res.status_int == 200
 217
 218         new_token_data = json.loads(token_res.body)
 219
 220         assert not 'error' in new_token_data
 221         assert 'access_token' in new_token_data
 222         assert 'token_type' in new_token_data
 223         assert 'expires_in' in new_token_data
 224         assert type(new_token_data['expires_in']) == int
 225         assert new_token_data['expires_in'] > 0