Starlark libraries

This page is the reference for the Starlark modules available to custom integration scripts, with a runnable example for each. Save any example as a .star file and run it with the runZero CLI:

runzero script --filename example.star

Load only the names you use; every module is available via load(...). The names report_assets, report_asset, pager, max_pages, OPTIONS_TLS, and OPTIONS_HTTP are predeclared and never need a load().

load("runzero.types", "ImportAsset", "NetworkInterface", "Service",
                      "ServiceProtocolData", "Software", "Vulnerability",
                      "to_custom_attributes")
load("kwargs", "require", "has", "get", "get_string", "get_bool", "get_int",
               "get_float", "get_list", "get_url_base", "get_http_tls",
               "get_http_options")
load("http", http_get="get", http_post="post", "head", "put", "patch", "delete",
             "get_json", "post_json", "url_encode", "url_parse", "url_join",
             "multipart", "bearer", "basic", "oauth2_token")
load("requests", "Session", "Cookie")
load("net", "ip_address", "network_interface", "normalize_mac",
            "ip_network", "ip_in_network", "resolve",
            "routable_ip", "routable_ips", "clean_hostname", "clean_hostnames",
            "mac_key")
load("coerce", "as_text", "as_dict", "as_list", "dicts",
               "as_int", "as_float", "as_bool", "dedupe")
load("json", json_encode="encode", json_decode="decode")
load("jsonstream", "iter_array", "iter_lines")
load("csv", csv_read="read_all", csv_write="write_dicts")
load("xml", xml_parse="parse")
load("re", re_match="match", re_find_all="find_all", re_sub="sub")
load("time", "now", "parse_time", "parse_ts", "parse_duration", "sleep")
load("uuid", "new_uuid")
load("base64", base64_encode="encode", base64_decode="decode")
load("hex", hex_encode="encode", hex_decode="decode")
load("base32", b32_encode="encode", b32_decode="decode")
load("crypto", "sha256", "hmac_sha256", "sign_v4", "random_hex")
load("jwt", jwt_encode="encode", jwt_decode="decode")
load("gzip", gzip_compress="compress", gzip_decompress="decompress")
load("flatten_json", "flatten")
load("runzero.progress", progress_report="report", progress_info="info")
load("socket", "tcp", "udp", "tls")
load("runzero.ssh", ssh_dial="dial")
load("runzero.smb", smb_dial="dial")
load("runzero.winrm", winrm_dial="dial")
load("runzero.wmi", wmi_dial="dial")
load("runzero.sql", sql_connect="connect")

runzero.types

Constructors for the objects a script imports. Every keyword argument accepts both camelCase (osVersion) and snake_case (os_version) spellings.

ImportAsset(id, ...) — id is required and must be a stable, unique string (up to 1024 characters). Optional fields: hostnames, domain, os, osVersion, manufacturer, model, deviceType, assetType, tags, firstSeenTS, lastSeenTS, networkInterfaces, services, software, vulnerabilities, customAttributes, runZeroID, trust_device_type, trust_os, trust_os_version. Merge policy is no longer set per asset: declare matchBehavior in the CONFIG block instead — passing it to ImportAsset is an error.
NetworkInterface(macAddress=None, ipv4Addresses=[], ipv6Addresses=[]) — or build one with net.network_interface, which handles mixed and messy input.
Service(address, port, transport, ...) — the first three are required; vendor, product, version, protocolData, and customAttributes are optional.
ServiceProtocolData(name, attributes={}) — protocol-specific details attached to a Service.
Software(id, ...) — optional vendor, product, version, update, cpe23, serviceAddress, serviceTransport, servicePort, installedAt, installedSize, installedFrom, and more.
Vulnerability(id, ...) — optional name, description, solution, cve (must match CVE-YYYY-NNNN), cpe23, cvss2BaseScore, cvss3BaseScore, severityRank/riskRank (0=Info through 4=Critical), severityScore/riskScore, exploitable, service linkage, and timestamps.
to_custom_attributes(value, ...) — coerces arbitrary nested data into the string -> string map required by customAttributes. Nested dicts are flattened with separator (default "."), lists joined with list_join (default ","), empty values dropped unless drop_empty=False, and prefix, exclude, max_key, max_value, and max_entries are available for fine control. The platform caps custom attributes at 1024 entries, with keys up to 256 and values up to 1024 characters.

