Merge pull request #435 from metabrainz/mismatch-oauth-url

Provide a user-friendly error if oauth params are incorrect

critiquebrainz/db/oauth_client.py

@@ -30,11 +30,12 @@ def create(*, user_id, name, desc, website, redirect_uri):
         }
     """
     with db.engine.connect() as connection:
-        connection.execute(sqlalchemy.text("""
+        row = connection.execute(sqlalchemy.text("""
             INSERT INTO oauth_client (client_id, client_secret, redirect_uri,
                         user_id, name, "desc", website)
                  VALUES (:client_id, :client_secret, :redirect_uri,
                         :user_id, :name, :desc, :website)
+                  RETURNING client_id, client_secret, redirect_uri, user_id, name, "desc", website
         """), {
             "client_id": generate_string(CLIENT_ID_LENGTH),
             "client_secret": generate_string(CLIENT_SECRET_LENGTH),
@@ -44,6 +45,7 @@ def create(*, user_id, name, desc, website, redirect_uri):
             "website": website,
             "desc": desc,
         })
+        return dict(row.fetchone())
 
 
 def update(*, client_id, name=None, desc=None, website=None, redirect_uri=None):

critiquebrainz/frontend/testing.py

@@ -1,4 +1,5 @@
 import os
+from critiquebrainz.ws.oauth import oauth
 
 from flask_testing import TestCase
 
@@ -10,6 +11,7 @@ class FrontendTestCase(TestCase):
 
     def create_app(self):
         app = create_app(config_path=os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'test_config.py'))
+        oauth.init_app(app)
         return app
 
     def setUp(self):

critiquebrainz/frontend/views/oauth.py

@@ -1,4 +1,6 @@
 from flask import Blueprint, render_template, redirect, request
+from werkzeug.exceptions import BadRequest
+from critiquebrainz.ws.oauth.exceptions import OAuthError
 from flask_login import login_required, current_user
 
 import critiquebrainz.db.oauth_client as db_oauth_client
@@ -19,12 +21,18 @@ def authorize_prompt():
     state = request.args.get('state')
 
     if request.method == 'GET':  # Client requests access
-        oauth.validate_authorization_request(client_id, response_type, redirect_uri, scope)
-        client = db_oauth_client.get_client(client_id)
-        return render_template('oauth/prompt.html', client=client, scope=scope,
-                               cancel_url=build_url(redirect_uri, dict(error='access_denied')))
+        try:
+            oauth.validate_authorization_request(client_id, response_type, redirect_uri, scope)
+            client = db_oauth_client.get_client(client_id)
+            return render_template('oauth/prompt.html', client=client, scope=scope,
+                                cancel_url=build_url(redirect_uri, dict(error='access_denied')))
+        except OAuthError as e:
+            raise BadRequest(e.desc)
 
     if request.method == 'POST':  # User grants access to the client
-        oauth.validate_authorization_request(client_id, response_type, redirect_uri, scope)
-        code = oauth.generate_grant(client_id, current_user.id, redirect_uri, scope)
-        return redirect(build_url(redirect_uri, dict(code=code, state=state)))
+        try:
+            oauth.validate_authorization_request(client_id, response_type, redirect_uri, scope)
+            code = oauth.generate_grant(client_id, current_user.id, redirect_uri, scope)
+            return redirect(build_url(redirect_uri, dict(code=code, state=state)))
+        except OAuthError as e:
+            raise BadRequest(e.desc)

critiquebrainz/frontend/views/test/test_oauth.py

@@ -0,0 +1,77 @@
+from flask import url_for
+import critiquebrainz.db.oauth_client as db_oauth_client
+import critiquebrainz.db.oauth_grant as db_oauth_grant
+import critiquebrainz.db.users as db_users
+from critiquebrainz.frontend.testing import FrontendTestCase
+
+from urllib.parse import urlparse, parse_qs
+
+
+class OauthTestCase(FrontendTestCase):
+    def setUp(self):
+        from critiquebrainz.db.user import User
+        self.user = User(db_users.get_or_create(2, "9371e5c7-5995-4471-a5a9-33481f897f9c", new_user_data={
+            "display_name": u"User",
+        }))
+        self.oauthclient = db_oauth_client.create(
+            user_id=self.user.id, 
+            name='An oauth app',
+            desc='This is a great client', 
+            website='https://example.com', 
+            redirect_uri='https://example.com/redirect'
+        )
+
+
+    def test_invalid_clientid(self):
+        self.temporary_login(self.user)
+        response = self.client.get(url_for('oauth.authorize_prompt', response_type='code', client_id='not-an-id', redirect_uri='x', scope='x', state='x'))
+        assert response.status_code == 400
+        assert "400 Bad Request: Client authentication failed." in response.text
+
+    def test_invalid_redirect_uri(self):
+        self.temporary_login(self.user)
+        client_id = self.oauthclient["client_id"]
+        response = self.client.get(url_for('oauth.authorize_prompt', response_type='code', client_id=client_id, redirect_uri='x', scope='x', state='x'))
+        assert response.status_code == 400
+        assert "400 Bad Request: Invalid redirect uri." in response.text
+
+    def test_invalid_scope(self):
+        self.temporary_login(self.user)
+        client_id = self.oauthclient["client_id"]
+        redirect_uri = self.oauthclient["redirect_uri"]
+        response = self.client.get(url_for('oauth.authorize_prompt', response_type='code', client_id=client_id, redirect_uri=redirect_uri, scope='x', state='x'))
+        assert response.status_code == 400
+        assert "400 Bad Request: The requested scope is invalid, unknown, or malformed." in response.text
+
+    def test_valid_oauth_request(self):
+        self.temporary_login(self.user)
+        client_id = self.oauthclient["client_id"]
+        redirect_uri = self.oauthclient["redirect_uri"]
+        response = self.client.get(url_for('oauth.authorize_prompt', response_type='code', client_id=client_id, redirect_uri=redirect_uri, scope='review', state='x'))
+        assert response.status_code == 200
+        assert "Do you want to give access to your account to An oauth app?" in response.text
+
+    def test_approve_invalid_parameter(self):
+        """Same endpoint, but a POST with an invalid client id"""
+        self.temporary_login(self.user)
+        response = self.client.post(url_for('oauth.authorize_prompt', response_type='code', client_id='x', redirect_uri='y', scope='review', state='x'))
+        assert response.status_code == 400
+        assert "400 Bad Request: Client authentication failed." in response.text
+
+    def test_approve_application(self):
+        """A POST to approve an oauth authorization"""
+        self.temporary_login(self.user)
+        client_id = self.oauthclient["client_id"]
+        redirect_uri = self.oauthclient["redirect_uri"]
+        response = self.client.post(url_for('oauth.authorize_prompt', response_type='code', client_id=client_id, redirect_uri=redirect_uri, scope='review', state='x'))
+        assert response.status_code == 302
+        # This is the redirect URL that we set in the oauth client
+        assert response.location.startswith('https://example.com/redirect?code=')
+
+        redirect_location = urlparse(response.location)
+        query = parse_qs(redirect_location.query)
+        assert query['state'] == ['x']
+        code = query['code'][0]
+
+        grants = db_oauth_grant.list_grants(client_id=client_id, code=code)
+        assert len(grants) == 1

pytest.ini

@@ -1,3 +1,3 @@
 [pytest]
 testpaths = critiquebrainz
-addopts = --cov=critiquebrainz
+addopts = --cov=critiquebrainz --no-cov-on-fail