Some improvements

Author: NexSabre

packet_helper_core/__init__.py

@@ -1,3 +1,3 @@
-from packet_helper_core.core import Core
+from packet_helper_core.decoder import Decoder
 from packet_helper_core.packet_data import PacketData
 from packet_helper_core.packet_data_scapy import PacketDataScapy

packet_helper_core/core.py

@@ -0,0 +1,30 @@
+from packet_helper_core.packet_data import PacketData
+from packet_helper_core.packet_data_scapy import PacketDataScapy
+from packet_helper_core.utils.utils import decode_hex
+
+
+class Decoder:
+    """
+    Class Decoder is just a wrapper to create a handy-shortcut
+    for preparing a data from hex string
+    """
+
+    def __init__(self, hex_string: str) -> None:
+        self.hex_string = hex_string.replace(" ", "")
+
+        self.__decoded_by_tshark, self.__decoded_by_scapy = (None,) * 2
+
+    def run(self, extend_with_scapy: bool = True) -> None:
+        self.__decoded_by_tshark = PacketData(raw=str(decode_hex(self.hex_string)))
+        if extend_with_scapy:
+            self.__decoded_by_scapy = PacketDataScapy(
+                raw=self.hex_string, packet_data=self.tshark_data
+            )
+
+    @property
+    def tshark_data(self) -> PacketData | None:
+        return self.__decoded_by_tshark
+
+    @property
+    def scapy_data(self) -> PacketDataScapy | None:
+        return self.__decoded_by_scapy

packet_helper_core/decoder.py

@@ -11,17 +11,15 @@ class PacketData:
     _data_layer: list[str] = field(default_factory=list)
 
     def __post_init__(self):
-        self.raw_array = self.raw.split("\n")
-        self.length = self.raw_array[0].replace(")", "").split()[2]
-        self.array = self.raw_array[1:]
+        self.array = self.raw.split("\n")[1:]
 
-        self.header = self.compose_header()
-        self.body = self.compose_body()
-        self.body2 = self.compose_body_list()
+        self.header = self.__compose_header()
+        self.body = self.__compose_body()
+        self.body2 = self.__compose_body_list()
 
-        self.update_header()
+        self.__update_header()
 
-    def compose_header(self):
+    def __compose_header(self):
         return [
             a.replace("Layer", "").replace(":", "").replace(" ", "")
             for a in self.array
@@ -35,7 +33,7 @@ def __is_data_element(self, layer_fragment: str) -> bool:
             return True
         return False
 
-    def compose_body(self) -> dict[str, list[str]]:
+    def __compose_body(self) -> dict[str, list[str]]:
         temp_body_dict: dict[str, list[str]] = {}
         actual_layer: str = ""
         for x in self.array:
@@ -51,7 +49,7 @@ def compose_body(self) -> dict[str, list[str]]:
             temp_body_dict[actual_layer].append(x)
         return temp_body_dict
 
-    def compose_body_list(self) -> list[list[str]]:
+    def __compose_body_list(self) -> list[list[str]]:
         temp_body_dict = []
         line = []
         ckhsum_flag = False
@@ -82,12 +80,12 @@ def compose_body_list(self) -> list[list[str]]:
 
         if ckhsum_flag:
             for y in temp_body_dict:
-                self.chksum_verification(y)
+                self.__chksum_verification(y)
 
         temp_body_dict.append(data_found)
         return temp_body_dict
 
-    def chksum_verification(self, element) -> None:
+    def __chksum_verification(self, element) -> None:
         chksum_status: ChecksumStatus = ChecksumStatus()
         for x in element:
             x = x.lower()
@@ -104,7 +102,7 @@ def chksum_verification(self, element) -> None:
             chksum_status.verify()
             self.chksum_list.append(chksum_status)
 
-    def update_header(self):
+    def __update_header(self) -> None:
         """Update header with data layer which is 'hidden' in the tshark output"""
         if self._data_layer:
             self.header.append("RAW")

packet_helper_core/packet_data.py

@@ -7,13 +7,12 @@ def from_sh_list(packet_list) -> BasePacket:
     imported_all = importlib.import_module("scapy.all")
 
     def remove_none():
-        return {k: v for k, v in _value.items() if v is not None}
+        return {k: v for k, v in layer.get(_key, {}).items() if v is not None}
 
     new_packet = None
     for layer in packet_list:
         if isinstance(layer, dict):
             _key = [x for x in layer.keys()][0]
-            _value = layer.get(_key)
             _value = remove_none()
             if _key == "Ethernet":
                 _key = "Ether"

packet_helper_core/utils/conversion.py

@@ -2,7 +2,7 @@
 
 from packet_helper_core import PacketData, PacketDataScapy
 from packet_helper_core.checksum_status import ChecksumStatus
-from packet_helper_core.core import Core
+from packet_helper_core.decoder import Decoder
 from scapy.all import IP, TCP, Ether  # noqa
 from scapy_helper import get_hex
 
@@ -13,21 +13,21 @@ def test_core_post_init():
         "IP",
         "TCP",
     ]