Limits worth knowing: up to 256 network interfaces and 100 hostnames (260 characters each) per asset; up to 100 tags. hostnames and tags silently drop empty strings, but a None entry in hostnames is an error that aborts the run — hostnames=[device.get("hostname")] fails whenever the field is absent. Pass hostname fields through net.clean_hostnames, which also screens out placeholder names.

load("runzero.types", "ImportAsset", "Service", "ServiceProtocolData",
                      "Software", "Vulnerability", "to_custom_attributes")
load("net", "network_interface")

def main(*args, **kwargs):
    nic = network_interface(
        mac="9C:B6:D0:F1:D2:04",
        ips=["192.0.2.5", "fe80::1%eth0", "[2001:db8::1]:443"],
    )
    svc = Service(
        address="192.0.2.5",
        port=443,
        transport="tcp",
        vendor="nginx",
        product="nginx",
        version="1.25.3",
        protocolData=[
            ServiceProtocolData(name="http", attributes={"server": "nginx"}),
            ServiceProtocolData(name="tls", attributes={"subject": "CN=acme"}),
        ],
    )
    asset = ImportAsset(
        id="device-123",
        hostnames=["web1.acme.local"],
        os="Linux",
        osVersion="6.1",
        deviceType="Server",
        manufacturer="Acme",
        model="Rack-42",
        tags=["prod", "web"],
        networkInterfaces=[nic],
        services=[svc],
        software=[Software(id="pkg-1", vendor="F5", product="NGINX", version="1.25.3",
                           serviceAddress="192.0.2.5")],
        vulnerabilities=[Vulnerability(id="vuln-1", name="Example finding",
                                       cve="CVE-2023-0001", severityRank=3,
                                       severityScore=8.1, riskRank=3, riskScore=8.1)],
        customAttributes=to_custom_attributes({
            "name": "web1",
            "labels": ["prod", "edge"],       # joined with ","
            "sys": {"os": "linux", "ver": 5}, # flattened with "."
            "empty": "",                      # dropped
        }),
    )
    return [asset]

kwargs

Typed, validating accessors over the **kwargs dict passed to main. Values arrive as strings; these helpers coerce them and apply defaults.

require(kwargs, *keys) — error if any key is missing or blank.
has(kwargs, key) — True when the key is present and non-empty.
get(kwargs, key, default="") / get_string(...) — aliases.
get_bool(kwargs, key, default=False) — accepts true/false, 1/0, yes/no, y/n, on/off.
get_int(kwargs, key, default=0) / get_float(kwargs, key, default=0.0)
get_list(kwargs, key, default=[], sep=",") — splits a delimited string, or passes a list through.
get_url_base(kwargs, key="url", default="") — extracts scheme + host from a URL kwarg, dropping any path or query.
get_http_tls(kwargs, prefix="tls_") — collects the OPTIONS_TLS include into a tls= dict.
get_http_options(kwargs, prefix="http_", tls_prefix="tls_", headers=None) — collects the OPTIONS_HTTP and OPTIONS_TLS includes into a dict of keyword arguments for any http function.

load("kwargs", "require", "has", "get_string", "get_bool", "get_int", "get_list")

def main(*args, **kwargs):
    require(kwargs, "client_id", "client_secret")
    client_id = get_string(kwargs, "client_id")
    page_size = get_int(kwargs, "page_size", default=100)
    include_offline = get_bool(kwargs, "include_offline", default=False)
    regions = get_list(kwargs, "regions", default=[])
    print(client_id, page_size, include_offline, regions)
    if has(kwargs, "region"):
        print("region override:", get_string(kwargs, "region"))

runzero script --filename example.star --kwargs client_id=abc --kwargs client_secret=xyz --kwargs include_offline=yes --kwargs regions=us-east,us-west

coerce

Total conversions for messy API data: nothing in this module raises, so a malformed or missing field degrades to a default instead of aborting the import.

