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()