Fix illegal shell characters

This change checks if the user input is legal.
if illegal, we raise an InvalidValue excepiton.

Story: 2010004
Task: 45128

Change-Id: Ib81646b8f8a01fcbc31d033ec205491b76a7b755

releasenotes/notes/fix-illegal-value-be1acadc8c54c224.yaml

@@ -0,0 +1,6 @@
+---
+fixes:
+  - |
+    check if the user input is legal, currently, trove may have a 
+    RCE vulnerability. more details see:
+    `Stroy 2010004 <https://storyboard.openstack.org/#!/story/2010004>`__

trove/backup/service.py

@@ -88,6 +88,9 @@ class BackupController(wsgi.Controller):
         swift_container = data.get('swift_container')
         restore_from = data.get('restore_from')
 
+        if swift_container:
+            utils.validate_command(swift_container)
+
         context.notification = notification.DBaaSBackupCreate(
             context, request=req)
 

trove/common/exception.py

@@ -61,6 +61,10 @@ class InvalidRPCConnectionReuse(TroveError):
     message = _("Invalid RPC Connection Reuse.")
 
 
+class InvalidValue(TroveError):
+    message = _("The value is not allowed: %(value)s.")
+
+
 class NotFound(TroveError):
 
     message = _("Resource %(uuid)s cannot be found.")

trove/common/utils.py

@@ -17,6 +17,7 @@
 from collections import abc
 import inspect
 import os
+import shlex
 import shutil
 import uuid
 import urllib.parse as urlparse
@@ -423,3 +424,13 @@ def req_to_text(req):
         parts.extend([b'', safe_encode(req.body)])
 
     return b'\r\n'.join(parts).decode(req.charset)
+
+
+def validate_command(string):
+    """
+    Check if the string is legal for command
+
+    raise invalidvalue if illegal
+    """
+    if string != shlex.quote(string):
+        raise exception.InvalidValue(value=string)

trove/tests/unittests/common/test_utils.py

@@ -186,3 +186,19 @@ class TestUtils(trove_testtools.TestCase):
         expected = ('GET / HTTP/1.0\r\nHost: localhost:80\r\n'
                     'X-Auth-Project-Id: \u6d4b\u8bd5')
         self.assertEqual(expected, utils.req_to_text(req))
+
+    def test_validate_command(self):
+        string1 = "hello_world"
+        string2 = "hello world"
+        string3 = "hello@world_123"
+        string4 = "example.com/databse/mysql:5.7"
+        string5 = 'test --db-user="$(touch /rce_successful.txt)"'
+        self.assertIsNone(utils.validate_command(string1))
+        self.assertRaises(exception.InvalidValue,
+                          utils.validate_command,
+                          string2)
+        self.assertIsNone(utils.validate_command(string3))
+        self.assertIsNone(utils.validate_command(string4))
+        self.assertRaises(exception.InvalidValue,
+                          utils.validate_command,
+                          string5)