as_text(value, default="", join="") — any value to a string; lists are joined with join when set.
as_dict(value) — a dict, or {} for anything else.
as_list(value, wrap=True) — a list; scalars are wrapped in a one-element list unless wrap=False.
dicts(value) — the dict elements of a list, dropping everything else.
as_int(value, default=0) / as_float(value, default=0.0) — numbers from ints, floats, or numeric strings.
as_bool(value, default=False) — the usual true/false spellings.
dedupe(values, fold_case=False) — order-preserving deduplication.

load("coerce", "as_text", "as_int", "as_list", "dicts")

def main(*args, **kwargs):
    device = {"name": None, "port": "8443", "ips": "192.0.2.5",
              "disks": [{"id": 1}, "oops", {"id": 2}]}
    print("name:", as_text(device["name"], default="unknown"))
    print("port:", as_int(device["port"]))
    print("ips:", as_list(device["ips"]))    # a scalar becomes a one-item list
    print("disks:", dicts(device["disks"]))  # non-dict entries dropped

http

Stateless HTTP requests, JSON convenience wrappers, auth helpers, and URL utilities.

Verbs — get(url, headers=None, params=None, timeout=60, insecure_skip_verify=False, tls=None) and head(...) take the same arguments; post, put, patch, and delete additionally accept body= (bytes/string) or json= (auto-encoded dict). All verbs default to a 60-second timeout, and params= replaces any query string already present in the URL. Each returns a response struct with status_code (int), status (string), headers (values are lists of strings), and body.

get_json / post_json — the common fetch-check-decode pattern in one call, returning a (data, err) tuple. err is None on success, or a short string: "status 401: ..." for an HTTP failure, or the transport error text for a connection failure. A 2xx response with an empty body decodes to None. Both also accept insecure_skip_verify= and tls=. Transient failures are retried by default: retries defaults to 3, with exponential backoff (retry_backoff, default 1 second, doubling up to retry_max_backoff, default 30 seconds) on transient statuses (408, 425, 429, 500, 502, 503, 504 by default; override with retry_on=[...]), honoring Retry-After headers. Pass retries=0 to opt out — do so for non-idempotent writes.

Auth helpers — bearer(token) and basic(username, password) format Authorization header values. oauth2_token(token_url, client_id, client_secret, scope=None, audience=None, grant_type="client_credentials", extra=None, headers=None, timeout=60, insecure_skip_verify=False, tls=None) performs an OAuth token exchange and returns the access token string.

URL utilities — url_encode(params) builds a query string; url_parse(url) returns a struct (scheme, host, hostname, port, path, query, fragment, …) or None; url_join(base, ref) resolves a relative reference (useful for pagination next links); multipart(fields) builds a multipart/form-data body and returns (body_bytes, content_type).

load("http", "get_json", "url_parse", "url_join", "bearer")

def main(*args, **kwargs):
    url = "https://hacker-news.firebaseio.com/v0/topstories.json"
    data, err = get_json(url, headers={"Accept": "application/json"}, retries=2)
    if err:
        print("fetch failed:", err)
        return None
    print("top story IDs:", data[:5])

    u = url_parse(url)
    print("host:", u.host, "path:", u.path)
    print("next page:", url_join(url, "/v0/beststories.json"))
    print("auth header:", bearer("example-token"))

In an integration, collect the option-suite settings once and splat them into every call:

http_options = get_http_options(kwargs, headers={"Authorization": bearer(token)})
data, err = get_json(url, params={"limit": 100}, **http_options)

The tls= argument accepts a dict with the keys insecure, server_name, ca_pem, client_cert_pem, client_key_pem, and thumbprints (SHA-256 pins); unknown keys are an error. Use the raw verbs instead of get_json when you need response headers, cookies, or the status code directly.

Where an egress filter applies (console-hosted runs, or the CLI’s --starlark-allow-cidrs/--starlark-block-cidrs flags), a request to a refused address aborts the script rather than returning an err tuple.

requests

A stateful HTTP Session with sticky headers and a cookie jar. Verbs (get, post, put, patch, delete, head) accept headers, cookies, params, body, json, and timeout (default 60 seconds), and return the same response struct as the http module.

Note that Session does not accept the tls= dict; only Session(insecure_skip_verify=True) is available. Prefer the http module when the TLS option suite matters.

