Added support for entering namespaces using absolute paths

README.rst

@@ -38,6 +38,10 @@ Example usage from Python:
         # output network interfaces as seen from within the mypid's net NS:
         subprocess.check_output(['ip', 'a'])
 
+    # or enter an arbitrary namespace:
+    with Namespace('/var/run/netns/foo', 'net'):
+        # output network interfaces as seen from within the mypid's net NS:
+        subprocess.check_output(['ip', 'a'])
 
 .. _nsenter: http://man7.org/linux/man-pages/man1/nsenter.1.html
 .. _blog post "Entering Kernel Namespaces from Python": http://tech.zalando.com/posts/entering-kernel-namespaces-with-python.html

nsenter/__init__.py

@@ -22,20 +22,28 @@ class Namespace(object):
     """A context manager for entering namespaces
 
     Args:
-        pid: The PID for the owner of the namespace to enter
+        pid: The PID for the owner of the namespace to enter, or an absolute
+             path to a file which represents a namespace handle.
+
         ns_type: The type of namespace to enter must be one of
-            mnt ipc net pid user uts
+                 mnt ipc net pid user uts.  If pid is an absolute path, this
+                 much match the type of namespace it represents
+
         proc: The path to the /proc file system.  If running in a container
-            the host proc file system may be binded mounted in a different
-            location
+              the host proc file system may be binded mounted in a different
+              location
 
     Raises:
         IOError: A non existent PID was provided
         ValueError: An improper ns_type was provided
         OSError: Unable to enter or exit the namespace
 
     Example:
-        with Namespace(<pid>, <ns_type>):
+        with Namespace(916, 'net'):
+            #do something in the namespace
+            pass
+
+        with Namespace('/var/run/netns/foo', 'net'):
             #do something in the namespace
             pass
     """
@@ -53,7 +61,13 @@ def __init__(self, pid, ns_type, proc='/proc'):
         self.ns_type = ns_type
         self.proc = proc
 
-        self.target_fd = self._nsfd(pid, ns_type).open()
+        # if it's numeric, then it's a pid, else assume it's a path
+        try:
+            pid = int(pid)
+            self.target_fd = self._nsfd(pid, ns_type).open()
+        except ValueError:
+            self.target_fd = Path(pid).open()
+
         self.target_fileno = self.target_fd.fileno()
 
         self.parent_fd = self._nsfd('self', ns_type).open()
@@ -82,7 +96,9 @@ def _close_files(self):
             self.target_fd.close()
         except:
             pass
-        self.parent_fd.close()
+
+        if self.parent_fd is not None:
+            self.parent_fd.close()
 
     def __enter__(self):
         self._log.debug('Entering %s namespace %s', self.ns_type, self.pid)

tests.py

@@ -1,69 +1,135 @@
+import ctypes
+import os
+import subprocess
+import tempfile
 import unittest
 
-import subprocess, os, errno
+from nsenter import Namespace, NAMESPACE_NAMES
 
-from nsenter import Namespace, NAMESPACE_NAMES 
 
 class TestNamespaces(unittest.TestCase):
 
+    _libc = ctypes.CDLL('libc.so.6', use_errno=True)
+
     def setUp(self):
         """Spawn a child process so we have a PID to enter"""
-        
+
         self._child = subprocess.Popen(['/bin/cat'])
 
     def tearDown(self):
         """SIGTERM the child process"""
-        
+
         self._child.terminate()
         self._child.wait()
 
-    def test_namespaces_except_user(self):
-        """Test entering all namespaces execept user
-        """
+    def test_namespace_non_exist_path(self):
+        """Test entering a non-existent path"""
+
+        def do_test():
+            fd, filename = tempfile.mkstemp()
+            os.close(fd)
+            os.remove(filename)
+
+            with Namespace(filename, 'net'):
+                pass
+
+        self.assertRaises(IOError, do_test)
+
+    def test_namespace_plain_file_path(self):
+        """Test entering a plain file path"""
+
+        fd, filename = tempfile.mkstemp()
+        os.close(fd)
+
+        def do_test():
+            with Namespace(filename, 'net'):
+                pass
+
+        self.assertRaises(OSError, do_test)
+
+        os.remove(filename)
+
+    def test_namespace_directory_path(self):
+        """Test entering a directory path"""
+
+        def do_test():
+            with Namespace('/tmp', 'net'):
+                pass
+
+        self.assertRaises(IOError, do_test)
+
+    @unittest.skipIf(os.geteuid() != 0, "Must be root to bind mount")
+    def test_namespace_good_path(self):
+        """Test entering an arbirtrary namespace"""
+
+        try:
+            # get the path to it's network namespace
+            ns_path = os.path.join('/proc', str(self._child.pid), 'ns', 'net')
+
+            # bind mount it to a temp location
+            fd, filename = tempfile.mkstemp()
+            os.close(fd)
+
+            assert self._libc.mount(ns_path.encode('ascii'), filename.encode('ascii'), 0, 4096, 0) == 0
+
+            # enter the bind mount
+            with Namespace(filename, 'net'):
+                pass
+
+        finally:
+            # ensure we clean up the bind
+            self._libc.umount(filename.encode('ascii'))
+            os.remove(filename)
+
+    @unittest.skipIf(os.geteuid() != 0, "Must be root to setns()")
+    def test_namespaces_as_root(self):
+        """Test entering all namespaces the pid has as root"""
+
+        for name in filter(lambda x: x != 'user', NAMESPACE_NAMES):
+            if os.path.exists(os.path.join('/proc', str(self._child.pid), 'ns', name)):
+                with Namespace(self._child.pid, name):
+                    pass
+
+    @unittest.skipIf(os.geteuid() == 0, "Must not be root to trigger OSError")
+    def test_namespaces_except_user_as_normal(self):
+        """Test entering all namespaces execept user as non-root"""
 
-        #Can't use the assertRaises context manager in python2.6
         def do_test():
             for name in filter(lambda x: x != 'user', NAMESPACE_NAMES):
                 with Namespace(self._child.pid, name):
                     pass
-
-            #if we aren't root (technically: CAP_SYS_ADMIN)
-            #then we'll get OSError (EPERM) for all our tests
-            if os.geteuid() != 0:
-               self.assertRaises(OSError, do_test)
-            else:
-                do_test()
 
+        self.assertRaises(OSError, do_test)
+
+    @unittest.skipIf(os.geteuid() != 0, "Must be root to setns()")
     def test_user_namespace(self):
         """Test entering a non-existent namespace"""
-        
+
         def do_test():
             with Namespace(self._child.pid, 'user'):
                 pass
 
-        #This process doesn't have a user namespace
-        #So this will OSError(EINVAL)
-        self.assertRaises(OSError, do_test)
+        # this will railse a IOError on python2 and OSError on python 3
+        # as the file for this namespace does not exist!
+        self.assertRaises((IOError, OSError), do_test)
 
     def test_bad_namespace(self):
         """Test entering a bad namespace type"""
-        
+
         def do_test():
             with Namespace(self._child.pid, 'foo'):
                 pass
         self.assertRaises(ValueError, do_test)
 
     def test_bad_pid(self):
-        """Test entering bad pid's name space"""
-        
+        """Test entering bad pid's namespace"""
+
         def do_test():
             with Namespace('foo', 'net'):
                 pass
 
         self.assertRaises(IOError, do_test)
 
 
-
 if __name__ == '__main__':
     unittest.main()
-