File: //lib64/mft/python_tools/mlxste/loaders/raw.py
# Copyright (C) Nov 2020 Mellanox Technologies Ltd. All rights reserved.
#
# This software is available to you under a choice of one of two
# licenses. You may choose to be licensed under the terms of the GNU
# General Public License (GPL) Version 2, available from the file
# COPYING in the main directory of this source tree, or the
# OpenIB.org BSD license below:
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
# - Redistributions of source code must retain the above
# copyright notice, this list of conditions and the following
# disclaimer.
#
# - Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following
# disclaimer in the documentation and/or other materials
# provided with the distribution.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
# --
import binascii
import re
from .loader import PacketLoader
from ..constants import PacketType
from ..packet import RawPacket
__all__ = ['RawPacketLoader', 'RawHexPacketLoader']
class RawPacketLoader(PacketLoader):
"""A concrete class that represents packet loader for raw data packets."""
__product_type__ = PacketType.RAW.value
def read_packets_file(self):
"""Reads and return the content of the input file.
:return: The content of the input file.
:rtype: bytes
"""
with open(self.path, 'rb') as f:
return f.read().rstrip()
def get_packets_buffers(self):
"""Returns list of packets buffers from the input file.
:return: list of packets buffers.
:rtype: List[bytes]
"""
buffer = self.read_packets_file()
return [buffer]
def __iter__(self):
for buffer in self.get_packets_buffers():
yield RawPacket(buffer)
class RawHexPacketLoader(RawPacketLoader):
"""A concrete class that represents packet loader for hexadecimal
packets."""
__product_type__ = PacketType.HEX.value
@classmethod
def hex_to_bin(cls, buffer):
"""Return the binary data represented by hexadecimal buffer string.
:param bytes buffer: An hexadecimal buffer string.
:return: Binary data.
:rtype: bytes
"""
try:
buffer = re.sub(rb'\s+', b'', buffer)
return binascii.a2b_hex(buffer)
except binascii.Error as err:
error_format = 'invalid packet buffer: {}, error: {}'
raise ValueError(error_format.format(buffer, str(err).lower()))
def get_packets_buffers(self):
"""Returns list of packets buffers from the input file.
:return: list of packets buffers.
:rtype: List[bytes]
"""
buffers = self.read_packets_file().splitlines()
return [self.hex_to_bin(buffer) for buffer in buffers]