Merge lp:~james-w/lava-dashboard/oauth into lp:lava-dashboard

[James Westby]

Hi,

Another inital code push for discussion, this time adding oauth support.

I looked at the existing django apps for doing this, and didn't really
like the approach of any of them. django-oauth requires separate decorators
on views that can and should accept oauth signed requests. django-piston
is closer I think, but implies lots of other things that don't really
fit in to the current architecture. If we want to move towards something
like django-piston provides then we should drop this and go that way I think.

The approach I took is to provide a middleware that sets request.user
if the request is oauth signed. This means that django's existing auth
framework continues to work, and you can just use it's existing decorators
etc. to do access control, and use request.user to know who is making
a request.

There are some issues though:

1. This relies on an external project that is unpackaged at this time.
2. That external project ships a patched embedded copy of python-oauth, though
I don't know what the patches are for.
3. That project requires consumers to be pre-registered, and I'm not sure
we want that. It would be possible to work around it, but would require
some work.
4. I'm not sure I have the Resource stuff correct in this branch.
5. I'm not convinced that I have thought through all the corners and so there
may be security holes.
6. There is nothing so far for the view to know if the request is oauth,
which consumer it is etc., and no support for differing token access levels.
We won't need any of that right away, but if we want that then django-piston
may be the way to go rather than adding all of that.

Thanks,

James

[James Westby (james-w)]

Oh, and django-oauth's tests are not isolated, so running all the tests
in the project currently fails.

Thanks,

James

[Zygmunt Krynicki (zyga)]

This branch could be re-targeted at linaro-django-xmlrpc. If we are going to do oauth (eventually we might) then that's the logical place it should go to.

Unmerged revisions

73. By James Westby on 2010-10-08

Allow request.user to be missing.

72. By James Westby on 2010-10-08

Add a middleware on top of django-oauth that puts the user on the request as normal.

=== modified file 'dashboard_server/default_settings.py'
--- dashboard_server/default_settings.py	2010-10-02 21:25:06 +0000
+++ dashboard_server/default_settings.py	2010-10-08 19:37:59 +0000
@@ -103,6 +103,7 @@
     'django.middleware.common.CommonMiddleware',
     'django.contrib.sessions.middleware.SessionMiddleware',
     'django.contrib.auth.middleware.AuthenticationMiddleware',
+    'oauth_middleware.OAuthMiddleware',
     'django.middleware.transaction.TransactionMiddleware',
 )
 
@@ -129,6 +130,8 @@
     'django.contrib.sites',
     'django.contrib.databrowse',
     'django.contrib.humanize',
+    'oauth_provider',
+    'oauth_middleware',
     'dashboard_app',
 )
 
 
