Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.fields = set()
offset = 0
unpacker = "!"
discovered_fields = []
# Iterate through all fields of this template and build the unpack format string
# See https://www.iana.org/assignments/ipfix/ipfix.xhtml
for index, field in enumerate(template):
field_type_id = field.id
field_length = field.length
offset += field_length
# Here, reduced-size encoding of fields blocks the usage of IPFIXFieldTypes.get_type_unpack.
# See comment in IPFIXFieldTypes.get_type_unpack for more information.
field_type = IPFIXFieldTypes.by_id(field_type_id) # type: Optional[FieldType]
if not field_type and type(field) is not TemplateFieldEnterprise:
# This should break, since the exporter seems to use a field identifier
# which is not standardized by IANA.
raise NotImplementedError("Field type with ID {} is not implemented".format(field_type_id))
datatype = field_type.type # type: str
discovered_fields.append((field_type.name, field_type_id))
# Catch fields which are meant to be raw bytes and skip the rest
if IPFIXDataTypes.is_bytes(datatype):
unpacker += "{}s".format(field_length)
continue
# Go into int, uint, float types
issigned = IPFIXDataTypes.is_signed(datatype)
isfloat = IPFIXDataTypes.is_float(datatype)
def data(self):
return {
IPFIXFieldTypes.by_id(key)[1]: value for (key, value) in self.fields
}