server: token revocation

* Adds token revocation endpoint.  (closes inveniosoftware#53)

Signed-off-by: Jiri Kuncar <[email protected]>

invenio_oauth2server/models.py

@@ -360,6 +360,11 @@ class Token(db.Model):
     is_internal = db.Column(db.Boolean, default=False)
     """Determines if token is an internally generated token."""
 
+    def delete(self):
+        """Delete token by revoke token handler."""
+        db.session.delete(self)
+        db.session.commit()
+
     @property
     def scopes(self):
         """Return all scopes."""

invenio_oauth2server/provider.py

@@ -108,3 +108,9 @@ def save_token(token, request, *args, **kwargs):
     db.session.add(tok)
     db.session.commit()
     return tok
+
+
+@oauth2.clientgetter
+def get_client(client_id, *args, **kwargs):
+    """Get client for token revocation."""
+    return Client.query.get_or_none(client_id)

invenio_oauth2server/views/server.py

@@ -131,6 +131,13 @@ def access_token():
     return None
 
 
+@blueprint.route('/revoke', methods=['POST'])
+@oauth2.revoke_handler
+def revoke_token():
+    """Revoke an access or refresh token."""
+    return {}
+
+
 @blueprint.route('/errors')
 def errors():
     """Error view in case of invalid oauth requests."""

tests/test_server.py

@@ -26,8 +26,11 @@
 
 from __future__ import absolute_import, print_function
 
+from flask import url_for
 from flask_principal import AnonymousIdentity
 
+from invenio_oauth2server.models import Token
+
 
 def test_user_identity_init(resource_fixture):
     """Test that user identity is loaded properly when a token is used."""
@@ -42,3 +45,31 @@ def test_user_identity_init(resource_fixture):
         request_res = client.get(app.url_for_test0resource_token)
         assert request_res.status_code == 200
         assert app.identity.user.id == app.user_id
+
+
+def test_token_revocation(models_fixture):
+    """Test that user token can not be used after revocation."""
+    import base64
+
+    app = models_fixture
+
+    def _base64(text):
+        return base64.b64encode(text.decode('utf-8')).encode('utf-8')
+
+    auth_code = _base64(u'client_test_u1c1:client_test_u1c1')
+
+    with app.test_client() as client:
+        tok = Token.query.filter_by(
+            refresh_token='dev_refresh_2').first()
+        assert tok is not None
+
+        revoke_url = '/oauth/revoke'
+        args = 'token=dev_access_1'
+        res = client.post(revoke_url, query_string=args, headers={
+            'Authorization': 'Basic %s' % auth_code,
+        })
+        assert res.status_code == 200
+
+        tok = Token.query.filter_by(
+            access_token='dev_access_1').first()
+        assert tok is None