=== added directory 'oauth_middleware'
=== added file 'oauth_middleware/__init__.py'
--- oauth_middleware/__init__.py	1970-01-01 00:00:00 +0000
+++ oauth_middleware/__init__.py	2010-10-08 19:37:59 +0000
@@ -0,0 +1,90 @@
+from django.conf import settings
+from django.http import HttpResponseForbidden
+
+from oauth.oauth import (
+    OAuthError,
+    OAuthRequest,
+    OAuthServer,
+    OAuthSignatureMethod_PLAINTEXT,
+    OAuthSignatureMethod_HMAC_SHA1,
+    )
+
+from oauth_provider.decorators import CheckOAuth
+from oauth_provider.stores import DataStore as _DataStore
+
+
+def get_oauth_auth_params(request):
+    # Django converts Authorization header in HTTP_AUTHORIZATION
+    auth_header = {}
+    if 'Authorization' in request.META:
+        auth_header = {'Authorization': request.META['Authorization']}
+    elif 'HTTP_AUTHORIZATION' in request.META:
+        auth_header =  {'Authorization': request.META['HTTP_AUTHORIZATION']}
+    # Don't include extra parameters when request.method is POST and 
+    # request.MIME['CONTENT_TYPE'] is "application/x-www-form-urlencoded" 
+    # (See http://oauth.net/core/1.0a/#consumer_req_param).
+    # But there is an issue with Django's test Client and custom content types
+    # so an ugly test is made here, if you find a better solution...
+    parameters = {}
+    if request.method == "POST" and \
+        (request.META.get('CONTENT_TYPE') == "application/x-www-form-urlencoded" \
+            or request.META.get('SERVER_NAME') == 'testserver'):
+        parameters = dict(request.REQUEST.items())
+    return {"method": request.method,
+            "uri": request.build_absolute_uri(),
+            "auth_header": auth_header,
+            "parameters": parameters,
+            "query_string": request.META.get('QUERY_STRING', ''),
+           }
+
+
+class DataStore(_DataStore):
+
+    def lookup_consumer(self, key):
+        # Do something here to allow arbitrary consumers based on a
+        # settings key. Also need to override other methods in that case
+        # to avoid key checks
+        return super(DataStore, self).lookup_consumer(key)
+
+
+def authenticate(method=None, uri=None, auth_header=None,
+                 parameters=None, query_string=None):
+    oauth_request = OAuthRequest.from_request(method,
+                                      uri,
+                                      headers=auth_header,
+                                      parameters=parameters,
+                                      query_string=query_string)
+    if oauth_request is None:
+        # Do we want to 403 here, it's half an oauth signed request?
+        return None
+    OAUTH_SIGNATURE_METHODS = getattr(settings, 'OAUTH_SIGNATURE_METHODS', ['plaintext', 'hmac-sha1'])
+    oauth_server = OAuthServer(DataStore(oauth_request))
+    if 'plaintext' in OAUTH_SIGNATURE_METHODS:
+        oauth_server.add_signature_method(OAuthSignatureMethod_PLAINTEXT())
+    if 'hmac-sha1' in OAUTH_SIGNATURE_METHODS:
+        oauth_server.add_signature_method(OAuthSignatureMethod_HMAC_SHA1())
+    consumer, token, parameters = oauth_server.verify_request(oauth_request)
+    # Handle resource names?
+    return token.user
+
+
+class OAuthMiddleware(object):
+
+    def process_request(self, request):
+        oauth_auth_params = get_oauth_auth_params(request)
+        try:
+            user = authenticate(**oauth_auth_params)
+        except OAuthError, e:
+            return HttpResponseForbidden(content=str(e))
+        if user:
+            # If the user is already authenticated and that user doesn't
+            # match the one that the headers are for then something
+            # is wrong, so give an Unauthenticated response
+            if getattr(request, "user", None) and request.user.is_authenticated():
+                if request.user != user:
+                    return HttpResponseForbidden()
+            # User is valid. Overwrite request.user with the user that
+            # signed the request, which will prevent any
+            # django.contrib.auth checks, and so sessions will be
+            # invalid.
+            request.user = user
 