-    core_results = Core(get_hex(Ether() / IP() / TCP()))
+    decoder = Decoder(get_hex(Ether() / IP() / TCP()))
+    decoder.run()
 
-    assert isinstance(core_results.hex_string, str), "Should be String"
-    assert isinstance(
-        core_results.scapy_data, PacketDataScapy
-    ), "Should be PacketDataScapy"
-    assert isinstance(core_results.tshark_data, PacketData), "Should be PacketData"
-    assert core_results.scapy_data.header == expected_headers, "Should be properly decoded"
-    assert core_results.tshark_data.header == expected_headers, "Should be properly decoded"
+    assert isinstance(decoder.hex_string, str), "Should be String"
+    assert isinstance(decoder.scapy_data, PacketDataScapy), "Should be PacketDataScapy"
+    assert isinstance(decoder.tshark_data, PacketData), "Should be PacketData"
+    assert decoder.scapy_data.header == expected_headers, "Should be properly decoded"
+    assert decoder.tshark_data.header == expected_headers, "Should be properly decoded"
 
 
 def test_core_chksum_verification():
-    core_results = Core(get_hex(Ether() / IP() / IP() / TCP()))
-    assert core_results.tshark_data.chksum_list
-    assert len(core_results.tshark_data.chksum_list) == 4
+    decoder = Decoder(get_hex(Ether() / IP() / IP() / TCP()))
+    decoder.run()
+    assert decoder.tshark_data.chksum_list
+    assert len(decoder.tshark_data.chksum_list) == 4
 
 
 @pytest.mark.parametrize(
@@ -48,22 +48,24 @@ def test_core_chksum_verification():
 def test_negative_core_chksum_verification_with_wrong_chksum(
     packet: str, position_to_check: int, expected_chksum_value: str
 ):
-    core_results2 = Core(packet)
+    decoder = Decoder(packet)
+    decoder.run()
     assert (
-        core_results2.tshark_data.chksum_list[position_to_check].chksum
+        decoder.tshark_data.chksum_list[position_to_check].chksum
         == expected_chksum_value
     )
 
 
 def test_ethernet_ip_udp_dns():
-    core_result = Core(
+    decoder = Decoder(
         "00E01CCCCCC2001F33D9736108004500008000004000401124550A0A01010"
         "A0A01040035DB66006C2D2E795681800001000200020000046D61696C0870"
         "617472696F747302696E0000010001C00C0005000100002A4B0002C011C01"
         "10001000100002A4C00044A358C99C011000200010001438C0006036E7332"
         "C011C011000200010001438C0006036E7331C011"
     )
-    chksum_obj: ChecksumStatus = core_result.tshark_data.chksum_list[2]
+    decoder.run()
+    chksum_obj: ChecksumStatus = decoder.tshark_data.chksum_list[2]
 
     assert chksum_obj.chksum == "0x2d2e"
     assert chksum_obj.chksum_calculated == "0x2d2d"

tests/test_core.py renamed to tests/test_decoder.py

@@ -1,4 +1,8 @@
+from typing import Any
+
 import pytest
+from scapy.base_classes import BasePacket
+
 from packet_helper_core.utils.conversion import from_sh_list
 from scapy.packet import Packet
 from scapy_helper import get_hex, to_list
@@ -10,8 +14,11 @@
     "packet", (SIMPLE_IP_IN_IP_PACKET, SIMPLE_IP_IN_IP_PACKET / SIMPLE_IP_IN_IP_PACKET)
 )
 def test_from_sh_list(packet: Packet) -> None:
-    packet_list = to_list(packet)
-    new_packet = from_sh_list(packet_list)
+    packet_list: list[dict[str, Any]] = to_list(packet)
+    packet_generated_from_scapy_helper = from_sh_list(packet_list)
 
-    assert get_hex(packet) == get_hex(new_packet)
-    assert packet_list == to_list(new_packet)
+    assert isinstance(packet_generated_from_scapy_helper, BasePacket)
+    assert get_hex(packet) == get_hex(
+        packet_generated_from_scapy_helper
+    ), "Packets should return same hex results"
+    assert packet_list == to_list(packet_generated_from_scapy_helper)