safepyrun

Run Python with audit-hook checks, approved tools, and a shared namespace

from fastcore.test import expect_fail,test_eq
import string,types,subprocess

Helpers and setup

When RunPython needs a caller’s namespace, _find_frame_dict looks for sentinel in the globals of each stack frame. If the stack search fails (for instance, inside asyncio.gather), it tries the namespace stored in _rp_globals. Its remaining fallbacks are __main__ and then this module’s globals. An empty sentinel skips the search and prefers __main__.

_test_sentinel = True
d = _find_frame_dict('_test_sentinel')
assert '_test_sentinel' in d
d2 = _find_frame_dict('nonexistent_sentinel_xyz')
assert d2 is not None

_test_sentinel = True
d = _find_frame_dict('_test_sentinel')
assert '_test_sentinel' in d
d3 = _find_frame_dict('')
assert d3 is not None

---

source

find_var

def find_var(
    var:str
):

Search for var in all frames of the call stack

find_var('_test_sentinel')

True

Builtins and wrappers

---

source

freeze_mon_policy

def freeze_mon_policy(
    p
):

Freeze monitoring policy for one run

A host can change mon_disable_policy to tune which sys.monitoring call sites stop producing events. Each run gets its own snapshot from freeze_mon_policy; changes to the configuration don’t alter that snapshot.

---

source

on_call

def on_call(
    caller, callee, fn, code, off, data, calls
):

Fast monitoring callback to decide if event should be DISABLEd

For this cheap performance check, on_call looks at caller and callee names, name pairs, prefixes, suffixes, and any configured predicates. It doesn’t inspect arguments. When a policy needs args or kwargs, it can check the audit events that fastaudit raises for sys.monitoring C calls.

---

source

frame_args

def frame_args(
    fr, obj:NoneType=None
):

Introspection helper used before audit denies an op; check whether it happened inside an approved call

Here’s what frame_args recovers from a plain function: positional arguments, keyword-only arguments, and extra **kwargs.

def some_tool(path, text, *, overwrite=False, **kwargs): return frame_args(currentframe())
args,kwargs = some_tool('path', 'text', overwrite=True, a='b')
test_eq(args, ['path', 'text'])
test_eq(kwargs, {'overwrite': True, 'a': 'b'})

For a method, passing the bound object removes self from the positional arguments:

class F:
    def some_meth(self, path, text, *, overwrite=False, **kwargs): return frame_args(currentframe(), self)
args,kwargs = F().some_meth('path', 'text', overwrite=True, a='b')
test_eq(args, ['path', 'text'])
test_eq(kwargs, {'overwrite': True, 'a': 'b'})

list(_prefix_keys('a.b.cc'))

['a.b.*', 'a.*']

test_eq(_ctx_check(..., 'save', None, ['x'], {}, {}), None)
test_eq(_ctx_check({'save'}, 'save', None, ['x'], {}, {}), True)
test_eq(_ctx_check({'load'}, 'save', None, ['x'], {}, {}), False)
test_eq(_ctx_check({'load', ...}, 'save', None, ['x'], {}, {}), None)

seen = []
def ok(obj, a, kw, data): seen.append((obj, a, kw, data))
def bad(obj, a, kw, data): raise Exception('denied')

seen.clear()
test_eq(_ctx_check({('save', ok)}, 'save', None, ['x'], {'force': True}, {'d': 1}), True)
test_eq(seen, [(None, ['x'], {'force': True}, {'d': 1})])
with expect_fail(): _ctx_check({('save', bad)}, 'save', None, ['x'], {}, {})

class T: pass
t = T()
seen.clear()
test_eq(_ctx_check({('save', ok)}, 'save', t, [t, 'x'], {'force': True}, {'d': 1}), True)
test_eq(seen, [(t, ['x'], {'force': True}, {'d': 1})])

The registration checks distinguish three answers:

• True: an explicit registration allows this call.
• False: no registration matched.
• None: an allow-all (...) entry matched, but we still need to check explicit policies.

