ref(controller): switch to token-based authn

When pulling out CSRF tokens from the authentication requirements, we
are still relying on cookies. This is great from a CSRF attacker's
standpoint as attackers can easily `<iframe>` the site that utilizes the
API, generate a POST request and re-use the existing cookie. Using
token-based authn fixes this because there will be no existing authn
cookie in the request. With this change, we are also not vulnerable to
XSRF attacks either because we do not store a validated login cookie
in the browser (yay!).

Author: Matthew Fisher

api/models.py

@@ -15,18 +15,19 @@
 import threading
 
 from django.conf import settings
-from django.contrib.auth.models import User
+from django.contrib.auth import get_user_model
 from django.core.exceptions import ValidationError
 from django.db import models
 from django.db.models import Max
-from django.db.models.signals import post_delete
-from django.db.models.signals import post_save
+from django.db.models.signals import post_delete, post_save
+from django.dispatch import receiver
 from django.utils.encoding import python_2_unicode_compatible
 from django_fsm import FSMField, transition
 from django_fsm.signals import post_transition
 from docker.utils import utils
 from json_field.fields import JSONField
 import requests
+from rest_framework.authtoken.models import Token
 
 from api import fields
 from registry import publish_release
@@ -830,6 +831,13 @@ def _etcd_purge_domains(**kwargs):
 post_delete.connect(_log_domain_removed, sender=Domain, dispatch_uid='api.models.log')
 
 
+# automatically generate a new token on creation
+@receiver(post_save, sender=get_user_model())
+def create_auth_token(sender, instance=None, created=False, **kwargs):
+    if created:
+        Token.objects.create(user=instance)
+
+
 # save FSM transitions as they happen
 def _save_transition(**kwargs):
     kwargs['instance'].save()
@@ -851,7 +859,7 @@ def _save_transition(**kwargs):
 if _etcd_client:
     post_save.connect(_etcd_publish_key, sender=Key, dispatch_uid='api.models')
     post_delete.connect(_etcd_purge_key, sender=Key, dispatch_uid='api.models')
-    post_delete.connect(_etcd_purge_user, sender=User, dispatch_uid='api.models')
+    post_delete.connect(_etcd_purge_user, sender=get_user_model(), dispatch_uid='api.models')
     post_save.connect(_etcd_publish_domains, sender=Domain, dispatch_uid='api.models')
     post_delete.connect(_etcd_purge_domains, sender=Domain, dispatch_uid='api.models')
     post_save.connect(_etcd_create_app, sender=App, dispatch_uid='api.models')

api/tests/test_api_middleware.py

@@ -5,7 +5,9 @@
 """
 from __future__ import unicode_literals
 
+from django.contrib.auth.models import User
 from django.test import TestCase
+from rest_framework.authtoken.models import Token
 
 from deis import __version__
 
@@ -17,16 +19,17 @@ class APIMiddlewareTest(TestCase):
     fixtures = ['tests.json']
 
     def setUp(self):
-        self.assertTrue(
-            self.client.login(username='autotest', password='password'))
+        self.user = User.objects.get(username='autotest')
+        self.token = Token.objects.get(user=self.user).key
 
     def test_x_deis_version_header_good(self):
         """
         Test that when the version header is sent, the request is accepted.
         """
         response = self.client.get(
             '/api/apps',
-            HTTP_X_DEIS_VERSION=__version__.rsplit('.', 1)[0]
+            HTTP_X_DEIS_VERSION=__version__.rsplit('.', 1)[0],
+            HTTP_AUTHORIZATION='token {}'.format(self.token),
         )
         self.assertEqual(response.status_code, 200)
 
@@ -36,13 +39,15 @@ def test_x_deis_version_header_bad(self):
         """
         response = self.client.get(
             '/api/apps',
-            HTTP_X_DEIS_VERSION='1234.5678'
+            HTTP_X_DEIS_VERSION='1234.5678',
+            HTTP_AUTHORIZATION='token {}'.format(self.token),
         )
         self.assertEqual(response.status_code, 405)
 
     def test_x_deis_version_header_not_present(self):
         """
         Test that when the version header is not present, the request is accepted.
         """