load("requests", "Session", "Cookie")
load("json", json_decode="decode")

def main(*args, **kwargs):
    session = Session()
    session.headers.set("Accept", "application/json")
    session.headers.set("User-Agent", "runZero-Example/1.0")

    url = "https://hacker-news.firebaseio.com/v0/topstories.json"
    response = session.get(url)
    if response.status_code == 200:
        data = json_decode(response.body)
        print("top story IDs:", data[:5])
    else:
        print("request failed with status:", response.status_code)

session.headers exposes get(key) and set(key, value) (value=None deletes a header). session.cookies exposes get(url), set(url, cookies), and clear(); Cookie(name, value, path="", domain="", secure=False, ...) builds a full cookie object for cookies.set.

net

IP, MAC, and DNS helpers, plus identity screens that keep placeholder values out of asset matching.

ip_address(s) — validates one address; the result exposes .version (4 or 6) and stringifies to canonical form.
network_interface(mac=None, ips=None, ipv4=None, ipv6=None) — builds a NetworkInterface from messy input: mixed v4/v6 lists, addr:port and [addr]:port suffixes, and %zone IDs are handled, duplicates dropped, and up to 99 addresses kept per family. Returns None when nothing usable remains, so if nic: guards work.
normalize_mac(s, preserve_bits=True) — canonical lowercase colon form from any common format (colons, dashes, Cisco dotted, bare hex); returns None for unparseable input. Pass preserve_bits=False to clear the locally-administered bit, the historical matching form; network_interface stores the asset MAC with the bit cleared regardless.
mac_key(s) — normalize_mac plus rejection of values that cannot identify a device: all-zero and broadcast addresses and synthetic ip-<addr> placeholders return None.
routable_ip(value, exclude=None) / routable_ips(values, exclude=None) — the canonical identity-usable address (or None), rejecting loopback, unspecified, link-local, multicast, and broadcast addresses while keeping RFC1918, CGNAT, and ULA space. Accepts bracketed, host:port, %zone, and CIDR-suffixed input. exclude=[...] overrides the exclusion list; exclude=[] disables it.
clean_hostname(value, extra=None, max_length=253) / clean_hostnames(values, ...) — a usable hostname or None, rejecting placeholders (localhost, unknown, none, -, n/a), IP-shaped values, all-numeric names, over-long names, and illegal DNS characters. extra=[...] adds source-specific placeholders; the plural form also deduplicates case-insensitively.
ip_network(cidr) — struct with cidr, version, prefix, network, broadcast, netmask, and a contains(ip) method.
ip_in_network(ip, cidr) — one-shot membership check; False for malformed or mixed-family input.
resolve(host, timeout=10) — A/AAAA lookup returning a list of IP addresses; returns an empty list (never an error) for unresolvable input.

load("net", "ip_address", "network_interface", "normalize_mac",
            "ip_network", "ip_in_network", "resolve",
            "routable_ip", "clean_hostnames", "mac_key")

def main(*args, **kwargs):
    addr = ip_address("192.0.2.5")
    print("ip:", addr, "version:", addr.version)
    print("mac:", normalize_mac("9CB6.D0F1.D204"))

    net10 = ip_network("10.0.0.0/8")
    print("contains:", net10.contains("10.1.2.3"))
    print("in network:", ip_in_network("10.1.2.3", "10.0.0.0/8"))

    for ip in resolve("localhost"):
        print("resolved:", ip, ip.version)

    nic = network_interface(mac="9c-b6-d0-f1-d2-04",
                            ips=["192.0.2.5", "fe80::1%eth0", "[2001:db8::1]:443"])
    print("nic:", nic)

    print("routable:", routable_ip("127.0.0.1"), routable_ip("10.1.2.3:8443"))
    print("hostnames:", clean_hostnames(["web1.acme.local", "localhost", None]))
    print("mac key:", mac_key("00:00:00:00:00:00"))  # None: not an identity

json

load("json", json_encode="encode", json_decode="decode")

def main(*args, **kwargs):
    data = {"name": "runZero", "features": ["scan", "API", "integrations"]}
    encoded = json_encode(data)
    print("encoded:", encoded)
    decoded = json_decode(encoded)
    print("decoded name:", decoded["name"])

