Skip to content

Latest commit

 

History

History
271 lines (211 loc) · 11.3 KB

File metadata and controls

271 lines (211 loc) · 11.3 KB

UDP Scanner

A threaded UDP port scanner that infers open/closed/filtered state from replies and ICMP behaviour, using socket, argparse, and optional service-specific probes.

Project Overview

UDP is connectionless, so there is no handshake to confirm a port is open. Instead we send a datagram and interpret what comes back: a UDP reply means open, an ICMP "port unreachable" (surfaced as ConnectionRefusedError) means closed, and silence means open|filtered. This project sends probes to each port concurrently and classifies the result, mirroring how nmap -sU reasons about UDP.

[!warning] Authorized use only UDP scanning is noisy and can disrupt services. Scan only hosts you own or are authorized to test. Examples target 127.0.0.1.

Objectives

  • Send UDP probes and interpret the responses — and the silences.
  • Explain why UDP scanning is inherently unreliable compared with TCP.
  • Use ICMP port-unreachable messages to infer closed ports.
  • Apply protocol-specific payloads to elicit replies.
  • Report open, closed, and open-filtered states honestly.

Requirements

Item Detail
Python 3.9 or newer
Dependencies socket, argparse (standard library)
Privileges None for basic probing; root needed to read ICMP directly
Target 127.0.0.1 or a lab VM you control
Note Results are less definitive than a TCP scan — this is inherent to UDP

Architecture

  • CLIargparse for host, port range, threads, timeout, and retries.
  • Prober — sends a service-appropriate payload (e.g. a DNS query to 53) or an empty datagram.
  • Classifier — maps recvfrom() data → open, ConnectionRefusedError → closed, timeout → open|filtered.
  • SchedulerThreadPoolExecutor runs probes concurrently.
  • Reporter — prints per-port state; open/filtered ports are the interesting ones.
for each port ─▶ sendto(probe) ─▶ recvfrom() ?
                    ├─ data        → open
                    ├─ ICMP unreach→ closed
                    └─ timeout     → open|filtered

Directory Structure

udp-scanner/
├── udp_scanner.py      # CLI + scan logic
└── payloads.py         # (optional) per-port probe datagrams

Implementation Plan

  1. Send a probe. Create a SOCK_DGRAM socket, set a timeout, and sendto() an empty payload.
  2. Read the reply. A response means the port is open. Handle socket.timeout as open|filtered.
  3. Detect closed ports. On Linux a ConnectionRefusedError on a connected UDP socket indicates an ICMP port-unreachable, meaning closed.
  4. Add protocol payloads. Many services ignore empty datagrams. Send a real DNS query to 53, an NTP request to 123, an SNMP get to 161.
  5. Retry. UDP is lossy. Probe each port two or three times before concluding.
  6. Rate-limit. ICMP unreachable messages are rate-limited by the kernel; scanning too fast produces false "open|filtered" results.
  7. Report three states honestly rather than collapsing to open/closed.

Complete Example Implementation

#!/usr/bin/env python3
"""udp_scanner.py - Threaded UDP port scanner (educational).

UDP results are inherently uncertain; treat 'open|filtered' as 'unknown'.
Only scan systems you own or are authorized to test.
"""
import argparse
import socket
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

# A well-formed DNS standard query for example.com helps ports like 53 answer.
DNS_PROBE = bytes.fromhex(
    "abcd0100000100000000000007" "6578616d706c6503636f6d0000010001"
)
PROBES = {53: DNS_PROBE}


def scan_udp(host: str, port: int, timeout: float, retries: int) -> tuple[int, str]:
    """Return (port, state) where state is open, closed, or open|filtered."""
    payload = PROBES.get(port, b"\x00")
    for _ in range(retries):
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            sock.settimeout(timeout)
            try:
                sock.sendto(payload, (host, port))
                sock.recvfrom(1024)      # any reply => service answered
                return port, "open"
            except socket.timeout:
                continue                 # retry; could still be filtered
            except ConnectionRefusedError:
                return port, "closed"    # ICMP port unreachable
            except OSError as exc:
                return port, f"error ({exc})"
    return port, "open|filtered"


