Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 66 additions & 6 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@ name: CI

on:
push:
branches: [ "master", "main" ]
branches: [ "master", "main", "codex/**" ]
pull_request:
branches: [ "master", "main" ]
workflow_dispatch:

jobs:
lint:
Expand Down Expand Up @@ -32,9 +33,11 @@ jobs:
run: mypy --package wireshark_mcp --ignore-missing-imports --no-namespace-packages

test:
runs-on: ubuntu-latest
runs-on: ${{ matrix.os }}
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest, windows-latest, macos-latest]
python-version: ["3.10", "3.11", "3.12", "3.13"]

steps:
Expand All @@ -50,15 +53,72 @@ jobs:
python -m pip install --upgrade pip
pip install .[dev]

- name: Run tests with coverage
run: |
pytest tests/ --cov=wireshark_mcp --cov-report=term-missing -v

- name: Compile check
run: python -m compileall src/

integration-linux:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6

- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.12"

- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install .[dev]

- name: Install TShark
run: |
sudo apt-get update
sudo env DEBIAN_FRONTEND=noninteractive apt-get install -y tshark
tshark -v

- name: Run tests with coverage
- name: Run Linux TShark integration smoke tests
run: |
pytest tests/ --cov=wireshark_mcp --cov-report=term-missing -v
pytest tests/test_client.py -k "real_tshark" -v

- name: Compile check
run: python -m compileall src/
package-smoke:
runs-on: ${{ matrix.os }}
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest, windows-latest, macos-latest]

steps:
- uses: actions/checkout@v6

- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: "3.12"

- name: Install build tools
run: |
python -m pip install --upgrade pip
pip install build

- name: Build package
run: python -m build

- name: Validate packaged skill files
run: |
python -c "import glob, os, zipfile; wheel = max(glob.glob('dist/*.whl'), key=os.path.getmtime); names = set(zipfile.ZipFile(wheel).namelist()); assert any(name.endswith('wireshark_mcp/skills/wireshark-traffic-analysis/SKILL.md') for name in names), 'Skill package missing from wheel'"

- name: Install built wheel
run: |
python -c "import glob, os, subprocess, sys; wheel = max(glob.glob('dist/*.whl'), key=os.path.getmtime); subprocess.check_call([sys.executable, '-m', 'pip', 'install', '--force-reinstall', wheel])"

- name: Run packaged CLI smoke tests
run: |
wireshark-mcp --version
python -m wireshark_mcp.server --version
wireshark-mcp --config
wireshark-mcp --doctor
15 changes: 15 additions & 0 deletions .github/workflows/publish.yml
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,21 @@ jobs:

- name: Build package
run: python -m build

- name: Validate built wheel contents
run: |
python -c "import glob, os, zipfile; wheel = max(glob.glob('dist/*.whl'), key=os.path.getmtime); names = set(zipfile.ZipFile(wheel).namelist()); assert any(name.endswith('wireshark_mcp/skills/wireshark-traffic-analysis/SKILL.md') for name in names), 'Skill package missing from wheel'"

- name: Install built wheel
run: |
python -c "import glob, os, subprocess, sys; wheel = max(glob.glob('dist/*.whl'), key=os.path.getmtime); subprocess.check_call([sys.executable, '-m', 'pip', 'install', '--force-reinstall', wheel])"

- name: Run release smoke tests
run: |
wireshark-mcp --version
python -m wireshark_mcp.server --version
wireshark-mcp --config
wireshark-mcp --doctor

- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
Expand Down
38 changes: 29 additions & 9 deletions src/wireshark_mcp/tshark/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,18 @@ def _require_tool(self, tool_name: str) -> dict[str, Any]:
},
}

def _get_checked_tool_path(self, tool_name: str) -> str:
"""Return a tool path after availability has already been validated."""
tool_path = self._tool_paths.get(tool_name)
if not self._tool_is_available(tool_path):
raise RuntimeError(f"{tool_name} tool not available")
assert tool_path is not None
return tool_path

def _select_capture_backend_path(self) -> str:
"""Return the resolved executable for the preferred capture backend."""
return self._get_checked_tool_path(self._select_capture_backend())

# --- Validation Methods ---

