feat: validate tshark -e field names before running analysis

This commit is contained in:
YueGuobin 2026-05-12 14:27:16 +08:00
parent 9c23f9037c
commit 32c3b4cd35
No known key found for this signature in database

View File

@ -30,6 +30,7 @@ Analyzes packets from an active GNS3 capture using tshark with user-provided arg
The LLM constructs tshark commands based on protocol knowledge from packet analysis skills. The LLM constructs tshark commands based on protocol knowledge from packet analysis skills.
""" """
import json
import logging import logging
import os import os
import subprocess import subprocess
@ -44,6 +45,9 @@ from gns3server.agent.gns3_copilot.gns3_client.context_helpers import (
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Cache of valid tshark field names, loaded lazily from `tshark -G fields`
_tshark_valid_fields: set | None = None
class PacketAnalysisTool(BaseTool): class PacketAnalysisTool(BaseTool):
""" """
@ -87,6 +91,82 @@ class PacketAnalysisTool(BaseTool):
alone instead. alone instead.
""" """
@classmethod
def _load_valid_tshark_fields(cls) -> set:
"""
Load and cache valid tshark field names from `tshark -G fields`.
Returns:
Set of valid field names, or empty set on failure.
"""
global _tshark_valid_fields
if _tshark_valid_fields is not None:
return _tshark_valid_fields
try:
result = subprocess.run(
["tshark", "-G", "fields"],
capture_output=True,
text=True,
timeout=30,
)
fields = set()
for line in result.stdout.splitlines():
parts = line.split("\t")
if len(parts) >= 3:
fields.add(parts[2])
_tshark_valid_fields = fields
logger.debug(f"Loaded {len(fields)} valid tshark field names")
except Exception as e:
logger.warning(f"Could not load tshark field names: {e}")
_tshark_valid_fields = set()
return _tshark_valid_fields
def _validate_tshark_args(self, tshark_args: str) -> str | None:
"""
Validate tshark arguments before execution.
Checks `-e` field names against the tshark field registry.
Returns an error JSON string if invalid, or None if valid.
Args:
tshark_args: tshark command arguments string
Returns:
Error JSON string, or None if validation passes.
"""
import shlex
try:
args = shlex.split(tshark_args)
except ValueError as e:
return json.dumps({"error": f"Invalid tshark_args syntax: {e}"})
valid_fields = self._load_valid_tshark_fields()
if not valid_fields:
return None
invalid_fields = []
i = 0
while i < len(args):
if args[i] == "-e" and i + 1 < len(args):
field = args[i + 1]
if field not in valid_fields:
invalid_fields.append(field)
i += 2
else:
i += 1
if invalid_fields:
return json.dumps({
"error": f"Invalid tshark field names: {', '.join(invalid_fields)}",
"hint": "Use packet_analysis_skills tool to look up valid field names for the protocol",
"invalid_fields": invalid_fields,
})
return None
def _run( def _run(
self, self,
project_id: str, project_id: str,
@ -119,6 +199,11 @@ class PacketAnalysisTool(BaseTool):
if not tshark_args or not tshark_args.strip(): if not tshark_args or not tshark_args.strip():
return '{"error": "tshark_args is required"}' return '{"error": "tshark_args is required"}'
# Pre-validate tshark field names before downloading capture
validation_error = self._validate_tshark_args(tshark_args)
if validation_error:
return validation_error
temp_file = None temp_file = None
try: try:
# Download capture file # Download capture file