An explicit policy gets its say before a blanket allowance. For example, a (name, policy) registration can allow the method or raise PermissionError even when 'pkg.*': ... also covers it.

_ctx_check handles one registration set. A direct name match succeeds. For a matching validator tuple, it passes (obj, args, kwargs, data) to the validator. One passing validator is enough; if all matching validators deny the call, it combines their messages in a PermissionError. An allow-all entry returns None. Module, class, tracked-call, and frame-call checks all use this logic.

_call_allowed looks up one CallInfo. It tries the module, module-prefix keys, the bound instance, and the class/MRO registrations. _prefix_keys('a.b.c'), for example, yields a.b.* and a.*. An explicit success returns immediately. Otherwise it remembers any allow-all match and returns None only after checking the remaining levels. Both async tracked calls and the stack-frame fallback use this lookup.

_ctx_allowed applies that lookup to every call in a DenyInfo. It returns True for an explicit success, or for an allow-all match if no explicit policy has decided the outcome. A blanket allowance anywhere in the stack can therefore permit the operation.

---

source

DenyInfo

def DenyInfo(
    event, args, frame, msg, data, calls, frame_args
):

---

source

CallInfo

def CallInfo(
    fn:NoneType=None, args:tuple=(), kwargs:NoneType=None, module:NoneType=None, qualname:NoneType=None,
    name:NoneType=None, frame:NoneType=None, source:NoneType=None
):

---

source

RawDenyInfo

def RawDenyInfo(
    args, frame, msg, calls, frame_args
):

A pre_deny callback receives a DenyInfo. Its public attributes are event, data, call, args, kwargs, calls, native_calls, tracked_calls, and frame_calls; raw keeps the lower-level audit details.

A native callee has neither a Python frame nor a tracked call. For a fastaudit.call event, DenyInfo makes a CallInfo from the dotted callee name and puts it in native_calls. That lets the same lookup check extension-class and module registrations.

Several callable objects can share one __call__ implementation. Think of fastspec’s dynamically generated OpFunc operations: allowing one operation shouldn’t require allowing the whole class. _call_allowed compares the bound object with registry keys by identity, so allow(op) applies to that object alone:

class _Op:
    async def __call__(self, x): return x

op1,op2 = _Op(),_Op()
tst = dict(pytools={op1: {'__call__'}})
test_eq(_call_allowed(CallInfo(fn=_Op.__call__, args=(op1,'x'), module=__name__, qualname='_Op.__call__'), tst), True)
test_eq(_call_allowed(CallInfo(fn=_Op.__call__, args=(op2,'x'), module=__name__, qualname='_Op.__call__'), tst), False)

class _Unhashable:
    __hash__ = None
    def save(self): pass
test_eq(_call_allowed(CallInfo(fn=_Unhashable.save, args=(_Unhashable(),), module=__name__, qualname='_Unhashable.save'), tst), False)

---

source

before_deny

def before_deny(
    event, args, frame, msg, data, calls, pre_deny:NoneType=None, _frame_args:function=frame_args
):

Check whether a possibly-denied audit event happened inside an approved call.

When fastaudit is about to deny an operation, it calls before_deny. This builds a DenyInfo and asks pre_deny(info) first. A non-None answer is final; otherwise _ctx_allowed checks the registered pytools. A successful check returns True. Without one, the adapter returns None and leaves the denial to the audit layer.

Check direct allow-by-name:

tst_data = dict(pytools={sys.modules[__name__]: {'_bd_allowed'}}, ok_dests=set())
def _bd_allowed(): return before_deny(None, None, currentframe(), None, data=tst_data, calls=[])
def _bd_denied (): return before_deny(None, None, currentframe(), None, data=tst_data, calls=[])

assert _bd_allowed()
assert not _bd_denied()

Check the registered method passes its object, recovered args, kwargs, and allowed destinations:

_seen = []
def _bd_check(obj, args, kw, data): _seen.append((obj, args, kw, data['ok_dests']))

