Tests: added privileged credential tests.

test/test_go_isolation.py

@@ -19,16 +19,21 @@ class TestGoIsolation(TestApplicationGo):
 
         return unit if not complete_check else unit.complete()
 
+    def unpriv_creds(self):
+        nobody_uid = pwd.getpwnam('nobody').pw_uid
+
+        try:
+            nogroup_gid = grp.getgrnam('nogroup').gr_gid
+            nogroup = 'nogroup'
+        except:
+            nogroup_gid = grp.getgrnam('nobody').gr_gid
+            nogroup = 'nobody'
+
+        return (nobody_uid, nogroup_gid, nogroup)
+
     def isolation_key(self, key):
         return key in self.available['features']['isolation'].keys()
 
-    def conf_isolation(self, isolation):
-        self.assertIn(
-            'success',
-            self.conf(isolation, 'applications/ns_inspect/isolation'),
-            'configure isolation',
-        )
-
     def test_isolation_values(self):
         self.load('ns_inspect')
 
@@ -40,56 +45,155 @@ class TestGoIsolation(TestApplicationGo):
                     obj['NS'][ns.upper()], ns_value, '%s match' % ns
                 )
 
-    def test_isolation_user(self):
+    def test_isolation_unpriv_user(self):
         if not self.isolation_key('unprivileged_userns_clone'):
             print('unprivileged clone is not available')
             raise unittest.SkipTest()
 
-        self.load('ns_inspect')
-
-        user_id = pwd.getpwnam('nobody').pw_uid
-
-        try:
-            group_id = grp.getgrnam('nogroup').gr_gid
-        except:
-            group_id = grp.getgrnam('nobody').gr_gid
-
-        obj = self.getjson()['body']
-
-        self.assertTrue(obj['UID'] != 0, 'uid not zero')
-        self.assertTrue(obj['GID'] != 0, 'gid not zero')
-
         if self.is_su:
-            self.assertEqual(obj['UID'], user_id, 'uid match')
+            print('privileged tests, skip this')
-            self.assertEqual(obj['GID'], group_id, 'gid match')
+            raise unittest.SkipTest()
-        else:
-            self.assertEqual(obj['UID'], self.uid, 'uid match')
-            self.assertEqual(obj['GID'], self.gid, 'gid match')
 
-        self.conf_isolation({"namespaces": {"credential": True}})
+        self.load('ns_inspect')
+        obj = self.getjson()['body']
+
+        self.assertEqual(obj['UID'], self.uid, 'uid match')
+        self.assertEqual(obj['GID'], self.gid, 'gid match')
+
+        self.load('ns_inspect', isolation={'namespaces': {'credential': True}})
 
         obj = self.getjson()['body']
 
-        # default uid and gid maps current user to nobody
+        nobody_uid, nogroup_gid, nogroup = self.unpriv_creds()
-        self.assertEqual(obj['UID'], user_id, 'uid nobody')
-        self.assertEqual(obj['GID'], group_id, 'gid nobody')
 
