Dramatically improved processing of Micropub tokens which supports both the Authorization header and the access_token field approaches

lemonauth/tokens.py

@@ -1,7 +1,61 @@
 from jose import jwt
 
 from datetime import datetime, timedelta
+from django.contrib.auth import get_user_model
 from django.conf import settings
+from django.http import HttpResponse
+
+
+def auth(request):
+    if 'HTTP_AUTHORIZATION' in request.META:
+        auth = request.META.get('HTTP_AUTHORIZATION').split(' ')
+        if auth[0] != 'Bearer':
+            return HttpResponse(
+                'authorisation with {0} not supported'.format(auth[0]),
+                content_type='text/plain',
+                status=400
+            )
+        if len(auth) != 2:
+            return HttpResponse(
+                'invalid Bearer auth format, must be Bearer <token>',
+                content_type='text/plain',
+                status=400
+            )
+        token = auth[1]
+    elif 'access_token' in request.POST:
+        token = request.POST.get('access_token')
+    elif 'access_token' in request.GET:
+        token = request.GET.get('access_token')
+    else:
+        return HttpResponse(
+            'authorisation required',
+            content_type='text/plain',
+            status=401
+        )
+
+    try:
+        token = decode(token)
+    except Exception as e:
+        return HttpResponse(
+            'invalid micropub token',
+            content_type='text/plain',
+            status=403,
+        )
+
+    return MicropubToken(token)
+
+
+class MicropubToken:
+    def __init__(self, tok):
+        self.user = get_user_model().objects.get(pk=tok['uid'])
+        self.client = tok['cid']
+        self.scope = tok['sco']
+
+        self.me = self.user.full_url
+        self.scopes = self.scope.split(' ')
+
+    def __contains__(self, scope):
+        return scope in self.scopes
 
 
 def encode(payload):

lemonauth/views/token.py

@@ -11,22 +11,13 @@ from lemoncurry import utils
 @method_decorator(csrf_exempt, name='dispatch')
 class TokenView(View):
     def get(self, req):
-        token = req.META.get('HTTP_AUTHORIZATION', '').split(' ')
-        if not token:
-            return utils.bad_req('missing Authorization header')
-        if token[0] != 'Bearer':
-            return utils.bad_req('only Bearer auth is supported')
-        try:
-            token = tokens.decode(token[1])
-        except Exception:
-            return utils.forbid('invalid token')
-
-        user = get_user_model().objects.get(pk=token['uid'])
-        me = urljoin(utils.origin(req), user.url)
+        token = tokens.auth(req)
+        if hasattr(token, 'content'):
+            return token
         res = {
-            'me': me,
-            'client_id': token['cid'],
-            'scope': token['sco'],
+            'me': token.me,
+            'client_id': token.client,
+            'scope': token.scope,
         }
         return utils.choose_type(req, res)