extend the tests to include tests for python

Author: Anga205

.vscode/c_cpp_properties.json

@@ -0,0 +1,16 @@
+{
+    "configurations": [
+        {
+            "name": "Linux",
+            "includePath": [
+                "${workspaceFolder}/**"
+            ],
+            "defines": [],
+            "compilerPath": "/usr/bin/gcc",
+            "cStandard": "c17",
+            "cppStandard": "gnu++17",
+            "intelliSenseMode": "linux-gcc-x64"
+        }
+    ],
+    "version": 4
+}

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "C_Cpp.errorSquiggles": "disabled"
+}

integration_test.go

@@ -104,6 +104,38 @@ func TestContainerizationAPISecurityIntegration(t *testing.T) {
 	}
 }
 
+func TestContainerizationAPISecurityIntegrationPython3(t *testing.T) {
+	if config.Config.Globals.ENABLE_QUEUE {
+		t.Fatal("ENABLE_QUEUE must be false for integration tests")
+	}
+
+	h := integrationHarness{baseURL: testServer.URL}
+	h.apiPort = mustExtractPort(t, h.baseURL)
+	if !python3SupportAvailable(t, h.baseURL) {
+		t.Skip("python3 execution is not supported by the API yet")
+	}
+
+	cases := []struct {
+		name string
+		run  func(*testing.T, integrationHarness)
+	}{
+		{name: "file privacy across request IDs", run: testFilesystemIsolationPython3},
+		{name: "disk spammer is terminated and data is reclaimed", run: testDiskCleanupPython3},
+		{name: "fork bomb does not poison subsequent requests", run: testForkBombContainmentPython3},
+		{name: "network namespace blocks localhost bridge", run: testNetworkIsolationPython3},
+		{name: "memory hard limit triggers oom kill", run: testMemoryHardLimitPython3},
+		{name: "io flood is bounded and returns before timeout", run: testIOFloodResiliencePython3},
+		{name: "signal trap cannot survive forced timeout", run: testSignalTrapTimeoutPython3},
+		{name: "orphan grandchild is reaped after request exits", run: testOrphanReapingPython3},
+		{name: "inode bomb does not poison host temp filesystem", run: testInodeExhaustionPython3},
+		{name: "privileged reboot syscall is denied", run: testPrivilegedSyscallDeniedPython3},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) { tc.run(t, h) })
+	}
+}
+
 func testFilesystemIsolation(t *testing.T, h integrationHarness) {
 	writeCode := mustLoadSampleCode(t, "file_privacy_write_read.c", nil)
 	writeResp := callSimpleExecute(t, h.baseURL, buildCRequest(writeCode, 4, 32768))
@@ -266,6 +298,155 @@ func testPrivilegedSyscallDenied(t *testing.T, h integrationHarness) {
 	}
 }
 
