analyze_tcp_retransmissions

Instructions

Implementation Reference

• src/mcpcap/modules/tcp.py:106-132 (handler)

  The public tool method 'analyze_tcp_retransmissions' in TCPModule. Delegates to analyze_packets with analysis_type='retransmissions', which routes to _analyze_retrans.

  def analyze_tcp_retransmissions(
      self,
      pcap_file: str,
      server_ip: str | None = None,
      threshold: float = 0.02,
  ) -> dict[str, Any]:
      """
      Analyze TCP retransmission patterns.
  
      Args:
          pcap_file: HTTP URL or absolute local file path to PCAP file
          server_ip: Optional filter for server IP address
          threshold: Retransmission rate threshold (default: 2%)
  
      Returns:
          A structured dictionary containing:
          - total_retransmissions: Total number of retransmissions
          - retransmission_rate: Overall retransmission rate
          - by_connection: Per-connection retransmission statistics
          - summary: Worst connections and threshold violations
      """
      return self.analyze_packets(
          pcap_file,
          analysis_type="retransmissions",
          server_ip=server_ip,
          threshold=threshold,
      )

• src/mcpcap/modules/tcp.py:409-471 (handler)

  Private helper _analyze_retrans that performs the actual retransmission analysis logic. Groups TCP packets by connection, calculates retransmission counts and rates per connection, sorts by rate descending, and returns retransmission metrics including threshold comparison.

  def _analyze_retrans(
      self, pcap_file: str, tcp_packets: list, all_packets: list
  ) -> dict[str, Any]:
      """Analyze TCP retransmissions."""
      threshold = self._analysis_kwargs.get("threshold", 0.02)
  
      # Group by connection
      connections = defaultdict(list)
      for pkt in tcp_packets:
          conn_key = self._get_connection_key(pkt)
          connections[conn_key].append(pkt)
  
      by_connection = []
      total_retrans = 0
      worst_rate = 0
      worst_conn = ""
  
      for conn_key, pkts in connections.items():
          src_ip, src_port, dst_ip, dst_port = conn_key
          conn_str = f"{src_ip}:{src_port} <-> {dst_ip}:{dst_port}"
  
          conn_info = self._analyze_single_connection(conn_key, pkts)
          retrans_count = conn_info["retransmissions"]
          total_retrans += retrans_count
  
          retrans_rate = retrans_count / len(pkts) if len(pkts) > 0 else 0
  
          by_connection.append(
              {
                  "connection": conn_str,
                  "retrans_count": retrans_count,
                  "total_packets": len(pkts),
                  "retrans_rate": retrans_rate,
              }
          )
  
          if retrans_rate > worst_rate:
              worst_rate = retrans_rate
              worst_conn = conn_str
  
      # Sort by retransmission rate
      by_connection.sort(key=lambda x: x["retrans_rate"], reverse=True)
  
      overall_rate = total_retrans / len(tcp_packets) if len(tcp_packets) > 0 else 0
      connections_above_threshold = sum(
          1 for c in by_connection if c["retrans_rate"] > threshold
      )
  
      return {
          "file": pcap_file,
          "analysis_timestamp": datetime.now().isoformat(),
          "total_packets": len(tcp_packets),
          "total_retransmissions": total_retrans,
          "retransmission_rate": overall_rate,
          "threshold": threshold,
          "exceeds_threshold": overall_rate > threshold,
          "by_connection": by_connection[:10],  # Top 10
          "summary": {
              "worst_connection": worst_conn,
              "worst_retrans_rate": worst_rate,
              "connections_above_threshold": connections_above_threshold,
          },
      }

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

  Registration of the tool via FastMCP's @tool decorator in _register_tools, passing the module's method reference.

  self.mcp.tool(module.analyze_tcp_retransmissions)

• src/mcpcap/modules/tcp.py:106-132 (schema)

  The function signature and docstring define the input schema (pcap_file: str, server_ip: str|None, threshold: float) and output schema (dict with total_retransmissions, retransmission_rate, by_connection, summary).

  def analyze_tcp_retransmissions(
      self,
      pcap_file: str,
      server_ip: str | None = None,
      threshold: float = 0.02,
  ) -> dict[str, Any]:
      """
      Analyze TCP retransmission patterns.
  
      Args:
          pcap_file: HTTP URL or absolute local file path to PCAP file
          server_ip: Optional filter for server IP address
          threshold: Retransmission rate threshold (default: 2%)
  
      Returns:
          A structured dictionary containing:
          - total_retransmissions: Total number of retransmissions
          - retransmission_rate: Overall retransmission rate
          - by_connection: Per-connection retransmission statistics
          - summary: Worst connections and threshold violations
      """
      return self.analyze_packets(
          pcap_file,
          analysis_type="retransmissions",
          server_ip=server_ip,
          threshold=threshold,
      )