The module also provides encode_indent and indent for pretty-printed output.

jsonstream

Stream large JSON documents without materializing them. iter_array(body, path=None) iterates the elements of an array (optionally at a dot-separated path inside the document); iter_lines(body) iterates NDJSON / JSON-lines input. Both accept strings or bytes — pass a response body directly.

load("jsonstream", "iter_array", "iter_lines")

def main(*args, **kwargs):
    big_json = '{"data": {"items": [{"id": 1}, {"id": 2}]}}'
    for item in iter_array(big_json, path="data.items"):
        print("streamed id:", item["id"])

    for line in iter_lines('{"a": 1}\n{"a": 2}'):
        print("ndjson:", line["a"])

csv

read_all(text, delimiter=",", comment="", header=True) returns a list of dicts keyed by the header row (or a list of lists with header=False); read_rows is the header-less form. write_all(rows) and write_dicts(rows, fields=None) serialize back to CSV strings.

load("csv", csv_read="read_all", csv_write="write_dicts")

def main(*args, **kwargs):
    rows = csv_read("id,name\n1,web1\n2,web2\n")
    print("first row:", rows[0]["name"])
    print("as csv:", csv_write(rows))

xml

parse(text) returns an element tree. Elements expose tag, text, tail, attrib, and children, plus the methods find(path), find_all(path), get(name, default=""), and text_all().

load("xml", xml_parse="parse")

def main(*args, **kwargs):
    doc = xml_parse("<devices><device><name>web1</name></device></devices>")
    print("name:", doc.find("device/name").text)
    print("count:", len(doc.find_all("device")))

re

Regular expressions using Go RE2 syntax. match(pattern, string) and search(...) return a struct (.match, .start, .end, .groups, and .named for named groups) or None. Also available: find_all, find_all_groups, sub(pattern, repl, string, count=-1), split, escape, and compile(pattern) for a reusable pattern object.

load("re", re_match="match", re_find_all="find_all", re_sub="sub")

def main(*args, **kwargs):
    matches = re_find_all(r"id=(\d+)", "id=10 id=20")
    print("matches:", matches)
    print("cleaned:", re_sub(r"\s+", " ", "a   b\tc"))
    m = re_match(r"(?P<major>\d+)\.(?P<minor>\d+)", "5.1")
    print("named groups:", m.named)

time

The standard Starlark time module plus a sandbox-aware sleep.

parse_time(s) — parses RFC 3339 strings into a time value with year, month, day, hour, minute, second, unix, and unix_nano fields, plus format(layout) and in_location(name) methods. It raises on input it does not recognize, and a raise from a builtin aborts the whole script — use it for literals, and parse_ts for API-supplied values.
parse_ts(value, default=None, assume_utc=True, clamp_to_now=True, unit="s") — a never-raising timestamp parser for values a vendor API returns: epoch ints, floats, and numeric strings (unit selects s/ms/us/ns) plus a wide set of datetime layouts, with zone-less values read as UTC. Non-positive epochs yield default, and future values are clamped to now, because the platform drops an ImportAsset whose first- or last-seen timestamp is ahead of the clock.
parse_duration(s) — parses "90m", "1h30m", "250ms", and similar into a duration with hours, minutes, seconds, and related fields. Time and duration values support arithmetic.
now() — the current time. from_timestamp(sec) converts a Unix timestamp.
sleep(duration) — accepts a duration string or value, and respects the task deadline.

load("time", "now", "parse_time", "parse_ts", "parse_duration", "sleep")

def main(*args, **kwargs):
    t = parse_time("2026-08-14T15:00:00Z")
    print("unix:", t.unix, "year:", t.year)
    print("formatted:", t.format("2006-01-02"))

    seen = parse_ts("2026-08-14 15:00:00")   # zone-less, read as UTC
    print("api time:", seen)
    print("bad input:", parse_ts("not-a-date"))  # None, never a crash

    d = parse_duration("90m")
    print("minutes:", d.minutes)
    print("90 minutes from now:", now() + d)
    sleep("250ms")

uuid

load("uuid", "new_uuid")

def main(*args, **kwargs):
    print("generated UUID:", new_uuid())

base64, hex, and base32