-        response = self.client.get('/api/apps')
+        response = self.client.get('/api/apps',
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)

api/tests/test_app.py

@@ -9,8 +9,10 @@
 import json
 import os.path
 
-from django.test import TestCase
 from django.conf import settings
+from django.contrib.auth.models import User
+from django.test import TestCase
+from rest_framework.authtoken.models import Token
 
 from api.models import App
 
@@ -21,8 +23,8 @@ class AppTest(TestCase):
     fixtures = ['tests.json']
 
     def setUp(self):
-        self.assertTrue(
-            self.client.login(username='autotest', password='password'))
+        self.user = User.objects.get(username='autotest')
+        self.token = Token.objects.get(user=self.user).key
         # provide mock authentication used for run commands
         settings.SSH_PRIVATE_KEY = '<some-ssh-private-key>'
 
@@ -35,39 +37,46 @@ def test_app(self):
         Test that a user can create, read, update and delete an application
         """
         url = '/api/apps'
-        response = self.client.post(url)
+        response = self.client.post(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
         app_id = response.data['id']  # noqa
         self.assertIn('id', response.data)
         self.assertIn('url', response.data)
         self.assertEqual(response.data['url'], '{app_id}.deisapp.local'.format(**locals()))
-        response = self.client.get('/api/apps')
+        response = self.client.get('/api/apps',
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data['results']), 1)
         url = '/api/apps/{app_id}'.format(**locals())
-        response = self.client.get(url)
+        response = self.client.get(url,
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)
         body = {'id': 'new'}
-        response = self.client.patch(url, json.dumps(body), content_type='application/json')
+        response = self.client.patch(url, json.dumps(body), content_type='application/json',
+                                     HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 405)
-        response = self.client.delete(url)
+        response = self.client.delete(url,
+                                      HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 204)
 
     def test_app_override_id(self):
         body = {'id': 'myid'}
         response = self.client.post('/api/apps', json.dumps(body),
-                                    content_type='application/json')
+                                    content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
         body = {'id': response.data['id']}
         response = self.client.post('/api/apps', json.dumps(body),
-                                    content_type='application/json')
+                                    content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertContains(response, 'App with this Id already exists.', status_code=400)
         return response
 
     def test_app_actions(self):
         url = '/api/apps'
         body = {'id': 'autotest'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
         app_id = response.data['id']  # noqa
         # test logs
@@ -78,20 +87,23 @@ def test_app_actions(self):
         if os.path.exists(path):
             os.remove(path)
         url = '/api/apps/{app_id}/logs'.format(**locals())
-        response = self.client.get(url)
+        response = self.client.get(url,
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 204)
         self.assertEqual(response.data, 'No logs for {}'.format(app_id))
         # write out some fake log data and try again
         with open(path, 'a') as f:
             f.write(FAKE_LOG_DATA)
-        response = self.client.get(url)
+        response = self.client.get(url,
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.data, FAKE_LOG_DATA)
         os.remove(path)
         # test run
         url = '/api/apps/{app_id}/run'.format(**locals())
         body = {'command': 'ls -al'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.data[0], 0)
         # delete file for future runs
@@ -101,12 +113,14 @@ def test_app_release_notes_in_logs(self):
         """Verifies that an app's release summary is dumped into the logs."""
         url = '/api/apps'
         body = {'id': 'autotest'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
         app_id = response.data['id']  # noqa
         path = os.path.join(settings.DEIS_LOG_DIR, app_id + '.log')
         url = '/api/apps/{app_id}/logs'.format(**locals())
-        response = self.client.get(url)
+        response = self.client.get(url,
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertIn('autotest created initial release', response.data)
         self.assertEqual(response.status_code, 200)
         # delete file for future runs
@@ -116,28 +130,33 @@ def test_app_errors(self):
         app_id = 'autotest-errors'
         url = '/api/apps'
         body = {'id': 'camelCase'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertContains(response, 'App IDs can only contain [a-z0-9-]', status_code=400)
         url = '/api/apps'
         body = {'id': 'deis'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertContains(response, "App IDs cannot be 'deis'", status_code=400)
         body = {'id': app_id}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
         app_id = response.data['id']  # noqa
         url = '/api/apps/{app_id}'.format(**locals())
-        response = self.client.delete(url)
+        response = self.client.delete(url,
+                                      HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEquals(response.status_code, 204)
         for endpoint in ('containers', 'config', 'releases', 'builds'):
             url = '/api/apps/{app_id}/{endpoint}'.format(**locals())
-            response = self.client.get(url)
+            response = self.client.get(url,
+                                       HTTP_AUTHORIZATION='token {}'.format(self.token))
             self.assertEquals(response.status_code, 404)
 
     def test_app_structure_is_valid_json(self):
         """Application structures should be valid JSON objects."""
         url = '/api/apps'
-        response = self.client.post(url)
+        response = self.client.post(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
         app_id = response.data['id']
         self.assertIn('structure', response.data)
@@ -146,57 +165,59 @@ def test_app_structure_is_valid_json(self):
         app.structure = {'web': 1}
         app.save()
         url = '/api/apps/{}'.format(app_id)
-        response = self.client.get(url)
+        response = self.client.get(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertIn('structure', response.data)
         self.assertEqual(response.data['structure'], {"web": 1})
 
     def test_admin_can_manage_other_apps(self):
         """Administrators of Deis should be able to manage all applications.
         """
         # log in as non-admin user and create an app
-        self.assertTrue(
-            self.client.login(username='autotest2', password='password'))
+        user = User.objects.get(username='autotest2')
+        token = Token.objects.get(user=user)
         app_id = 'autotest'
         url = '/api/apps'
         body = {'id': app_id}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(token))
         # log in as admin, check to see if they have access
-        self.assertTrue(
-            self.client.login(username='autotest', password='password'))
         url = '/api/apps/{}'.format(app_id)
-        response = self.client.get(url)
+        response = self.client.get(url,
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)
         # check app logs
         url = '/api/apps/{app_id}/logs'.format(**locals())
-        response = self.client.get(url)
+        response = self.client.get(url,
+                                   HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)
         self.assertIn('autotest2 created initial release', response.data)
         # run one-off commands
         url = '/api/apps/{app_id}/run'.format(**locals())
         body = {'command': 'ls -al'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.data[0], 0)
         # delete the app
         url = '/api/apps/{}'.format(app_id)
-        response = self.client.delete(url)
+        response = self.client.delete(url,
+                                      HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 204)
 
     def test_admin_can_see_other_apps(self):
         """If a user creates an application, the administrator should be able
         to see it.
         """
         # log in as non-admin user and create an app
-        self.assertTrue(
-            self.client.login(username='autotest2', password='password'))
+        user = User.objects.get(username='autotest2')
+        token = Token.objects.get(user=user)
         app_id = 'autotest'
         url = '/api/apps'
         body = {'id': app_id}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(token))
         # log in as admin
-        self.assertTrue(
-            self.client.login(username='autotest', password='password'))
-        response = self.client.get(url)
+        response = self.client.get(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.data['count'], 1)
 
     def test_run_without_auth(self):
@@ -205,13 +226,15 @@ def test_run_without_auth(self):
         settings.SSH_PRIVATE_KEY = ''
         url = '/api/apps'
         body = {'id': 'autotest'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 201)
         app_id = response.data['id']  # noqa
         # test run
         url = '/api/apps/{app_id}/run'.format(**locals())
         body = {'command': 'ls -al'}
-        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEquals(response.status_code, 400)
         self.assertEquals(response.data, 'Support for admin commands is not configured')