check for invalid hostnames on update_attributes

Adds the hostname validation and username validation to the
update_attributes code for users to prevent bad data getting
into the system and breaking in unexpected places.

Change-Id: Ib501357e45e99945ab972ceed044efc1e2377de5
Closes-Bug: 1235845

trove/extensions/mysql/models.py

@@ -142,12 +142,20 @@ class User(object):
                           user_attrs):
         load_and_verify(context, instance_id)
         client = create_guest_client(context, instance_id)
-        user_name = user_attrs.get('name')
-        host_name = user_attrs.get('host')
-        user = user_name or username
-        host = host_name or hostname
+
+        user_changed = user_attrs.get('name')
+        host_changed = user_attrs.get('host')
+
+        validate = guest_models.MySQLUser()
+        if host_changed:
+            validate.host = host_changed
+        if user_changed:
+            validate.name = user_changed
+
+        user = user_changed or username
+        host = host_changed or hostname
         userhost = "%s@%s" % (user, host)
-        if user_name or host_name:
+        if user_changed or host_changed:
             existing_users, _nadda = Users.load_with_client(
                 client,
                 limit=1,

trove/extensions/mysql/service.py

@@ -144,8 +144,11 @@ class UserController(wsgi.Controller):
             raise exception.BadRequest(msg=str(e))
         if not user:
             raise exception.UserNotFound(uuid=id)
-        models.User.update_attributes(context, instance_id, username, hostname,
-                                      user_attrs)
+        try:
+            models.User.update_attributes(context, instance_id, username,
+                                          hostname, user_attrs)
+        except (ValueError, AttributeError) as e:
+            raise exception.BadRequest(msg=str(e))
         return wsgi.Result(None, 202)
 
     def update_all(self, req, body, tenant_id, instance_id):

trove/tests/api/users.py

@@ -270,6 +270,28 @@ class TestUsers(object):
         self.dbaas.users.delete(instance_info.id, "testuser2",
                                 hostname=hostname2)
 
+    @test()
+    def test_updateduser_newhost_invalid(self):
+        # Ensure invalid hostnames/usernames aren't allowed to enter the system
+        users = []
+        username = "testuser1"
+        hostname1 = "192.168.0.1"
+        users.append({"name": username, "password": "password",
+                      "host": hostname1, "databases": []})
+        self.dbaas.users.create(instance_info.id, users)
+        hostname1 = hostname1.replace('.', '%2e')
+        assert_raises(exceptions.BadRequest,
+                      self.dbaas.users.update_attributes, instance_info.id,
+                      username, {"host": "badjuju"}, hostname1)
+        assert_equal(400, self.dbaas.last_http_code)
+
+        assert_raises(exceptions.BadRequest,
+                      self.dbaas.users.update_attributes, instance_info.id,
+                      username, {"name": " bad username   "}, hostname1)
+        assert_equal(400, self.dbaas.last_http_code)
+
+        self.dbaas.users.delete(instance_info.id, username, hostname=hostname1)
+
     @test()
     def test_cannot_change_rootpassword(self):
         # Cannot change password for a root user