base64 provides encode/decode plus raw_encode/raw_decode (unpadded), url_encode/url_decode (URL-safe alphabet), and raw_url_encode/raw_url_decode (both); its decode functions return strings. hex provides encode/decode, and base32 provides encode/decode plus unpadded raw_encode/raw_decode; their decode functions return bytes.

load("base64", base64_encode="encode", base64_decode="decode")
load("hex", hex_encode="encode", hex_decode="decode")
load("base32", b32_encode="encode")

def main(*args, **kwargs):
    enc = base64_encode("user:pass")
    print("base64:", enc, "->", base64_decode(enc))
    print("hex:", hex_encode("hi"))
    print("base32:", b32_encode("hi"))

crypto

Hashes, HMAC, AWS request signing, and random values.

Hashes: sha1, sha256, sha512, md5 — accept strings, return hex.
HMAC: hmac_sha1/hmac_sha256/hmac_sha512(key, data, output="hex") and the generic hmac(algorithm, key, data, output="hex"); output may be "hex", "base64", "base64_raw", or "bytes".
sign_v4(method, url, headers, body, access_key, secret_key, region, service, session_token=None, timestamp=None) — returns the AWS Signature V4 headers (Authorization, X-Amz-Date, X-Amz-Content-Sha256, and X-Amz-Security-Token when a session token is given).
random_bytes(n) / random_hex(n) — cryptographically secure random output.

load("crypto", "sha256", "hmac_sha256", "random_hex", "sign_v4")

def main(*args, **kwargs):
    print("sha256:", sha256("test"))
    print("hmac:", hmac_sha256("key", "message"))
    print("random:", random_hex(8))

    headers = sign_v4("GET", "https://ec2.us-east-1.amazonaws.com/?Action=DescribeInstances",
                      {}, "", "AKIDEXAMPLE", "EXAMPLESECRET", "us-east-1", "ec2")
    print("signed headers:", sorted(headers.keys()))

jwt

encode(claims, key, algorithm="HS256", headers=None), decode(token, key, algorithms=None) (verifies the signature), and decode_unverified(token) (returns {"header": ..., "claims": ...} without verification). HS, RS, PS, ES, and EdDSA algorithm families are supported; the none algorithm is rejected.

load("jwt", jwt_encode="encode", jwt_decode="decode", "decode_unverified")

def main(*args, **kwargs):
    token = jwt_encode({"sub": "runzero", "exp": 1893456000}, "example-signing-key")
    print("claims:", jwt_decode(token, "example-signing-key"))
    print("header:", decode_unverified(token)["header"])

gzip

compress(data) and decompress(data) operate on bytes and return bytes. Decompression is bounded at 1 GiB of output and a 500:1 expansion ratio to guard against decompression bombs.

load("gzip", gzip_compress="compress", gzip_decompress="decompress")

def main(*args, **kwargs):
    original = bytes("Hello, runZero!")
    compressed = gzip_compress(original)
    print("compressed length:", len(compressed))
    print("round trip:", gzip_decompress(compressed))

flatten_json

flatten(input, separator="_", root_keys_to_ignore=None) flattens nested structures into a single-level dict. For building customAttributes, prefer to_custom_attributes, which also stringifies values and applies the platform limits.

load("flatten_json", "flatten")

def main(*args, **kwargs):
    flat = flatten({"a": {"b": 1, "c": 2}, "d": 3})
    print("flattened:", flat)   # {"a_b": 1, "a_c": 2, "d": 3}

runzero.progress

Surface progress and log lines from a running task in the runZero console.

report(pct, msg="") — percentage is clamped to 0–100; calls within 250 ms are coalesced; messages are truncated to 256 bytes.
info(msg) / warn(msg) — emit log lines through the task logger.

load("runzero.progress", progress_report="report", progress_info="info",
                         progress_warn="warn")

def main(*args, **kwargs):
    progress_report(0, "starting sync")
    progress_info("fetched page 1")
    progress_warn("retrying after 429")
    progress_report(100, "done")

report_assets, report_asset, and pager (predeclared)

Stream ImportAsset values to runZero while the script runs, instead of returning them all from main. No load() is required. See streaming large imports for the pagination pattern.

