| import json |
| import os |
| import platform |
| import resource |
| import secrets |
| import signal |
| import struct |
| import subprocess |
| import tempfile |
| from pathlib import Path |
| from common import ROOT |
|
|
|
|
| def seccomp_program(): |
| if platform.machine() != 'x86_64': |
| raise RuntimeError('Sandbox syscall profile requires x86_64') |
| instructions = [(0x20,0,0,4),(0x15,1,0,0xc000003e),(0x06,0,0,0x80000000), |
| (0x20,0,0,0),(0x45,0,1,0x40000000),(0x06,0,0,0x00050001)] |
| for number in [41,42,43,44,45,46,47,49,50,53,56,57,58,101,165,166,272,288,298,304,308,310,311,321,323,428,429,430,431,432,435,442]: |
| instructions += [(0x15,0,1,number),(0x06,0,0,0x00050001)] |
| instructions.append((0x06,0,0,0x7fff0000)) |
| return b''.join(struct.pack('HBBI',*instruction) for instruction in instructions) |
|
|
|
|
| def limits(): |
| resource.setrlimit(resource.RLIMIT_CPU,(2,2)) |
| resource.setrlimit(resource.RLIMIT_AS,(384*1024**2,384*1024**2)) |
| resource.setrlimit(resource.RLIMIT_FSIZE,(65536,65536)) |
| resource.setrlimit(resource.RLIMIT_NOFILE,(64,64)) |
| resource.setrlimit(resource.RLIMIT_CORE,(0,0)) |
|
|
|
|
| def check_code(code,entry_point,test_code): |
| base = ROOT/'tmp/coding_sandbox' |
| base.mkdir(parents=True,exist_ok=True) |
| nonce = secrets.token_hex(16) |
| with tempfile.TemporaryDirectory(prefix='case-',dir=base) as directory: |
| work = Path(directory) |
| (work/'payload.json').write_text(json.dumps({'code':code,'entry_point':entry_point,'test_code':test_code})) |
| (work/'worker.py').write_bytes((ROOT/'src/code_worker.py').read_bytes()) |
| (work/'filter.bpf').write_bytes(seccomp_program()) |
| with (work/'filter.bpf').open('rb') as filter_file, (work/'stdout.log').open('wb') as stdout, (work/'stderr.log').open('wb') as stderr: |
| command = ['bwrap','--unshare-all','--die-with-parent','--new-session','--cap-drop','ALL', |
| '--ro-bind','/usr','/usr','--symlink','usr/lib','/lib','--symlink','usr/lib64','/lib64', |
| '--proc','/proc','--dev','/dev','--size','16777216','--tmpfs','/tmp', |
| '--ro-bind',str(work),'/work','--chdir','/tmp','--clearenv','--setenv','PATH','/usr/bin', |
| '--ro-bind',str(ROOT/'tools/python/cpython-3.12.13-linux-x86_64-gnu'),'/runtime', |
| '--setenv','HOME','/tmp','--setenv','PYTHONHASHSEED','0','--seccomp',str(filter_file.fileno()), |
| '/runtime/bin/python3.12','-I','-S','/work/worker.py',nonce] |
| try: |
| process = subprocess.run(command,stdout=stdout,stderr=stderr,timeout=4,pass_fds=(filter_file.fileno(),)) |
| except subprocess.TimeoutExpired: |
| return {'passed':False,'error_type':'Timeout','error':'wall_timeout_4s'} |
| content = (work/'stdout.log').read_bytes()[:65536].decode(errors='replace') |
| marker = '__WSP_RESULT__'+nonce |
| matches = [line[len(marker):] for line in content.split('\n') if line.startswith(marker)] |
| if process.returncode == 0 and len(matches)==1: |
| return json.loads(matches[0]) |
| error = (work/'stderr.log').read_bytes()[:1000].decode(errors='replace') |
| if 'bwrap:' in error: |
| raise RuntimeError('Sandbox infrastructure failure: '+error[:500]) |
| return {'passed':False,'error_type':'ProcessFailure','returncode':process.returncode,'error':error[:300]} |
|
|
