subnet_scan

Instructions

Tool call template

Replace each placeholder with a value matching the input schema. Only required parameters are included; add optional parameters as needed.

{
  "id": 1,
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "arguments": {
      "subnet": "<subnet>"
    },
    "name": "subnet_scan"
  }
}

Implementation Reference

• src/sounding/server.py:452-497 (handler)

  The subnet_scan MCP tool handler: discovers live hosts on a local subnet by probing TCP ports 22, 80, and 443. Validates subnet via validate_subnet, enforces /20 size cap, and uses asyncio semaphore (50) for concurrency-limited probing.

  async def subnet_scan(subnet: str) -> dict:
      """Discover live hosts on a local subnet by probing common ports.
  
      Only allows RFC 1918 private subnets for safety.
      Probes TCP ports 22, 80, and 443 on each host.
      """
      subnet = validate_subnet(subnet)
      network = ipaddress.IPv4Network(subnet, strict=False)
  
      # Safety: cap at /20 (4096 hosts).
      if network.num_addresses > 4096:
          raise ValueError("Subnet too large — maximum /20 (4096 addresses)")
  
      probe_ports = [22, 80, 443]
  
      async def _probe(ip: str) -> dict | None:
          for port in probe_ports:
              try:
                  reader, writer = await asyncio.wait_for(
                      asyncio.open_connection(ip, port),
                      timeout=1,
                  )
                  writer.close()
                  await writer.wait_closed()
                  return {"ip": ip, "open_port": port}
              except (OSError, asyncio.TimeoutError):
                  continue
          return None
  
      # Run probes with concurrency limit.
      sem = asyncio.Semaphore(50)
  
      async def _limited_probe(ip: str) -> dict | None:
          async with sem:
              return await _probe(ip)
  
      hosts = [str(ip) for ip in network.hosts()]
      results = await asyncio.gather(*[_limited_probe(h) for h in hosts])
      found = [r for r in results if r is not None]
  
      return {
          "subnet": subnet,
          "hosts_scanned": len(hosts),
          "hosts_found": len(found),
          "hosts": found,
      }

• src/sounding/validators.py:190-211 (schema)

  The validate_subnet validator: ensures the subnet is a valid IPv4 CIDR and belongs to RFC 1918 private ranges (10.0.0.0/8, 172.16.0.0/12, 192.168.x.x/16). Raises ValueError for public or malformed subnets.

  def validate_subnet(subnet: str) -> str:
      """Validate a CIDR subnet — only RFC 1918 private ranges permitted.
  
      Returns the cleaned subnet string.
      Raises ``ValueError`` for public or malformed subnets.
      """
      subnet = subnet.strip()
      if not subnet:
          raise ValueError("Subnet must not be empty")
  
      try:
          network = ipaddress.IPv4Network(subnet, strict=False)
      except (ipaddress.AddressValueError, ValueError) as exc:
          raise ValueError(f"Invalid subnet: {subnet!r} — {exc}") from exc
  
      if not any(network.subnet_of(priv) for priv in _PRIVATE_NETWORKS):
          raise ValueError(
              f"Subnet {subnet} is not within RFC 1918 private ranges. "
              "Only 10.0.0.0/8, 172.16.0.0/12, and 192.168.x.x/16 are allowed."
          )
  
      return subnet

• src/sounding/server.py:451-452 (registration)

  The @mcp.tool() decorator registers subnet_scan as an MCP tool with FastMCP.

  @mcp.tool()
  async def subnet_scan(subnet: str) -> dict:

• src/sounding/validators.py:30-34 (helper)

  Private network definitions used by validate_subnet to restrict subnet_scan to RFC 1918 ranges only.

  _PRIVATE_NETWORKS = [
      ipaddress.IPv4Network("10.0.0.0/8"),
      ipaddress.IPv4Network("172.16.0.0/12"),
      ipaddress.IPv4Network(f"192.168.{0}.0/16"),
  ]