port_scan

Instructions

Implementation Reference

• src/sounding/server.py:246-290 (handler)

  The `port_scan` tool handler function, decorated with @mcp.tool(). It validates the host, defaults to common ports, limits to 100 ports, enforces rate-limiting (1 scan/sec), then concurrently scans each port via asyncio.open_connection with a 2s timeout. Returns a dict with host, ports_scanned, open_ports, and all_results.

  async def port_scan(host: str, ports: Optional[list[int]] = None) -> dict:
      """Scan common TCP ports on a host.
  
      Safety limits: max 100 ports, rate-limited to one scan per second.
      """
      global _last_scan_time
  
      host = validate_host(host)
      if ports is None:
          ports = DEFAULT_PORTS
      if len(ports) > 100:
          raise ValueError("port_scan is limited to 100 ports per call")
      for p in ports:
          if not validate_port(p):
              raise ValueError(f"Invalid port in list: {p}")
  
      # Rate limiting — one scan per second.
      async with _scan_lock:
          now = time.monotonic()
          wait = 1.0 - (now - _last_scan_time)
          if wait > 0:
              await asyncio.sleep(wait)
          _last_scan_time = time.monotonic()
  
      async def _check(p: int) -> dict:
          try:
              reader, writer = await asyncio.wait_for(
                  asyncio.open_connection(host, p),
                  timeout=2,
              )
              writer.close()
              await writer.wait_closed()
              return {"port": p, "state": "open"}
          except (OSError, asyncio.TimeoutError):
              return {"port": p, "state": "closed"}
  
      results = await asyncio.gather(*[_check(p) for p in ports])
      open_ports = [r for r in results if r["state"] == "open"]
  
      return {
          "host": host,
          "ports_scanned": len(ports),
          "open_ports": open_ports,
          "all_results": results,
      }

• src/sounding/server.py:245-246 (registration)

  The `@mcp.tool()` decorator that registers `port_scan` as an MCP tool on the FastMCP instance.

  @mcp.tool()
  async def port_scan(host: str, ports: Optional[list[int]] = None) -> dict:

• src/sounding/server.py:35-43 (helper)

  Rate-limit state variables (`_last_scan_time`, `_scan_lock`) and the default port list (`DEFAULT_PORTS`) used by port_scan.

  # Rate-limit state for port_scan.
  _last_scan_time: float = 0.0
  _scan_lock = asyncio.Lock()
  
  # Default common ports for port_scan.
  DEFAULT_PORTS = [
      21, 22, 23, 25, 53, 80, 110, 111, 135, 139, 143, 443, 445,
      993, 995, 1723, 3306, 3389, 5432, 5900, 8080, 8443,
  ]

• src/sounding/server.py:246-246 (schema)

  Type signature for port_scan: takes `host: str` and optional `ports: Optional[list[int]]`, returns `dict`.

  async def port_scan(host: str, ports: Optional[list[int]] = None) -> dict: