Login via OAuth2

Author: stefankoegl

mygpo/moauth/__init__.py

@@ -0,0 +1,7 @@
+from django.contrib import admin
+
+from . import models
+
+@admin.register(models.AuthRequest)
+class AuthRequestAdmin(admin.ModelAdmin):
+    list_display = ('created', 'state')

mygpo/moauth/admin.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class AuthConfig(AppConfig):
+    name = 'auth'

mygpo/moauth/apps.py

@@ -0,0 +1,58 @@
+import requests
+
+from django.db import IntegrityError
+from django.contrib.auth import get_user_model
+from django.contrib.auth.backends import ModelBackend
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+class OAuth2Backend(ModelBackend):
+    """ OAuth2 authentication backend
+
+    Authenticates based on token info URL; Uses Users from the ModelBackend """
+
+    def authenticate(self, token_info_url=None):
+        logger.info('Authenticating user from "%s"', token_info_url)
+        if token_info_url is None:
+            return
+
+        token = self._get_token_info(token_info_url)
+        username = token['user']['login']
+        return self._get_user(username)
+
+    def _get_token_info(self, token_info_url):
+        """ Retrieves token info and returns the username """
+
+        headers = {
+            'Accept': 'application/json'
+        }
+
+        r = requests.get(token_info_url, headers=headers)
+        token = r.json()
+        #{
+        #    'token': '62b6a03b16a5453f810cf6d32ac975f8',
+        #    'app': {
+        #        'url': None,
+        #        'name': 'gpodder.net',
+        #        'client_id': 'Nb0QLDW2psFSXfGwmCvJ1ElhITu9P3Kg'
+        #    },
+        #    'created_at': '2016-02-07T12:42:14.140Z',
+        #    'user': {
+        #        'login': 'stefan'
+        #    },
+        #    'scopes': [
+        #        'actions:add',
+        #        'podcastlists'
+        #    ]
+        #}
+        return token
+
+    def _get_user(self, username):
+        """ Get user based on username """
+        User = get_user_model()
+        try:
+            return User.objects.create(username=username)
+        except IntegrityError as ie:
+            return User.objects.get(username__iexact=username)

mygpo/moauth/backends.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.9.2 on 2016-02-07 12:13
+from __future__ import unicode_literals
+
+import datetime
+import django.contrib.postgres.fields
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='AuthRequest',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('scopes', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(blank=True, max_length=64), size=None)),
+                ('state', models.CharField(max_length=32)),
+                ('created', models.DateTimeField(default=datetime.datetime.utcnow)),
+            ],
+        ),
+    ]

mygpo/moauth/migrations/0001_initial.py

@@ -0,0 +1,13 @@
+from datetime import datetime
+
+from django.db import models
+from django.contrib.postgres.fields import ArrayField
+
+
+class AuthRequest(models.Model):
+
+    scopes = ArrayField(models.CharField(max_length=64, blank=True))
+
+    state = models.CharField(max_length=32)
+
+    created = models.DateTimeField(default=datetime.utcnow)

mygpo/moauth/migrations/__init__.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

mygpo/moauth/models.py

@@ -0,0 +1,16 @@
+from django.conf.urls import url
+
+from . import views
+
+
+urlpatterns = [
+
+    url(r'^oauth/login$',
+        views.InitiateOAuthLogin.as_view(),
+        name='login-oauth'),
+
+    url(r'^oauth/callback$',
+        views.OAuthCallback.as_view(),
+        name='oauth-callback'),
+
+]

mygpo/moauth/tests.py