class _BDT:
    def save(self, path, *, overwrite=False): return before_deny(None, None, currentframe(), None, data=tst_data, calls=[])
tst_data = dict(pytools={_BDT: [('save', _bd_check)]}, ok_dests={'/tmp'})

t = _BDT()
test_eq(t.save('/tmp/x', overwrite=True), True)
test_eq(_seen, [(t, ['/tmp/x'], dict(overwrite=True), {'/tmp'})])

def _deny(obj, a, kw, data): raise Exception('denied')

info = types.SimpleNamespace(
    calls=[
        CallInfo(module='matplotlib', qualname='pyplot.plot', name='matplotlib.pyplot.plot'),
        CallInfo(module='danger', qualname='save', name='danger.save')],
    data=dict(pytools={'matplotlib.*': ..., 'danger.*': {('save', _deny)}}))

with expect_fail(): _ctx_allowed(info)

This test supplies a native call’s dotted name directly. There is no Python frame or tracked call for the callee, but its extension-class or module registration must still apply:

class _NativeT:
    def op(self): ...

tst_data = dict(pytools={_NativeT: {'op'}}, ok_dests=set())
assert before_deny('fastaudit.call', ('caller', f'{__name__}._NativeT.op', 'chain'), currentframe(), 'msg', tst_data, [])
assert not before_deny('fastaudit.call', ('caller', f'{__name__}._NativeT.other', 'chain'), currentframe(), 'msg', tst_data, [])

Main implementation

---

source

srcfn

def srcfn(
    src
):

Stores src in linecache under <python_{i%10}>, returns the name.

srcfn(''),srcfn('')

('<python_0>', '<python_1>')

---

source

__run_python

async def __run_python(
    code:str, g:NoneType=None, ok_dests:NoneType=None
):

Each _run_python call gives fastaudit a policy bundle: approved pytools, allowed write destinations, and a snapshot of the monitoring configuration.

For input(), it uses the host’s current builtins.input, not the hook that happened to exist at import time. Kernel hosts can therefore install interactive stdin routing after importing safepyrun.

---

source

set_data

def set_data(
    **kw
):

Set default entries for the per-run policy data dict; per-RunPython kwargs win

Use set_data for policy data that should apply to every RunPython instance. For example, a host or user config can set ok_urls once rather than supplying it on each call. Calling set_data raises a safepyrun.set_data audit event; code running under the policy cannot use it to change those defaults.

A tool call must finish its background work before leaving the audit context. _run_python records the existing tasks with asyncio.all_tasks(), then waits for new tasks before returning. That includes tasks they spawn in turn. A failure in a background task makes the tool call raise too.

Without this wait, work started with asyncio.create_task could outlive the policy that allowed the tool to run.

res = []
async def bg_job():
    await asyncio.sleep(0.1)
    res.append('finished')

r = await _run_python("asyncio.create_task(bg_job())\n'returned'", g=dict(asyncio=asyncio, bg_job=bg_job), ok_dests=())
test_eq(r, 'returned')
test_eq(res, ['finished'])  # bg task completed before the call returned

async def bg_bad():
    await asyncio.sleep(0.01)
    subprocess.run(['ls'])
    
with expect_fail(PermissionError, 'subprocess.Popen blocked'):
    await _run_python("asyncio.create_task(bg_bad())", g=dict(asyncio=asyncio, bg_bad=bg_bad), ok_dests=())

A host can inject helpers such as python and allow into the user namespace. The runner doesn’t copy those names back out after execution. User code therefore cannot replace a helper with a trojan for the host to call later.

with expect_fail(PermissionError):
    await _run_python("allow_imports.add('evil')", g={'allow_imports': allow_imports})
allow_imports.discard('evil')

Importing this module gives user code access to mutable host policy: __pytools__, allow_imports, and mon_disable_policy. _run_python fingerprints that state before execution and checks it again at the end. If it changed, the call raises. A run must not quietly loosen permissions for the next call.