-        self.conf_isolation(
+        # unprivileged unit map itself to nobody in the container by default
-            {
+        self.assertEqual(obj['UID'], nobody_uid, 'uid of nobody')
-                "namespaces": {"credential": True},
+        self.assertEqual(obj['GID'], nogroup_gid, 'gid of %s' % nogroup)
-                "uidmap": [
+
-                    {"container": user_id, "host": self.uid, "size": 1}
+        self.load(
-                ],
+            'ns_inspect',
-                "gidmap": [
+            user='root',
-                    {"container": group_id, "host": self.gid, "size": 1}
+            isolation={'namespaces': {'credential': True}},
-                ],
-            }
         )
 
         obj = self.getjson()['body']
 
-        self.assertEqual(obj['UID'], user_id, 'uid match')
+        self.assertEqual(obj['UID'], 0, 'uid match user=root')
-        self.assertEqual(obj['GID'], group_id, 'gid match')
+        self.assertEqual(obj['GID'], 0, 'gid match user=root')
+
+        self.load(
+            'ns_inspect',
+            user='root',
+            group=nogroup,
+            isolation={'namespaces': {'credential': True}},
+        )
+
+        obj = self.getjson()['body']
+
+        self.assertEqual(obj['UID'], 0, 'uid match user=root group=nogroup')
+        self.assertEqual(
+            obj['GID'], nogroup_gid, 'gid match user=root group=nogroup'
+        )
+
+        self.load(
+            'ns_inspect',
+            user='root',
+            group='root',
+            isolation={
+                'namespaces': {'credential': True},
+                'uidmap': [{'container': 0, 'host': self.uid, 'size': 1}],
+                'gidmap': [{'container': 0, 'host': self.gid, 'size': 1}],
+            },
+        )
+
+        obj = self.getjson()['body']
+
+        self.assertEqual(obj['UID'], 0, 'uid match uidmap')
+        self.assertEqual(obj['GID'], 0, 'gid match gidmap')
+
+    def test_isolation_priv_user(self):
+        if not self.is_su:
+            print('unprivileged tests, skip this')
+            raise unittest.SkipTest()
+
+        self.load('ns_inspect')
+
+        nobody_uid, nogroup_gid, nogroup = self.unpriv_creds()
+
+        obj = self.getjson()['body']
+
+        self.assertEqual(obj['UID'], nobody_uid, 'uid match')
+        self.assertEqual(obj['GID'], nogroup_gid, 'gid match')
+
+        self.load('ns_inspect', isolation={'namespaces': {'credential': True}})
+
+        obj = self.getjson()['body']
+
+        # privileged unit map app creds in the container by default
+        self.assertEqual(obj['UID'], nobody_uid, 'uid nobody')
+        self.assertEqual(obj['GID'], nogroup_gid, 'gid nobody')
+
+        self.load(
+            'ns_inspect',
+            user='root',
+            isolation={'namespaces': {'credential': True}},
+        )
+
+        obj = self.getjson()['body']
+
+        self.assertEqual(obj['UID'], 0, 'uid nobody user=root')
+        self.assertEqual(obj['GID'], 0, 'gid nobody user=root')
+
+        self.load(
+            'ns_inspect',
+            user='root',
+            group=nogroup,
+            isolation={'namespaces': {'credential': True}},
+        )
+
+        obj = self.getjson()['body']
+
+        self.assertEqual(obj['UID'], 0, 'uid match user=root group=nogroup')
+        self.assertEqual(
+            obj['GID'], nogroup_gid, 'gid match user=root group=nogroup'
+        )
+
+        self.load(
+            'ns_inspect',
+            user='root',
+            group='root',
+            isolation={
+                'namespaces': {'credential': True},
+                'uidmap': [{'container': 0, 'host': 0, 'size': 1}],
+                'gidmap': [{'container': 0, 'host': 0, 'size': 1}],
+            },
+        )
+
+        obj = self.getjson()['body']
+
+        self.assertEqual(obj['UID'], 0, 'uid match uidmap user=root')
+        self.assertEqual(obj['GID'], 0, 'gid match gidmap user=root')
+
+        # map 65535 uids
+        self.load(
+            'ns_inspect',
+            user='nobody',
+            isolation={
+                'namespaces': {'credential': True},
+                'uidmap': [
+                    {'container': 0, 'host': 0, 'size': nobody_uid + 1}
+                ],
+            },
+        )
+
+        obj = self.getjson()['body']
+
+        self.assertEqual(
+            obj['UID'], nobody_uid, 'uid match uidmap user=nobody'
+        )
+        self.assertEqual(
+            obj['GID'], nogroup_gid, 'gid match uidmap user=nobody'
+        )
 
     def test_isolation_mnt(self):
         if not self.isolation_key('mnt'):
@@ -100,9 +204,9 @@ class TestGoIsolation(TestApplicationGo):
             print('unprivileged clone is not available')
             raise unittest.SkipTest()
 
-        self.load('ns_inspect')
+        self.load(
-        self.conf_isolation(
+            'ns_inspect',
-            {"namespaces": {"mount": True, "credential": True}}
+            isolation={'namespaces': {'mount': True, 'credential': True}},
         )
 
         obj = self.getjson()['body']
@@ -132,12 +236,14 @@ class TestGoIsolation(TestApplicationGo):
             print('pid namespace is not supported')
             raise unittest.SkipTest()
 
-        if not self.isolation_key('unprivileged_userns_clone'):
+        if not (self.is_su or self.isolation_key('unprivileged_userns_clone')):
-            print('unprivileged clone is not available')
+            print('requires root or unprivileged_userns_clone')
             raise unittest.SkipTest()
 
-        self.load('ns_inspect')
+        self.load(
-        self.conf_isolation({"namespaces": {"pid": True, "credential": True}})
+            'ns_inspect',
+            isolation={'namespaces': {'pid': True, 'credential': True}},
+        )
 
         obj = self.getjson()['body']
 
@@ -163,7 +269,7 @@ class TestGoIsolation(TestApplicationGo):
             else:
                 namespaces[ns] = False
 
-        self.conf_isolation({"namespaces": namespaces})
+        self.load('ns_inspect', isolation={'namespaces': namespaces})
 
         obj = self.getjson()['body']