report_asset(asset) — report exactly one asset (or None, a no-op), returning 1 or 0. Streaming batches internally, so per-record reporting is the preferred pattern; there is nothing to gain by accumulating a page into a list first.
report_assets(...) — accepts a single asset, several positional assets, or a list/tuple, returning the count.
pager(label="pages", limit=0) — a loop guard for pagination. p.next() is the while condition; reaching the page ceiling (CONFIG maxPages, default 1,000,000) is an error naming the label, never a silent stop. limit= may lower the effective ceiling but not raise it, and p.page is the 1-based current page.
max_pages() — the effective page ceiling.

load("runzero.types", "ImportAsset")

def main(*args, **kwargs):
    a = ImportAsset(id="asset-1")
    b = ImportAsset(id="asset-2")

    report_asset(a)              # one asset; the preferred streaming pattern
    report_assets(a, b)          # several positional assets
    n = report_assets([a, b])    # a list, returning the count
    print("reported", n)

    p = pager(label="devices")
    while p.next():
        print("page", p.page)
        break                    # a real script would fetch until the API runs dry
    return None

socket

Raw TCP, UDP, and TLS connections for simple line- or byte-oriented protocols.

tcp(host, port, timeout=30, tls=None, insecure_skip_verify=False, server_name=None) — tls= accepts a bool or the same dict of TLS overrides as the http module (insecure, server_name, ca_pem, client_cert_pem, client_key_pem, thumbprints), so kwargs.get_http_tls() output can be splatted straight into a socket.
udp(host, port, timeout=30)
tls(host, port, timeout=30, insecure_skip_verify=False, server_name=None, tls=None)

Sockets expose send(data, timeout=None), recv(max=..., timeout=None) (max defaults to 16 MiB, per-call ceiling 200 MiB), recv_exact(n, ...), recv_line(...) and recv_until(delim, ...) (max defaults to 1 MiB), starttls(...) (TCP only, also accepts tls=), set_timeout(seconds), and close(), plus the attributes local_addr, remote_addr, is_tls, closed, and network. Always close() sockets when done.

On an Explorer task, sockets can reach the internal addresses visible to that Explorer, the same as the direct-protocol modules — scope the Explorer and the credential to the intended system. On console-hosted runs, private and internal addresses are blocked for every module, sockets included.

load("socket", "tls")

def main(*args, **kwargs):
    host = kwargs.get("host", "example.com")
    sock = tls(host, 443, timeout=10)
    print("connected to:", sock.remote_addr, "tls:", sock.is_tls)
    sock.close()

Direct-protocol modules

For sources without a REST API, these modules open authenticated connections from the Explorer the task runs on, and can reach internal addresses visible to that Explorer. They return session objects — always close() them when done: a script can hold at most 256 open connections and sessions at once, and network reads across every module share a 4 GiB per-run budget. Scripts using these modules should declare "validationMode": "compile" since --validate cannot dial a real endpoint.

The examples below read their targets from --kwargs so they can run against a test system:

runzero script --filename example.star --kwargs host=192.0.2.10 --kwargs username=svc --kwargs password=...

runzero.ssh

dial(host, username, password=None, private_key=None, private_key_passphrase=None, host_key=None, host_keys=None, insecure_ignore_host_key=False, port=22, timeout=30). There is no trust-on-first-use: pin the expected host key(s) — in authorized_keys or PEM form, for example from ssh-keyscan — or explicitly pass insecure_ignore_host_key=True; combining pins with the insecure flag is an error. Password auth also answers keyboard-interactive prompts, and timeouts clamp at 600 seconds.

Sessions provide:

run(command, stdin=None, timeout=0) — returns (stdout, stderr, exit_code); a timeout of 0 uses the session timeout. stdin is capped at 1 MiB and each output stream at 16 MiB; run_command is an alias.
stream(command, stdin=None, timeout=0) — starts a long-lived remote command and returns a stream with send(data), recv(max=..., timeout=None) (max defaults to 1 MiB; b"" on EOF), close_stdin(), close(), and the attributes exit_code (None until exit), stderr, and closed. Output is consumed incrementally with no per-stream cap, so a stream can process more than run can return.
open_unix(path) / open_tcp(host, port) — port forwarding to a remote Unix socket or a TCP address reachable from the remote host, returning a full socket object. This is how a script reaches a service that only listens locally on the remote system, such as a management socket or a loopback-only listener.

