✅ 🔅 ℹ️

ℹ️ minor updates
🔅 added internal.functions.py
ℹ️ updated GHA cache with id
ℹ️ updated test.pcapng
🔅 changed how pcaps are loaded
✅ pcap_http_streams method added

.github/workflows/tests_multi_os.yml

@@ -18,6 +18,7 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
     - uses: actions/cache@v1
+      id: cache
       with:
         path: ~/.cache/pip
         key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
@@ -60,6 +61,7 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
     - uses: actions/cache@v1
+      id: cache
       with:
         path: ~\AppData\Local\pip\Cache
         key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
@@ -97,6 +99,7 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
     - uses: actions/cache@v1
+      id: cache
       with:
         path: ~/Library/Caches/pip
         key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}

README.md

@@ -1,5 +1,5 @@
 <p align="center">
-    <img src="https://raw.githubusercontent.com/securisec/chepy/master/logo.png" width="50%">
+    <img src="https://raw.githubusercontent.com/securisec/chepy/master/logo.png" width="65%">
 </p>
 
 ![](https://github.com/securisec/chepy/workflows/tests/badge.svg)

chepy/core.py

@@ -18,6 +18,7 @@
 
 from .modules.exceptions import PrintException
 from .modules.internal.colors import YELLOW, CYAN, GREEN, MAGENTA
+from .modules.internal.functions import full_duplex
 
 
 class ChepyDecorators(object):
@@ -79,7 +80,7 @@ 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_file = None
+        self._pcap_sessions = None
 
         #: Log level
         self.log_level = logging.INFO
@@ -935,9 +936,9 @@ def read_pcap(self):
         Returns:
             Chepy: The Chepy object. 
         """
-        pcap_file = scapy_utils.PcapReader(self.state)
+        pcap_file = scapy_utils.rdpcap(self.state)
         self.state = GREEN("Pcap loaded")
-        self._pcap_file = pcap_file
+        self._pcap_sessions = pcap_file.sessions(full_duplex)
         return self
 
     @ChepyDecorators.call_stack

chepy/modules/internal/functions.py

@@ -0,0 +1,62 @@
+import scapy.all as scapy
+
+
+def full_duplex(p):  # pragma: no cover
+    """Create a full duplex stream from packets
+    `Reference <https://gist.github.com/MarkBaggett/d8933453f431c111169158ce7f4e2222>`__
+    """
+    sess = "Other"
+    if "Ether" in p:
+        if "IP" in p:
+            if "TCP" in p:
+                sess = str(
+                    sorted(
+                        [
+                            "TCP",
+                            p[scapy.IP].src,
+                            p[scapy.TCP].sport,
+                            p[scapy.IP].dst,
+                            p[scapy.TCP].dport,
+                        ],
+                        key=str,
+                    )
+                )
+            elif "UDP" in p:
+                sess = str(
+                    sorted(
+                        [
+                            "UDP",
+                            p[scapy.IP].src,
+                            p[scapy.UDP].sport,
+                            p[scapy.IP].dst,
+                            p[scapy.UDP].dport,
+                        ],
+                        key=str,
+                    )
+                )
+            elif "ICMP" in p:
+                sess = str(
+                    sorted(
+                        [
+                            "ICMP",
+                            p[scapy.IP].src,
+                            p[scapy.IP].dst,
+                            p[scapy.ICMP].code,
+                            p[scapy.ICMP].type,
+                            p[scapy.ICMP].id,
+                        ],
+                        key=str,
+                    )
+                )
+            else:
+                sess = str(
+                    sorted(
+                        ["IP", p[scapy.IP].src, p[scapy.IP].dst, p[scapy.IP].proto],
+                        key=str,
+                    )
+                )
+        elif "ARP" in p:
+            sess = str(sorted(["ARP", p[scapy.ARP].psrc, p[scapy.ARP].pdst], key=str))
+        # else:
+        #     sess = p.sprintf("Ethernet type=%04xr,Ether.type%")
+    return sess

chepy/modules/pcap.py

@@ -1,4 +1,6 @@
+import scapy.all as scapy
 import scapy.layers.dns as scapy_dns
+import scapy.layers.http as scapy_http
 from scapy.utils import PcapNgReader, PcapReader
 
 from ..core import ChepyCore, ChepyDecorators
@@ -21,12 +23,58 @@ def pcap_dns_queries(self):
             ]
         """
         hold = []
-        for index, packet in enumerate(self._pcap_file):
-            try:
-                hold.append(
-                    {"frame": index + 1, "query": packet[scapy_dns.DNS].qd.qname}
-                )
-            except:
-                continue
+        sessions = self._pcap_sessions
+        for session in sessions:
+            packets = sessions.get(session)
+            for packet in packets:
+                if not packet.haslayer(scapy.DNSRR):
+                    continue
+                dns = packet.getlayer("DNS")
+                query = packet.getlayer("DNS").qd.qname
+                hold.append(query)
+        self.state = hold
+        return self
+
+    @ChepyDecorators.call_stack
+    def pcap_http_streams(self):
+        """Get a dict of HTTP req/res 
+
+        This method does full fully assemble when data exceeds a 
+        certain threshold. 
+        
+        Returns:
+            Chepy: The Chepy object. 
+        """
+        hold = []
+        sessions = self._pcap_sessions
+        for session in sessions:
+            packets = sessions.get(session)
+            req_res = {"request": {}, "response": {}}
+            for packet in packets:
+                if not packet.haslayer(scapy_http.HTTP):
+                    continue
+                if packet.haslayer(scapy_http.HTTPRequest):
+                    req_res["request"]["headers"] = packet.getlayer(
+                        scapy_http.HTTPRequest
+                    ).fields
+                    if packet.haslayer(scapy_http.Raw):
+                        req_res["request"]["payload"] = packet.getlayer(
+                            scapy_http.Raw
+                        ).load
+                    else:
+                        req_res["request"]["payload"] = {}
+                if packet.haslayer(scapy_http.HTTPResponse):
+                    req_res["response"]["headers"] = packet.getlayer(
+                        scapy_http.HTTPResponse
+                    ).fields
+                    if packet.haslayer(scapy_http.Raw):
+                        req_res["response"]["payload"] = packet.getlayer(
+                            scapy_http.Raw
+                        ).load
+                    else:  # pragma: no cover
+                        req_res["response"]["payload"] = {}
+            if len(req_res.get("request")):
+                hold.append(req_res)
+
         self.state = hold
         return self

docs/examples.md

@@ -1,7 +1,7 @@
 
 # Examples
 
-## Solving a CTF channel
+## Solving a StormCTF channel
 We are given a file called encoding which contains the following string.
 
 ```

tests/test_pcap.py

@@ -10,6 +10,17 @@ def test_pcap_dns():
             .pcap_dns_queries()
             .o
         )
-        == 6
+        == 3
+    )
+
+def test_pcap_http_streams():
+    assert (
+        len(
+            Chepy("tests/files/test.pcapng")
+            .read_pcap()
+            .pcap_http_streams()
+            .o
+        )
+        == 4
     )