def _validate_file(self, filepath: str) -> dict[str, Any]:
Expand Down Expand Up @@ -212,6 +224,7 @@ async def check_capabilities(self) -> dict[str, Any]:
async def get_version(tool_path: str | None) -> dict[str, Any]:
if not self._tool_is_available(tool_path):
return {"available": False}
assert tool_path is not None
try:
proc = await asyncio.create_subprocess_exec(
tool_path,
Expand All @@ -234,7 +247,7 @@ async def get_version(tool_path: str | None) -> dict[str, Any]:

async def list_interfaces(self) -> str:
"""List interfaces (-D)."""
backend = self.dumpcap_path if self._tool_is_available(self.dumpcap_path) else self.tshark_path
backend = self._select_capture_backend_path()
return await self._run_command([backend, "-D"])

# --- Capture Management ---
Expand All @@ -253,7 +266,7 @@ async def capture_packets(
if not output_validation["success"]:
return json.dumps(output_validation)

backend = self.dumpcap_path if self._tool_is_available(self.dumpcap_path) else self.tshark_path
backend = self._select_capture_backend_path()
cmd = [backend, "-i", interface, "-w", output_file]

if capture_filter:
Expand Down Expand Up @@ -666,7 +679,8 @@ async def get_file_info(self, pcap_file: str) -> str:
if not required["success"]:
return json.dumps(required)

return await self._run_command([self.capinfos_path, pcap_file])
capinfos_path = self._get_checked_tool_path("capinfos")
return await self._run_command([capinfos_path, pcap_file])

async def merge_pcap_files(self, output_file: str, input_files: list[str]) -> str:
"""Mergecap: Merge multiple pcaps."""
Expand All @@ -683,7 +697,8 @@ async def merge_pcap_files(self, output_file: str, input_files: list[str]) -> st
if not output_validation["success"]:
return json.dumps(output_validation)

cmd = [self.mergecap_path, "-w", output_file] + input_files
mergecap_path = self._get_checked_tool_path("mergecap")
cmd = [mergecap_path, "-w", output_file] + input_files
return await self._run_command(cmd)

async def editcap_trim(
Expand All @@ -706,7 +721,8 @@ async def editcap_trim(
if not output_validation["success"]:
return json.dumps(output_validation)

cmd = [self.editcap_path]
editcap_path = self._get_checked_tool_path("editcap")
cmd = [editcap_path]
if start_time:
cmd.extend(["-A", start_time])
if stop_time:
Expand Down Expand Up @@ -745,7 +761,8 @@ async def editcap_split(
}
)

cmd = [self.editcap_path]
editcap_path = self._get_checked_tool_path("editcap")
cmd = [editcap_path]
if packets_per_file > 0:
cmd.extend(["-c", str(packets_per_file)])
if seconds_per_file > 0:
Expand All @@ -767,7 +784,8 @@ async def editcap_time_shift(self, input_file: str, output_file: str, seconds: f
if not output_validation["success"]:
return json.dumps(output_validation)

cmd = [self.editcap_path, "-t", str(seconds), input_file, output_file]
editcap_path = self._get_checked_tool_path("editcap")
cmd = [editcap_path, "-t", str(seconds), input_file, output_file]
return await self._run_command(cmd)

async def editcap_deduplicate(self, input_file: str, output_file: str, duplicate_window: int = 5) -> str:
Expand All @@ -784,7 +802,8 @@ async def editcap_deduplicate(self, input_file: str, output_file: str, duplicate
if not output_validation["success"]:
return json.dumps(output_validation)

cmd = [self.editcap_path, "-D", str(duplicate_window), input_file, output_file]
editcap_path = self._get_checked_tool_path("editcap")
cmd = [editcap_path, "-D", str(duplicate_window), input_file, output_file]
return await self._run_command(cmd)

async def text2pcap_import(
Expand All @@ -808,7 +827,8 @@ async def text2pcap_import(
if not output_validation["success"]:
return json.dumps(output_validation)

cmd = [self.text2pcap_path]
text2pcap_path = self._get_checked_tool_path("text2pcap")
cmd = [text2pcap_path]
if timestamp_format:
cmd.extend(["-t", timestamp_format])
if ascii_mode:
Expand Down
5 changes: 5 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,11 @@ def _validate_file(self, filepath: str) -> dict[str, Any]:
return super()._validate_file(filepath)
return {"success": True}

@staticmethod
def _tool_is_available(tool_path: str | None) -> bool:
"""Treat any configured mock command name as available."""
return bool(tool_path)

async def _run_command(
self,
cmd: list[str],
Expand Down
37 changes: 37 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for TSharkClient core functionality."""

import json
import shutil

import pytest

Expand Down Expand Up @@ -106,6 +107,27 @@ def test_client_prefers_env_tool_paths(self, monkeypatch) -> None:
assert client.dumpcap_path == "/opt/wireshark/dumpcap"
assert client.text2pcap_path == "/opt/wireshark/text2pcap"

def test_describe_capabilities_reports_capture_backend_fallback(self, mock_client) -> None:
capabilities = mock_client.describe_capabilities()
assert capabilities["_meta"]["capture_backend"] == "dumpcap"
assert capabilities["dumpcap"]["requirement"] == "optional"

mock_client.dumpcap_path = None
mock_client._tool_paths["dumpcap"] = None

fallback_capabilities = mock_client.describe_capabilities()
assert fallback_capabilities["_meta"]["capture_backend"] == "tshark"

@pytest.mark.asyncio
async def test_check_capabilities_detects_real_tshark_when_installed(self) -> None:
if shutil.which("tshark") is None:
pytest.skip("tshark not installed on this host")

result = await TSharkClient().check_capabilities()

assert result["success"]
assert result["data"]["tshark"]["available"] is True


class TestRunCommand:
"""Tests for _run_command error handling."""
Expand Down Expand Up @@ -140,6 +162,21 @@ async def test_binary_whitelist_allows_windows_exe_names_case_insensitive(self,


class TestSuiteBehavior:
@pytest.mark.asyncio
async def test_list_interfaces_prefers_dumpcap_when_available(self, mock_client) -> None:
result = await mock_client.list_interfaces()
assert "dumpcap" in result
assert mock_client._last_cmd[0] == "dumpcap"

@pytest.mark.asyncio
async def test_list_interfaces_falls_back_to_tshark(self, mock_client) -> None:
mock_client.dumpcap_path = None
mock_client._tool_paths["dumpcap"] = None

result = await mock_client.list_interfaces()
assert "tshark" in result
assert mock_client._last_cmd[0] == "tshark"

@pytest.mark.asyncio
async def test_capture_prefers_dumpcap_when_available(self, mock_client) -> None:
result = await mock_client.capture_packets("en0", "/tmp/out.pcapng", duration=10)
Expand Down
2 changes: 1 addition & 1 deletion tests/test_files.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,6 @@ async def test_text2pcap_import_command(self, mock_client: MockTSharkClient) ->
ascii_mode=True,
)
assert "text2pcap" in result
assert '-t %H:%M:%S.%f' in result
assert "-t %H:%M:%S.%f" in result
assert "-a" in result
assert "-E ether" in result
52 changes: 52 additions & 0 deletions tests/test_installer.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@
_get_linux_config_home,
_get_mcp_servers_dict,
_get_python_executable,
_iter_wireshark_search_dirs,
_join_path,
_read_json_config,
_render_codex_toml_block,
_write_json_config,
Expand All @@ -28,6 +30,7 @@ def test_get_python_executable_returns_string(self):
assert len(result) > 0

def test_get_python_executable_in_venv(self, tmp_path, monkeypatch):
monkeypatch.setattr("wireshark_mcp.installer.sys.platform", "linux")
venv_dir = tmp_path / "venv"
bin_dir = venv_dir / "bin"
bin_dir.mkdir(parents=True)
Expand Down Expand Up @@ -276,10 +279,35 @@ def test_uninstall_removes_codex_toml_block(self, tmp_path):


class TestPlatformConfigs:
def test_join_path_uses_target_platform_separators(self):
assert _join_path("/Users/tester", "Library", "Claude", platform="darwin") == "/Users/tester/Library/Claude"
assert (
_join_path(r"C:\Users\tester", "AppData", "Roaming", platform="win32") == r"C:\Users\tester\AppData\Roaming"
)

def test_linux_config_home_uses_xdg(self, monkeypatch):
monkeypatch.setenv("XDG_CONFIG_HOME", "/tmp/xdg-config")
assert _get_linux_config_home("/home/tester") == "/tmp/xdg-config"

def test_mac_client_configs_use_application_support(self, monkeypatch):
monkeypatch.setattr("wireshark_mcp.installer.sys.platform", "darwin")
monkeypatch.setattr("wireshark_mcp.installer.os.path.expanduser", lambda _: "/Users/tester")

configs = _get_client_configs()

assert configs["Claude"] == (
"/Users/tester/Library/Application Support/Claude",
"claude_desktop_config.json",
)
assert configs["Zed"] == (
"/Users/tester/Library/Application Support/Zed",
"settings.json",
)
assert configs["VS Code"] == (
"/Users/tester/Library/Application Support/Code/User",
"settings.json",
)

def test_linux_client_configs_use_xdg(self, monkeypatch):
monkeypatch.setattr("wireshark_mcp.installer.sys.platform", "linux")
monkeypatch.setattr("wireshark_mcp.installer.os.path.expanduser", lambda _: "/home/tester")
Expand All @@ -306,3 +334,27 @@ def test_windows_client_configs_include_supported_paths(self, monkeypatch):
r"C:\Users\tester\AppData\Roaming\Code\User",
"settings.json",
)

def test_iter_wireshark_search_dirs_for_macos(self, monkeypatch):
monkeypatch.setattr("wireshark_mcp.installer.sys.platform", "darwin")
monkeypatch.setattr("wireshark_mcp.installer.os.path.expanduser", lambda _: "/Users/tester")

search_dirs = _iter_wireshark_search_dirs()

assert "/Applications/Wireshark.app/Contents/MacOS" in search_dirs
assert "/Applications/Wireshark.app/Contents/Helpers" in search_dirs
assert "/Users/tester/Applications/Wireshark.app/Contents/MacOS" in search_dirs
assert "/opt/homebrew/bin" in search_dirs

def test_iter_wireshark_search_dirs_for_windows(self, monkeypatch):
monkeypatch.setattr("wireshark_mcp.installer.sys.platform", "win32")
monkeypatch.setattr("wireshark_mcp.installer.os.path.expanduser", lambda _: r"C:\Users\tester")
monkeypatch.setenv("LOCALAPPDATA", r"C:\Users\tester\AppData\Local")
monkeypatch.setenv("PROGRAMFILES", r"C:\Program Files")
monkeypatch.setenv("PROGRAMFILES(X86)", r"C:\Program Files (x86)")

search_dirs = _iter_wireshark_search_dirs()

assert r"C:\Program Files\Wireshark" in search_dirs
assert r"C:\Program Files (x86)\Wireshark" in search_dirs
assert r"C:\Users\tester\AppData\Local\Programs\Wireshark" in search_dirs
Loading