Implement a token endpoint - currently all tokens last forever and can't be revoked, but I can add revocation later without too much trouble

lemonauth/tokens.py

@@ -25,3 +25,13 @@ def gen_auth_code(req):
         code['sco'] = ' '.join(req.POST.getlist('scope'))
 
     return encode(code)
+
+
+def gen_token(code):
+    tok = {
+        'uid': code['uid'],
+        'cid': code['cid'],
+        'sco': code['sco'],
+        'iat': datetime.utcnow(),
+    }
+    return encode(tok).decode('utf-8')

lemonauth/urls.py

@@ -7,4 +7,5 @@ urlpatterns = [
     url('^logout$', views.logout, name='logout'),
     url('^indie$', views.IndieView.as_view(), name='indie'),
     url('^indie/approve$', views.indie_approve, name='indie_approve'),
+    url('^token$', views.TokenView.as_view(), name='token'),
 ]

lemonauth/views/__init__.py

@@ -1,3 +1,4 @@
 from .login import login
 from .logout import logout
 from .indie import IndieView, approve as indie_approve
+from .token import TokenView

lemonauth/views/token.py

@@ -0,0 +1,58 @@
+from django.contrib.auth import get_user_model
+from django.views import View
+from django.utils.decorators import method_decorator
+from django.views.decorators.csrf import csrf_exempt
+from urllib.parse import urljoin
+
+from .. import tokens
+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)
+        res = {
+            'me': me,
+            'client_id': token['cid'],
+            'scope': token['sco'],
+        }
+        return utils.choose_type(req, res)
+
+    def post(self, req):
+        post = req.POST
+        try:
+            code = tokens.decode(post.get('code'))
+        except Exception:
+            return utils.forbid('invalid auth code')
+
+        if code['typ'] != 'code':
+            return utils.bad_req(
+                'this endpoint only supports response_type=code'
+            )
+        if code['cid'] != post.get('client_id'):
+            return utils.forbid('client id did not match')
+        if code['uri'] != post.get('redirect_uri'):
+            return utils.forbid('redirect uri did not match')
+
+        user = get_user_model().objects.get(pk=code['uid'])
+        me = urljoin(utils.origin(req), user.url)
+        if me != post.get('me'):
+            return utils.forbid('me did not match')
+
+        return utils.choose_type(req, {
+            'access_token': tokens.gen_token(code),
+            'me': me,
+            'scope': code['sco'],
+        })

lemoncurry/utils.py

@@ -1,7 +1,7 @@
 import json
 from accept_types import get_best_match
 from django.conf import settings
-from django.http import HttpResponse
+from django.http import HttpResponse, JsonResponse
 from django.http import HttpResponseForbidden, HttpResponseBadRequest
 from os.path import join
 from shorturls import default_converter as converter
@@ -28,14 +28,6 @@ def uri(request):
     return origin(request) + request.path
 
 
-def choose_type(request, content, reps):
-    accept = request.META.get('HTTP_ACCEPT', '*/*')
-    type = get_best_match(accept, reps.keys())
-    if type:
-        return reps[type](content)
-    return HttpResponse(status=406)
-
-
 def form_encoded_response(content):
     return HttpResponse(
         urlencode(content),
@@ -43,6 +35,20 @@ def form_encoded_response(content):
     )
 
 
+REPS = {
+    'application/x-www-form-urlencoded': form_encoded_response,
+    'application/json': JsonResponse,
+}
+
+
+def choose_type(request, content, reps=REPS):
+    accept = request.META.get('HTTP_ACCEPT', '*/*')
+    type = get_best_match(accept, reps.keys())
+    if type:
+        return reps[type](content)
+    return HttpResponse(status=406)
+
+
 def shortlink(obj):
     prefix = ShortURL(None).get_prefix(obj)
     tinyid = converter.from_decimal(obj.pk)