Add PathFilter to rootwrap.

PathFilter is a type of filter that allows to check
if path arguments of a command resolve to file system
paths within given directories.

Fixes bug 1098568.

Change-Id: Ie2686ad2ff114075c6d8d804031b6e3fa60a43ca

2 files changed, 165 insertions, 0 deletions

diff --git a/openstack/common/rootwrap/filters.py b/openstack/common/rootwrap/filters.py
index eadda25..d9618af 100644
--- a/openstack/common/rootwrap/filters.py
+++ b/openstack/common/rootwrap/filters.py
@@ -88,6 +88,52 @@ class RegExpFilter(CommandFilter):
         return False
 
 
+class PathFilter(CommandFilter):
+    """Command filter checking that path arguments are within given dirs
+
+        One can specify the following constraints for command arguments:
+            1) pass     - pass an argument as is to the resulting command
+            2) some_str - check if an argument is equal to the given string
+            3) abs path - check if a path argument is within the given base dir
+
+        A typical rootwrapper filter entry looks like this:
+            # cmdname: filter name, raw command, user, arg_i_constraint [, ...]
+            chown: PathFilter, /bin/chown, root, nova, /var/lib/images
+
+    """
+
+    def match(self, userargs):
+        command, arguments = userargs[0], userargs[1:]
+
+        equal_args_num = len(self.args) == len(arguments)
+        exec_is_valid = super(PathFilter, self).match(userargs)
+        args_equal_or_pass = all(
+            arg == 'pass' or arg == value
+            for arg, value in zip(self.args, arguments)
+            if not os.path.isabs(arg)  # arguments not specifying abs paths
+        )
+        paths_are_within_base_dirs = all(
+            os.path.commonprefix([arg, os.path.realpath(value)]) == arg
+            for arg, value in zip(self.args, arguments)
+            if os.path.isabs(arg)  # arguments specifying abs paths
+        )
+
+        return (equal_args_num and
+                exec_is_valid and
+                args_equal_or_pass and
+                paths_are_within_base_dirs)
+
+    def get_command(self, userargs, exec_dirs=[]):
+        command, arguments = userargs[0], userargs[1:]
+
+        # convert path values to canonical ones; copy other args as is
+        args = [os.path.realpath(value) if os.path.isabs(arg) else value
+                for arg, value in zip(self.args, arguments)]
+
+        return super(PathFilter, self).get_command([command] + args,
+                                                   exec_dirs)
+
+
 class DnsmasqFilter(CommandFilter):
     """Specific filter for the dnsmasq call (which includes env)"""
 
diff --git a/tests/unit/test_rootwrap.py b/tests/unit/test_rootwrap.py
index 8ee8e72..ea6ccbb 100644
--- a/tests/unit/test_rootwrap.py
+++ b/tests/unit/test_rootwrap.py
@@ -19,6 +19,9 @@ import logging
 import logging.handlers
 import os
 import subprocess
+import uuid
+
+import fixtures
 
 from openstack.common.rootwrap import filters
 from openstack.common.rootwrap import wrapper
@@ -198,3 +201,119 @@ class RootwrapTestCase(utils.BaseTestCase):
         raw.set('DEFAULT', 'syslog_log_level', 'INFO')
         config = wrapper.RootwrapConfig(raw)
         self.assertEqual(config.syslog_log_level, logging.INFO)
