Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 22 additions & 2 deletions dlms_cosem/protocol/acse/aare.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,26 @@ def application_context_name(self) -> acse_base.AppContextName:
@property
def protocol_version(self) -> int:
return 0

@classmethod
def _pop_ber_length(cls, source_bytes: bytearray) -> int:
"""Pop a BER length field from data and return the decoded length."""
first = source_bytes.pop(0)

# Short form length, value is in the first byte.
if first < 0x80:
return first

num_octets = first & 0x7F
if num_octets == 0:
raise ValueError("Indefinite BER length is not supported for ACSE APDUs")

if len(source_bytes) < num_octets:
raise ValueError("Insufficient data to decode BER long-form length")

length_bytes = bytes(source_bytes[:num_octets])
del source_bytes[:num_octets]
return int.from_bytes(length_bytes, byteorder="big")

@classmethod
def from_bytes(cls, source_bytes: bytes):
Expand All @@ -192,7 +212,7 @@ def from_bytes(cls, source_bytes: bytes):
if not aare_tag == cls.TAG:
raise ValueError("Bytes are not an AARQ APDU. TAg is not int(96)")

aare_length = aare_data.pop(0)
aare_length = cls._pop_ber_length(aare_data)

if not len(aare_data) == aare_length:
raise ValueError(
Expand All @@ -216,7 +236,7 @@ def from_bytes(cls, source_bytes: bytes):
f"Could not find object with tag {object_tag} "
f"in AARQ definition"
)
object_length = aare_data.pop(0)
object_length = cls._pop_ber_length(aare_data)
object_data = bytes(aare_data[:object_length])
aare_data = aare_data[object_length:]

Expand Down