analyze_icmp_packets

Instructions

Implementation Reference

• src/mcpcap/modules/icmp.py:20-42 (handler)

  The main handler function for the 'analyze_icmp_packets' tool. Delegates to BaseModule.analyze_packets() which handles both remote URLs and local files, then calls _analyze_protocol_file().

  def analyze_icmp_packets(self, pcap_file: str) -> dict[str, Any]:
      """
      Analyze ICMP packets from a PCAP file and return comprehensive analysis results.
  
      FILE UPLOAD LIMITATION: This MCP tool cannot process files uploaded through
      Claude's web interface. Files must be accessible via URL or local file path.
  
      SUPPORTED INPUT FORMATS:
      - Remote files: "https://example.com/capture.pcap"
      - Local files: "/absolute/path/to/capture.pcap"
  
      UNSUPPORTED:
      - Files uploaded through Claude's file upload feature
      - Base64 file content
      - Relative file paths
  
      Args:
          pcap_file: HTTP URL or absolute local file path to PCAP file
  
      Returns:
          A structured dictionary containing ICMP packet analysis results
      """
      return self.analyze_packets(pcap_file)

• src/mcpcap/modules/icmp.py:44-94 (handler)

  Core analysis logic: reads PCAP with Scapy, filters ICMP packets, applies max_packets limit, analyzes each packet via _analyze_icmp_packet(), generates statistics, and returns structured results.

  def _analyze_protocol_file(self, pcap_file: str) -> dict[str, Any]:
      """Perform the actual ICMP packet analysis on a local PCAP file."""
      try:
          packets = rdpcap(pcap_file)
          icmp_packets = [pkt for pkt in packets if pkt.haslayer(ICMP)]
  
          if not icmp_packets:
              return {
                  "file": pcap_file,
                  "total_packets": len(packets),
                  "icmp_packets_found": 0,
                  "message": "No ICMP packets found in this capture",
              }
  
          # Apply max_packets limit if specified
          packets_to_analyze = icmp_packets
          limited = False
          if self.config.max_packets and len(icmp_packets) > self.config.max_packets:
              packets_to_analyze = icmp_packets[: self.config.max_packets]
              limited = True
  
          packet_details = []
          for i, pkt in enumerate(packets_to_analyze, 1):
              packet_info = self._analyze_icmp_packet(pkt, i)
              packet_details.append(packet_info)
  
          # Generate statistics
          stats = self._generate_statistics(packet_details)
  
          result = {
              "file": pcap_file,
              "analysis_timestamp": datetime.now().isoformat(),
              "total_packets": len(packets),
              "icmp_packets_found": len(icmp_packets),
              "icmp_packets_analyzed": len(packet_details),
              "statistics": stats,
              "packets": packet_details,
          }
  
          if limited:
              result["note"] = (
                  f"Analysis limited to first {self.config.max_packets} ICMP packets due to --max-packets setting"
              )
  
          return result
  
      except Exception as e:
          return {
              "error": f"Error reading PCAP file '{pcap_file}': {str(e)}",
              "file": pcap_file,
          }

