mcpacket

Instructions

Implementation Reference

• src/mcpcap/modules/tcp.py:134-162 (handler)

  The main handler function for the 'analyze_traffic_flow' tool. It accepts pcap_file, server_ip, and optional server_port, then delegates to analyze_packets with analysis_type='traffic_flow' to perform the analysis.

  def analyze_traffic_flow(
      self,
      pcap_file: str,
      server_ip: str,
      server_port: Optional[int] = None,
  ) -> dict[str, Any]:
      """
      Analyze bidirectional traffic flow characteristics.
  
      Identifies traffic direction, asymmetry, RST sources, and data transfer patterns.
  
      Args:
          pcap_file: HTTP URL or absolute local file path to PCAP file
          server_ip: Server IP address (required)
          server_port: Optional filter for server port
  
      Returns:
          A structured dictionary containing:
          - client_to_server: Client-to-server traffic statistics
          - server_to_client: Server-to-client traffic statistics
          - analysis: Asymmetry analysis and interpretations
      """
      return self.analyze_packets(
          pcap_file,
          analysis_type="traffic_flow",
          server_ip=server_ip,
          server_port=server_port,
      )

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

  Registration of the 'analyze_traffic_flow' tool in the MCP server within the TCP module registration block.

  self.mcp.tool(module.analyze_tcp_retransmissions)
  self.mcp.tool(module.analyze_traffic_flow)

• src/mcpcap/modules/tcp.py:472-570 (helper)

  Core helper function implementing the traffic flow analysis logic, calculating bidirectional statistics, asymmetry, and RST source analysis.

  def _analyze_flow(
      self, pcap_file: str, tcp_packets: list, all_packets: list
  ) -> dict[str, Any]:
      """Analyze traffic flow."""
      server_ip = self._analysis_kwargs.get("server_ip")
      server_port = self._analysis_kwargs.get("server_port")
  
      if not server_ip:
          return {"error": "server_ip is required for traffic flow analysis"}
  
      client_to_server = {
          "packet_count": 0,
          "byte_count": 0,
          "syn_count": 0,
          "rst_count": 0,
          "fin_count": 0,
          "data_packets": 0,
          "retransmissions": 0,
      }
  
      server_to_client = {
          "packet_count": 0,
          "byte_count": 0,
          "syn_count": 0,
          "rst_count": 0,
          "fin_count": 0,
          "data_packets": 0,
          "retransmissions": 0,
      }
  
      client_seqs = set()
      server_seqs = set()
  
      for pkt in tcp_packets:
          src_ip, dst_ip = self._extract_ips(pkt)
          tcp = pkt[TCP]
          flags = tcp.flags
  
          # Determine direction
          is_client_to_server = dst_ip == server_ip
          if server_port:
              is_client_to_server = tcp.dport == server_port
  
          stats = client_to_server if is_client_to_server else server_to_client
          seqs = client_seqs if is_client_to_server else server_seqs
  
          stats["packet_count"] += 1
          stats["byte_count"] += len(pkt)
  
          if flags & 0x02:
              stats["syn_count"] += 1
          if flags & 0x04:
              stats["rst_count"] += 1
          if flags & 0x01:
              stats["fin_count"] += 1
          if len(tcp.payload) > 0:
              stats["data_packets"] += 1
  
          # Retransmissions
          seq = tcp.seq
          if seq in seqs and len(tcp.payload) > 0:
              stats["retransmissions"] += 1
          seqs.add(seq)
  
      # Analysis
      total_client = client_to_server["packet_count"]
      total_server = server_to_client["packet_count"]
      asymmetry_ratio = total_client / total_server if total_server > 0 else 0
  
      # Determine primary RST source
      client_rst = client_to_server["rst_count"]
      server_rst = server_to_client["rst_count"]
      if client_rst > server_rst:
          rst_source = "client"
          interpretation = f"Client sends all RST packets ({client_rst} vs {server_rst}). Server responds normally. Suggests client-side issue (possibly firewall)."
      elif server_rst > client_rst:
          rst_source = "server"
          interpretation = "Server sends more RST packets. Suggests server-side rejection or service issue."
      else:
          rst_source = "balanced"
          interpretation = "Balanced RST distribution."
  
      return {
          "file": pcap_file,
          "analysis_timestamp": datetime.now().isoformat(),
          "server": f"{server_ip}:{server_port or 'any'}",
          "client_to_server": client_to_server,
          "server_to_client": server_to_client,
          "analysis": {
              "asymmetry_ratio": asymmetry_ratio,
              "primary_rst_source": rst_source,
              "data_flow_direction": "client_heavy"
              if asymmetry_ratio > 1.2
              else "server_heavy"
              if asymmetry_ratio < 0.8
              else "balanced",
              "interpretation": interpretation,
          },
      }

• src/mcpcap/modules/tcp.py:171-202 (helper)

  Helper function that reads the PCAP file, filters TCP packets, and routes 'traffic_flow' analysis to the _analyze_flow method.

  def _analyze_protocol_file(self, pcap_file: str) -> dict[str, Any]:
      """Perform the actual TCP packet analysis on a local PCAP file."""
      try:
          packets = rdpcap(pcap_file)
          tcp_packets = [pkt for pkt in packets if pkt.haslayer(TCP)]
  
          if not tcp_packets:
              return {
                  "file": pcap_file,
                  "total_packets": len(packets),
                  "tcp_packets_found": 0,
                  "message": "No TCP packets found in this capture",
              }
  
          # Apply filtering if specified
          filtered_packets = self._apply_filters(
              tcp_packets,
              self._analysis_kwargs.get("server_ip"),
              self._analysis_kwargs.get("server_port"),
          )
  
          # Route to appropriate analysis method
          if self._analysis_type == "connections":
              return self._analyze_connections(pcap_file, filtered_packets, packets)
          elif self._analysis_type == "anomalies":
              return self._analyze_anomalies(pcap_file, filtered_packets, packets)
          elif self._analysis_type == "retransmissions":
              return self._analyze_retrans(pcap_file, filtered_packets, packets)
          elif self._analysis_type == "traffic_flow":
              return self._analyze_flow(pcap_file, filtered_packets, packets)
          else:
              return {"error": f"Unknown analysis type: {self._analysis_type}"}