g = {}
await _run_python("python = 'trojan'\nallow = 'trojan'\nkeep = 42", g=g)
test_eq(g['keep'], 42)
assert 'python' not in g and 'allow' not in g

_check_user_code checks the submitted source before execution. It rejects banned-module imports, calls to exec, eval, or compile, and references to importlib. With ban_defs=True, it also rejects function and class definitions.

These source rules apply to the submitted code, not to the implementations of host-provided tools. Banning pkg also bans pkg.submodule, but not pkg_extra. A ban on pkg.submodule leaves the rest of pkg available. Ordinary imports outside ban_imports remain available.

Some libraries, such as httpcore, wrap our permission errors in another exception. We walk the exception chain to recover the PermissionError.

---

source

RunPython

def RunPython(
    g:NoneType=None, sentinel:NoneType=None, ok_dests:fastcore.xtras.Unset=UNSET,
    ban_imports:frozenset=frozenset({'fastaudit', 'importlib', 'socket', 'safepyrun'}), ban_defs:bool=True,
    pre_deny:NoneType=None, yolo:bool=False, **kwargs
):

Execute Python with audit-hook safety checks and access to LLM tools, returning last expression. import works in the usual way. All builtins are available. Multiline code blocks can be used. By default, defining functions or classes is not allowed (ban_defs=True); construct RunPython with ban_defs=False to permit them.

Sandbox: an audit hook blocks risky operations by default (e.g. network), and socket/importlib imports are banned. To permit one, the user must allow() a function that performs it from the real Python process; sandboxed code cannot allow() itself. NB: Locals are exported back to the caller’s namespace.

Use RunPython to execute code under the audit-hook policy. Pass a namespace as g, or let it use IPython’s user namespace or _find_frame_dict outside IPython. Omitting ok_dests uses the configured default_ok_dests. Pass a directory list to restrict paths, () to allow none, or None to remove path restrictions. Other audit checks and source restrictions still apply.

python = RunPython()

await python('[]')

[]

async def f(): return 1

await python('await f()')

1

Pass g={} to start with a clean namespace:

g = {}
await RunPython(g=g)('_ns_test_var = 42')
test_eq(list(g.keys()), ['_ns_test_var'])
assert '_ns_test_var' not in globals()

---

source

create_python_magic

def create_python_magic(
    shell:NoneType=None, python:NoneType=None, *, g:NoneType=None, sentinel:NoneType=None,
    ok_dests:fastcore.xtras.Unset=UNSET,
    ban_imports:frozenset=frozenset({'fastaudit', 'importlib', 'socket', 'safepyrun'}), ban_defs:bool=True,
    pre_deny:NoneType=None, yolo:bool=False
):

Create magic

create_python_magic()

%%py
print('tt')

tt

%%py
type('t')

str

%%py
a = 1
a+=2
a

3

Unpacking is allowed:

%%py
a = [1,2,3]
print(*a)

1 2 3

def f(): warnings.warn('a warning')

%%py
print("asdf")
f()
1+1

asdf

/var/folders/51/b2_szf2945n072c0vj2cyty40000gn/T/kernmini_7541/3833129470.py:1: UserWarning: a warning
  def f(): warnings.warn('a warning')

2

With the default ban_defs=True, submitted code cannot define classes or functions:

with expect_fail(PermissionError):
    await python('class A:\n    def __init__(self): print("hi")')

with expect_fail(PermissionError):
    await python('def f(): print("hi")')

with expect_fail(PermissionError): await python('os.system("ls")')
with expect_fail(PermissionError): _check_user_code(ast.parse('from safepyrun.core import RunPython'), {'safepyrun'}, True)

%%py
print(os.unlink)
print(type(os.unlink))
print(os.unlink.__qualname__)

<built-in function unlink>
<class 'builtin_function_or_method'>
unlink

class C: ...
c = C()

%%py
isinstance(C, type)

True

%%py
isinstance(c, type)

False

async def f(): return 1

%%py
await f()

1

allow_write_types

o = SimpleNamespace(x=1)