+
+
+class PathFilterTestCase(utils.BaseTestCase):
+    def setUp(self):
+        super(PathFilterTestCase, self).setUp()
+
+        tmpdir = fixtures.TempDir('/tmp')
+        self.useFixture(tmpdir)
+
+        self.f = filters.PathFilter('/bin/chown', 'root', 'nova', tmpdir.path)
+
+        gen_name = lambda: str(uuid.uuid4())
+
+        self.SIMPLE_FILE_WITHIN_DIR = os.path.join(tmpdir.path, 'some')
+        self.SIMPLE_FILE_OUTSIDE_DIR = os.path.join('/tmp', 'some')
+        self.TRAVERSAL_WITHIN_DIR = os.path.join(tmpdir.path, 'a', '..',
+                                                 'some')
+        self.TRAVERSAL_OUTSIDE_DIR = os.path.join(tmpdir.path, '..', 'some')
+
+        self.TRAVERSAL_SYMLINK_WITHIN_DIR = os.path.join(tmpdir.path,
+                                                         gen_name())
+        os.symlink(os.path.join(tmpdir.path, 'a', '..', 'a'),
+                   self.TRAVERSAL_SYMLINK_WITHIN_DIR)
+
+        self.TRAVERSAL_SYMLINK_OUTSIDE_DIR = os.path.join(tmpdir.path,
+                                                          gen_name())
+        os.symlink(os.path.join(tmpdir.path, 'a', '..', '..', '..', 'etc'),
+                   self.TRAVERSAL_SYMLINK_OUTSIDE_DIR)
+
+        self.SYMLINK_WITHIN_DIR = os.path.join(tmpdir.path, gen_name())
+        os.symlink(os.path.join(tmpdir.path, 'a'), self.SYMLINK_WITHIN_DIR)
+
+        self.SYMLINK_OUTSIDE_DIR = os.path.join(tmpdir.path, gen_name())
+        os.symlink(os.path.join('/tmp', 'some_file'), self.SYMLINK_OUTSIDE_DIR)
+
+    def test_argument_pass_constraint(self):
+        f = filters.PathFilter('/bin/chown', 'root', 'pass', 'pass')
+
+        args = ['chown', 'something', self.SIMPLE_FILE_OUTSIDE_DIR]
+        self.assertTrue(f.match(args))
+
+    def test_argument_equality_constraint(self):
+        f = filters.PathFilter('/bin/chown', 'root', 'nova', '/tmp/spam/eggs')
+
+        args = ['chown', 'nova', '/tmp/spam/eggs']
+        self.assertTrue(f.match(args))
+
+        args = ['chown', 'quantum', '/tmp/spam/eggs']
+        self.assertFalse(f.match(args))
+
+    def test_wrong_arguments_number(self):
+        args = ['chown', '-c', 'nova', self.SIMPLE_FILE_WITHIN_DIR]
+        self.assertFalse(self.f.match(args))
+
+    def test_wrong_exec_command(self):
+        args = ['wrong_exec', self.SIMPLE_FILE_WITHIN_DIR]
+        self.assertFalse(self.f.match(args))
+
+    def test_match(self):
+        args = ['chown', 'nova', self.SIMPLE_FILE_WITHIN_DIR]
+        self.assertTrue(self.f.match(args))
+
+    def test_match_traversal(self):
+        args = ['chown', 'nova', self.TRAVERSAL_WITHIN_DIR]
+        self.assertTrue(self.f.match(args))
+
+    def test_match_symlink(self):
+        args = ['chown', 'nova', self.SYMLINK_WITHIN_DIR]
+        self.assertTrue(self.f.match(args))
+
+    def test_match_traversal_symlink(self):
+        args = ['chown', 'nova', self.TRAVERSAL_SYMLINK_WITHIN_DIR]
+        self.assertTrue(self.f.match(args))
+
+    def test_reject(self):
+        args = ['chown', 'nova', self.SIMPLE_FILE_OUTSIDE_DIR]
+        self.assertFalse(self.f.match(args))
+
+    def test_reject_traversal(self):
+        args = ['chown', 'nova', self.TRAVERSAL_OUTSIDE_DIR]
+        self.assertFalse(self.f.match(args))
+
+    def test_reject_symlink(self):
+        args = ['chown', 'nova', self.SYMLINK_OUTSIDE_DIR]
+        self.assertFalse(self.f.match(args))
+
+    def test_reject_traversal_symlink(self):
+        args = ['chown', 'nova', self.TRAVERSAL_SYMLINK_OUTSIDE_DIR]
+        self.assertFalse(self.f.match(args))
+
+    def test_get_command(self):
+        args = ['chown', 'nova', self.SIMPLE_FILE_WITHIN_DIR]
+        expected = ['/bin/chown', 'nova', self.SIMPLE_FILE_WITHIN_DIR]
+
+        self.assertEqual(expected, self.f.get_command(args))
+
+    def test_get_command_traversal(self):
+        args = ['chown', 'nova', self.TRAVERSAL_WITHIN_DIR]
+        expected = ['/bin/chown', 'nova',
+                    os.path.realpath(self.TRAVERSAL_WITHIN_DIR)]
+
+        self.assertEqual(expected, self.f.get_command(args))
+
+    def test_get_command_symlink(self):
+        args = ['chown', 'nova', self.SYMLINK_WITHIN_DIR]
+        expected = ['/bin/chown', 'nova',
+                    os.path.realpath(self.SYMLINK_WITHIN_DIR)]
+
+        self.assertEqual(expected, self.f.get_command(args))
+
+    def test_get_command_traversal_symlink(self):
+        args = ['chown', 'nova', self.TRAVERSAL_SYMLINK_WITHIN_DIR]
+        expected = ['/bin/chown', 'nova',
+                    os.path.realpath(self.TRAVERSAL_SYMLINK_WITHIN_DIR)]
+
+        self.assertEqual(expected, self.f.get_command(args))