Wireshark Dissector for Starcom over IP

StarcomIP.lua

-- Refrences retrieved April 5 2023
-- StarComIP Encapsulation Spec: https://buyandsell.gc.ca/cds/public/2017/10/06/9d0480c5678292ea69a866e314b6e152/starcom_ip_implementation.pdf
-- StarCom Application Protocol Spec: https://buyandsell.gc.ca/cds/public/2017/11/08/e026edc197b91636a748c12daea4d4ca/ABES.PROD.PW__HN.B445.E73523.EBSU003.PDF
--
-- StarcomIP Wireshark LUA Dissector (C) William Hudson 2023
--
-- Lua 5.2
--

starcom = Proto("StarCom", "StarCom over IP")

opcodes = {
    [0] = "RESET",
    [1] = "SET DEVICE NUMBER",
    [2] = "SET POINT NUMBER",
    [3] = "ALARM STATUS",
    [4] = "HARDWARE STATUS",
    [5] = "DATE/TIME",
    [6] = "PRINT DATA-LOGGER TEXT",
    [7] = "DATA-LOGGER TEXT XOFF",
    [8] = "DATA-LOGGER TEXT XON"
}

field_signatureBytes = ProtoField.bytes("starcom.sig_bytes", "Signature Bytes", base.SPACE)
field_messageLength = ProtoField.uint16("starcom.msg_len", "Message Length", base.DEC)
field_applicationMessage = ProtoField.bytes("starcom.message", "StarCom Application Message", base.SPACE)
field_opcode = ProtoField.uint8('starcom.opcode', "Message Type", base.DEC, opcodes)
field_dataLength = ProtoField.uint8('starcom.data_len', "Data Count", base.DEC)
field_msgData = ProtoField.bytes("starcom.data", "Data", base.SPACE)
field_pointNum = ProtoField.uint16("starcom.point", "StarCom Point", base.DEC)
field_deviceNum = ProtoField.uint16("starcom.device", "StarCom Device", base.DEC)
-- alarm status value
field_asv_detection = ProtoField.bool("starcom.alarm.detection", "Alarm Status (Detection)", 8, nil, 0x01, "detection alarm")
field_asv_tamper = ProtoField.bool("starcom.alarm.tamper", "Alarm Status (Tamper)", 8, nil, 0x02, "tamper alarm")
field_asv_fail = ProtoField.bool("starcom.alarm.fail", "Alarm Status (Fail)", 8, nil, 0x04, "fail alarm")
-- hardware status value
field_hsv_rom = ProtoField.bool("starcom.hardware.rom_error", "Hardware Status (ROM Error)", 8, nil, 0x01, "rom error")
field_hsv_ram = ProtoField.bool("starcom.hardware.ram_error", "Hardware Status (RAM Error)", 8, nil, 0x02, "ram error")
field_hsv_data = ProtoField.bool("starcom.hardware.data_error", "Hardware Status (Data Error)", 8, nil, 0x04, "data error")
field_hsv_device = ProtoField.bool("starcom.hardware.device_fault", "Hardware Status (Device Fault)", 8, nil, 0x08, "device fault")
field_hsv_power = ProtoField.bool("starcom.hardware.power_fault", "Hardware Status (Power Fault)", 8, nil, 0x10, "power fault")

starcom.fields = {
    field_signatureBytes,
    field_messageLength,
    field_applicationMessage,
    field_opcode,
    field_dataLength,
    field_msgData,
    field_pointNum,
    field_deviceNum,
    field_asv_detection,
    field_asv_tamper,
    field_asv_fail,
    field_hsv_rom,
    field_hsv_ram,
    field_hsv_data,
    field_hsv_device,
    field_hsv_power
}

expert_oob_length = ProtoExpert.new("starcom.oob_length", "StarCom message data length out-of-bounds", expert.group.MALFORMED, expert.severity.ERROR)
expert_bad_length = ProtoExpert.new("starcom.bad_length", "StarCom message data length out-of-spec", expert.group.PROTOCOL, expert.severity.ERROR)
expert_bad_data = ProtoExpert.new("starcom.bad_data", "StarCom message data out-of-spec", expert.group.PROTOCOL, expert.severity.ERROR)
expert_reset = ProtoExpert.new("starcom.reset", "StarCom RESET message", expert.group.SEQUENCE, expert.severity.WARN)
expert_hardware_fault = ProtoExpert.new("starcom.hardware_fault", "StarCom message indicating a hardware fault", expert.group.RESPONSE_CODE, expert.severity.WARN)
expert_in_alarm = ProtoExpert.new("starcom.in_alarm", "StarCom message indicating an alarm state", expert.group.RESPONSE_CODE, expert.severity.NOTE)

starcom.experts = {
    expert_oob_length,
    expert_bad_length,
    expert_bad_data,
    expert_reset,
    expert_in_alarm,
    expert_hardware_fault
}

local HEADER_SIZE = 4
local LEN_OFFSET = 2
local LEN_LEN = 2
local SIG_OFFSET = 0 
local SIG_LEN = 2
local SIG_BYTES_E034 = "\224\52"