• src/mcpcap/modules/icmp.py:96-190 (handler)

  Helper that analyzes a single ICMP packet: extracts IP src/dst/version/TTL, ICMP type/code/type_name/ID/sequence/checksum, and code descriptions for Destination Unreachable and Time Exceeded types.

  def _analyze_icmp_packet(self, packet, packet_num: int) -> dict[str, Any]:
      """Analyze a single ICMP packet and extract relevant information."""
      info = {
          "packet_number": packet_num,
          "timestamp": datetime.fromtimestamp(float(packet.time)).isoformat(),
      }
  
      # Basic IP information
      if packet.haslayer(IP):
          ip_layer = packet[IP]
          info.update(
              {
                  "src_ip": ip_layer.src,
                  "dst_ip": ip_layer.dst,
                  "ip_version": 4,
                  "ttl": ip_layer.ttl,
                  "packet_size": len(packet),
              }
          )
      elif packet.haslayer(IPv6):
          ipv6_layer = packet[IPv6]
          info.update(
              {
                  "src_ip": ipv6_layer.src,
                  "dst_ip": ipv6_layer.dst,
                  "ip_version": 6,
                  "hop_limit": ipv6_layer.hlim,
                  "packet_size": len(packet),
              }
          )
  
      # ICMP information
      if packet.haslayer(ICMP):
          icmp_layer = packet[ICMP]
  
          # Map ICMP types to human-readable names
          icmp_types = {
              0: "Echo Reply",
              3: "Destination Unreachable",
              4: "Source Quench",
              5: "Redirect",
              8: "Echo Request",
              11: "Time Exceeded",
              12: "Parameter Problem",
              13: "Timestamp Request",
              14: "Timestamp Reply",
              15: "Information Request",
              16: "Information Reply",
          }
  
          # Map destination unreachable codes
          dest_unreach_codes = {
              0: "Network Unreachable",
              1: "Host Unreachable",
              2: "Protocol Unreachable",
              3: "Port Unreachable",
              4: "Fragmentation Required",
              5: "Source Route Failed",
          }
  
          # Map time exceeded codes
          time_exceeded_codes = {
              0: "TTL Exceeded in Transit",
              1: "Fragment Reassembly Time Exceeded",
          }
  
          icmp_type = icmp_layer.type
          icmp_code = icmp_layer.code
  
          info.update(
              {
                  "icmp_type": icmp_type,
                  "icmp_code": icmp_code,
                  "icmp_type_name": icmp_types.get(
                      icmp_type, f"Unknown Type ({icmp_type})"
                  ),
                  "icmp_id": getattr(icmp_layer, "id", None),
                  "icmp_seq": getattr(icmp_layer, "seq", None),
                  "checksum": icmp_layer.chksum,
              }
          )
  
          # Add code descriptions for specific types
          if icmp_type == 3:  # Destination Unreachable
              info["icmp_code_name"] = dest_unreach_codes.get(
                  icmp_code, f"Unknown Code ({icmp_code})"
              )
          elif icmp_type == 11:  # Time Exceeded
              info["icmp_code_name"] = time_exceeded_codes.get(
                  icmp_code, f"Unknown Code ({icmp_code})"
              )
          else:
              info["icmp_code_name"] = f"Code {icmp_code}"
  
      return info

• src/mcpcap/modules/icmp.py:192-243 (handler)

  Helper that generates aggregate statistics: ICMP type counts, unique source/destination IPs, echo request/reply pair tracking, and unreachable destination tracking.

  def _generate_statistics(self, packets: list[dict[str, Any]]) -> dict[str, Any]:
      """Generate statistics from analyzed ICMP packets."""
      stats = {
          "icmp_types": {},
          "unique_sources": set(),
          "unique_destinations": set(),
          "echo_pairs": {},
          "unreachable_destinations": set(),
      }
  
      for pkt in packets:
          # Count ICMP types
          if "icmp_type_name" in pkt:
              type_name = pkt["icmp_type_name"]
              stats["icmp_types"][type_name] = (
                  stats["icmp_types"].get(type_name, 0) + 1
              )
  
          # Track unique IPs
          if "src_ip" in pkt:
              stats["unique_sources"].add(pkt["src_ip"])
          if "dst_ip" in pkt:
              stats["unique_destinations"].add(pkt["dst_ip"])
  
          # Track echo request/reply pairs
          if pkt.get("icmp_type") in [0, 8] and pkt.get("icmp_id") is not None:
              echo_id = pkt["icmp_id"]
              if echo_id not in stats["echo_pairs"]:
                  stats["echo_pairs"][echo_id] = {"requests": 0, "replies": 0}
  
              if pkt["icmp_type"] == 8:  # Echo Request
                  stats["echo_pairs"][echo_id]["requests"] += 1
              elif pkt["icmp_type"] == 0:  # Echo Reply
                  stats["echo_pairs"][echo_id]["replies"] += 1
  
          # Track unreachable destinations
          if pkt.get("icmp_type") == 3:  # Destination Unreachable
              if "dst_ip" in pkt:
                  stats["unreachable_destinations"].add(pkt["dst_ip"])
  
      # Convert sets to lists for JSON serialization
      return {
          "icmp_type_counts": stats["icmp_types"],
          "unique_sources_count": len(stats["unique_sources"]),
          "unique_destinations_count": len(stats["unique_destinations"]),
          "unique_sources": list(stats["unique_sources"]),
          "unique_destinations": list(stats["unique_destinations"]),
          "echo_sessions": len(stats["echo_pairs"]),
          "echo_pairs": stats["echo_pairs"],
          "unreachable_destinations_count": len(stats["unreachable_destinations"]),
          "unreachable_destinations": list(stats["unreachable_destinations"]),
      }

• src/mcpcap/core/server.py:57-58 (registration)

  Registration of analyze_icmp_packets as an MCP tool via FastMCP's tool() decorator within the MCPServer._register_tools() method.

  elif module_name == "icmp":
      self.mcp.tool(module.analyze_icmp_packets)