Run Scans

Python scripting gives you the same scanning capabilities as the CLI. Configure a probe, decide where targets come from, and iterate over the asynchronous stream of results.

Configure a probe

import pyrmap

icmp = pyrmap.ProbeConfig.icmp()
tcp_443 = pyrmap.ProbeConfig.tcp(port=443)
dns = pyrmap.ProbeConfig.dns(domain="example.com", query_type="AAAA")

Each helper mirrors the corresponding CLI subcommand:

• icmp()
• tcp(port=None)
• udp(port=None)
• tcp_synack(port=None)
• dns(port=None, domain=None, query_type=None)
• ntp(port=None)

Launch a scan from a file

import asyncio
import pyrmap

async def scan_icmp():
    scanner = pyrmap.Scanner(
        pyrmap.ProbeConfig.icmp(),
        pps=1_500,
        timeout=6,
        interface="en0",
    )

    stream = await scanner.scan_from_file(
        "targets.txt",
        unique=True,
        budget=None,
    )

    async for result in stream:
        print(result["target"], result.get("alias"))

asyncio.run(scan_icmp())

• Scanner accepts defaults for interface, packets-per-second, timeout, worker count, and an optional remote address.
• scan_from_file streams a newline-delimited file of IPv6 addresses. unique=True deduplicates targets before probing.
• Each iteration yields a dictionary of tags (for example target, probe, alias, port, latency_ms, dns_response_code).

Scan a Python list

async def scan_small_batch(targets):
    scanner = pyrmap.Scanner(pyrmap.ProbeConfig.tcp(port=22))
    stream = await scanner.scan(targets, unique=False)

    replies = []
    async for result in stream:
        replies.append(result)
    return replies

Scanner.scan creates a temporary target file behind the scenes so you can pass lists or other iterable data.

Remote workers

Start a worker on the remote host with rmap serve (see the Remote Scans usage page). Then pass the address to the scanner or to the individual call:

scanner = pyrmap.Scanner(
    pyrmap.ProbeConfig.icmp(),
    remote="worker01.example.net:50051",
)

stream = await scanner.scan_from_file("targets.txt")

Override the destination per call with scanner.scan_from_file(..., remote="backup:50051").

Quick helper: scan_local_stream

When you just want a stream without instantiating Scanner, call the module-level function:

stream = await pyrmap.scan_local_stream(
    targets_path="targets.txt",
    interface=None,
    probe_config=pyrmap.ProbeConfig.ntp(port=123),
    pps=800,
    timeout=6,
    unique=True,
    budget=None,
    workers=64,
)

async for result in stream:
    handle(result)

scan_local_stream always runs locally; set up a full Scanner when you need remote execution.

Handling timeouts and errors

• The async iterator skips internal errors automatically. If the generator encounters a fatal error, it raises RuntimeError on the next iteration.
• Increase timeout for probes that expect multi-packet exchanges (DNS or NTP).
• Combine unique=True with budget=<N> to rate-limit opportunistic scans or to stop after a fixed number of successful probes.