Merge pull request #2136 from bacongobbler/remove_csrf_token

ref(controller): remove csrf token; switch to token-based authn

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
@@ -831,6 +832,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()
@@ -852,7 +860,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')