Merge with server

.github/workflows/ci.yml

@@ -23,7 +23,7 @@ jobs:
           sudo apt-get install -y --allow-change-held-packages --force-yes tshark
       - name: Run pytest
         run: |
-          pytest
+          make test
   verify-code:
     name: Verify code w/Black&Flake8
     runs-on: ubuntu-latest

.gitignore

@@ -10,3 +10,5 @@
 venv/
 .mypy_cache/
 .pytest_cache/
+
+.DS_Store/

Makefile

@@ -25,3 +25,8 @@ build:
 	@echo "Building..."
 	python3 setup.py sdist bdist_wheel --universal
 	@echo "Building... Done"
+
+.PHONY: test
+test:
+	mkdir -p static && touch static/index.html
+	PYTHONPATH=${PWD} pytest tests/integration

README.md

@@ -17,7 +17,7 @@ sudo apt-get -y install wireshark
 sudo apt-get install -y --allow-change-held-packages --force-yes tshark
 ```
 
-Recommended *Python 3.8* (as minimal). 
+Recommended *Python 3.11* (as minimal). 
 
 To verify that all works, try to run test using `pytest` (in root directory of this package):
 

api/main.py

@@ -0,0 +1,52 @@
+from fastapi import FastAPI, Request, status
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+from starlette.templating import _TemplateResponse  # noqa
+
+from api.routers.api.create import api as api_create
+from api.routers.api.hex import api as api_hex
+from api.routers.api.info import api as api_info
+from api.routers.api.packets import api as api_packets
+from api.routers.version import version
+
+app = FastAPI(
+    title="Packet Helper Next",
+    description="Packet Helper API helps you to decode hex into packets with description 🚀",
+    version="0.1",
+    license_info={
+        "name": "GPL v2.0",
+        "url": "https://github.com/PacketHelper/packet-helper-next/blob/main/LICENSE",
+    },
+)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+app.mount("/static", StaticFiles(directory="static"), name="static")
+
+# rest api
+# /api/*
+router_api_config = {"prefix": "/api", "tags": ["api"]}
+for api_router in (api_create, api_hex, api_info, api_packets):
+    app.include_router(api_router, **router_api_config)
+
+# /*
+app.include_router(version)
+
+templates = Jinja2Templates(directory="static")
+
+
+@app.get("/", include_in_schema=False, status_code=status.HTTP_200_OK)
+def get_root(request: Request) -> _TemplateResponse:
+    """Return Vue single page"""
+    return templates.TemplateResponse("index.html", {"request": request})
+
+
+@app.get("/hex/{hex_string}", include_in_schema=False, status_code=status.HTTP_200_OK)
+def get_hex(request: Request) -> _TemplateResponse:
+    """Return specific path for Vue single-page"""
+    return templates.TemplateResponse("index.html", {"request": request})

api/models/creator_packets.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class CreatorPacketsRequest(BaseModel):
+    packets: list[Any]
+
+
+class CreatorPacketsResponse(BaseModel):
+    packets: list[dict[str, Any]] | None
+
+
+class CreatorPacketsObjectsRequest(BaseModel):
+    packets: list[dict[str, Any]]
+
+
+class CreatorPacketsObjectsResponse(BaseModel):
+    builtpacket: dict[str, str]  # FIXME rename => built_packet

api/models/decoded_hex.py

@@ -0,0 +1,17 @@
+from typing import Literal
+
+from pydantic import BaseModel
+
+from core.models.scapy_response import ScapyResponse
+
+
+class HexSummary(BaseModel):
+    length: int
+    length_unit: Literal["B", "b"]
+    hexdump: str
+
+
+class DecodedHexResponse(BaseModel):
+    hex: str
+    summary: HexSummary
+    structure: list[ScapyResponse]

api/models/info_response.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+
+
+class InfoResponse(BaseModel):
+    version: str
+    revision: str | None
+
+
+class VersionResponse(BaseModel):
+    packethelper: str  # FIXME rename => packet_helper
+    framework: str

api/routers/api/create.py

@@ -0,0 +1,25 @@
+from fastapi import APIRouter, HTTPException, status
+from core.utils.conversion import from_sh_list
+from scapy_helper import get_hex
+
+from api.models.creator_packets import (
+    CreatorPacketsObjectsRequest,
+    CreatorPacketsObjectsResponse,
+)
+
+api = APIRouter()
+
+
+@api.post("/create", status_code=status.HTTP_201_CREATED, tags=["api"])
+def post_api_create(
+    request: CreatorPacketsObjectsRequest,
+) -> CreatorPacketsObjectsResponse:
+    _hex = None
+    try:
+        _hex = get_hex(from_sh_list(request.packets))
+    except AttributeError as error:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"error": f"Layer is not supported {str(error).split()[-1]}"},
+        )
+    return CreatorPacketsObjectsResponse(builtpacket={"hex": _hex})

api/routers/api/hex.py

@@ -0,0 +1,56 @@
+import functools
+
+import pydantic
+from fastapi import APIRouter, HTTPException, status
+from core.decoders.scapy_data import ScapyData
+from core.decoders.tshark_data import TSharkData
+from core.decoders.decode_string import decode_string
+from scapy_helper import hexdump
+
+from api.models.decoded_hex import DecodedHexResponse
+from core.models.scapy_response import ScapyResponse
+
+api = APIRouter()
+
+
+@api.get("/hex/{hex_string}", status_code=status.HTTP_200_OK, tags=["api"])
+def get_api_hex(hex_string: str) -> DecodedHexResponse:
+    @functools.cache
+    def prepare_api_response(hex_to_decode: str) -> list[ScapyResponse]:
+        packet = decode_string(hex_to_decode)
+        packet_data = TSharkData(packet)
+        scapy_data = ScapyData(hex_to_decode, packet_data)
+
+        return scapy_data.packet_structure
+
+    h = " ".join(
+        [
+            "".join([hex_string[e - 1], hex_string[e]])
+            for e, _ in enumerate(hex_string)
+            if e % 2
+        ]
+    )
+
+    try:
+        response = DecodedHexResponse(
+            hex=hex_string,
+            summary={
+                "length": len(h.split()),
+                "length_unit": "B",
+                "hexdump": hexdump(h, dump=True),
+            },
+            structure=prepare_api_response(hex_string),
+        )
+    except IndexError:
+        raise HTTPException(
+            status.HTTP_400_BAD_REQUEST,
+            detail={
+                "error": f"Hex <{hex_string}> is incorrect. Is packet length is correct?"
+            },
+        )
+    except pydantic.error_wrappers.ValidationError as ve:
+        raise HTTPException(
+            status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"error": f"Incorrect response from engine: <{ve}>"},
+        )
+    return response

api/routers/api/info.py

@@ -0,0 +1,18 @@
+from os import getenv
+
+from fastapi import APIRouter, status
+
+from api.models.info_response import InfoResponse
+
+api = APIRouter()
+
+
+@api.get(
+    "/info",
+    description="Get information about packet helper version and revision",
+    status_code=status.HTTP_200_OK,
+    tags=["api"],
+)
+def get_info() -> InfoResponse:
+    ph_version = getenv("PH_VERSION", "v1.0.0:00000000").split(":")
+    return InfoResponse(version=ph_version[0], revision=ph_version[1])

api/routers/api/packets.py

@@ -0,0 +1,28 @@
+import importlib
+
+from fastapi import APIRouter, HTTPException, status
+from scapy_helper import to_list
+
+from api.models.creator_packets import CreatorPacketsRequest, CreatorPacketsResponse
+
+api = APIRouter()
+
+
+@api.post("/packets", status_code=status.HTTP_201_CREATED, tags=["api"])
+def post_api_packets(request: CreatorPacketsRequest) -> CreatorPacketsResponse:
+    imported_all = importlib.import_module("scapy.all")
+    packet = None
+    try:
+        for protocol in request.packets:
+            new_layer = imported_all.__getattribute__(protocol)
+            if packet is None:
+                packet = new_layer()
+                continue
+            packet /= new_layer()
+    except AttributeError as error:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"error": f"Layer is not supported {str(error).split()[-1]}"},
+        )
+
+    return CreatorPacketsResponse(packets=to_list(packet))

api/routers/version.py

@@ -0,0 +1,14 @@
+from fastapi import APIRouter, status
+
+
+from api.models.info_response import VersionResponse
+
+version = APIRouter()
+
+
+@version.get(
+    "/version", status_code=status.HTTP_200_OK, include_in_schema=False, deprecated=True
+)
+def get_version() -> VersionResponse:
+    """Return information about version of the Packet Helper"""
+    return VersionResponse(packethelper="0.1", framework="fastapi")

core/decoders/decode_string.py

@@ -0,0 +1,30 @@
+import functools
+
+from pyshark import InMemCapture
+from pyshark.packet.packet import Packet
+
+
+@functools.cache
+def decode_string(hex_str: str) -> Packet:
+    """
+    Decode string (in string-hex format) using 'InMemCapture' using a tshark.
+    """
+    _custom_params = [
+        "-o",
+        "tcp.check_checksum:TRUE",
+        "-o",
+        "ip.check_checksum:TRUE",
+        "-o",
+        "stt.check_checksum:TRUE",
+        "-o",
+        "udp.check_checksum:TRUE",
+        "-o",
+        "wlan.check_checksum:TRUE",
+    ]
+    # only interested with the first packet
+    packet = InMemCapture(custom_parameters=_custom_params)
+    decoded_packet: Packet = packet.parse_packet(
+        bytes.fromhex(hex_str.replace(" ", ""))
+    )
+    packet.close()
+    return decoded_packet

core/decoders/scapy_data.py

@@ -0,0 +1,58 @@
+from dataclasses import dataclass
+
+from scapy.packet import Packet
+from scapy_helper import get_hex
+
+from core.decoders.tshark_data import TSharkData
+from core.models.scapy_response import ScapyResponse
+from core.utils.scapy_reader import scapy_reader
+
+
+@dataclass
+class ScapyData:
+    raw: str
+    packet_data: TSharkData
+
+    def __post_init__(self):
+        self.headers: list[str] = [x.replace("\r", "") for x in self.packet_data.header]
+        self.scapy_headers: list[Packet] = scapy_reader(self.raw)
+
+        self.full_scapy_representation_headers: list[str] = [
+            repr(x) for x in self.scapy_headers
+        ]
+        self.single_scapy_representation_headers = [
+            f"{x.split(' |')[0]}>" for x in self.full_scapy_representation_headers
+        ]
+
+        self.packet_structure = self.__make_structure()
+
+    def __make_structure(self) -> list[ScapyResponse]:
+        scapy_responses: list[ScapyResponse] = []
+
+        for index, header in enumerate(self.scapy_headers):
+            scapy_header = self.scapy_headers[index].copy()
+            scapy_header.remove_payload()  # payload is not necessary for our usage in this case
+            scapy_data_dict: ScapyResponse = ScapyResponse(
+                **{
+                    "name": header.name,
+                    "bytes_record": str(header),
+                    "hex_record": get_hex(header),
+                    "hex_record_full": get_hex(scapy_header),
+                    "length": len(header),
+                    "length_unit": "B",
+                    "representation": f"{repr(header).split(' |')[0]}>",
+                    "representation_full": repr(header),
+                }
+            )
+
+            # RAW elements on the end are added to the last package as data!
+            try:
+                scapy_data_dict.tshark_name = self.packet_data.body2[index][0]
+                scapy_data_dict.tshark_raw_summary = self.packet_data.body2[index][1:]
+            except IndexError:
+                break
+
+            scapy_data_dict.chksum_status = self.packet_data.chksum_list[index]
+
+            scapy_responses.append(scapy_data_dict)
+        return scapy_responses