✅ 🔅 ℹ️

🔅 added new core attribute _pcap_read to hold PcapReader generator
ℹ️ minor updates
✅ pcap_payload method added
✅ split_by_n method added

chepy/core.py

@@ -80,6 +80,8 @@ def __init__(self, *data):
         #: Holds all the methods that are called/chanined and their args
         self._stack = list()
         #: Holder for scapy pcap reader
+        self._pcap_read = None
+        #: Holder for scapy assembled sessions
         self._pcap_sessions = None
 
         #: Log level
@@ -937,6 +939,7 @@ def read_pcap(self):
             Chepy: The Chepy object. 
         """
         pcap_file = scapy_utils.rdpcap(self.state)
+        self._pcap_read = scapy_utils.PcapReader(self.state)
         self.state = GREEN("Pcap loaded")
         self._pcap_sessions = pcap_file.sessions(full_duplex)
         return self

chepy/modules/pcap.py

@@ -78,3 +78,24 @@ def pcap_http_streams(self):
 
         self.state = hold
         return self
+
+    @ChepyDecorators.call_stack
+    def pcap_payload(self, layer: str):
+        """Get an array of payloads based on provided layer
+        
+        Args:
+            layer (str): A valid Scapy layer. 
+        
+        Returns:
+            Chepy: The Chepy object. 
+        """
+        assert hasattr(scapy, layer), "Not a valid Scapy layer"
+        hold = []
+        for packet in self._pcap_read:
+            if not packet.haslayer(layer):
+                continue
+            check_raw = packet.haslayer("Raw")
+            if check_raw:
+                hold.append(packet.getlayer(scapy.Raw).load)
+        self.state = hold
+        return self

chepy/modules/utils.py

@@ -163,6 +163,19 @@ def split_by(self, pattern: str = "\n", trim=True):
             self.state = re.split(pattern, self._convert_to_str())
         return self
 
+    @ChepyDecorators.call_stack
+    def split_by_n(self, n: int):
+        """Split a string by n characters.
+        
+        Args:
+            n (int): [description]
+        
+        Returns:
+            Chepy: The Chepy object. 
+        """
+        self.state = re.findall(".{1," + str(n) + "}", self._convert_to_str())
+        return self
+
     @ChepyDecorators.call_stack
     def unique(self):
         """Get an array of unique list items

requirements.txt

@@ -27,5 +27,4 @@ requests
 requests-toolbelt
 scapy
 ua-parser
-ujson
-tqdm
+ujson

tests/test_pcap.py

@@ -13,14 +13,13 @@ def test_pcap_dns():
         == 3
     )
 
+
 def test_pcap_http_streams():
-    assert (
-        len(
-            Chepy("tests/files/test.pcapng")
-            .read_pcap()
-            .pcap_http_streams()
-            .o
-        )
-        == 4
-    )
+    assert len(Chepy("tests/files/test.pcapng").read_pcap().pcap_http_streams().o) == 4
+
+
+def test_pcap_payload():
+    assert Chepy("tests/files/test.pcapng").read_pcap().pcap_payload(
+        layer="ICMP"
+    ).o == [b"secret", b"message"]
 

tests/test_utils.py

@@ -67,6 +67,10 @@ def test_split_by():
     assert len(Chepy("some lol random lolol data").split_by("lo", trim=False).o) == 4
 
 
+def test_split_by_n():
+    assert Chepy("some string").split_by_n(2).o[2] == " s"
+
+
 def test_unique():
     assert len(Chepy('["a", "a", 1]').str_list_to_list().unique().o) == 2