Setting the file. One moment.
Document Limits · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary
helpers/ _document_limits.py
Python · 95 lines · 4 KB
13 if sys.platform == "linux" :
14 import resource
15
16 resource.setrlimit(resource. RLIMIT_AS , ( MEMORY_BYTES , MEMORY_BYTES ))
17 resource.setrlimit(resource. RLIMIT_CPU , ( CPU_SECONDS , CPU_SECONDS ))
18 resource.setrlimit(resource. RLIMIT_CORE , ( 0 , 0 ))
19 resource.setrlimit(resource. RLIMIT_FSIZE , ( 0 , 0 ))
20 return None
21 if sys.platform != "win32" :
22 raise OSError ( "unsupported limit platform" )
23
24 import ctypes
25 from ctypes import wintypes as w
26
27 class Basic ( ctypes . Structure ):
28 _fields_ = [
29 ( "process_time" , ctypes.c_longlong), ( "job_time" , ctypes.c_longlong),
30 ( "flags" , w. DWORD ), ( "min_ws" , ctypes.c_size_t),
31 ( "max_ws" , ctypes.c_size_t), ( "active" , w. DWORD ),
32 ( "affinity" , ctypes.c_size_t), ( "priority" , w. DWORD ),
33 ( "scheduling" , w. DWORD ),
34 ]
35
36 class IO ( ctypes . Structure ):
37 _fields_ = [(name, ctypes.c_ulonglong) for name in
38 ( "read_ops" , "write_ops" , "other_ops" , "read_bytes" ,
39 "write_bytes" , "other_bytes" )]
40
41 class Extended ( ctypes . Structure ):
42 _fields_ = [
43 ( "basic" , Basic), ( "io" , IO ), ( "process_memory" , ctypes.c_size_t),
44 ( "job_memory" , ctypes.c_size_t), ( "peak_process" , ctypes.c_size_t),
45 ( "peak_job" , ctypes.c_size_t),
46 ]
47
48 kernel = ctypes.WinDLL( "kernel32" , use_last_error = True )
49 kernel.CreateJobObjectW.argtypes = [ctypes.c_void_p, w. LPCWSTR ]
50 kernel.CreateJobObjectW.restype = w. HANDLE
51 kernel.SetInformationJobObject.argtypes = [w. HANDLE , ctypes.c_int,
52 ctypes.c_void_p, w. DWORD ]
53 kernel.SetInformationJobObject.restype = w. BOOL
54 kernel.GetCurrentProcess.restype = w. HANDLE
55 kernel.AssignProcessToJobObject.argtypes = [w. HANDLE , w. HANDLE ]
56 kernel.AssignProcessToJobObject.restype = w. BOOL
57 kernel.CloseHandle.argtypes = [w. HANDLE ]
58 kernel.CloseHandle.restype = w. BOOL
59 job = kernel.CreateJobObjectW( None , None )
60 if not job:
61 raise ctypes.WinError(ctypes.get_last_error())
62 limits = Extended()
63 # PROCESS_TIME | ACTIVE_PROCESS | PROCESS_MEMORY | KILL_ON_JOB_CLOSE.
64 limits.basic.flags = 0x 2 | 0x 8 | 0x 100 | 0x 2000
65 limits.basic.process_time = CPU_SECONDS * 10_000_000
66 limits.basic.active = 1
67 limits.process_memory = MEMORY_BYTES
68 if not kernel.SetInformationJobObject(job, 9 , ctypes.byref(limits),
69 ctypes.sizeof(limits)):
70 error = ctypes.get_last_error()
71 kernel.CloseHandle(job)
72 raise ctypes.WinError(error)
73 if not kernel.AssignProcessToJobObject(job, kernel.GetCurrentProcess()):
74 error = ctypes.get_last_error()
75 kernel.CloseHandle(job)
76 raise ctypes.WinError(error)
77 # Keep the handle until process exit: closing it terminates this worker.
78 return job
79
80
81 def deny_side_effects (event, args):
82 """Defense in depth, not a sandbox for arbitrary Python/native code."""
83 if event.startswith( "socket." ) or event in {
84 "subprocess.Popen" , "os.system" , "os.posix_spawn" , "os.fork" ,
85 "os.remove" , "os.unlink" , "os.rename" , "os.replace" , "os.mkdir" , "os.rmdir" ,
86 "os.truncate" , "os.chmod" , "os.chown" , "os.lchown" , "os.utime" ,
87 "os.link" , "os.symlink" , "os.chflags" , "os.setxattr" , "os.removexattr" ,
88 }:
89 raise PermissionError ( "Document side effect denied" )
90 if event == "open" :
91 _, mode, flags = args
92 if (mode and any (char in mode for char in "wax+" )) or (
93 flags & (os. O_WRONLY | os. O_RDWR | os. O_CREAT | os. O_TRUNC )
94 ):
95 raise PermissionError ( "Document write denied" )