kaitai_tests.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from kaitaistruct import * \n",
    "from scapy.all import * \n",
    "from scapy.contrib.modbus import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "\n",
    "cwd = Path().resolve()\n",
    "print(cwd)\n",
    "\n",
    "lib_path = os.path.abspath(os.path.join(cwd))\n",
    "sys.path.append(lib_path)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "counter = 0\n",
    "def callback(pkt):\n",
    "    global counter\n",
    "    print(counter) \n",
    "    pkt.show()\n",
    "    counter += 1\n",
    "    \n",
    "packets = sniff(offline=lib_path + \"/data/pcaps/cap3.pcapng\", prn=callback, count=100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(lib_path+\"/data/binary/pkt\", \"wb+\") as file:\n",
    "    file.write(bytes(packets[0]))\n",
    "    file.write(bytes(packets[1]))\n",
    "    file.write(bytes(packets[2]))\n",
    " "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(lib_path+\"/data/binary/only_modbus\", \"wb\") as file:\n",
    "    file.write(bytes(packets[0]['TCP'].payload))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(cwd)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from kaitaistruct import KaitaiStream, BytesIO\n",
    "from src.kaitai_py.ethernet_frame import EthernetFrame\n",
    "from src.kaitai_py.udp_datagram import UdpDatagram\n",
    "from src.kaitai_py.tcp_segment import TcpSegment\n",
    "from src.kaitai_py.icmp_packet import IcmpPacket\n",
    "from src.kaitai_py.ipv6_packet import Ipv6Packet\n",
    "from src.kaitai_py.ipv4_packet import Ipv4Packet\n",
    "\n",
    "raw = (bytes(packets[2]))\n",
    "\n",
    "data = EthernetFrame(KaitaiStream(BytesIO(raw)))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data\n",
    "#data = EthernetFrame.from_file(lib_path+\"/kaitai/dane/pkt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('ttl:', data.body.ttl)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data.body.body"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data.body.body.body"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\n",
    "from __future__ import print_function\n",
    "import collections\n",
    "import textwrap\n",
    "from optparse import OptionParser\n",
    "import codecs as c\n",
    "\n",
    "from pymodbus.factory import ClientDecoder, ServerDecoder\n",
    "from pymodbus.transaction import ModbusSocketFramer\n",
    "from pymodbus.transaction import ModbusBinaryFramer\n",
    "from pymodbus.transaction import ModbusAsciiFramer\n",
    "from pymodbus.transaction import ModbusRtuFramer\n",
    "from pymodbus.compat import  IS_PYTHON3\n",
    "\n",
    "\n",
    "class Decoder(object):\n",
    "\n",
    "    def __init__(self, framer, encode=False):\n",
    "        \"\"\" Initialize a new instance of the decoder\n",
    "        :param framer: The framer to use\n",
    "        :param encode: If the message needs to be encoded\n",
    "        \"\"\"\n",
    "        self.framer = framer\n",
    "        self.encode = encode\n",
    "        self.name = None\n",
    "        self.unit = None\n",
    "    \n",
    "    def decode(self, message):\n",
    "        \"\"\" Attempt to decode the supplied message\n",
    "        :param message: The messge to decode\n",
    "        \"\"\"\n",
    "\n",
    "        value = message if self.encode else c.encode(message, 'hex_codec')\n",
    "        print(\"=\"*80)\n",
    "        print(\"Decoding Message %s\" % value)\n",
    "        print(\"=\"*80)\n",
    "        decoders = [\n",
    "            self.framer(ServerDecoder(), client=None),\n",
    "            self.framer(ClientDecoder(), client=None)\n",
    "        ]\n",
    "        for decoder in decoders:\n",
    "            print(\"%s\" % decoder.decoder.__class__.__name__)\n",
    "            print(\"-\"*80)\n",
    "            try:\n",
    "                decoder.addToFrame(message)\n",
    "                if decoder.checkFrame():\n",
    "                    self.unit = decoder._header.get(\"uid\", 0x00)\n",
    "                    unit = decoder._header.get(\"uid\", 0x00)\n",
    "                    decoder.advanceFrame()\n",
    "                    decoder.processIncomingPacket(message, self.report, unit)\n",
    "                else:\n",
    "                    self.check_errors(decoder, message)\n",
    "            except Exception as ex:\n",
    "                self.check_errors(decoder, message)\n",
    "\n",
    "    def check_errors(self, decoder, message):\n",
    "        \"\"\" Attempt to find message errors\n",
    "        :param message: The message to find errors in\n",
    "        \"\"\"\n",
    "        print(\"Unable to parse message - {} with {}\".format(message,\n",
    "                                                                decoder))\n",
    "\n",
    "    def report(self, message):\n",
    "        \"\"\" The callback to print the message information\n",
    "        :param message: The message to print\n",
    "        \"\"\"\n",
    "        \n",
    "        print(\"%-15s = %s\" % ('name', message.__class__.__name__))\n",
    "        for (k, v) in message.__dict__.items():\n",
    "            if isinstance(v, dict):\n",
    "                print(\"%-15s =\" % k)\n",
    "                for kk,vv in v.items():\n",
    "                    print(\"  %-12s => %s\" % (kk, vv))\n",
    "\n",
    "            elif isinstance(v, collections.Iterable):\n",
    "                print(\"%-15s =\" % k)\n",
    "                value = str([int(x) for x  in v])\n",
    "                for line in textwrap.wrap(value, 60):\n",
    "                    print(\"%-15s . %s\" % (\"\", line))\n",
    "            else:\n",
    "                print(\"%-15s = %s\" % (k, hex(v)))\n",
    "        print(\"%-15s = %s\" % ('documentation', message.__doc__))\n",
    "\n",
    "\n",
    "# -------------------------------------------------------------------------- #\n",
    "# and decode our message\n",
    "# -------------------------------------------------------------------------- #\n",
    "\n",
    "def get_messages(option):\n",
    "    \"\"\" A helper method to generate the messages to parse\n",
    "    :param options: The option manager\n",
    "    :returns: The message iterator to parse\n",
    "    \"\"\"\n",
    "    if option.message:\n",
    "        if option.transaction:\n",
    "            msg = \"\"\n",
    "            for segment in option.message.split():\n",
    "                segment = segment.replace(\"0x\", \"\")\n",
    "                segment = \"0\" + segment if len(segment) == 1 else segment\n",
    "                msg = msg + segment\n",
    "            option.message = msg\n",
    "\n",
    "        if not option.ascii:\n",
    "            if not IS_PYTHON3:\n",
    "                option.message = option.message.decode('hex')\n",
    "            else:\n",
    "                option.message = c.decode(option.message.encode(), 'hex_codec')\n",
    "        yield option.message\n",
    "    elif option.file:\n",
    "        with open(option.file, \"r\") as handle:\n",
    "            for line in handle:\n",
    "                if line.startswith('#'): continue\n",
    "                if not option.ascii:\n",
    "                    line = line.strip()\n",
    "                    line = line.decode('hex')\n",
    "                yield line\n",
    "\n",
    "\n",
    "def main():\n",
    "    \"\"\" \n",
    "        The main runner function\n",
    "    \"\"\"\n",
    "    #option = get_options()\n",
    "\n",
    "\n",
    "    framer = ModbusSocketFramer\n",
    "    decoder = Decoder(framer)\n",
    "    decoder.decode(data.body.body.body)\n",
    "\n",
    "\n",
    "main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}