API - update OAuth2 scopes
This commit is contained in:
@ -6,6 +6,7 @@ import pytest
|
||||
from flask import Flask
|
||||
|
||||
from fittrackee.oauth2.client import check_scope, create_oauth_client
|
||||
from fittrackee.oauth2.exceptions import InvalidOAuth2Scopes
|
||||
from fittrackee.oauth2.models import OAuth2Client
|
||||
from fittrackee.users.models import User
|
||||
|
||||
@ -15,7 +16,7 @@ TEST_METADATA = {
|
||||
'client_name': random_string(),
|
||||
'client_uri': random_string(),
|
||||
'redirect_uris': [random_domain()],
|
||||
'scope': 'read write',
|
||||
'scope': 'profile:read',
|
||||
}
|
||||
|
||||
|
||||
@ -127,13 +128,21 @@ class TestCreateOAuth2Client:
|
||||
def test_oauth_client_has_expected_scope(
|
||||
self, app: Flask, user_1: User
|
||||
) -> None:
|
||||
scope = 'write'
|
||||
scope = 'workouts:write'
|
||||
client_metadata: Dict = {**TEST_METADATA, 'scope': scope}
|
||||
|
||||
oauth_client = create_oauth_client(client_metadata, user_1)
|
||||
|
||||
assert oauth_client.scope == scope
|
||||
|
||||
def test_it_raises_error_when_scope_is_invalid(
|
||||
self, app: Flask, user_1: User
|
||||
) -> None:
|
||||
client_metadata: Dict = {**TEST_METADATA, 'scope': random_string()}
|
||||
|
||||
with pytest.raises(InvalidOAuth2Scopes):
|
||||
create_oauth_client(client_metadata, user_1)
|
||||
|
||||
def test_oauth_client_has_expected_token_endpoint_auth_method(
|
||||
self, app: Flask, user_1: User
|
||||
) -> None:
|
||||
@ -165,19 +174,23 @@ class TestOAuthCheckScopes:
|
||||
@pytest.mark.parametrize(
|
||||
'input_scope', ['', 1, random_string(), [random_string(), 'readwrite']]
|
||||
)
|
||||
def test_it_returns_read_if_scope_is_invalid(
|
||||
def test_it_raises_error_when_scope_is_invalid(
|
||||
self, input_scope: Any
|
||||
) -> None:
|
||||
assert check_scope(input_scope) == 'read'
|
||||
with pytest.raises(InvalidOAuth2Scopes):
|
||||
check_scope(input_scope)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'input_scope,expected_scope',
|
||||
[
|
||||
('read', 'read'),
|
||||
('read ' + random_string(), 'read'),
|
||||
('write', 'write'),
|
||||
('write read', 'write read'),
|
||||
('write read ' + random_string(), 'write read'),
|
||||
('profile:read', 'profile:read'),
|
||||
('profile:read ' + random_string(), 'profile:read'),
|
||||
('profile:write', 'profile:write'),
|
||||
('profile:read profile:write', 'profile:read profile:write'),
|
||||
(
|
||||
'profile:write profile:read ' + random_string(),
|
||||
'profile:write profile:read',
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_it_return_only_valid_scopes(
|
||||
|
@ -83,6 +83,30 @@ class TestOAuthClientCreation(ApiTestCaseMixin):
|
||||
error_message=f'OAuth client metadata missing keys: {missing_key}',
|
||||
)
|
||||
|
||||
def test_it_returns_error_when_scope_is_invalid(
|
||||
self, app: Flask, user_1: User
|
||||
) -> None:
|
||||
invalid_scope = self.random_string()
|
||||
metadata: Dict = {
|
||||
**TEST_OAUTH_CLIENT_METADATA,
|
||||
'scope': invalid_scope,
|
||||
}
|
||||
client, auth_token = self.get_test_client_and_auth_token(
|
||||
app, user_1.email
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
self.route,
|
||||
data=json.dumps(metadata),
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {auth_token}'),
|
||||
)
|
||||
|
||||
self.assert_400(
|
||||
response,
|
||||
error_message=('OAuth client invalid scopes'),
|
||||
)
|
||||
|
||||
def test_it_creates_oauth_client(self, app: Flask, user_1: User) -> None:
|
||||
client, auth_token = self.get_test_client_and_auth_token(
|
||||
app, user_1.email
|
||||
|
@ -1,58 +1,28 @@
|
||||
from json import dumps
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from werkzeug.test import TestResponse
|
||||
|
||||
from fittrackee.users.models import User
|
||||
|
||||
from ..mixins import ApiTestCaseMixin
|
||||
from ..utils import random_short_id, random_string
|
||||
|
||||
|
||||
class OAuth2ScopesTestCase(ApiTestCaseMixin):
|
||||
def assert_expected_response(
|
||||
self, response: TestResponse, client_scope: str, endpoint_scope: str
|
||||
) -> None:
|
||||
if client_scope == endpoint_scope:
|
||||
self.assert_not_insufficient_scope_error(response)
|
||||
else:
|
||||
self.assert_insufficient_scope(response)
|
||||
|
||||
|
||||
class TestOAuth2ScopesWithReadAccess(OAuth2ScopesTestCase):
|
||||
scope = 'read'
|
||||
|
||||
class TestOAuth2Scopes(ApiTestCaseMixin):
|
||||
@pytest.mark.parametrize(
|
||||
'endpoint_url',
|
||||
'endpoint_url,scope',
|
||||
[
|
||||
'/api/auth/profile',
|
||||
'/api/records',
|
||||
'/api/sports',
|
||||
'/api/sports/1',
|
||||
f'/api/stats/{random_string()}/by_sport',
|
||||
f'/api/stats/{random_string()}/by_time',
|
||||
'/api/users/test',
|
||||
'/api/workouts',
|
||||
f'/api/workouts/{random_short_id()}',
|
||||
f'/api/workouts/{random_short_id()}/chart_data',
|
||||
f'/api/workouts/{random_short_id()}/chart_data/segment/1',
|
||||
f'/api/workouts/{random_short_id()}/gpx',
|
||||
f'/api/workouts/{random_short_id()}/gpx/download',
|
||||
f'/api/workouts/{random_short_id()}/gpx/segment/1',
|
||||
('/api/auth/profile', 'profile:read'),
|
||||
('/api/workouts', 'workouts:read'),
|
||||
],
|
||||
)
|
||||
def test_access_to_get_endpoints(
|
||||
self, app: Flask, user_1: User, endpoint_url: str
|
||||
def test_oauth_client_can_access_authorized_endpoints(
|
||||
self, app: Flask, user_1: User, endpoint_url: str, scope: str
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1, scope=self.scope
|
||||
)
|
||||
) = self.create_oauth_client_and_issue_token(app, user_1, scope=scope)
|
||||
|
||||
response = client.get(
|
||||
endpoint_url,
|
||||
@ -60,208 +30,29 @@ class TestOAuth2ScopesWithReadAccess(OAuth2ScopesTestCase):
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_expected_response(
|
||||
response, client_scope=self.scope, endpoint_scope='read'
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'endpoint_url',
|
||||
['/api/users'],
|
||||
)
|
||||
def test_access_to_endpoints_as_admin(
|
||||
self, app: Flask, user_1_admin: User, endpoint_url: str
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1_admin, scope=self.scope
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
endpoint_url,
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_expected_response(
|
||||
response, client_scope=self.scope, endpoint_scope='read'
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'endpoint_url',
|
||||
[
|
||||
'/api/auth/picture',
|
||||
'/api/auth/profile/edit',
|
||||
'/api/auth/profile/edit/preferences',
|
||||
'/api/auth/profile/edit/sports',
|
||||
'/api/workouts',
|
||||
'/api/workouts/no_gpx',
|
||||
],
|
||||
)
|
||||
def test_access_post_endpoints(
|
||||
self, app: Flask, user_1: User, endpoint_url: str
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1, scope=self.scope
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
endpoint_url,
|
||||
data=dumps(dict()),
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_expected_response(
|
||||
response, client_scope=self.scope, endpoint_scope='write'
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'endpoint_url',
|
||||
[
|
||||
'/api/auth/profile/edit/account',
|
||||
'/api/workouts/0',
|
||||
],
|
||||
)
|
||||
def test_access_to_patch_endpoints(
|
||||
self, app: Flask, user_1: User, endpoint_url: str
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1, scope=self.scope
|
||||
)
|
||||
|
||||
response = client.patch(
|
||||
endpoint_url,
|
||||
data=dumps(dict()),
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_expected_response(
|
||||
response, client_scope=self.scope, endpoint_scope='write'
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'endpoint_url',
|
||||
[
|
||||
'/api/config',
|
||||
'/api/sports/1',
|
||||
f'/api/users/{random_string()}',
|
||||
],
|
||||
)
|
||||
def test_access_to_patch_endpoints_as_admin(
|
||||
self, app: Flask, user_1_admin: User, endpoint_url: str
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1_admin, scope=self.scope
|
||||
)
|
||||
|
||||
response = client.patch(
|
||||
endpoint_url,
|
||||
data=dumps(dict()),
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_expected_response(
|
||||
response, client_scope=self.scope, endpoint_scope='write'
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'endpoint_url',
|
||||
[
|
||||
'/api/auth/picture',
|
||||
'/api/auth/profile/reset/sports/1',
|
||||
f'/api/users/{random_string()}',
|
||||
'/api/workouts/0',
|
||||
],
|
||||
)
|
||||
def test_access_to_delete_endpoints(
|
||||
self, app: Flask, user_1: User, endpoint_url: str
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1, scope=self.scope
|
||||
)
|
||||
user_1.picture = random_string()
|
||||
|
||||
response = client.delete(
|
||||
endpoint_url,
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_expected_response(
|
||||
response, client_scope=self.scope, endpoint_scope='write'
|
||||
)
|
||||
|
||||
|
||||
class TestOAuth2ScopesWithWriteAccess(TestOAuth2ScopesWithReadAccess):
|
||||
scope = 'write'
|
||||
|
||||
|
||||
class TestOAuth2ScopesWithReadAndWriteAccess(ApiTestCaseMixin):
|
||||
scope = 'read write'
|
||||
|
||||
def test_client_can_access_endpoint_with_read_scope(
|
||||
self, app: Flask, user_1: User
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1, scope=self.scope
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
'/api/auth/profile',
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_not_insufficient_scope_error(response)
|
||||
|
||||
def test_client_with_read_can_access_endpoints_with_write_scope(
|
||||
self, app: Flask, user_1: User
|
||||
@pytest.mark.parametrize(
|
||||
'endpoint_url,scope',
|
||||
[
|
||||
('/api/auth/profile', 'workouts:read'),
|
||||
('/api/workouts', 'profile:read'),
|
||||
],
|
||||
)
|
||||
def test_oauth_client_can_not_access_unauthorized_endpoints(
|
||||
self, app: Flask, user_1: User, endpoint_url: str, scope: str
|
||||
) -> None:
|
||||
(
|
||||
client,
|
||||
oauth_client,
|
||||
access_token,
|
||||
_,
|
||||
) = self.create_oauth_client_and_issue_token(
|
||||
app, user_1, scope=self.scope
|
||||
)
|
||||
) = self.create_oauth_client_and_issue_token(app, user_1, scope=scope)
|
||||
|
||||
response = client.post(
|
||||
'/api/auth/picture',
|
||||
data=dumps(dict()),
|
||||
response = client.get(
|
||||
endpoint_url,
|
||||
content_type='application/json',
|
||||
headers=dict(Authorization=f'Bearer {access_token}'),
|
||||
)
|
||||
|
||||
self.assert_not_insufficient_scope_error(response)
|
||||
self.assert_insufficient_scope(response)
|
||||
|
Reference in New Issue
Block a user