diff --git a/ChangeLog.md b/ChangeLog.md index 61c008d..eca978c 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -8,7 +8,7 @@ Generated by [`auto-changelog`](https://github.com/CookPete/auto-changelog). ##### Bug Fixes -- Fix WatchList.stats() and add regression tests — remove early return that bypassed lock [`481a78a`](https://github.com/craigerl/aprsd/commit/481a78a) +- Fix factory() unsafe deserialization — validate _type against known packet allowlist before globals() lookup [`2e64eac`](https://github.com/craigerl/aprsd/commit/2e64eac) - Fix APRSISDriver.is_configured() always returning True when driver is disabled [`a9ef65f`](https://github.com/craigerl/aprsd/commit/a9ef65f) diff --git a/aprsd/packets/core.py b/aprsd/packets/core.py index 5e988db..e842e25 100644 --- a/aprsd/packets/core.py +++ b/aprsd/packets/core.py @@ -780,11 +780,28 @@ def is_mice_packet(packet: dict[Any, Any]) -> bool: return get_packet_type(packet) == PACKET_TYPE_MICE +# Allowlist of class names that factory() may deserialise from disk. +# Built lazily from TYPE_LOOKUP so it stays in sync automatically. +_KNOWN_PACKET_TYPE_NAMES: set[str] = set() + + +def _known_packet_type_names() -> set[str]: + global _KNOWN_PACKET_TYPE_NAMES + if not _KNOWN_PACKET_TYPE_NAMES: + _KNOWN_PACKET_TYPE_NAMES = {cls.__name__ for cls in TYPE_LOOKUP.values()} | { + 'UnknownPacket' + } + return _KNOWN_PACKET_TYPE_NAMES + + def factory(raw_packet: dict[Any, Any]) -> type[Packet]: """Factory method to create a packet from a raw packet string.""" raw = raw_packet if '_type' in raw: - cls = globals()[raw['_type']] + type_name = raw['_type'] + if type_name not in _known_packet_type_names(): + raise ValueError(f'Unknown packet type {type_name!r} in saved data') + cls = globals()[type_name] return cls.from_dict(raw) raw['raw_dict'] = raw.copy() diff --git a/tests/packets/test_packet.py b/tests/packets/test_packet.py index f935899..cf7df68 100644 --- a/tests/packets/test_packet.py +++ b/tests/packets/test_packet.py @@ -73,3 +73,55 @@ class TestPacket(unittest.TestCase): restored = packets.factory(json_dict) self.assertEqual(restored.from_call, packet.from_call) self.assertEqual(restored.to_call, packet.to_call) + + +class TestFactory(unittest.TestCase): + """Tests for packets.factory() — especially the _type allowlist.""" + + def test_factory_known_type_roundtrip(self): + """factory() with a known _type deserialises correctly.""" + raw = 'KFAKE>APZ100::KMINE :Hello{99' + pkt_dict = aprslib.parse(raw) + pkt_dict['format'] = 'message' + original = packets.factory(pkt_dict) + # Serialise to JSON dict (adds _type) then round-trip through factory + json_dict = json.loads(original.to_json()) + self.assertIn('_type', json_dict) + restored = packets.factory(json_dict) + self.assertIsInstance(restored, packets.MessagePacket) + self.assertEqual(restored.from_call, original.from_call) + + def test_factory_unknown_type_raises(self): + """factory() must raise ValueError for an unknown _type value. + + Regression test for the unsafe-deserialization fix: before the fix, + globals()[raw['_type']] was called without validation, allowing any + string from disk to look up arbitrary module-global names. + """ + crafted = {'_type': 'os'} # 'os' exists in module globals via imports + with self.assertRaises(ValueError): + packets.factory(crafted) + + def test_factory_arbitrary_string_raises(self): + """factory() rejects completely arbitrary _type strings.""" + crafted = {'_type': 'EvilClass'} + with self.assertRaises(ValueError): + packets.factory(crafted) + + def test_factory_empty_type_raises(self): + """factory() rejects an empty _type string.""" + crafted = {'_type': ''} + with self.assertRaises(ValueError): + packets.factory(crafted) + + def test_factory_allowlist_covers_all_type_lookup_classes(self): + """Every class in TYPE_LOOKUP must be in the factory() allowlist.""" + from aprsd.packets.core import TYPE_LOOKUP, _known_packet_type_names + + allowlist = _known_packet_type_names() + for cls in TYPE_LOOKUP.values(): + self.assertIn( + cls.__name__, + allowlist, + f'{cls.__name__} is in TYPE_LOOKUP but missing from the allowlist', + )