+func testFilesystemIsolationPython3(t *testing.T, h integrationHarness) {
+	writeCode := mustLoadSampleCode(t, "file_privacy_write_read.py", nil)
+	writeResp := callSimpleExecute(t, h.baseURL, buildPython3Request(writeCode, 4, 32768))
+	if !strings.Contains(writeResp.Output, "SecretData123") {
+		t.Fatalf("step A did not return secret in stdout; stdout=%q stderr=%q", writeResp.Output, writeResp.Error)
+	}
+
+	readCode := mustLoadSampleCode(t, "file_privacy_read_only.py", nil)
+	readResp := callSimpleExecute(t, h.baseURL, buildPython3Request(readCode, 4, 32768))
+	if !strings.Contains(strings.ToLower(readResp.Error), "no such file or directory") {
+		t.Fatalf("step B unexpectedly accessed file from another sandbox; stdout=%q stderr=%q", readResp.Output, readResp.Error)
+	}
+}
+
+func testDiskCleanupPython3(t *testing.T, h integrationHarness) {
+	beforeFree := mustFreeBytes(t, os.TempDir())
+	beforeCount := countSandboxTempDirs(t)
+
+	code := mustLoadSampleCode(t, "disk_spammer.py", nil)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 2, 32768))
+	stderrLower := strings.ToLower(resp.Error)
+	if !containsAny(stderrLower, []string{"execution timed out", "memory limit exceeded"}) {
+		t.Fatalf("disk spammer was not terminated as expected; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+
+	time.Sleep(400 * time.Millisecond)
+	assertDiskReclaimed(t, beforeFree, mustFreeBytes(t, os.TempDir()))
+	assertNoSandboxLeak(t, beforeCount, countSandboxTempDirs(t))
+}
+
+func testForkBombContainmentPython3(t *testing.T, h integrationHarness) {
+	bombCode := mustLoadSampleCode(t, "fork_bomb.py", nil)
+	_ = callSimpleExecute(t, h.baseURL, buildPython3Request(bombCode, 2, 32768))
+
+	helloCode := mustLoadSampleCode(t, "hello_world.py", nil)
+	helloResp := callSimpleExecute(t, h.baseURL, buildPython3Request(helloCode, 4, 32768))
+	if !strings.Contains(helloResp.Output, "Hello World") {
+		t.Fatalf("follow-up request failed after fork bomb; stdout=%q stderr=%q", helloResp.Output, helloResp.Error)
+	}
+	if strings.Contains(strings.ToLower(helloResp.Error), "resource temporarily unavailable") {
+		t.Fatalf("follow-up request indicates host PID exhaustion; stderr=%q", helloResp.Error)
+	}
+}
+
+func testNetworkIsolationPython3(t *testing.T, h integrationHarness) {
+	replacements := map[string]string{"__API_PORT__": strconv.Itoa(h.apiPort)}
+	code := mustLoadSampleCode(t, "network_localhost_bridge.py", replacements)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 3, 32768))
+
+	combined := strings.ToLower(resp.Output + "\n" + resp.Error)
+	if strings.Contains(combined, "connected") {
+		t.Fatalf("localhost bridge unexpectedly succeeded; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+	if !containsAny(combined, []string{"connection refused", "network is unreachable", "no route to host"}) {
+		t.Fatalf("network isolation did not produce expected connect error; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+}
+
+func testMemoryHardLimitPython3(t *testing.T, h integrationHarness) {
+	code := mustLoadSampleCode(t, "memory_bomb.py", nil)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 3, 16384))
+	if !containsAny(strings.ToLower(resp.Error), []string{"memory limit exceeded", "killed", "execution timed out"}) {
+		t.Fatalf("memory bomb did not terminate as expected; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+
+	execDur, err := time.ParseDuration(resp.CPUTime)
+	if err != nil {
+		t.Fatalf("failed to parse cpu_time %q: %v", resp.CPUTime, err)
+	}
+	if execDur > 500*time.Millisecond {
+		t.Fatalf("memory enforcement was too slow: cpu_time=%s stderr=%q", resp.CPUTime, resp.Error)
+	}
+}
+
+func testIOFloodResiliencePython3(t *testing.T, h integrationHarness) {
+	code := mustLoadSampleCode(t, "io_spam.py", nil)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 1, 32768))
+
+	stderrLower := strings.ToLower(resp.Error)
+	if !containsAny(stderrLower, []string{"execution timed out", "killed", "memory limit exceeded"}) {
+		t.Fatalf("io flood did not terminate with expected error; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+
+	if strings.TrimSpace(resp.Error) == "" {
+		t.Fatalf("io flood returned empty stderr; expected bounded but non-empty error output")
+	}
+
+	const maxErrorBytes = (1 << 20) + 4096
+	if len(resp.Error) > maxErrorBytes {
+		t.Fatalf("stderr exceeded resilience cap: got=%d bytes cap=%d", len(resp.Error), maxErrorBytes)
+	}
+
+	assertDurationNotExcessive(t, resp.CPUTime, 3*time.Second, "io flood request")
+}
+
+func testSignalTrapTimeoutPython3(t *testing.T, h integrationHarness) {
+	code := mustLoadSampleCode(t, "signal_trap.py", nil)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 1, 32768))
+
+	if !strings.Contains(strings.ToLower(resp.Error), "execution timed out") {
+		t.Fatalf("signal trap did not timeout as expected; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+
+	assertDurationWindow(t, resp.CPUTime, 900*time.Millisecond, 2500*time.Millisecond, "signal trap timeout")
+}
+
+func testOrphanReapingPython3(t *testing.T, h integrationHarness) {
+	code := mustLoadSampleCode(t, "orphan_maker.py", nil)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 2, 32768))
+	assertDurationNotExcessive(t, resp.CPUTime, 3*time.Second, "orphan maker request")
+
+	time.Sleep(400 * time.Millisecond)
+	if cnt := countProcessesByComm(t, "orphanpygc"); cnt > 0 {
+		t.Fatalf("orphan grandchild leaked to host after request completion: count=%d stderr=%q", cnt, resp.Error)
+	}
+}
+
+func testInodeExhaustionPython3(t *testing.T, h integrationHarness) {
+	beforeFree := mustFreeBytes(t, os.TempDir())
+	beforeCount := countSandboxTempDirs(t)
+
+	code := mustLoadSampleCode(t, "inode_bomb.py", nil)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 4, 32768))
+
+	combinedLower := strings.ToLower(resp.Output + "\n" + resp.Error)
+	if !containsAny(combinedLower, []string{"inode bomb completed", "no space left", "disk quota exceeded"}) {
+		t.Fatalf("inode exhaustion test produced unexpected result; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+
+	time.Sleep(400 * time.Millisecond)
+	assertDiskReclaimed(t, beforeFree, mustFreeBytes(t, os.TempDir()))
+	assertNoSandboxLeak(t, beforeCount, countSandboxTempDirs(t))
+	mustCreateAndDeleteTempFile(t)
+}
+
+func testPrivilegedSyscallDeniedPython3(t *testing.T, h integrationHarness) {
+	code := mustLoadSampleCode(t, "try_reboot.py", nil)
+	resp := callSimpleExecute(t, h.baseURL, buildPython3Request(code, 3, 32768))
+	assertDurationNotExcessive(t, resp.CPUTime, 3*time.Second, "privileged reboot syscall")
+
+	combinedLower := strings.ToLower(resp.Output + "\n" + resp.Error)
+	if strings.Contains(combinedLower, "reboot succeeded unexpectedly") {
+		t.Fatalf("privileged reboot syscall unexpectedly succeeded; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+	if !containsAny(combinedLower, []string{"operation not permitted", "permission denied", "bad system call", "killed", "hangup"}) {
+		t.Fatalf("expected privileged syscall denial signal was not observed; stdout=%q stderr=%q", resp.Output, resp.Error)
+	}
+}
+
 func mustLoadSampleCode(t *testing.T, fileName string, replacements map[string]string) string {
 	t.Helper()
 	path := filepath.Join(sampleCodeDir, fileName)
@@ -285,6 +466,43 @@ func buildCRequest(code string, timeoutSec, maxMemoryKB uint) simpleExecuteReque
 	return simpleExecuteRequest{Language: "c", Code: code, Timeout: timeoutSec, MaxMemory: maxMemoryKB}
 }
 
+func buildPython3Request(code string, timeoutSec, maxMemoryKB uint) simpleExecuteRequest {
+	return simpleExecuteRequest{Language: "python3", Code: code, Timeout: timeoutSec, MaxMemory: maxMemoryKB}
+}
+
+func python3SupportAvailable(t *testing.T, baseURL string) bool {
+	t.Helper()
+	req := buildPython3Request("print('py-ok')", 1, 32768)
+	body, err := json.Marshal(req)
+	if err != nil {
+		return false
+	}
+
+	httpReq, err := http.NewRequest(http.MethodPost, baseURL+"/simple-execute", bytes.NewReader(body))
+	if err != nil {
+		return false
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+
+	httpResp, err := httpClient.Do(httpReq)
+	if err != nil {
+		return false
+	}
+	defer httpResp.Body.Close()
+
+	rawBody, err := io.ReadAll(httpResp.Body)
+	if err != nil || httpResp.StatusCode != http.StatusOK {
+		return false
+	}
+
+	var parsed simpleExecuteResponse
+	if err := json.Unmarshal(rawBody, &parsed); err != nil {
+		return false
+	}
+
+	return strings.Contains(parsed.Output, "py-ok")
+}
+
 func callSimpleExecute(t *testing.T, baseURL string, req simpleExecuteRequest) simpleExecuteResponse {
 	t.Helper()
 	body, err := json.Marshal(req)

sample_code_for_tests/disk_spammer.py

@@ -0,0 +1,6 @@
+buf = b'A' * (1024 * 1024)
+
+with open('/disk_spam.bin', 'wb', buffering=0) as f:
+    while True:
+        f.write(buf)
+        f.flush()

sample_code_for_tests/file_privacy_read_only.py

@@ -0,0 +1,8 @@
+import sys
+
+try:
+    with open('/root/test.txt', 'r', encoding='utf-8') as f:
+        print(f.read(), end='')
+except FileNotFoundError as exc:
+    print(exc, file=sys.stderr)
+    raise SystemExit(1)

sample_code_for_tests/file_privacy_write_read.py

@@ -0,0 +1,8 @@
+import os
+
+os.makedirs('/root', exist_ok=True)
+with open('/root/test.txt', 'w', encoding='utf-8') as f:
+    f.write('SecretData123')
+
+with open('/root/test.txt', 'r', encoding='utf-8') as f:
+    print(f.read(), end='')

sample_code_for_tests/fork_bomb.py

@@ -0,0 +1,14 @@
+import os
+import signal
+import sys
+
+while True:
+    try:
+        pid = os.fork()
+    except OSError as exc:
+        print(exc, file=sys.stderr)
+        raise SystemExit(0)
+
+    if pid == 0:
+        while True:
+            signal.pause()

sample_code_for_tests/hello_world.py

@@ -0,0 +1 @@
+print('Hello World')

sample_code_for_tests/inode_bomb.py

@@ -0,0 +1,7 @@
+import sys
+
+for i in range(10000):
+    with open(f'file_{i}.txt', 'w', encoding='utf-8'):
+        pass
+
+print('inode bomb completed', file=sys.stderr)

sample_code_for_tests/io_spam.py

@@ -0,0 +1,8 @@
+import sys
+import time
+
+line = 'IO_FLOOD:' + ('A' * 4086) + '\n'
+while True:
+    sys.stderr.write(line)
+    sys.stderr.flush()
+    time.sleep(0.001)