await python('''
d = {}
d["x"] = 1
o.x = 2
d["x"],o.x''')

(1, 2)

Config

At import time, safepyrun loads {xdg_config_home}/safepyrun/config.py. If the file is missing, it creates one containing default_ok_dests = ['.', '/tmp']. It never overwrites an existing config. Use this file to choose write defaults and extend the allowlists without changing the package.

The config runs with safepyrun.core globals available. You can call allow, set_data, and allow_write_types without importing them. The policy classes AllowPolicy, PathWritePolicy, PosAllowPolicy, and OpenWritePolicy are available too, along with the standard library modules that safepyrun.core imports.

The default path is ~/.config/safepyrun/config.py on Linux and macOS. Set XDG_CONFIG_HOME to use another config directory. For example:

# Add pandas tools
allow({pandas.DataFrame: ['head', 'describe', 'info', 'shape']})

# Allow writes under these directories by default
default_ok_dests = ['.', '/tmp']

# Default policy data for allow policies (e.g. chk_url-style URL checks)
set_data(ok_urls={'http://example.org'})

Calls to allow and set_data register their changes directly. The loader also copies back an assignment to default_ok_dests; other assignments don’t change the module’s globals.

Config errors propagate and stop initialization. Fix the config before retrying startup.

Examples

%%py
a = {"b":1}
list(a.items())

[('b', 1)]

%%py
Path().exists()

True

%%py
os.path.join('/foo', 'bar', 'baz.py')

'/foo/bar/baz.py'

%%py
a_=3

a_

3

%%py
aa_='33'

%%py
len(aa_)

2

def g(): ...

%%py
inspect.getsource(g)

'def g(): ...\n'

with expect_fail(PermissionError): await python("os.unlink('/foo/bar')")

async def agen():
    for x in [1,2]: yield x

%%py
res = []
async for x in agen(): res.append(x)
res

[1, 2]

import asyncio
async def fetch(n): return n * 10

%%py
print(string.ascii_letters)
await asyncio.gather(fetch(1), fetch(2), fetch(3))

abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ

[10, 20, 30]

import numpy as np

%%py
np.array([1,2,3]).sum()

np.int64(6)

Allow policy examples

python2 = RunPython(ok_dests=['/tmp'])

await python2("Path('/tmp/test_write.txt').write_text('hello')")

5

await python2("open('/tmp/test_open.txt', 'w').write('hi')")

2

with expect_fail(PermissionError): await python2("Path('/etc/evil.txt').write_text('bad')")
with expect_fail(PermissionError): await python2("open('/root/bad.txt', 'w')")

await python2("open('/etc/passwd', 'r').read(10)")

'##\n# User '

await python2("import shutil; shutil.copy('/tmp/test_write.txt', '/tmp/test_copy.txt')")

'/tmp/test_copy.txt'

with expect_fail(PermissionError): await python2("import shutil; shutil.copy('/tmp/test_write.txt', '/root/bad.txt')")
with expect_fail(PermissionError): await python("Path('/tmp/test.txt').write_text('nope')")

python_cwd = RunPython(ok_dests=['.'])

# Writing to cwd should work
await python_cwd("Path('test_cwd_ok.txt').write_text('hello')")

5

python_ro = RunPython(ok_dests=())
with expect_fail(PermissionError): await python_ro("Path('test_cwd_ok.txt').write_text('hello')")
python_un = RunPython(ok_dests=None)
await python_un("Path('test_cwd_ok.txt').write_text('hello')")
with expect_fail(PermissionError): await python_un("subprocess.run(['echo', 'hi'])")

Path('test_cwd_ok.txt').unlink(missing_ok=True)

# Writing to /tmp should be blocked (not in ok_dests)
with expect_fail(PermissionError): await python_cwd("Path('/tmp/nope.txt').write_text('bad')")

# Parent traversal should be blocked
with expect_fail(PermissionError): await python_cwd("Path('../escape.txt').write_text('bad')")

