# check
## Safe tests
The model writes the tests, so the notebook does not trust them.
[`safe_test`](https://vedicreader.github.io/abhyasa/check.html#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`](https://vedicreader.github.io/abhyasa/check.html#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.
------------------------------------------------------------------------
source
### safe_test
``` python
def safe_test(
t:str, fname:str
)->bool:
```
*Whether `t` is one plain `assert` that uses only `fname`, literals and
the safe builtins.*
------------------------------------------------------------------------
source
### validate_exercise
``` python
def validate_exercise(
r
)->fastcore.basics.AttrDict:
```
*Check the model’s exercise and raise `ValueError` with the reason if
invalid.*
[`safe_test`](https://vedicreader.github.io/abhyasa/check.html#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.
``` python
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')
```
True
[`validate_exercise`](https://vedicreader.github.io/abhyasa/check.html#validate_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.
``` python
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)
```
``` python
{ '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`.
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#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.
------------------------------------------------------------------------
source
### time_limit
``` python
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.
``` python
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.21
## Run and share results
[`Report`](https://vedicreader.github.io/abhyasa/check.html#report) is
an `AttrDict`, a dictionary whose keys can also be read as attributes.
`found` says whether the function is callable. `timeout` says whether
any check ran out of time or the run was interrupted. `checks` holds
each result; `passed` is true only when a found function passes every
check. Each check has a zero-based index `i`, a label, a status, `got`,
error text, a `hidden` flag and elapsed milliseconds `ms`. The status is
`pass`, `fail` (a false assert), `error` (another exception) or
`timeout`.
------------------------------------------------------------------------
source
### public
``` python
def public(
report
)->dict:
```
*Return page-safe results without test source or expected values.*
------------------------------------------------------------------------
source
### Report
``` python
def Report(
*args, **kwargs
):
```
*Results for one learner function.*
[`run_asserts`](https://vedicreader.github.io/abhyasa/check.html#run_asserts)
gives each test `secs` seconds within a shared `total` budget. Tests
left after that budget expires become timeouts without running.
`KeyboardInterrupt` stops the run and marks the current and remaining
checks as timeouts. Printed output and error output are captured and
discarded. For a single equality assert with a call on the left, the
call runs once and its value becomes `got`, clipped to 200 characters.
------------------------------------------------------------------------
source
### run_asserts
``` python
def run_asserts(
fn, fname, tests, *, secs:float=2.0, total:float=10.0
)->__main__.Report:
```
*Run tests with per-test and total time limits, capturing output and
actual values.*
A report holds one result for each test. `got` is the value returned by
the call, written as Python text. `say` gives a short label.
``` python
def double(n): return n * 2
result = run_asserts(double, 'double', ['assert double(2) == 4', dict(test='assert double(3) == 7', say='Three')])
test_eq([c.status for c in result.checks], ['pass', 'fail'])
test_eq([c.got for c in result.checks], ['4', '6'])
assert not result.passed
result
```
``` python
{ 'checks': [ { 'error': '',
'got': '4',
'hidden': False,
'i': 0,
'label': 'Test 1',
'ms': 0.25216699577867985,
'status': 'pass'},
{ 'error': 'AssertionError: ',
'got': '6',
'hidden': False,
'i': 1,
'label': 'Three',
'ms': 0.16520917415618896,
'status': 'fail'}],
'found': True,
'timeout': False}
```
A hidden test keeps its expected value private.
[`public`](https://vedicreader.github.io/abhyasa/check.html#public)
makes a page-safe dictionary. It omits test text and keeps only the
exception class for a hidden error.
``` python
private_test = 'assert double(3) == 912839'
private_result = run_asserts(double, 'double', [dict(test=private_test, say='Private case', hidden=True)])
page = public(private_result)
assert private_test not in json.dumps(page) and '912839' not in json.dumps(page)
test_eq(page['checks'][0]['error'], 'AssertionError')
page
```
{'v': 1,
'found': True,
'timeout': False,
'checks': [{'i': 0,
'label': 'Private case',
'status': 'fail',
'got': '6',
'error': 'AssertionError',
'hidden': True,
'ms': 0.14683394692838192}]}
## Notebook test cells
[`load_tests`](https://vedicreader.github.io/abhyasa/check.html#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.
------------------------------------------------------------------------
source
### load_tests
``` python
def load_tests(
nb
)->tuple:
```
*Return ordered setup sources and labelled test cells from a notebook.*
[`run_cells`](https://vedicreader.github.io/abhyasa/check.html#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.
------------------------------------------------------------------------
source
### run_cells
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#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.
------------------------------------------------------------------------
source
### snippet
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#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`](https://vedicreader.github.io/abhyasa/check.html#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 + 2` seconds, 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 + 1` seconds of CPU time and
files of at most `runner.file_mb` MB (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-exec` also applies
`SANDBOX_PROFILE` to 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.
------------------------------------------------------------------------
source
### parse
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#_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`](https://vedicreader.github.io/abhyasa/check.html#_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`](https://vedicreader.github.io/abhyasa/check.html#_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`](https://vedicreader.github.io/abhyasa/check.html#_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`](https://vedicreader.github.io/abhyasa/check.html#_split_future)
refuses it with a `ValueError`.
[`_start`](https://vedicreader.github.io/abhyasa/check.html#_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:
``` python
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:
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#_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.
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#_feed) does
not raise when the child closed its input, or when the file is closed
already.
``` python
r, w = os.pipe(); os.close(r)
f = os.fdopen(w, 'wb'); f.close()
_feed(f, b'print(1)')
```
[`run_isolated`](https://vedicreader.github.io/abhyasa/check.html#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.
------------------------------------------------------------------------
source
### run_isolated
``` python
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.
``` python
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.
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#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.
``` python
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`](https://vedicreader.github.io/abhyasa/check.html#parse) reads
only the line that has it. A line without the nonce does not count, even
when it looks like a result.
``` python
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.
``` python
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.
``` python
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.
``` python
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'], r
```
Plain `logging` works in learner code. The code sets up `logging` and
writes a warning, and the check passes. `sys._getframe` stays refused.
``` python
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'], r
```
The `python` option runs the given interpreter. The test code reports
its own version, and the version must match.
``` python
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.
``` python
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.
``` python
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.
``` python
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 = real
```
The timer still stops a slow test after the guard refuses ctypes. The
test must finish in less than the total time.
``` python
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'], r
```
The timer of a worker thread uses `PyThreadState_SetAsyncExc`. The test
runs
[`time_limit`](https://vedicreader.github.io/abhyasa/check.html#time_limit)
in a thread of a closed process with the guard, and the timer must stop
a slow loop.
``` python
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.
``` python
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.
``` python
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 == 0
```
Hostile 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.
``` python
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()
```