Sessions also expose host, port, username, and close().

load("runzero.ssh", ssh_dial="dial")

def main(*args, **kwargs):
    host = kwargs.get("host")
    if not host:
        print("pass --kwargs host=... --kwargs username=... --kwargs password=... to run")
        return None
    session = ssh_dial(host=host, username=kwargs["username"],
                       password=kwargs.get("password"),
                       insecure_ignore_host_key=True)
    stdout, stderr, code = session.run("uname -a")
    session.close()
    print("exit:", code, "output:", stdout)

runzero.smb

dial(host, username, password="", domain="", nt_hash=None, port=445, timeout=30) — DOMAIN\user shorthand is accepted in username, and timeouts clamp at 600 seconds. Sessions provide list_shares() and mount(share); a mounted share offers read(path, limit=0) (single reads capped at 64 MiB), list(path="/"), stat(path), exists(path), unmount(), close(), and a name attribute.

Access refusals are values, not errors: mount() and list() return None on access denied — distinct from an empty list — so a script can skip a share it cannot read and keep going. read() and stat() still raise; exists() is stat’s soft counterpart.

load("runzero.smb", smb_dial="dial")

def main(*args, **kwargs):
    host = kwargs.get("host")
    if not host:
        print("pass --kwargs host=... --kwargs username=... --kwargs password=... to run")
        return None
    session = smb_dial(host=host, username=kwargs["username"], password=kwargs["password"])
    print("shares:", session.list_shares())
    session.close()

runzero.winrm

dial(host, username, password, port=0, https=False, insecure_skip_verify=False, ca_cert=None, auth="ntlm", timeout=60) (port 0 selects the default, 5985 or 5986 with https=True). Sessions provide run(command) (alias run_command), run_powershell(script) (alias run_ps), and wql(query, namespace="root/cimv2"), plus the attributes host, port, and https.

load("runzero.winrm", winrm_dial="dial")

def main(*args, **kwargs):
    host = kwargs.get("host")
    if not host:
        print("pass --kwargs host=... --kwargs username=... --kwargs password=... to run")
        return None
    session = winrm_dial(host, kwargs["username"], kwargs["password"])
    stdout, stderr, code = session.run_powershell("Get-ComputerInfo | Select-Object OsName")
    session.close()
    print(stdout)

runzero.wmi

dial(host, username, password, transport="tcp", namespace="//./root/cimv2", target_name="", timeout=60, port=0, smb_port=445) (port 0 selects the default, 135). Sessions provide query(query, limit=0, page=100) returning a list of dicts.

load("runzero.wmi", wmi_dial="dial")

def main(*args, **kwargs):
    host = kwargs.get("host")
    if not host:
        print("pass --kwargs host=... --kwargs username=... --kwargs password=... to run")
        return None
    session = wmi_dial(host, kwargs["username"], kwargs["password"])
    rows = session.query("SELECT Caption, Version FROM Win32_OperatingSystem")
    session.close()
    print(rows)

runzero.sql

connect(driver, dsn, timeout=30, max_open_conns=4, max_idle_conns=4, conn_max_lifetime=0) with drivers postgres, mysql, and mssql (and common aliases). Sessions provide query(query, params=None, limit=100000, timeout=0) returning a list of dicts (row limit up to 1,000,000), and exec(query, params=None, timeout=0) returning a dict with rows_affected and last_insert_id. A timeout of 0 uses the session timeout; timeouts clamp at 600 seconds. DSNs are restricted to network connections: file-reading options (such as sslkey, passfile, or allowAllFiles) and Unix-socket or named-pipe transports are rejected.

load("runzero.sql", sql_connect="connect")

def main(*args, **kwargs):
    dsn = kwargs.get("dsn")
    if not dsn:
        print("pass --kwargs dsn=postgres://user:pass@host:5432/db to run")
        return None
    db = sql_connect("postgres", dsn)
    rows = db.query("SELECT version()")
    db.close()
    print(rows)

Updated August 24, 2026

← Custom Integration Scripts

Dragos →