# Sneaky traversal via subdir/../../ should also be blocked
with expect_fail(PermissionError): await python_cwd("Path('subdir/../../escape.txt').write_text('bad')")

allow

Plain Python needs no registration when it has no guarded side effects. allow matters when a callable needs an operation the audit policy would otherwise deny, such as a subprocess, network request, write, or other guarded audit event:

def pure(name): return f"Hello, {name}!"
test_eq(await python('pure("World")'), 'Hello, World!')

class A:
    def f(self): ...

def g(self:A): ...

import fastcore.basics

__pytools__.pop(fastcore.basics, None);

with expect_fail(PermissionError): await python("patch(g)")

assert await python("patch(g)")

def trusted_echo(): return subprocess.run(['echo', 'hi'], capture_output=True, text=True)

with expect_fail(PermissionError): await python("trusted_echo().stdout")
with expect_fail(PermissionError): await python("import subprocess; subprocess.run(['echo', 'hi'])")

allow(trusted_echo)
test_eq((await python("trusted_echo().stdout")), "hi\n")
with expect_fail(PermissionError): await python("import subprocess; subprocess.run(['echo', 'hi'])")

class _MethT:
    def echo (self): return run('echo hi')
    def echo2(self): return run('echo hi')

t = _MethT()
with expect_fail(PermissionError): await python("t.echo()")
with expect_fail(PermissionError): await python("t.echo2()")
allow(_MethT.echo)
test_eq(await python("t.echo()"), "hi")

allow({_MethT:['echo2']})
test_eq(await python("t.echo2()"), "hi")

@patch
def echo3(self:_MethT): return run('echo hi')
with expect_fail(PermissionError): await python("t.echo3()")
allow({_MethT:['echo3']})
test_eq(await python("t.echo3()"), "hi")

Here are three callable instances with the same __call__. Registering op_a leaves op_b and op_c unapproved, just as we’d want for separate operations on a generated client:

class _OpT:
    def __init__(self,name): store_attr()
    def __call__(self): return run('echo hi')

op_a,op_b,op_c = _OpT('a'),_OpT('b'),_OpT('c')
with expect_fail(PermissionError): await python("op_a()")
allow(op_a)
test_eq(await python("op_a()"), "hi")
with expect_fail(PermissionError): await python("op_b()")

Registering every operation gets tedious when a client keeps generating new ones. A policy can instead inspect the object, for example checking an OpenAPIClient’s base_url. Policies are additive: AllowPolicy raises only when none permits the call. These examples check the operation’s name:

__pytools__.pop(op_a,None);

class AllowOpTA(AllowPolicy):
    def __call__(self, obj, args, kwargs, data):
        if 'a' not in obj.name : raise PermissionError(f'Only "a" is allowed not: {obj.name}')
allow({_OpT:[('__call__', AllowOpTA())]})

test_eq(await python("op_a()"), "hi")
with expect_fail(PermissionError): await python("op_b()")

class AllowOpTB(AllowPolicy):
    def __call__(self, obj, args, kwargs, data):
        if 'b' not in obj.name : raise PermissionError(f'Only "b" is allowed not: {obj.name}')
allow({_OpT:[('__call__', AllowOpTB())]})

test_eq(await python("op_b()"), "hi")
with expect_fail(PermissionError, 'Only "a" is allowed not: c; Only "b" is allowed not: c'): await python("op_c()")

import httpx

with expect_fail(PermissionError): await python('httpx.get("http://example.org")')

@allow
def getexample(): return httpx.get('http://example.org')

await python('getexample()')

<Response [200 OK]>

def testevent():
    sys.audit('python.testevent')
    return 'ok'

with expect_fail(PermissionError): await python('testevent()')

@allow
def testevent2():
    return testevent()

%%py
testevent2()

'ok'

def chk_url(obj, args, kw, data):
    url = args[0] if args else kw.get('url','')
    if url not in data.get('ok_urls', ()): raise PermissionError(url)

python_urls = RunPython(ok_urls={'http://example.org'})

