Merge pull request #2936 from bacongobbler/2681-validate-key-material

fix(controller): validate key material on upload

Author: Matthew Fisher

controller/api/models.py

@@ -5,6 +5,7 @@
 """
 
 from __future__ import unicode_literals
+import base64
 import etcd
 import importlib
 import logging
@@ -44,6 +45,14 @@ def log_event(app, msg, level=logging.INFO):
     app.log(msg)            # local filesystem
 
 
+def validate_base64(value):
+    """Check that value contains only valid base64 characters."""
+    try:
+        base64.b64decode(value.split()[1])
+    except Exception as e:
+        raise ValidationError(e)
+
+
 def validate_id_is_docker_compatible(value):
     """
     Check that the ID follows docker's image name constraints
@@ -825,7 +834,7 @@ class Key(UuidAuditedModel):
 
     owner = models.ForeignKey(settings.AUTH_USER_MODEL)
     id = models.CharField(max_length=128)
-    public = models.TextField(unique=True)
+    public = models.TextField(unique=True, validators=[validate_base64])
 
     class Meta:
         verbose_name = 'SSH Key'

controller/api/tests/test_key.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 """
 Unit tests for the Deis api app.
 
@@ -31,6 +32,10 @@
     "+MQnrrza0jaQ3QUAQ= autotest@autotesting comment"
 )
 
+BAD_KEY = (
+    "ssh-rsa foo_bar"
+)
+
 
 class KeyTest(TestCase):
 
@@ -63,12 +68,27 @@ def _check_key(self, pubkey):
         response = self.client.delete(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
         self.assertEqual(response.status_code, 204)
 
+    def _check_bad_key(self, pubkey):
+        """
+        Test that a user cannot add invalid SSH public keys
+        """
+        url = '/v1/keys'
+        body = {'id': 'mykey@box.local', 'public': pubkey}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 400)
+        return response
+
     def test_rsa_key(self):
         self._check_key(RSA_PUBKEY)
 
     def test_ecdsa_key(self):
         self._check_key(ECDSA_PUBKEY)
 
+    def test_bad_key(self):
+        response = self._check_bad_key(BAD_KEY)
+        self.assertEqual(response.data, {'public': ['Incorrect padding']})
+
     def _check_duplicate_key(self, pubkey):
         """
         Test that a user cannot add a duplicate key