diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/go/ns_inspect/app.go | 79 | ||||
| -rw-r--r-- | test/test_go_isolation.py | 135 | ||||
| -rw-r--r-- | test/unit/feature/isolation.py | 87 |
3 files changed, 301 insertions, 0 deletions
diff --git a/test/go/ns_inspect/app.go b/test/go/ns_inspect/app.go new file mode 100644 index 00000000..ebecbb00 --- /dev/null +++ b/test/go/ns_inspect/app.go @@ -0,0 +1,79 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http" + "nginx/unit" + "os" + "strconv" +) + +type ( + NS struct { + USER uint64 + PID uint64 + IPC uint64 + CGROUP uint64 + UTS uint64 + MNT uint64 + NET uint64 + } + + Output struct { + PID int + UID int + GID int + NS NS + } +) + +func abortonerr(err error) { + if err != nil { + panic(err) + } +} + +// returns: [nstype]:[4026531835] +func getns(nstype string) uint64 { + str, err := os.Readlink(fmt.Sprintf("/proc/self/ns/%s", nstype)) + if err != nil { + return 0 + } + + str = str[len(nstype)+2:] + str = str[:len(str)-1] + val, err := strconv.ParseUint(str, 10, 64) + abortonerr(err) + return val +} + +func handler(w http.ResponseWriter, r *http.Request) { + pid := os.Getpid() + out := &Output{ + PID: pid, + UID: os.Getuid(), + GID: os.Getgid(), + NS: NS{ + PID: getns("pid"), + USER: getns("user"), + MNT: getns("mnt"), + IPC: getns("ipc"), + UTS: getns("uts"), + NET: getns("net"), + CGROUP: getns("cgroup"), + }, + } + data, err := json.Marshal(out) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + + w.Write(data) +} + +func main() { + http.HandleFunc("/", handler) + unit.ListenAndServe(":7080", nil) +} diff --git a/test/test_go_isolation.py b/test/test_go_isolation.py new file mode 100644 index 00000000..780c2b03 --- /dev/null +++ b/test/test_go_isolation.py @@ -0,0 +1,135 @@ +import os +import json +import unittest +from unit.applications.lang.go import TestApplicationGo +from unit.feature.isolation import TestFeatureIsolation + + +class TestGoIsolation(TestApplicationGo): + prerequisites = {'modules': ['go'], 'features': ['isolation']} + + isolation = TestFeatureIsolation() + + @classmethod + def setUpClass(cls, complete_check=True): + unit = super().setUpClass(complete_check=False) + + TestFeatureIsolation().check(cls.available, unit.testdir) + + return unit if not complete_check else unit.complete() + + 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') + + obj = self.isolation.parsejson(self.get()['body']) + + for ns, ns_value in self.available['features']['isolation'].items(): + if ns.upper() in obj['NS']: + self.assertEqual( + obj['NS'][ns.upper()], ns_value, '%s match' % ns + ) + + def test_isolation_user(self): + if not self.isolation_key('unprivileged_userns_clone'): + print('unprivileged clone is not available') + raise unittest.SkipTest() + + self.load('ns_inspect') + obj = self.isolation.parsejson(self.get()['body']) + + self.assertTrue(obj['UID'] != 0, 'uid not zero') + self.assertTrue(obj['GID'] != 0, 'gid not zero') + self.assertEqual(obj['UID'], os.getuid(), 'uid match') + self.assertEqual(obj['GID'], os.getgid(), 'gid match') + + self.conf_isolation({"namespaces": {"credential": True}}) + + obj = self.isolation.parsejson(self.get()['body']) + + # default uid and gid maps current user to nobody + self.assertEqual(obj['UID'], 65534, 'uid nobody') + self.assertEqual(obj['GID'], 65534, 'gid nobody') + + self.conf_isolation( + { + "namespaces": {"credential": True}, + "uidmap": [ + {"container": 1000, "host": os.geteuid(), "size": 1} + ], + "gidmap": [ + {"container": 1000, "host": os.getegid(), "size": 1} + ], + } + ) + + obj = self.isolation.parsejson(self.get()['body']) + + # default uid and gid maps current user to root + self.assertEqual(obj['UID'], 1000, 'uid root') + self.assertEqual(obj['GID'], 1000, 'gid root') + + def test_isolation_mnt(self): + if not self.isolation_key('mnt'): + print('mnt namespace is not supported') + raise unittest.SkipTest() + + if not self.isolation_key('unprivileged_userns_clone'): + print('unprivileged clone is not available') + raise unittest.SkipTest() + + self.load('ns_inspect') + self.conf_isolation( + {"namespaces": {"mount": True, "credential": True}} + ) + + obj = self.isolation.parsejson(self.get()['body']) + + # all but user and mnt + allns = list(self.available['features']['isolation'].keys()) + allns.remove('user') + allns.remove('mnt') + + for ns in allns: + if ns.upper() in obj['NS']: + self.assertEqual( + obj['NS'][ns.upper()], + self.available['features']['isolation'][ns], + '%s match' % ns, + ) + + self.assertNotEqual( + obj['NS']['MNT'], self.isolation.getns('mnt'), 'mnt set' + ) + self.assertNotEqual( + obj['NS']['USER'], self.isolation.getns('user'), 'user set' + ) + + def test_isolation_pid(self): + if not self.isolation_key('pid'): + print('pid namespace is not supported') + raise unittest.SkipTest() + + if not self.isolation_key('unprivileged_userns_clone'): + print('unprivileged clone is not available') + raise unittest.SkipTest() + + self.load('ns_inspect') + self.conf_isolation({"namespaces": {"pid": True, "credential": True}}) + + obj = self.isolation.parsejson(self.get()['body']) + + self.assertEqual(obj['PID'], 1, 'pid of container is 1') + + +if __name__ == '__main__': + TestGoIsolation.main() diff --git a/test/unit/feature/isolation.py b/test/unit/feature/isolation.py new file mode 100644 index 00000000..9b06ab3c --- /dev/null +++ b/test/unit/feature/isolation.py @@ -0,0 +1,87 @@ +import os +import json +from unit.applications.proto import TestApplicationProto +from unit.applications.lang.go import TestApplicationGo +from unit.applications.lang.java import TestApplicationJava +from unit.applications.lang.node import TestApplicationNode +from unit.applications.lang.perl import TestApplicationPerl +from unit.applications.lang.php import TestApplicationPHP +from unit.applications.lang.python import TestApplicationPython +from unit.applications.lang.ruby import TestApplicationRuby + + +class TestFeatureIsolation(TestApplicationProto): + allns = ['pid', 'mnt', 'ipc', 'uts', 'cgroup', 'net'] + + def check(self, available, testdir): + test_conf = {"namespaces": {"credential": True}} + + module = '' + app = 'empty' + if 'go' in available['modules']: + module = TestApplicationGo() + + elif 'java' in available['modules']: + module = TestApplicationJava() + + elif 'node' in available['modules']: + module = TestApplicationNode() + app = 'basic' + + elif 'perl' in available['modules']: + module = TestApplicationPerl() + app = 'body_empty' + + elif 'php' in available['modules']: + module = TestApplicationPHP() + app = 'phpinfo' + + elif 'python' in available['modules']: + module = TestApplicationPython() + + elif 'ruby' in available['modules']: + module = TestApplicationRuby() + + if not module: + return + + module.testdir = testdir + module.load(app) + + resp = module.conf(test_conf, 'applications/' + app + '/isolation') + if 'success' not in resp: + return + + userns = self.getns('user') + if not userns: + return + + available['features']['isolation'] = {'user': userns} + + unp_clone_path = '/proc/sys/kernel/unprivileged_userns_clone' + if os.path.exists(unp_clone_path): + with open(unp_clone_path, 'r') as f: + if str(f.read()).rstrip() == '1': + available['features']['isolation'][ + 'unprivileged_userns_clone' + ] = True + + for ns in self.allns: + ns_value = self.getns(ns) + if ns_value: + available['features']['isolation'][ns] = ns_value + + def getns(self, nstype): + # read namespace id from symlink file: + # it points to: '<nstype>:[<ns id>]' + # # eg.: 'pid:[4026531836]' + nspath = '/proc/self/ns/' + nstype + data = None + + if os.path.exists(nspath): + data = int(os.readlink(nspath)[len(nstype) + 2 : -1]) + + return data + + def parsejson(self, data): + return json.loads(data.split('\n')[1]) |