@@ -0,0 +1,156 @@
+import requests
+from requests.auth import HTTPBasicAuth
+
+import urllib.parse
+
+from django.db import IntegrityError
+from django.core.urlresolvers import reverse
+from django.shortcuts import render
+from django.views.generic.base import RedirectView
+from django.views.generic.base import View
+from django.http import HttpResponseRedirect
+from django.contrib.sites.requests import RequestSite
+from django.contrib.auth import login, get_user_model, authenticate
+from django.conf import settings
+
+from mygpo.utils import random_token
+from . import models
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+AVAILABLE_SCOPES = [
+    'subscriptions',
+    'suggestions',
+    'account',
+    'favorites',
+    'podcastlists',
+    'apps:get',
+    'apps:sync',
+    'actions:get',
+    'actions:add',
+]
+
+class InitiateOAuthLogin(RedirectView):
+
+    def get_redirect_url(self):
+
+        client_id = settings.MYGPO_AUTH_CLIENT_ID
+        redir_uri = self._get_callback_url()
+        state = random_token()
+        response_type = 'code'
+
+        models.AuthRequest.objects.create(
+            scopes = AVAILABLE_SCOPES,
+            state = state,
+        )
+        logger.info('Initiated new new auth request "%s"', state)
+
+        scopes = AVAILABLE_SCOPES
+        qs = self._get_qs(client_id, redir_uri, scopes, state, response_type)
+        return _get_authorize_url('/authorize', qs)
+
+    def _get_qs(self, client_id, redirect_uri, scopes, state, response_type):
+        return urllib.parse.urlencode([
+            ('client_id', client_id),
+            ('redirect_uri', redirect_uri),
+            ('scope', ' '.join(scopes)),
+            ('state', state),
+            ('response_type', response_type),
+        ])
+
+    def _get_callback_url(self):
+        protocol = 'https' if self.request.is_secure() else 'http'
+        site = RequestSite(self.request)
+        domain = site.domain
+        view = reverse('oauth-callback')
+        return '{0}://{1}{2}'.format(protocol, domain, view)
+
+
+class OAuthCallback(View):
+    """ OAuth 2 callback handler
+
+    Gets and verifies token, logs in user """
+
+    def get(self, request):
+
+        if 'error' in self.request.GET:
+            # handle error
+            # error=server_error&error_description=An+unknown+error+occured
+            return
+
+        code = self.request.GET.get('code', None)
+        state = self.request.GET.get('state', None)
+
+        try:
+            authreq = models.AuthRequest.objects.get(state=state)
+        except models.AuthRequest.DoesNotExist:
+            # handle
+            return
+
+        access_token, token_info_url = self._get_access_token(code)
+
+        user = authenticate(token_info_url=token_info_url)
+        login(self.request, user)
+
+        return HttpResponseRedirect(reverse('home'))
+
+    def _get_access_token(self, code):
+        payload = {
+            'grant_type': 'authorization_code',
+            'code': code,
+            'client_id': settings.MYGPO_AUTH_CLIENT_ID,
+        }
+        auth = HTTPBasicAuth(settings.MYGPO_AUTH_CLIENT_ID,
+                             settings.MYGPO_AUTH_CLIENT_SECRET)
+
+        qs = self._get_qs(AVAILABLE_SCOPES)
+        token_url = _get_authorize_url('/token', qs)
+        r = requests.post(token_url, data=payload, auth=auth)
+        if r.status_code != 200:
+            return # handle error
+
+        resp = r.json()
+        access_token = resp['access_token']
+        expires_in = resp['expires_in']
+        token_type = resp['token_type']
+        scopes = resp['scope'].split(' ')
+        #{
+        #    'expires_in': 3599.995724,
+        #    'access_token': 'a46de116972b46e88481e7a082db60ca',
+        #    'token_type': 'Bearer',
+        #    'scope': 'podcastlists subscriptions suggestions apps:get actions:get account actions:add apps:sync favorites'
+        #}
+        logger.info(
+            'Received %s token "%s" for scopes "%s", expires in %f',
+            token_type, access_token, ' '.join(scopes), expires_in
+        )
+
+        token_info = r.links['https://gpodder.net/relation/token-info']['url']
+
+        # Reference Resolution
+        # https://tools.ietf.org/html/rfc3986#section-5
+        token_info_url = urllib.parse.urljoin(settings.MYGPO_AUTH_URL,
+                                              token_info)
+
+        return access_token, token_info_url
+
+        login(self.request, user)
+
+
+    def _get_qs(self, scopes):
+        return urllib.parse.urlencode([
+            ('scope', ' '.join(scopes)),
+        ])
+
+
+def _get_authorize_url(endpoint, qs):
+    r = urllib.parse.urlsplit(settings.MYGPO_AUTH_URL)
+    path = r.path
+    if path.endswith('/'):
+        path = path[:-1]
+
+    path = path + endpoint
+    parts = (r.scheme, r.netloc, path, qs, r.fragment)
+    return urllib.parse.urlunsplit(parts)