allow({httpx.get: chk_url})
with expect_fail(PermissionError): await python_urls('httpx.get("http://example.com")')
await python_urls('httpx.get("http://example.org")')

<Response [200 OK]>

To use the same policy data for new RunPython instances, put it in set_data. An instance can still override those defaults with its own kwargs:

set_data(ok_urls={'http://example.org'})
await RunPython()('httpx.get("http://example.org")')

<Response [200 OK]>

@allow
def _httpget(url): return httpx.get(url)

def httpget(url):
    sys.audit('myapp.httpget', url)
    return _httpget(url)

def url_allow(info):
    if info.event=='myapp.httpget' and info.raw.args[0] in info.data.get('ok_urls', ()): return True

python_urls2 = RunPython(pre_deny=url_allow, ok_urls={'http://example.org'})
with expect_fail(PermissionError): await python_urls2('httpget("http://example.com")')
await python_urls2('httpget("http://example.org")')

<Response [200 OK]>

Code inside a run cannot loosen policy for later runs. Calls to allow and set_data raise the audit events pyskills.allow and safepyrun.set_data. The policy denies those events like any other unapproved operation:

with expect_fail(PermissionError): await python('allow(print)')
with expect_fail(PermissionError): await python('set_data(ok_dests=["/"])')

Direct registry mutation raises no audit event to block up front. The fingerprint check must catch it at the end of the run, before any later call can use the changed policy:

with expect_fail(PermissionError, 'policy changed'): await python("__pytools__[str].add('lower')")
__pytools__.pop(str, None)

{'lower'}

Plots and Pandas

---

source

allow_matplotlib

def allow_matplotlib():

import matplotlib.pyplot as plt
from matplotlib.figure import Figure
from matplotlib.axes import Axes
from matplotlib.axis import Axis
from matplotlib.spines import Spine, SpinesProxy

allow_matplotlib()

%%py
fig, ax = plt.subplots(figsize=(3,2), dpi=100)
ax.plot([1,2,1], label='Triangle')
ax.plot([1,1.5,2], label='Rising')
ax.tick_params(labelsize=10)
ax.set_xlabel('X', fontsize=10)
ax.set_ylabel('Y', fontsize=10)
ax.spines[['top','right']].set_visible(False)
ax.yaxis.set_major_formatter('{x:.0f}')
ax.legend(fontsize=12);

python_tmp = RunPython(ok_dests=['/tmp'])
with expect_fail(PermissionError): await python_tmp("fig.savefig(os.path.expanduser('~/plot.png'))")
await python_tmp("fig.savefig('/tmp/plot.png')")

---

source

allow_pandas

def allow_pandas():

import pandas as pd

df = pd.DataFrame(data={'col1':[1,1,1,2,2],'col2':['George','Tim','Anna','Bob','Jon']})

python_pd = RunPython(ok_dests=['/tmp'])
with expect_fail(PermissionError): await python_pd('df.to_csv("/tmp/foo.csv")')

allow_pandas()

await python_pd('df.to_csv("/tmp/foo.csv")')
with expect_fail(PermissionError): await python_pd('df.to_csv("~/foo.csv")')

await python_pd('df.to_csv("/tmp/foo.csv")')
with expect_fail(PermissionError): await python_pd('df.to_csv("~/foo.csv")')
with expect_fail(PermissionError): await python_pd('df.to_json(path_or_buf="~/foo.json")')
await python_pd('df.to_json(path_or_buf="/tmp/foo.json")')

allow_pandas() doesn’t give pandas blanket permission. A write through an unregistered pandas method still needs to pass the ok_dests check, just like any other write:

await python_pd("pd.io.common.get_handle('/tmp/raw.txt','w').close()")
with expect_fail(PermissionError): await python_pd("pd.io.common.get_handle('~/raw.txt','w')")

Extension

---

source

load_ipython_extension

def load_ipython_extension(
    ip
):

---

source

cli

def cli(
    path:str, # Path to script, or '-' for stdin
):

Run a python script file in the safepyrun sandbox