=== added file 'oauth_middleware/models.py'
=== added file 'oauth_middleware/tests.py'
--- oauth_middleware/tests.py	1970-01-01 00:00:00 +0000
+++ oauth_middleware/tests.py	2010-10-08 19:37:59 +0000
@@ -0,0 +1,307 @@
+import time
+
+from django.conf import settings
+from django.contrib.auth.models import AnonymousUser, User
+from django.http import HttpRequest
+from django.test import TestCase
+
+from oauth.oauth import (
+    OAuthError,
+    OAuthRequest,
+    OAuthSignatureMethod_PLAINTEXT,
+    )
+from oauth_provider.models import (
+    Consumer,
+    Resource,
+    Token,
+    )
+
+from oauth_middleware import (
+    authenticate,
+    get_oauth_auth_params,
+    OAuthMiddleware,
+    )
+
+
+class TestRequest(HttpRequest):
+
+    def __init__(self, params=None):
+        super(TestRequest, self).__init__()
+        self.META["HTTP_HOST"] = "foo"
+        self._params = params or {}
+
+    def _get_request(self):
+        return self._params
+
+    REQUEST = property(_get_request)
+
+
+class GetOAuthAuthParamsTests(TestCase):
+
+    def get_request(self, params=None):
+        request = TestRequest(params=params)
+        return request
+
+    def test_empty_auth_header_by_default(self):
+        request = self.get_request()
+        params = get_oauth_auth_params(request)
+        self.assertEqual({}, params["auth_header"])
+
+    def test_auth_header_from_Authorization(self):
+        request = self.get_request()
+        request.META["Authorization"] = "foo"
+        params = get_oauth_auth_params(request)
+        self.assertEqual({"Authorization": "foo"}, params["auth_header"])
+
+    def test_auth_header_from_Http_Authorization(self):
+        request = self.get_request()
+        request.META["HTTP_AUTHORIZATION"] = "bar"
+        params = get_oauth_auth_params(request)
+        self.assertEqual({"Authorization": "bar"}, params["auth_header"])
+
+    def test_auth_header_from_Authorization_not_Http_Auth(self):
+        request = self.get_request()
+        request.META["Authorization"] = "foo"
+        request.META["HTTP_AUTHORIZATION"] = "bar"
+        params = get_oauth_auth_params(request)
+        self.assertEqual({"Authorization": "foo"}, params["auth_header"])
+
+    def test_parameters_empty_by_default(self):
+        request = self.get_request()
+        params = get_oauth_auth_params(request)
+        self.assertEqual({}, params["parameters"])
+
+    def test_parameters_with_form_post(self):
+        params = {"foo": "bar"}
+        request = self.get_request(params=params)
+        request.method = "POST"
+        request.META["CONTENT_TYPE"] = "application/x-www-form-urlencoded"
+        oauth_params = get_oauth_auth_params(request)
+        self.assertEqual(params, oauth_params["parameters"])
+
+    def test_parameters_with_post_to_test_server(self):
+        params = {"foo": "bar"}
+        request = self.get_request(params=params)
+        request.method = "POST"
+        request.META["SERVER_NAME"] = "testserver"
+        oauth_params = get_oauth_auth_params(request)
+        self.assertEqual(params, oauth_params["parameters"])
+
+    def test_parameters_includes_method(self):
+        request = self.get_request()
+        request.method = "PATCH"
+        oauth_params = get_oauth_auth_params(request)
+        self.assertEqual("PATCH", oauth_params["method"])
+
+    def test_parameters_includes_absolute_uri(self):
+        request = self.get_request()
+        oauth_params = get_oauth_auth_params(request)
+        self.assertEqual(request.build_absolute_uri(), oauth_params["uri"])
+
+    def test_parameters_includes_empty_query_string_by_default(self):
+        request = self.get_request()
+        oauth_params = get_oauth_auth_params(request)
+        self.assertEqual('', oauth_params["query_string"])
+
+    def test_parameters_includes_query_string(self):
+        query_string = "?foo=bar"
+        request = self.get_request()
+        request.META["QUERY_STRING"] = query_string
+        oauth_params = get_oauth_auth_params(request)
+        self.assertEqual(query_string, oauth_params["query_string"])
+
+
+class OAuthTestCase(TestCase):
+
+    def setUp(self):
+        self._original_signature_methods = getattr(
+            settings, 'OAUTH_SIGNATURE_METHODS', None)
+        super(OAuthTestCase, self).setUp()
+
+    def tearDown(self):
+        super(OAuthTestCase, self).tearDown()
+        if self._original_signature_methods is not None:
+            setattr(
+                settings, 'OAUTH_SIGNATURE_METHODS',
+                self._original_signature_methods)
+        elif getattr(settings, 'OAUTH_SIGNATURE_METHODS', None) is not None:
+            delattr(settings, 'OAUTH_SIGNATURE_METHODS')
+
+    def get_user(self, username=None):
+        if username is None:
+            username = "test-user"
+        user = User.objects.create(username=username)
+        user.save()
+        return user
+
+    def get_consumer(self):
+        return Consumer.objects.create_consumer("consumer")
+
+    def get_resource(self):
+        return Resource.objects.create(name="foo", url="bar")
+
+    def get_token(self, user, consumer, resource):
+        return Token.objects.create_token(
+            consumer, Token.ACCESS, time.time(), resource,
+            user=user)
+
+    def get_request(self, consumer, token, method, uri):
+        request = OAuthRequest.from_consumer_and_token(
+            consumer, token=token, http_method=method, http_url=uri)
+        request.sign_request(
+            OAuthSignatureMethod_PLAINTEXT(), consumer, token)
+        return request
+
+    def get_signed_request(self, user, method, uri):
+        consumer = self.get_consumer()
+        resource = self.get_resource()
+        token = self.get_token(user, consumer, resource)
+        request = OAuthRequest.from_consumer_and_token(
+            consumer, token=token, http_method=method, http_url=uri)
+        request.sign_request(
+            OAuthSignatureMethod_PLAINTEXT(), consumer, token)
+        return request
+
+
+class AuthenticateTests(OAuthTestCase):
+
+    def test_invalid_oauth_request_returns_None(self):
+        user = authenticate(uri="http://example.com")
+        self.assertEqual(None, user)
+
+    def test_invalid_oauth_params_raise_error(self):
+        self.assertRaises(
+            OAuthError, authenticate, uri="http://example.com",
+            auth_header={"Authorization": "OAuth foo"})
+
+    def test_authenticate_returns_user(self):
+        user = self.get_user()
+        method = "POST"
+        uri = "http://example.com"
+        request = self.get_signed_request(user, method, uri)
+        self.assertEqual(
+            user,
+            authenticate(
+                method=method, uri=uri,
+                auth_header=request.to_header()))
+
+    def test_authenticate_refuses_unwanted_methods(self):
+        settings.OAUTH_SIGNATURE_METHODS = ["hmac-sha1"]
+        user = self.get_user()
+        method = "POST"
+        uri = "http://example.com"
+        request = self.get_signed_request(user, method, uri)
+        self.assertRaises(
+            OAuthError, authenticate, method=method, uri=uri,
+                auth_header=request.to_header())
+
+    def test_authenticate_refuses_bad_signature(self):
+        user = self.get_user()
+        method = "POST"
+        uri = "http://example.com"
+        request = self.get_signed_request(user, method, uri)
+        request.set_parameter('oauth_signature', '000')
+        self.assertRaises(
+            OAuthError, authenticate, method=method, uri=uri,
+                auth_header=request.to_header())
+
+    def test_authenticate_refuses_unknown_token(self):
+        user = self.get_user()
+        consumer = self.get_consumer()
+        resource = self.get_resource()
+        token = self.get_token(user, consumer, resource)
+        method = "POST"
+        uri = "http://example.com"
+        request = self.get_request(consumer, token, method, uri)
+        request.set_parameter('oauth_signature', '000')
+        token.delete()
+        self.assertRaises(
+            OAuthError, authenticate, method=method, uri=uri,
+                auth_header=request.to_header())
+
+    def test_authenticate_refuses_unknown_user(self):
+        user = self.get_user()
+        consumer = self.get_consumer()
+        resource = self.get_resource()
+        token = self.get_token(user, consumer, resource)
+        method = "POST"
+        uri = "http://example.com"
+        request = self.get_request(consumer, token, method, uri)
+        request.set_parameter('oauth_signature', '000')
+        user.delete()
+        self.assertRaises(
+            OAuthError, authenticate, method=method, uri=uri,
+                auth_header=request.to_header())
+
+
+class OAuthMiddlewareTests(OAuthTestCase):
+
+    def test_nothing_happens_on_non_oauth_requests(self):
+        request = TestRequest()
+        request.user = "user"
+        middleware = OAuthMiddleware()
+        response = middleware.process_request(request)
+        self.assertEqual("user", request.user)
+        self.assertEqual(None, response)
+
+    def test_forbidden_on_bad_oauth_request(self):
+        request = TestRequest()
+        request.user = "user"
+        request.META["HTTP_AUTHORIZATION"] = (
+            "OAuth oauth_consumer_key=foo oauth_token=bar "
+            "oauth_signature=baz oauth_signature_method=zap "
+            "oauth_timestamp=zabg oauth_nonce=zot")
+        middleware = OAuthMiddleware()
+        response = middleware.process_request(request)
+        self.assertEqual(403, response.status_code)
+
+    def test_authenticated_user_set_on_request(self):
+        user = self.get_user()
+        request = TestRequest()
+        request.user = AnonymousUser()
+        method = "POST"
+        uri = "http://example.com"
+        oauth_request = self.get_signed_request(user, method, uri)
+        request.META["HTTP_AUTHORIZATION"] = oauth_request.to_header().values()[0]
+        middleware = OAuthMiddleware()
+        response = middleware.process_request(request)
+        self.assertEqual(None, response)
+        self.assertEqual(user, request.user)
+
+    def test_forbidden_if_request_already_authenticated(self):
+        user = self.get_user(username="test-user")
+        other_user = self.get_user(username="other-user")
+        request = TestRequest()
+        request.user = other_user
+        method = "POST"
+        uri = "http://example.com"
+        oauth_request = self.get_signed_request(user, method, uri)
+        request.META["HTTP_AUTHORIZATION"] = oauth_request.to_header().values()[0]
+        middleware = OAuthMiddleware()
+        response = middleware.process_request(request)
+        self.assertEqual(403, response.status_code)
+
+    def test_not_forbidden_if_same_user_authenticated(self):
+        user = self.get_user()
+        request = TestRequest()
+        request.user = user
+        method = "POST"
+        uri = "http://example.com"
+        oauth_request = self.get_signed_request(user, method, uri)
+        request.META["HTTP_AUTHORIZATION"] = oauth_request.to_header().values()[0]
+        middleware = OAuthMiddleware()
+        response = middleware.process_request(request)
+        self.assertEqual(None, response)
+        self.assertEqual(user, request.user)
+
+    def test_allows_no_user_on_request(self):
+        user = self.get_user()
+        request = TestRequest()
+        method = "POST"
+        uri = "http://example.com"
+        oauth_request = self.get_signed_request(user, method, uri)
+        request.META["HTTP_AUTHORIZATION"] = oauth_request.to_header().values()[0]
+        middleware = OAuthMiddleware()
+        response = middleware.process_request(request)
+        self.assertEqual(None, response)
+        self.assertEqual(user, request.user)
 
=== added file 'oauth_middleware/views.py'