Python Egress and Audits

Egress Policy

Interface	Purpose
runtime.egress.enable()	Enable egress interception.
runtime.egress.disable()	Disable egress interception.
runtime.egress.allow(hosts)	Add allowed host names or URLs.
runtime.egress.list()	List allowed hosts.
runtime.egress.remove(host)	Remove one allowed host.
runtime.egress.update_policy(policy)	Patch policy in one call.

Async

await runtime.egress.enable()
await runtime.egress.allow(["pypi.org", "*.pythonhosted.org"])

hosts = await runtime.egress.list()
print(hosts)

await runtime.egress.remove("pypi.org")
await runtime.egress.disable()

Sync

runtime.egress.enable()
runtime.egress.allow(["pypi.org", "*.pythonhosted.org"])

hosts = runtime.egress.list()
print(hosts)

runtime.egress.remove("pypi.org")
runtime.egress.disable()

allow() accepts host names or URLs. URLs are normalized to host names before being sent to the REST API.

Patch the policy in one call:

Async

from runta import EgressPolicy

await runtime.egress.update_policy(
    EgressPolicy(enabled=True, allowed_hosts=["api.github.com"])
)

Sync

from runta import EgressPolicy

runtime.egress.update_policy(
    EgressPolicy(enabled=True, allowed_hosts=["api.github.com"])
)

Audits

Interface	Purpose
runtime.egress.audit(limit=…)	Read recent egress audit events for the runtime.
runtime.audits.list(category=…, limit=…)	List generic audit records.

Async

egress_audits = await runtime.egress.audit(limit=100)
for audit in egress_audits:
    print(audit.action, audit.host, audit.path)

audits = await runtime.audits.list(category="egress", limit=20)
for audit in audits:
    print(audit.category, audit.payload)

Sync

egress_audits = runtime.egress.audit(limit=100)
for audit in egress_audits:
    print(audit.action, audit.host, audit.path)

audits = runtime.audits.list(category="egress", limit=20)
for audit in audits:
    print(audit.category, audit.payload)