def main() -> int:
    parser = argparse.ArgumentParser(description="Threaded UDP port scanner.")
    parser.add_argument("host", help="target host (must be authorized)")
    parser.add_argument("-p", "--ports", default="53,67,123,161",
                        help="comma-separated ports (default: 53,67,123,161)")
    parser.add_argument("-t", "--threads", type=int, default=50,
                        help="concurrent workers (default: 50)")
    parser.add_argument("--timeout", type=float, default=1.0,
                        help="per-probe timeout seconds (default: 1.0)")
    parser.add_argument("--retries", type=int, default=2,
                        help="probe retries before giving up (default: 2)")
    args = parser.parse_args()

    try:
        target_ip = socket.gethostbyname(args.host)
    except socket.gaierror:
        print(f"[!] Could not resolve {args.host}", file=sys.stderr)
        return 1

    ports = [int(p) for p in args.ports.split(",") if p.strip()]
    print(f"[*] UDP scanning {args.host} ({target_ip}) - {len(ports)} port(s)")
    with ThreadPoolExecutor(max_workers=args.threads) as pool:
        futures = [pool.submit(scan_udp, target_ip, p, args.timeout, args.retries)
                   for p in ports]
        for future in as_completed(futures):
            port, state = future.result()
            if state != "closed":
                print(f"[+] {port:>5}/udp {state}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
python udp_scanner.py 127.0.0.1 -p 53,123,161 --timeout 1

Output

[*] UDP scanning 127.0.0.1 (127.0.0.1) - 3 port(s)
[+]    53/udp open
[+]   123/udp open|filtered
[+]   161/udp open|filtered

Usage

# Scan common UDP ports on localhost
python3 udp_scanner.py 127.0.0.1 --ports 53,123,161

# Range with retries and a longer timeout
python3 udp_scanner.py 127.0.0.1 --start 50 --end 200 --retries 3 --timeout 2

# JSON output
python3 udp_scanner.py 127.0.0.1 --ports 53,123 --json

Example Output

$ python3 udp_scanner.py 127.0.0.1 --ports 53,123,161

PORT      STATE           NOTE
53/udp    open            responded to DNS query
123/udp   closed          ICMP port unreachable
161/udp   open|filtered   no response after 3 retries

[*] 1 open, 1 closed, 1 open|filtered

Error Handling

Condition Exception Response
No response within the timeout socket.timeout Report `open
ICMP port unreachable received ConnectionRefusedError Report closed
ICMP blocked by a firewall (silence) Falls back to `open
Host unresolvable socket.gaierror Report and exit 1
Sending to a broadcast address PermissionError Report that broadcast requires SO_BROADCAST; refuse by default
Kernel ICMP rate limiting (silence) Slow the scan; document that fast scans inflate `open

Security Considerations

[!warning] Authorized use only Only scan hosts you own or are explicitly authorized to test.

  • UDP results are genuinely ambiguous. open|filtered means "no evidence either way". Reporting it as open overstates your findings, and reporting it as closed hides real services. Say what you actually observed.
  • UDP is spoofable and amplification-prone. Services like DNS, NTP, and SNMP are classic reflection-amplification vectors. Never point a probe generator at a third party, and never craft packets with a forged source address.
  • Probing can trigger service behaviour. An SNMP or NTP probe elicits a real response from a real service; on fragile devices repeated probes can disrupt them.
  • Kernel ICMP rate limiting will mislead you. Scan slowly and retry, or you will report open ports that are simply closed ports whose ICMP replies were suppressed.
  • Prefer authenticated inventory data where you have it — UDP scanning is a last resort, not a first choice.

Testing

# Start a UDP echo listener and confirm the scanner sees it
python3 -c "
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(('127.0.0.1', 9998))
data, addr = s.recvfrom(1024); s.sendto(b'pong', addr); s.close()
" &
sleep 0.5
python3 udp_scanner.py 127.0.0.1 --ports 9998 | grep -q open && echo PASS
import socket
from unittest.mock import MagicMock, patch

import udp_scanner


def test_response_means_open():
    fake = MagicMock()
    fake.recvfrom.return_value = (b"pong", ("127.0.0.1", 9998))
    with patch("socket.socket") as mock_socket:
        mock_socket.return_value.__enter__.return_value = fake
        assert udp_scanner.probe("127.0.0.1", 9998) == "open"


def test_timeout_means_open_filtered():
    fake = MagicMock()
    fake.recvfrom.side_effect = socket.timeout
    with patch("socket.socket") as mock_socket:
        mock_socket.return_value.__enter__.return_value = fake
        assert udp_scanner.probe("127.0.0.1", 9999) == "open|filtered"

Cover all three states plus an unresolvable host.

Extension Ideas

  • More protocol probes — add SNMP (161), NTP (123), NetBIOS (137) payloads so silent-but-open services answer.
  • ICMP rate-limit awareness — Linux throttles port-unreachable messages; add pacing so "closed" isn't misread as "filtered".
  • Raw sockets — read ICMP directly (needs root) to distinguish filtered from open more reliably.
  • Merge with TCP — combine with [[Port-Scanner]] for a unified report.
  • asyncio datagram endpoints — scale to large ranges with loop.create_datagram_endpoint().

Troubleshooting

Symptom Likely cause Fix
Every port reports open|filtered ICMP blocked, or scanning too fast Slow down, increase retries, and confirm ICMP is permitted
Known-open port reports no response Service ignores empty datagrams Send a protocol-appropriate payload
Scan is extremely slow Long timeout multiplied by retries and port count Reduce the range; UDP scanning is inherently slow
Results differ between runs UDP is lossy and ICMP is rate-limited Retry and take the strongest signal
No ConnectionRefusedError ever raised Not using a connected UDP socket Call connect() before send() to receive ICMP errors
Differs from nmap -sU Nmap uses richer payloads and adaptive timing Expected — this is a simpler implementation

References

Related

  • [[Port-Scanner]]
  • [[Banner-Grabber]]
  • [[DNS-Enumeration]]
  • [[Socket-Programming-for-Networking]]
  • [[Argparse-Module]]
  • [[Mini-Projects/Readme|Mini-Projects]] — module index
  • [[Readme|Python for Security Professionals]] — course home