discover_and_query_printers.py

import argparse
import ipaddress
import socket
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Optional, Tuple, Dict

from pysnmp.hlapi import (
    SnmpEngine, CommunityData, UdpTransportTarget, ContextData,
    ObjectType, ObjectIdentity, getCmd, nextCmd
)

# ---- Printer-MIB OIDs ----
OID_PRINTER_NAME = "1.3.6.1.2.1.43.5.1.1.16.1"
OID_SUP_DESC     = "1.3.6.1.2.1.43.11.1.1.6.1"
OID_SUP_LEVEL    = "1.3.6.1.2.1.43.11.1.1.9.1"
OID_SUP_MAX      = "1.3.6.1.2.1.43.11.1.1.8.1"
UNKNOWN_LEVEL_VALUES = {-3, -2}

PRINTER_PORTS = [
    9100,  # JetDirect / RAW printing (very common)
    631,   # IPP
    515,   # LPD
    80,    # Web UI
    443,   # Web UI (HTTPS)
]

def tcp_port_open(ip: str, port: int, timeout: float) -> bool:
    try:
        with socket.create_connection((ip, port), timeout=timeout):
            return True
    except Exception:
        return False

def looks_like_printer(ip: str, timeout: float) -> List[int]:
    open_ports = []
    for p in PRINTER_PORTS:
        if tcp_port_open(ip, p, timeout):
            open_ports.append(p)
    return open_ports

def snmp_get(ip: str, community: str, oid: str, timeout: float, retries: int) -> Optional[str]:
    try:
        it = getCmd(
            SnmpEngine(),
            CommunityData(community, mpModel=1),
            UdpTransportTarget((ip, 161), timeout=timeout, retries=retries),
            ContextData(),
            ObjectType(ObjectIdentity(oid)),
        )
        errInd, errStat, _, varBinds = next(it)
        if errInd or errStat:
            return None
        for _, val in varBinds:
            return str(val)
        return None
    except Exception:
        return None

def snmp_walk(ip: str, community: str, base_oid: str, timeout: float, retries: int, max_rows: int = 256) -> List[Tuple[str, str]]:
    out = []
    try:
        for (errInd, errStat, _, varBinds) in nextCmd(
            SnmpEngine(),
            CommunityData(community, mpModel=1),
            UdpTransportTarget((ip, 161), timeout=timeout, retries=retries),
            ContextData(),
            ObjectType(ObjectIdentity(base_oid)),
            lexicographicMode=False,
        ):
            if errInd or errStat:
                break
            for oid, val in varBinds:
                oid_s = str(oid)
                if not oid_s.startswith(base_oid + ".") and oid_s != base_oid:
                    return out
                out.append((oid_s, str(val)))
                if len(out) >= max_rows:
                    return out
    except Exception:
        pass
    return out

def parse_index(oid: str, base: str) -> Optional[int]:
    if not oid.startswith(base + "."):
        return None
    tail = oid[len(base) + 1:]
    try:
        return int(tail)
    except ValueError:
        return None

def query_supplies(ip: str, community: str, timeout: float, retries: int):
    name = snmp_get(ip, community, OID_PRINTER_NAME, timeout, retries)
    if not name:
        return None

    desc_rows = snmp_walk(ip, community, OID_SUP_DESC, timeout, retries)
    lvl_rows  = snmp_walk(ip, community, OID_SUP_LEVEL, timeout, retries)
    max_rows  = snmp_walk(ip, community, OID_SUP_MAX, timeout, retries)

    desc_by_i: Dict[int, str] = {}
    lvl_by_i: Dict[int, int] = {}
    max_by_i: Dict[int, int] = {}

    for oid, val in desc_rows:
        i = parse_index(oid, OID_SUP_DESC)
        if i is not None:
            desc_by_i[i] = val

    for oid, val in lvl_rows:
        i = parse_index(oid, OID_SUP_LEVEL)
        if i is not None:
            try:
                lvl_by_i[i] = int(val)
            except ValueError:
                pass

    for oid, val in max_rows:
        i = parse_index(oid, OID_SUP_MAX)
        if i is not None:
            try:
                max_by_i[i] = int(val)
            except ValueError:
                pass

    supplies = []
    for i in sorted(set(desc_by_i) | set(lvl_by_i) | set(max_by_i)):
        desc = desc_by_i.get(i, f"Supply {i}")
        level_raw = lvl_by_i.get(i)
        max_raw = max_by_i.get(i)

        level = None if (level_raw is None or level_raw in UNKNOWN_LEVEL_VALUES) else level_raw
        max_cap = None if (max_raw is None or max_raw <= 0) else max_raw

        if level is not None and max_cap is not None:
            pct = max(0.0, min(100.0, (level / max_cap) * 100.0))
            supplies.append(f"{desc}: {pct:.0f}% ({level}/{max_cap})")
        elif level is not None:
            supplies.append(f"{desc}: level={level}")
        else:
            supplies.append(f"{desc}: unknown")

    return name, supplies

def main():
    ap = argparse.ArgumentParser(description="Discover printer-like devices by common ports, then query SNMP supplies.")
    ap.add_argument("--cidr", required=True, help="CIDR to scan, e.g. 192.168.50.0/24")
    ap.add_argument("--community", default="public", help="SNMP community (default: public)")
    ap.add_argument("--tcp-timeout", type=float, default=0.25, help="TCP connect timeout (default: 0.25)")
    ap.add_argument("--snmp-timeout", type=float, default=0.9, help="SNMP timeout (default: 0.9)")
    ap.add_argument("--snmp-retries", type=int, default=0, help="SNMP retries (default: 0)")
    ap.add_argument("--workers", type=int, default=256, help="Workers (default: 256)")
    args = ap.parse_args()

    net = ipaddress.ip_network(args.cidr, strict=False)
    targets = [str(ip) for ip in net.hosts()]

    # Step 1: find candidates by open ports
    candidates = []
    with ThreadPoolExecutor(max_workers=args.workers) as ex:
        futs = {ex.submit(looks_like_printer, ip, args.tcp_timeout): ip for ip in targets}
        for fut in as_completed(futs):
            ip = futs[fut]
            ports = fut.result()
            if ports:
                candidates.append((ip, ports))

    candidates.sort(key=lambda x: x[0])

    if not candidates:
        print("No hosts with common printer ports found on that subnet.")
        print("This strongly suggests VLAN/guest isolation or the printer is on a different subnet.")
        return

    print(f"Found {len(candidates)} printer-like hosts (by ports). Querying SNMP supplies...\n")

    # Step 2: try SNMP supplies on those candidates
    any_snmp = False
    for ip, ports in candidates:
        res = query_supplies(ip, args.community, args.snmp_timeout, args.snmp_retries)
        print("=" * 72)
        print(f"{ip}  open_ports={ports}")
        if not res:
            print("SNMP supplies: not available (SNMP disabled/blocked or community mismatch).")
            continue
        any_snmp = True
        name, supplies = res
        print(f"SNMP name: {name}")
        if supplies:
            for s in supplies:
                print(f"  - {s}")
        else:
            print("Supplies: (none reported)")

    if not any_snmp:
        print("\nNone of the printer-like hosts responded with Printer-MIB supplies over SNMP.")
        print("If you can open the printer web UI (http://<ip>), enable SNMP v1/v2c in its settings.")

if __name__ == "__main__":
    main()