safe_test
def safe_test(
t:str, fname:str
)->bool:Whether t is one plain assert that uses only fname, literals and the safe builtins.
check
Check a learner’s function with small tests. Validate test text, run it with time limits, and share results without test source or expected values.
Safe tests
The model writes the tests, so the notebook does not trust them. safe_test accepts only one assert, a statement that checks a condition. The assert can use the learner’s function, literals (fixed values such as 2), comparisons, arithmetic and a short list of builtins (Python’s built-in functions). It cannot use an import, a private attribute (a name that starts with _), open, exec or eval. validate_exercise checks the shape of the model’s reply. Tests run with only the safe builtins. This check limits the test text; it does not isolate the learner’s function from the rest of Python.
validate_exercise
def validate_exercise(
r
)->fastcore.basics.AttrDict:Check the model’s exercise and raise ValueError with the reason if invalid.
safe_test refuses an import, a name with two leading underscores, a call to anything but the function and the safe builtins, and code that is not one assert.
assert safe_test("assert rrf([['a'], ['b']], k=0) == ['a', 'b']", 'rrf')
assert safe_test("assert all(x in 'ab' for x in rrf([['a', 'b']]))", 'rrf')
for bad in ("import os", "assert __import__('os')", "assert rrf.__globals__", "assert open('/etc/passwd')", "assert exec('1')",
"rrf([])", "while True: pass", "assert rrf(1); import os", "assert (lambda: 1)()"):
assert not safe_test(bad, 'rrf'), bad
safe_test('assert rrf([]) == []', 'rrf')Truevalidate_exercise refuses an empty or a text tests, a missing name, and an unsafe test, with a reason. A test object is a dictionary with test text, an optional say label, and an optional hidden flag.
good = dict(task='t', fname='f', stub='def f(): ...', tests=['assert f() == 1'])
test_eq(validate_exercise(good).fname, 'f')
for bad, why in ((dict(good, tests=[]), 'list of 1 to 12'), (dict(good, tests='assert f()'), 'list of 1 to 12'),
({k: v for k, v in good.items() if k != 'fname'}, 'fname'), (dict(good, tests=['import os']), 'not one plain assert'), ([], 'JSON object')):
test_fail(lambda: validate_exercise(bad), contains=why)
validate_exercise(good){ 'fname': 'f',
'solution': '',
'stub': 'def f(): ...',
'task': 't',
'tests': ['assert f() == 1']}Named tests use the same safety check as text tests. Missing labels and hidden flags default to an empty label and False.
named = dict(good, tests=[dict(test='assert f() == 1', say='One', hidden=True)])
test_eq(validate_exercise(named).tests, named['tests'])
validate_exercise(named).tests[{'test': 'assert f() == 1', 'say': 'One', 'hidden': True}]Time limits
time_limit raises TimeoutError when the time is up. On the main thread it uses SIGALRM, an operating-system alarm. On another thread it starts a watchdog, a timer that asks Python to raise the error in the test thread. The watchdog cannot interrupt a call blocked in C code. It takes effect when that call returns to Python. The timer is always cancelled when the context ends.
time_limit
def time_limit(
secs:float=2.0
):Raise TimeoutError after secs; worker threads cannot interrupt blocked C calls.
A test that runs too long fails, and Python continues. The loop below has its own one-second limit, so even a broken timeout cannot hang this example.
def bounded_loop():
deadline, steps = time.monotonic() + 1., 0
while time.monotonic() < deadline:
steps += 1
started = time.monotonic()
with expect_fail(TimeoutError):
with time_limit(.2): bounded_loop()
elapsed = time.monotonic() - started
assert elapsed < 1.
round(elapsed, 2)0.21Notebook test cells
load_tests reads a notebook path or an in-memory notebook. A code cell with #| setup supplies a fixture. A cell with #| test: Label supplies a check; a bare #| test gets a numbered label. The abhyasa metadata can also mark tests, labels, and hidden checks. Comment directives take priority.
These cells are trusted author code, with full Python builtins. Do not pass model output to this runner.
load_tests
def load_tests(
nb
)->tuple:Return ordered setup sources and labelled test cells from a notebook.
run_cells runs setup once in a copy of the supplied namespace. Each check gets a fresh shallow copy of that base: new names do not pass between tests, but shared mutable objects are not copied. Setup and tests share one total time budget. Setup exceptions mark all checks with SetupError; an interrupt stops the run.
A failing top-level assert call(...) == expected records the actual value without calling the function twice. Hidden failures show only the exception class, not its message or the test source.
run_cells
def run_cells(
tests, ns, *, setup:tuple=(), secs:float=2.0, total:float=10.0
)->__main__.Report:Run trusted test cells in separate shallow copies of one setup namespace.
Checks in a user kernel
snippet builds standalone Python from the actual runner functions with inspect.getsource. It embeds test data as JSON, needs no installed abhyasa or fastcore, and prints one JSON result line. The line also has the nonce that you give, a random text that shows which line is the real result. Its definitions live in a private namespace, so no harness names remain in the user kernel. The kernel’s existing names, including __builtins__, stay unchanged.
snippet
def snippet(
kind, fname, tests, *, secs:NoneType=None, total:NoneType=None, setup:tuple=(), nonce:str=None
)->str:Build a standalone kernel check that prints one public JSON result with nonce. The time limits default to the settings runner.per_test and runner.total.
parse finds the last JSON result line in kernel stdout and rejects text that is not a public check result. With a nonce, it skips each line that does not have that nonce. Thus a stray result line, or a result line that the code copied, does not count. But the learner’s code runs in the same Python process as the checks. Thus the result is correct only for code that does not try to cheat.
run_isolated runs learner code and the same runner in a new Python process, in its own temporary folder. These limits apply:
- The process stops after
total + 2seconds, even when the code never finishes. The process starts in a new process group. At the time limit, all the processes of that group stop, also a process that the code started. - The first lines of the script set
total + 1seconds of CPU time and files of at mostrunner.file_mbMB (10 by default). A process that the system stops at the CPU limit gives a timeout. A process that stops on a different signal gives no result. - With
sandbox, an audit hook refuses the network, new processes, signals to other processes, frame walks, and file changes outside the temporary folder. On macOS,sandbox-execalso appliesSANDBOX_PROFILEto the process. - The report goes to a private pipe, not to stdout. Stdout is only the output of the learner code. The nonce comes in on a second pipe, so it is not in the source of the script.
This is a closed process with no network. It is not a security sandbox: code that uses ctypes can go around an audit hook.
parse
def parse(
text, nonce:str=None
)->dict:Read the last public check JSON line, or return a missing-result error. With nonce, only a line with that nonce counts.
The audit hook of the closed process. _guard runs only in the new process, before the learner code. It refuses each event that opens the network, starts a process or sends a signal to a process (os.kill, os.killpg and signal.pthread_kill). It also refuses a write, a delete or a move of a file outside the folder tmp. The runner stops the process group from the parent, so it does not need these calls.
The hook also refuses the calls that find a frame or an object of the runner: sys._getframe, sys._current_frames, sys.settrace, sys.setprofile, the gc object lists, and the tb_frame, gi_frame, cr_frame and ag_frame attributes. Thus learner code that uses inspect.stack or traceback gets a PermissionError. _guard first sets logging._srcfile to None, so logging.info and the like still work: logging then does not look for the frame of its caller.
On macOS, sandbox-exec applies this profile to the process. The profile refuses the network, new processes, signals to other processes, and file writes outside the temporary folder TMP. A signal to the process itself is allowed. Python can read the files that it needs to start.
_limits gives the first lines of the script. They set the CPU and file-size limits in the new process. On a system with no resource module, the process runs without these limits.
A from __future__ import must be the first statement of a script. _split_future cuts it off the code, so that the limits and the audit hook come after it.
It reads the statements from the syntax tree, not the lines. Code on the same line as the docstring or a future import, after a ;, would run before the guard. So _split_future refuses it with a ValueError.
_start runs in the new process before the learner code. It keeps the real exec, json.dumps and os.write, a copy of the builtins, and private copies of ast and contextlib. It reads the nonce from its pipe and closes that pipe. Then it defines the runner and starts a worker thread that waits for a lock. The nonce and the real functions are only in the closure of that thread. The learner code runs in the main thread, so a signal handler cannot get a frame of the worker. When the learner code ends, the last lines of the script open the lock. The worker runs the checks and writes the report to the result pipe.
If the learner code changes json, print, ast, contextlib or builtins, the runner does not see the change. If the learner code opens the lock early, the checks run early. They are still the real checks.
The head keeps only the docstring and the future imports. The body starts on the next line, so a decorator stays with its function:
src = '"doc"\nfrom __future__ import (\n annotations)\n@staticmethod\ndef f(): pass\n'
test_eq(_split_future(src), ('"doc"\nfrom __future__ import (\n annotations)', '@staticmethod\ndef f(): pass\n'))
test_eq(_split_future('x = 1\n'), ('', 'x = 1\n'))Code after a ; on the line of a future import is refused:
test_fail(lambda: _split_future('from __future__ import annotations; import socket; S = socket.socket()'), contains='own line')The parent keeps only a small part of what the process writes. _drain reads a pipe to its end in chunks and keeps at most cap bytes. It notes when there was more. Thus a process that prints in a fast loop cannot fill the memory of the app.
A pipe with 5000 bytes gives 1000 bytes to a reader with a cap of 1000, and the reader notes that there was more.
r, w = os.pipe()
got = _drain(r, 1000)
os.write(w, b'a' * 5000); os.close(w)
got.thread.join(2); os.close(r)
assert bytes(got.data) == b'a' * 1000 and got.over_feed does not raise when the child closed its input, or when the file is closed already.
r, w = os.pipe(); os.close(r)
f = os.fdopen(w, 'wb'); f.close()
_feed(f, b'print(1)')run_isolated puts it all together. Each run gets a new nonce and a new temporary folder. The parent reads the report from the result pipe. If the pipe has no report with the nonce, or a report larger than RESULT_MAX, the result is an error, never a pass. The parent keeps at most OUTPUT_KEEP bytes of stdout and of stderr.
On macOS, the sandbox-exec profile stops the network, signals to other processes, and all writes outside the temporary folder. It is the barrier. Without it (for example on Linux), the audit hook is a best-effort guard. Code that reads memory through ctypes cannot be fully stopped. On other systems, run the process in your own container.
run_isolated
def run_isolated(
kind, fname, source, tests, *, total:NoneType=None, setup:tuple=(), sandbox:bool=True, python:NoneType=None,
unix:bool=False
)->dict:Run source and its checks in a closed process with no network, using python (default sys.executable); with unix, it can use a local socket pair. total defaults to the setting runner.total, and each test gets runner.per_test seconds.
Small examples
A fixture and three checks keep their notebook order. Comments supply the first label; metadata makes the second check hidden; a bare directive gets a numbered label.
example_nb = new_nb([
mk_cell('#| setup\nnumber = 3'),
mk_cell('#| test: Double three\nassert double(number) == 6'),
mk_cell('assert double(number) == 918273', metadata={'abhyasa': {'test': True, 'say': 'Private', 'hidden': True}}),
mk_cell('#| test\nassert number == 3'),
])
example_setup, example_tests = load_tests(example_nb)
assert len(example_setup) == 1
assert [t['say'] for t in example_tests] == ['Double three', 'Private', 'Test 3']
assert [t['hidden'] for t in example_tests] == [False, True, False]
[t['say'] for t in example_tests]['Double three', 'Private', 'Test 3']The fixture is available in every check. The hidden failure shares its status and actual value, but not the private expected value.
cell_report = run_cells(example_tests, {'double': double}, setup=example_setup)
assert [c.status for c in cell_report.checks] == ['pass', 'fail', 'pass']
assert cell_report.checks[1].got == '6'
assert '918273' not in json.dumps(public(cell_report))
[(c.label, c.status, c.error) for c in cell_report.checks][('Double three', 'pass', ''),
('Private', 'fail', 'AssertionError'),
('Test 3', 'pass', '')]The same cell checks run in a kernel-like dictionary. The snippet prints one JSON line and leaves that dictionary unchanged. parse also gives a clear result when stdout contains no check report. In the closed process, stdout is only learner output, and the report goes to a pipe.
kernel_ns = {'double': double, '__builtins__': builtins.__dict__}
kernel_before = dict(kernel_ns)
kernel_stdout = io.StringIO()
with redirect_stdout(kernel_stdout):
exec(snippet('cell', 'double', example_tests, setup=example_setup), kernel_ns)
kernel_result = parse(kernel_stdout.getvalue())
assert kernel_ns == kernel_before
assert len(kernel_stdout.getvalue().splitlines()) == 1
assert [c['status'] for c in kernel_result['checks']] == ['pass', 'fail', 'pass']
assert parse('not JSON') == {'error': 'no check result'}
[(c['label'], c['status']) for c in kernel_result['checks']][('Double three', 'pass'), ('Private', 'fail'), ('Test 3', 'pass')]With a nonce, parse reads only the line that has it. A line without the nonce does not count, even when it looks like a result.
fake_line = json.dumps(dict(v=1, found=True, timeout=False, checks=[]))
assert parse(fake_line, 'n1') == {'error': 'no check result'}
assert parse(json.dumps(dict(nonce='n1', v=1, found=True, timeout=False, checks=[])) + '\n' + fake_line, 'n1')['found']A correct function passes in the closed process. Code that prints a false result line and stops does not pass.
assert run_isolated('assert', 'double', 'def double(n): return n * 2', ['assert double(2) == 4'])['checks'][0]['status'] == 'pass'
fake = f'print({fake_line!r})\nimport os; os._exit(0)'
assert run_isolated('assert', 'double', fake, ['assert double(2) == 4']) == {'error': 'no check result'}Learner code cannot make a false pass. Here the code changes json.dumps so that each check says “pass”, and the function is wrong. The runner keeps the real json.dumps, so the check fails.
patch = 'import json\n_d = json.dumps\njson.dumps = lambda o, **k: _d({**o, "checks": [{**c, "status": "pass"} for c in o["checks"]]} if isinstance(o, dict) and "checks" in o else o, **k)\n'
assert run_isolated('assert', 'double', patch + 'def double(n): return n', ['assert double(2) == 4'])['checks'][0]['status'] == 'fail'A frame walk does not find the nonce. The code tries to read its own frame, and the hook refuses it. The result has the error, not a pass.
walk = 'import sys\ndef double(n):\n f = sys._getframe()\n return n * 2'
r = run_isolated('assert', 'double', walk, ['assert double(2) == 4'])
assert r['checks'][0]['status'] == 'error' and 'PermissionError' in r['checks'][0]['error'], rPlain logging works in learner code. The code sets up logging and writes a warning, and the check passes. sys._getframe stays refused.
log = 'import logging\nlogging.basicConfig(level=logging.INFO)\ndef double(n):\n logging.warning("n is %s", n)\n logging.info("ok")\n return n * 2'
r = run_isolated('assert', 'double', log, ['assert double(2) == 4'])
assert r['checks'][0]['status'] == 'pass', r
r = run_isolated('assert', 'double', walk, ['assert double(2) == 4'])
assert r['checks'][0]['status'] == 'error' and 'PermissionError' in r['checks'][0]['error'], rThe python option runs the given interpreter. The test code reports its own version, and the version must match.
ver = run_isolated('assert', 'v', 'import sys\ndef v(): return sys.version', [f'assert v() == {sys.version!r}'], python=sys.executable)
assert ver['checks'][0]['status'] == 'pass'
assert run_isolated('assert', 'v', 'import sys\ndef v(): return sys.version', [f'assert v() == {sys.version!r}'])['checks'][0]['status'] == 'pass'By default, a local socket pair is refused. With unix=True, the code can send bytes across a pair.
pair = 'import socket\ndef ping():\n a, b = socket.socketpair()\n a.sendall(b"hi")\n return b.recv(2)'
assert run_isolated('assert', 'ping', pair, ['assert ping() == b"hi"'])['checks'][0]['error'].startswith('PermissionError')
r = run_isolated('assert', 'ping', pair, ['assert ping() == b"hi"'], unix=True, total=7.)
assert r['checks'][0]['status'] == 'pass'With unix=True, a connection, a bind, a name lookup and a send to an internet address are still refused.
net = '''import socket
def tcp():
s = socket.socket(); s.connect(("1.1.1.1", 80))
def udp():
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.sendto(b"x", ("1.1.1.1", 9))
def bind():
s = socket.socket(socket.AF_INET); s.bind(("127.0.0.1", 0))
def lookup(): socket.getaddrinfo("example.com", 80)
def ubind():
s = socket.socket(socket.AF_UNIX); s.bind("x.sock")
'''
for f in ('tcp', 'udp', 'bind', 'lookup', 'ubind'):
r = run_isolated('assert', f, net, [f'assert {f}() is None'], unix=True)
assert r['checks'][0]['status'] == 'error' and 'PermissionError' in r['checks'][0]['error'], f"{f}: {r}"Without sandbox-exec, the hook alone must refuse ctypes. The test sets the path of sandbox-exec to a missing file.
real, SANDBOX_EXEC = SANDBOX_EXEC, '/missing/sandbox-exec'
try:
bypass = 'import _ctypes, ctypes\nh = ctypes.pythonapi._handle\nfn = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int)(_ctypes.dlsym(h, "socket"))\nfn(2, 1, 0)'
for t in ('import ctypes; ctypes.CDLL(None)', 'import ctypes; ctypes.pythonapi.socket', 'import ctypes; ctypes.cdll.LoadLibrary("libc.dylib")', bypass):
for u in (False, True):
r = run_isolated('assert', 'f', 'def f():\n' + ''.join(f' {l}\n' for l in t.split('\n')), ['assert f() is None'], unix=u)
assert r['checks'][0]['status'] == 'error' and 'PermissionError' in r['checks'][0]['error'], (t, u, r)
assert run_isolated('assert', 'double', 'def double(n): return n * 2', ['assert double(2) == 4'])['checks'][0]['status'] == 'pass'
finally: SANDBOX_EXEC = realThe timer still stops a slow test after the guard refuses ctypes. The test must finish in less than the total time.
import time
t0 = time.monotonic()
r = run_isolated('assert', 'f', 'def f():\n while True: pass', ['assert f() is None'], total=6.)
assert time.monotonic() - t0 < 9
assert r.get('timeout') or 'Timeout' in r['checks'][0]['error'], rThe timer of a worker thread uses PyThreadState_SetAsyncExc. The test runs time_limit in a thread of a closed process with the guard, and the timer must stop a slow loop.
import tempfile
with tempfile.TemporaryDirectory() as d:
d = os.path.realpath(d)
script = '\n'.join(['import ctypes, signal, threading, time', 'from contextlib import contextmanager', f'TEST_SECONDS = {TEST_SECONDS!r}',
inspect.getsource(time_limit), inspect.getsource(_guard), f'_guard({d!r})',
'out = []',
'def work():',
' try:',
' with time_limit(.5):',
' while True: pass',
' except TimeoutError as e: out.append("stopped")',
't = threading.Thread(target=work); t.start(); t.join(5)',
'print(out, t.is_alive())'])
p = subprocess.run([sys.executable, '-I', '-B', '-c', script], capture_output=True, text=True, timeout=20, cwd=d)
assert p.stdout.strip() == "['stopped'] False", (p.stdout, p.stderr)A python that is not a Python, or a missing file, gives the error and no result. A path with a space works.
import tempfile
src = 'def double(n): return n * 2'
for bad in ('/usr/bin/true', '/missing/python'):
assert run_isolated('assert', 'double', src, ['assert double(2) == 4'], python=bad) == {'error': 'no check result'}, bad
with tempfile.TemporaryDirectory(prefix='a b ') as d:
link = os.path.join(d, 'py thon'); os.symlink(sys.executable, link)
assert run_isolated('assert', 'double', src, ['assert double(2) == 4'], python=link)['checks'][0]['status'] == 'pass'On macOS, the sandbox-exec profile alone (no audit hook) refuses an internet connect and a unix bind, and it allows a socket pair.
if sys.platform == 'darwin' and os.path.exists(SANDBOX_EXEC):
def prof(code): return subprocess.run([SANDBOX_EXEC, '-p', SANDBOX_PROFILE, '-D', 'TMP=/nonexistent', sys.executable, '-I', '-B', '-c', code], capture_output=True, text=True, timeout=20)
assert 'PermissionError' in prof('import socket; socket.socket().connect(("1.1.1.1", 80))').stderr
assert 'PermissionError' in prof('import socket; socket.socket(socket.AF_UNIX).bind("/tmp/abhyasa-x.sock")').stderr
assert prof('import socket; a, b = socket.socketpair(); a.send(b"x"); assert b.recv(1) == b"x"').returncode == 0Hostile code cannot stop another process. The hook refuses the call, and the result says PermissionError. The target here is a spare process, not this notebook.
spare = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'], start_new_session=True)
try:
for call in (f'os.kill({spare.pid}, 9)', f'os.killpg({spare.pid}, 9)'):
r = run_isolated('assert', 'f', f'import os\ndef f():\n {call}', ['assert f() is None'])
assert 'PermissionError' in r['checks'][0]['error'], r
time.sleep(.2)
assert spare.poll() is None
finally: spare.kill(); spare.wait()