-- I can use the PDU dissector to handle the signature check
-- is that sig bytes built into wireshark somehow?
-- see if heuristic thing is make sense?
function dissect_starcom(buffer, pinfo, root)
    length = buffer:len()
    if length == 0 then return end

    if buffer:raw(0,2) == SIG_BYTES_E034 then
        pinfo.cols.protocol = starcom.name

        local subtree = root:add(starcom, buffer(), "StarCom over IP")
        local msg_len = buffer(LEN_OFFSET,LEN_LEN)

        subtree:add(field_signatureBytes, buffer(SIG_OFFSET,SIG_LEN))
        subtree:add_le(field_messageLength, msg_len)
        subtree:append_text(", Len: "..msg_len:le_uint())

        -- TODO check IP wrapper len and make sure it's not OOB

        local DATA_HEADER_SIZE = 2
        local OPCODE_OFFSET = 0
        local OPCODE_LEN = 1
        local LEN_OFFSET = 1
        local LEN_LEN = 1
        local DATA_OFFSET = DATA_HEADER_SIZE
        local msg_offset = HEADER_SIZE
        local app_msg_tree = subtree:add(field_applicationMessage, buffer(msg_offset, msg_len:le_uint()))

        while msg_offset-DATA_HEADER_SIZE <= msg_len:le_uint() do
            local opcode = buffer(msg_offset+OPCODE_OFFSET,OPCODE_LEN)
            local opcode_tree = app_msg_tree:add(field_opcode, opcode)

            local data_len = buffer(msg_offset+LEN_OFFSET,LEN_LEN)
            local len_tree = app_msg_tree:add(field_dataLength, data_len)

            -- make sure the reported length is not too long for the buffer
            if msg_offset+DATA_OFFSET+data_len:uint() > length then
                app_msg_tree:add_proto_expert_info(expert_oob_length)
                return length -- stop.
            end

            local data, data_tree
            if data_len:uint() ~= 0 then
                data = buffer(msg_offset+DATA_OFFSET, data_len:uint())
                data_tree = app_msg_tree:add(field_msgData, data)
            end

            if opcode:uint() == 0 then
                opcode_tree:add_proto_expert_info(expert_reset)
                if data_len:uint() == 1 then
                    if data:uint() == 0 then
                        data_tree:add("Hard Reset (controller boot-up)"):set_generated(true)
                    elseif data:uint() == 1 then
                        data_tree:add("Software Reset (communication failure)"):set_generated(true)
                    else
                        data_tree:add_proto_expert_info(expert_bad_data)
                    end
                else
                    len_tree:add_proto_expert_info(expert_bad_length)
                end
            elseif opcode:uint() == 1 then
                if data_len:uint() == 1 then
                    data_tree:add(field_deviceNum, data)
                else
                    len_tree:add_proto_expert_info(expert_bad_length)
                end
            elseif opcode:uint() == 2 then
                if data_len:uint() == 2 then
                    data_tree:add_le(field_pointNum, data)
                else
                    len_tree:add_proto_expert_info(expert_bad_length)
                end
            elseif opcode:uint() == 3 then
                if data_len:uint() == 0 then
                    app_msg_tree:add(buffer(msg_offset+OPCODE_OFFSET, OPCODE_LEN+LEN_LEN), "Alarm Status Request"):set_generated(true)
                elseif data_len:uint() >= 1 then
                    if bit32.band(data(0,1):uint(), 0x7) ~= 0 then -- 0x7 bitmask 00000111
                        data_tree:add_proto_expert_info(expert_in_alarm)
                    end
                    data_tree:add(field_asv_detection, data)
                    data_tree:add(field_asv_tamper, data)
                    data_tree:add(field_asv_fail, data)
                end
            elseif opcode:uint() == 4 then
                if data_len:uint() == 0 then
                    app_msg_tree:add(buffer(msg_offset+OPCODE_OFFSET, OPCODE_LEN+LEN_LEN), "Hardware Status Request"):set_generated(true)
                elseif data_len:uint() >= 1 then
                    if bit32.band(data(0,1):uint(), 0x1f) ~= 0 then -- 0x1f bitmask 00011111
                        data_tree:add_proto_expert_info(expert_hardware_fault)
                    end
                    data_tree:add(field_hsv_rom, data)
                    data_tree:add(field_hsv_ram, data)
                    data_tree:add(field_hsv_data, data)
                    data_tree:add(field_hsv_device, data)
                    data_tree:add(field_hsv_power, data)
                end
            elseif opcode:uint() == 5 then
                if data_len:uint() == 0 then
                    app_msg_tree:add(buffer(msg_offset+OPCODE_OFFSET, OPCODE_LEN+LEN_LEN), "Date/Time Request"):set_generated(true)
                elseif data_len:uint() == 6 then
                    -- TODO date/time value
                else
                    len_tree:add_proto_expert_info(expert_bad_length)
                end
            elseif opcode:uint() == 6 then 
                -- TODO data logger text message
            end

            msg_offset = msg_offset + 2 + data_len:uint()
        end
    end
    return length
end

function get_starcomIP_length(buffer, pinfo, pdu_offset)
    local msg_len = buffer(pdu_offset+LEN_OFFSET,LEN_LEN):le_uint()
    return HEADER_SIZE + msg_len
end

function starcom.dissector(buffer, pinfo, root)
    dissect_tcp_pdus(buffer, root, HEADER_SIZE, get_starcomIP_length, dissect_starcom)
end

local tcp_port = DissectorTable.get("tcp.port")
tcp_port:add(4002, starcom)