"""Native OS boundary tests; assertions inspect actual Windows tokens/access.""" import base64 import json import os from pathlib import Path import socket import subprocess import sys import tempfile import unittest @unittest.skipUnless(sys.platform == 'win32', 'Windows AppContainer native integration') class WindowsAppContainerTests(unittest.TestCase): @classmethod def setUpClass(cls): from windows_appcontainer import AppContainer cls.source = Path(__file__).resolve().parent cls.runtime = Path(sys.executable).resolve().parent cls.sandbox = AppContainer([cls.source, cls.runtime], provision=True) def test_read_only_startup_never_mutates_permissions(self): from unittest.mock import patch from windows_appcontainer import AppContainer with patch('windows_appcontainer.subprocess.run', side_effect=AssertionError('Startup must not invoke icacls')): sandbox = AppContainer([self.source, self.runtime]) code, output = sandbox.run(sys.executable, ['-c', 'print("read-only-startup")']) self.assertEqual(code, 0) self.assertEqual(output.strip(), b'read-only-startup') def test_unprovisioned_runtime_fails_closed(self): from unittest.mock import patch from windows_appcontainer import AppContainer, SandboxError with tempfile.TemporaryDirectory(prefix='safetynet-unprovisioned-') as folder: with patch('windows_appcontainer.subprocess.run', side_effect=AssertionError('No automatic grants')): with self.assertRaisesRegex(SandboxError, 'provisioning is required'): AppContainer([folder]) def test_real_token_filesystem_network_and_secret_boundary(self): source = '''import ctypes,json,os,socket,sys from ctypes import wintypes as W k=ctypes.WinDLL('kernel32');a=ctypes.WinDLL('advapi32') k.GetCurrentProcess.restype=W.HANDLE a.OpenProcessToken.argtypes=[W.HANDLE,W.DWORD,ctypes.POINTER(W.HANDLE)] a.GetTokenInformation.argtypes=[W.HANDLE,ctypes.c_int,ctypes.c_void_p,W.DWORD,ctypes.POINTER(W.DWORD)] t=W.HANDLE();a.OpenProcessToken(k.GetCurrentProcess(),8,ctypes.byref(t));v=W.DWORD();n=W.DWORD();a.GetTokenInformation(t,29,ctypes.byref(v),4,ctypes.byref(n)) r={'appContainer':v.value==1,'secretAbsent':'INSPECTOR_TOKEN' not in os.environ,'echo':sys.stdin.buffer.read().decode()} for key,path,mode in [('privateDenied',sys.argv[1],'rb'),('runtimeWriteDenied',sys.argv[2],'wb')]: try:open(path,mode).close();r[key]=False except PermissionError:r[key]=True for key,kind in [('tcpDenied',socket.SOCK_STREAM),('udpDenied',socket.SOCK_DGRAM)]: try: s=socket.socket(type=kind);s.settimeout(2) if kind==socket.SOCK_STREAM:s.connect(('127.0.0.1',int(sys.argv[3]))) else:s.sendto(b'probe',('127.0.0.1',int(sys.argv[4]))) r[key]=False except OSError:r[key]=True finally:s.close() print(json.dumps(r)) ''' listener = socket.socket();listener.bind(('127.0.0.1',0));listener.listen() udp = socket.socket(type=socket.SOCK_DGRAM);udp.bind(('127.0.0.1',0));udp.settimeout(0.2) with tempfile.TemporaryDirectory(prefix='safetynet-private-boundary-') as folder: private = Path(folder)/'private.txt';private.write_text('synthetic private fixture') target = self.runtime/'appcontainer-unauthorized-write.txt' try: code, output = self.sandbox.run(sys.executable,['-c',source,str(private),str(target),str(listener.getsockname()[1]),str(udp.getsockname()[1])], b'pipe-only fixture') self.assertEqual(code,0,output.decode(errors='replace')) result=json.loads(output) for key in ['appContainer','secretAbsent','privateDenied','runtimeWriteDenied','tcpDenied']:self.assertTrue(result[key],key) self.assertEqual(result['echo'],'pipe-only fixture') self.assertFalse(target.exists()) # Windows firewall can silently discard UDP, so verify the live # receiver instead of trusting sendto's immediate return value. with self.assertRaises(socket.timeout): udp.recvfrom(1024) finally:listener.close();udp.close() def test_child_process_inherits_sandbox(self): child = "import ctypes,json;from ctypes import wintypes as W;k=ctypes.WinDLL('kernel32');a=ctypes.WinDLL('advapi32');k.GetCurrentProcess.restype=W.HANDLE;a.OpenProcessToken.argtypes=[W.HANDLE,W.DWORD,ctypes.POINTER(W.HANDLE)];a.GetTokenInformation.argtypes=[W.HANDLE,ctypes.c_int,ctypes.c_void_p,W.DWORD,ctypes.POINTER(W.DWORD)];t=W.HANDLE();a.OpenProcessToken(k.GetCurrentProcess(),8,ctypes.byref(t));v=W.DWORD();n=W.DWORD();a.GetTokenInformation(t,29,ctypes.byref(v),4,ctypes.byref(n));print(v.value)" parent='import subprocess,sys; result=subprocess.run([sys.executable,"-c",'+repr(child)+'],capture_output=True,text=True,timeout=5);print(result.stdout.strip());sys.exit(result.returncode)' code, output=self.sandbox.run(sys.executable,['-c',parent]) self.assertEqual(code,0,output.decode());self.assertEqual(output.strip(),b'1') def test_worker_cannot_open_broker_memory(self): # Test access rights without reading any actual parent memory or secrets. source='''import ctypes,json,sys from ctypes import wintypes as W k=ctypes.WinDLL('kernel32',use_last_error=True) k.OpenProcess.argtypes=[W.DWORD,W.BOOL,W.DWORD];k.OpenProcess.restype=W.HANDLE k.CloseHandle.argtypes=[W.HANDLE] handle=k.OpenProcess(0x0010,False,int(sys.argv[1])) result={'brokerMemoryReadDenied':not bool(handle),'error':ctypes.get_last_error()} if handle:k.CloseHandle(handle) print(json.dumps(result)) ''' code, output=self.sandbox.run(sys.executable,['-c',source,str(os.getpid())]) self.assertEqual(code,0,output.decode()) result=json.loads(output) self.assertTrue(result['brokerMemoryReadDenied']) self.assertEqual(result['error'],5) def test_real_parser_crosses_only_pipe(self): request={'route':'/inspect','data':base64.b64encode(b'Public text document').decode(),'policy':{'security':{},'dlp':{'rules':[], 'maxBytes':20971520}}} code,output=self.sandbox.run(sys.executable,[str(self.source/'sandbox_worker.py')],json.dumps(request).encode()) self.assertEqual(code,0,output.decode(errors='replace')) result=json.loads(output);self.assertEqual(result['decision'],'allow');self.assertTrue(result['coverage']['complete']) def test_real_yara_upload_and_download_inside_sandbox(self): rule={'id':'test-marker','source':'rule Block_Test { strings: $a = "SAFETYNET_BLOCK_TEST" condition: $a }'} policy={'security':{'yara':{'enabled':True,'rules':[rule],'maxScanBytes':1048576}},'dlp':{'maxBytes':1048576,'rules':[]}} for route,data,decision in [('/inspect',b'Public information','allow'),('/inspect',b'SAFETYNET_BLOCK_TEST','block'),('/scan-download',bytes([0,1,2,3,255]),'allow'),('/scan-download',b'SAFETYNET_BLOCK_TEST','block')]: request={'route':route,'data':base64.b64encode(data).decode(),'policy':policy} code,output=self.sandbox.run(sys.executable,[str(self.source/'sandbox_worker.py')],json.dumps(request).encode()) self.assertEqual(code,0,output.decode(errors='replace'));result=json.loads(output) self.assertEqual(result['decision'],decision,(route,result)) if route=='/inspect' and decision=='block':self.assertTrue(result['malware']['coverage']['complete']) else:self.assertTrue(result['coverage']['complete']) if decision=='block':self.assertEqual(result['ruleId'],'test-marker') def test_orphan_worker_is_killed_when_request_ends(self): import ctypes from ctypes import wintypes as W source='import subprocess,sys;child=subprocess.Popen([sys.executable,"-c","import time;time.sleep(60)"]);print(child.pid,flush=True)' code,output=self.sandbox.run(sys.executable,['-c',source]) self.assertEqual(code,0,output.decode());pid=int(output.strip()) kernel=ctypes.WinDLL('kernel32',use_last_error=True) kernel.OpenProcess.argtypes=[W.DWORD,W.BOOL,W.DWORD];kernel.OpenProcess.restype=W.HANDLE kernel.WaitForSingleObject.argtypes=[W.HANDLE,W.DWORD];kernel.CloseHandle.argtypes=[W.HANDLE] handle=kernel.OpenProcess(0x00100000,False,pid) if handle: try:self.assertEqual(kernel.WaitForSingleObject(handle,3000),0) finally:kernel.CloseHandle(handle) def test_large_binary_pipe_complete(self): import hashlib data=bytes(range(256))*65536 code,output=self.sandbox.run(sys.executable,['-c','import hashlib,sys;print(hashlib.sha256(sys.stdin.buffer.read()).hexdigest())'],data) self.assertEqual(code,0);self.assertEqual(output.strip().decode(),hashlib.sha256(data).hexdigest()) def test_worker_rejects_unsandboxed_execution(self): result=subprocess.run([sys.executable,str(self.source/'sandbox_worker.py')],input=b'{}',capture_output=True,timeout=5) self.assertNotEqual(result.returncode,0);self.assertEqual(result.stdout,b'');self.assertEqual(result.stderr,b'') def test_broker_dispatch_remains_authenticated_and_isolated(self): import queue import secrets import threading import urllib.error import urllib.request token=secrets.token_urlsafe(32) env={key:value for key,value in os.environ.items() if key.upper() in {'SYSTEMROOT','WINDIR','USERPROFILE','APPDATA','LOCALAPPDATA','TEMP','TMP'}} env.update({'INSPECTOR_TOKEN':token,'INSPECTOR_PORT':'0'}) process=subprocess.Popen([sys.executable,str(self.source/'server.py')],stdout=subprocess.PIPE,stderr=subprocess.PIPE,env=env,creationflags=subprocess.CREATE_NO_WINDOW) lines=queue.Queue();threading.Thread(target=lambda:lines.put(process.stdout.readline()),daemon=True).start() try: ready=json.loads(lines.get(timeout=40)) self.assertEqual(ready['containment']['parserIsolation']['kind'],'windows-appcontainer-v1') request={'data':base64.b64encode(b'Public allowed document').decode(),'policy':{'security':{},'dlp':{'maxBytes':20971520,'rules':[]}}} url='http://127.0.0.1:'+str(ready['port'])+'/inspect' body=json.dumps(request).encode() with self.assertRaises(urllib.error.HTTPError) as denied:urllib.request.urlopen(urllib.request.Request(url,data=body),timeout=5) self.assertEqual(denied.exception.code,401) with urllib.request.urlopen(urllib.request.Request(url,data=body,headers={'Authorization':'Bearer '+token}),timeout=30) as response:result=json.load(response) self.assertEqual(result['decision'],'allow');self.assertTrue(result['coverage']['complete']) finally: process.kill();process.wait(timeout=5) process.stdout.close();process.stderr.close() def test_deadline_and_response_bound_fail_closed(self): from windows_appcontainer import SandboxError with self.assertRaises(SandboxError):self.sandbox.run(sys.executable,['-c','import time;time.sleep(30)'],timeout=0.2) with self.assertRaises(SandboxError):self.sandbox.run(sys.executable,['-c','print("x"*10000)'],max_output=100) def test_real_compressed_office_macro_inside_sandbox(self): from test_office_macros import fixture, office, policy for source, expected in [('rule source_marker { strings: $m = "Hello from Python!" condition: $m }', 'block'), ('rule no_match { condition: false }', 'allow')]: request={'route':'/scan-download','data':base64.b64encode(office(fixture())).decode(),'policy':policy(source)} code, output=self.sandbox.run(sys.executable,[str(self.source/'sandbox_worker.py')],json.dumps(request).encode()) self.assertEqual(code,0,output.decode(errors='replace')) result=json.loads(output);self.assertEqual(result['decision'],expected) self.assertEqual(result['officeMacros']['modules'],5) self.assertNotIn('Hello from Python!',output.decode()) if __name__=='__main__':unittest.main()