From 5d86ceca64a7e9fa2edf7cc413ae7206825cc760 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 8 Oct 2026 14:51:48 -0400 Subject: [PATCH] fix(foundation): keep format='pcap' when an unavailable engine falls back to the default one (#1264) Extractor replaced format='pcap' with 'json' from the requested engine name, before run() found that engine missing and ran the default PCAP engine, so out.json was written where out.pcap was asked for. The trace format guard had the same flaw. - Factor the availability check out of run() into _select_engine(), and call it in __init__ before the format decisions, so both follow the engine that actually runs. A second call is silent, so each fallback still warns once. - The FormatWarning now names the normalised engine, as the trace one does. - test_extraction._traced patches _select_engine out with run, so its dict-frame cases keep testing the guard on hosts missing those engines. Tests: new test_extraction_engine_fallback_format_unit.py (fails without the fix: 29 failed, 3 passed); foundation/interface modules and the foundation unittest leg pass. --- pcapkit/foundation/extraction.py | 94 ++++++++---- tests/foundation/test_extraction.py | 10 +- ..._extraction_engine_fallback_format_unit.py | 140 ++++++++++++++++++ 3 files changed, 214 insertions(+), 30 deletions(-) create mode 100644 tests/foundation/test_extraction_engine_fallback_format_unit.py diff --git a/pcapkit/foundation/extraction.py b/pcapkit/foundation/extraction.py index 78c81deb10..d3ea4c8103 100644 --- a/pcapkit/foundation/extraction.py +++ b/pcapkit/foundation/extraction.py @@ -810,6 +810,57 @@ def run(self) -> 'None': # pylint: disable=inconsistent-return-statements the built-in PCAP-NG engine reads a PCAP-NG input to PCAP output. :rtype: None + """ + eng = self._select_engine() + if eng is not None: + logger.debug('using engine %s (%s)', eng.name, eng.module) + self._exeng = eng(self) + self._exeng.run() + self._open_output() + + # start iteration + self.record_frames() + return + + if self._magic in PCAP_Engine.MAGIC_NUMBER: + logger.debug('magic number %r identifies a PCAP file', self._magic) + self._exeng = cast('EngineBase[_P]', PCAP_Engine(self)) + elif self._magic in PCAPNG_Engine.MAGIC_NUMBER: + logger.debug('magic number %r identifies a PCAP-NG file', self._magic) + self._exeng = cast('EngineBase[_P]', PCAPNG_Engine(self)) + else: + raise FormatError(f'unknown file format: {self._magic!r}') + + # start engine + self._exeng.run() + self._open_output() + + # start iteration + self.record_frames() + + def _select_engine(self) -> 'Optional[Type[EngineBase]]': + """Settle which extraction engine will actually run. + + An engine that is unknown, not installed, or rules itself out via + :meth:`~pcapkit.foundation.engines.engine.EngineBase.unsupported_reason` + is replaced by the default one, and :attr:`self._exnam + ` is updated to match. + :meth:`__init__` calls this before choosing the output format, so that + the choice reflects the engine that runs rather than the one requested. + + Calling it again is silent: a fallback has already set the name to + ``'default'``, and an engine that passed the checks passes them again. + + Warns: + pcapkit.utilities.warnings.EngineWarning: If the extraction engine is not + available: its dependency is not installed, its + :meth:`~pcapkit.foundation.engines.engine.EngineBase.unsupported_reason` + rules it out, or the supplied engine is unknown. + + Returns: + The engine class to run, or :data:`None` for the default engine, + which :meth:`run` picks from the input's magic number. + """ logger.debug('requested extraction engine: %s', self._exnam) @@ -833,14 +884,7 @@ def run(self) -> 'None': # pylint: disable=inconsistent-return-statements EngineWarning, stacklevel=stacklevel()) self._exnam = 'default' elif self.import_test(eng.module, name=eng.name) is not None: # type: ignore[arg-type] - logger.debug('using engine %s (%s)', eng.name, eng.module) - self._exeng = eng(self) - self._exeng.run() - self._open_output() - - # start iteration - self.record_frames() - return + return eng else: # ``import_test`` has already warned that the package is absent; # warning again here would report one problem twice. @@ -850,22 +894,7 @@ def run(self) -> 'None': # pylint: disable=inconsistent-return-statements warn(f'unsupported extraction engine: {self._exnam}; ' 'using default engine instead', EngineWarning, stacklevel=stacklevel()) self._exnam = 'default' # using default/pcapkit engine - - if self._magic in PCAP_Engine.MAGIC_NUMBER: - logger.debug('magic number %r identifies a PCAP file', self._magic) - self._exeng = cast('EngineBase[_P]', PCAP_Engine(self)) - elif self._magic in PCAPNG_Engine.MAGIC_NUMBER: - logger.debug('magic number %r identifies a PCAP-NG file', self._magic) - self._exeng = cast('EngineBase[_P]', PCAPNG_Engine(self)) - else: - raise FormatError(f'unknown file format: {self._magic!r}') - - # start engine - self._exeng.run() - self._open_output() - - # start iteration - self.record_frames() + return None @staticmethod def import_test(engine: 'str', *, name: 'Optional[str]' = None) -> 'Optional[ModuleType]': @@ -1196,6 +1225,12 @@ def __init__(self, one, i.e. DPKT, Scapy, PyShark, PyPCAP, pcap-ct or PyPCAPFile. * If output file format is not supported. + In each case the engine is the one that will actually run: a + requested engine that is unavailable falls back to the default + engine first, which supports PCAP output. + + pcapkit.utilities.warnings.EngineWarning: If the requested extraction + engine is not available, c.f. :meth:`run`. pcapkit.utilities.warnings.AttributeWarning: If ``trace_analyse`` is requested while the extraction engine is PyShark. @@ -1207,14 +1242,20 @@ def __init__(self, if format is None: format = 'tree' + # NOTE: the engine is settled first, because the format checks below and + # in the trace settings depend on which engine *runs*: an unavailable + # one falls back to the default engine, which writes PCAP perfectly well. + self._exnam = cast('Engines', (engine or 'default').lower()) # extract using engine + self._select_engine() + # NOTE: the third-party engines' adapters hand the output writer each # frame as a plain :obj:`dict` built by ``packet2dict``, which carries # neither the frame's octets nor its record header, so there is nothing # for the PCAP writer to write. The flow tracer has the same limitation, # and is given the same replacement below. - if ((engine or 'default').lower() in ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile') + if (self._exnam in ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile') and format in ('pcap', 'cap')): - warn(f"'Extractor(engine={engine})' does not support 'format={format}'; " + warn(f"'Extractor(engine={self._exnam})' does not support 'format={format}'; " "using 'format=\"json\"' instead", FormatWarning, stacklevel=stacklevel()) format = 'json' @@ -1264,7 +1305,6 @@ def __init__(self, self._exptl = protocol or 'null' # extract til protocol self._exlyr = cast('Layers', (layer or 'none').lower()) # extract til layer - self._exnam = cast('Engines', (engine or 'default').lower()) # extract using engine self._exctx = ContextRegistry.make(context) # caller supplied context if reassembly: diff --git a/tests/foundation/test_extraction.py b/tests/foundation/test_extraction.py index ef05164e53..a8a2d7c238 100644 --- a/tests/foundation/test_extraction.py +++ b/tests/foundation/test_extraction.py @@ -1067,13 +1067,17 @@ def _traced(self, temp: pathlib.Path, tag: str, **kwargs: object): """An ``Extractor`` built for tracing, with ``run`` patched out. Nothing is extracted, so no engine needs to be installed -- the format - substitution under test happens in the constructor. Returns the extractor - and the patched module-level ``warn``. + substitution under test happens in the constructor. ``_select_engine`` is + patched out too, so the requested engine is taken as the one that runs; + otherwise an engine missing here would fall back to the default one, + which keeps PCAP (#1264). Returns the extractor and the patched + module-level ``warn``. """ from pcapkit.foundation.extraction import Extractor - with mock.patch.object(Extractor, 'run'): + with mock.patch.object(Extractor, 'run'), \ + mock.patch.object(Extractor, '_select_engine', return_value=None): with mock.patch('pcapkit.foundation.extraction.warn') as warn: extractor = Extractor(str(temp / 'sample.pcap'), str(temp / f'out-{tag}'), format='json', auto=False, nofile=True, trace=True, diff --git a/tests/foundation/test_extraction_engine_fallback_format_unit.py b/tests/foundation/test_extraction_engine_fallback_format_unit.py new file mode 100644 index 0000000000..5c5ea4697f --- /dev/null +++ b/tests/foundation/test_extraction_engine_fallback_format_unit.py @@ -0,0 +1,140 @@ +# -*- coding: utf-8 -*- +"""PCAP output survives a fallback from an unavailable engine to the default one. + +GitHub issue #1264: :class:`~pcapkit.foundation.extraction.Extractor` replaced +``format='pcap'`` with ``'json'`` from the *requested* engine name, before +:meth:`~pcapkit.foundation.extraction.Extractor.run` found the engine missing +and ran the default :class:`~pcapkit.foundation.engines.pcap.PCAP` engine -- +which writes PCAP -- so ``out.json`` appeared where ``out.pcap`` was asked for. +The trace format was replaced the same way. + +Each engine is made unavailable by patching its own check, so the result does +not depend on what this host has installed: ``unsupported_reason`` covers the +"not supported on this interpreter" path, and ``import_test`` the "not +installed" one. + +""" + +import contextlib +import importlib.util +import os +import sys +import tempfile +import unittest +import warnings +from unittest import mock + +from tests._support import reimport_once_per_class, sample_path + +#: Every engine whose frames cannot be written as PCAP. +ENGINES = ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile') + + +def _read(path: str) -> bytes: + with open(path, 'rb') as file: + return file.read() + + +class TestEngineFallbackKeepsPCAPOutput(unittest.TestCase): + """The format check follows the engine that runs, not the one requested.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _engine_class(name: str): + from pcapkit.corekit.module import ModuleDescriptor + from pcapkit.foundation.extraction import Extractor + + eng = Extractor.__engine__[name] + return eng.klass if isinstance(eng, ModuleDescriptor) else eng + + def _unavailable(self, name: str, how: str): + """Context manager making engine ``name`` unavailable ``how``.""" + klass = self._engine_class(name) + if how == 'unsupported': + return mock.patch.object(klass, 'unsupported_reason', return_value='unit test') + # A ``None`` entry in :data:`sys.modules` makes the import raise + # :exc:`ImportError`, so the real ``import_test`` runs and warns. + stack = contextlib.ExitStack() + stack.enter_context(mock.patch.object(klass, 'unsupported_reason', return_value=None)) + stack.enter_context(mock.patch.dict(sys.modules, {klass.module: None})) + return stack + + def _extract(self, **kwargs): + from pcapkit.interface import extract + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + extractor = extract(store=False, **kwargs) + return extractor, [w.message for w in caught] + + def test_pcap_output_is_kept_and_byte_exact(self) -> None: + from pcapkit.foundation.engines.pcap import PCAP + from pcapkit.utilities.warnings import EngineWarning, FormatWarning + + fin = sample_path('in.pcap') + for name in ENGINES: + for how in ('unsupported', 'not-installed'): + for fmt in ('pcap', 'cap'): + with self.subTest(engine=name, how=how, format=fmt), \ + tempfile.TemporaryDirectory() as tmp, self._unavailable(name, how): + extractor, caught = self._extract(fin=fin, fout=os.path.join(tmp, 'out'), + format=fmt, engine=name) + self.assertIsInstance(extractor.engine, PCAP) + self.assertEqual(extractor.format, 'pcap') + self.assertEqual(os.listdir(tmp), ['out.pcap']) + self.assertEqual(_read(os.path.join(tmp, 'out.pcap')), _read(fin)) + self.assertEqual([m for m in caught if isinstance(m, FormatWarning)], []) + # Settling the engine early must not report it twice. + engine_warnings = [str(m) for m in caught if isinstance(m, EngineWarning)] + self.assertEqual(len(engine_warnings), 1, engine_warnings) + self.assertIn('using default engine instead', engine_warnings[0]) + + def test_unknown_engine_keeps_pcap_output(self) -> None: + from pcapkit.utilities.warnings import EngineWarning + + fin = sample_path('in.pcap') + with tempfile.TemporaryDirectory() as tmp: + extractor, caught = self._extract(fin=fin, fout=os.path.join(tmp, 'out'), + format='pcap', engine='no-such-engine') + self.assertEqual(extractor.format, 'pcap') + self.assertEqual(_read(os.path.join(tmp, 'out.pcap')), _read(fin)) + self.assertEqual([str(m) for m in caught if isinstance(m, EngineWarning)], + ['unsupported extraction engine: no-such-engine; using default engine instead']) + + def test_trace_format_is_kept(self) -> None: + from pcapkit.utilities.warnings import FormatWarning + + fin = sample_path('in.pcap') + for name in ('dpkt', 'scapy', 'pyshark', 'pypcapfile'): + with self.subTest(engine=name), tempfile.TemporaryDirectory() as tmp, \ + self._unavailable(name, 'unsupported'): + extractor, caught = self._extract(fin=fin, fout=os.path.join(tmp, 'out'), + format='json', engine=name, trace=True, tcp=True, + trace_fout=os.path.join(tmp, 'trace')) + self.assertEqual([m for m in caught if isinstance(m, FormatWarning)], []) + flows = extractor.trace.tcp + self.assertTrue(flows) + for flow in flows: + self.assertTrue(flow.fpout.endswith('.pcap'), flow.fpout) + + def test_available_third_party_engine_still_falls_back_to_json(self) -> None: + # The other half of the contract: an engine that really runs keeps the + # existing replacement, and its warning names that engine. + from pcapkit.utilities.warnings import FormatWarning + + if importlib.util.find_spec('dpkt') is None: + self.skipTest('dpkt not installed') + with tempfile.TemporaryDirectory() as tmp: + extractor, caught = self._extract(fin=sample_path('in.pcap'), + fout=os.path.join(tmp, 'out'), format='pcap', engine='DPKT') + self.assertEqual(extractor.format, 'json') + self.assertEqual(os.listdir(tmp), ['out.json']) + self.assertEqual([str(m) for m in caught if isinstance(m, FormatWarning)], + ["'Extractor(engine=dpkt)' does not support 'format=pcap'; " + "using 'format=\"json\"' instead"]) + + +if __name__ == '__main__': + unittest.main()