Skip to content

Commit 8b93e5c

Browse files
author
Nils Weiss
committed
Implemented now (feasible)
1. Copilot: size_len=0 vs unspecified — Real bug. BERcodec_SEQUENCE.enc now defaults to size_len=None and only applies conf.ASN1_default_long_size when size_len is None, so size_len=0 can force short-form lengths. Regression test added. 2. guedou: drop unused helpers — Removed set_absent, SEQUENCE/CHOICE _m2i_ber, _register_choice, choice_order/choice_list, and unused _choice_* APIs. Logic inlined back into m2i / __init__ so this PR stays focused on codec-stem hooks. Kept on purpose • SEQUENCE OF holds_packets split — Needed for correct BER with tagged element fields (not for other encodings). Clarified the comment. • Prior encoding bugfixes (ASN1_Object/Packet in _encode_item) — Required so LDAP/Kerberos still work with the new encode path. asn1.uts, kerberos.uts, and ldap.uts all pass. Changes are uncommitted if you want a commit next. AI-Assisted: yes (Cursor)
1 parent 1f8da12 commit 8b93e5c

3 files changed

Lines changed: 57 additions & 98 deletions

File tree

scapy/asn1/ber.py

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -635,13 +635,14 @@ class BERcodec_SEQUENCE(BERcodec_Object[Union[bytes, List[BERcodec_Object[Any]]]
635635
tag = ASN1_Class_UNIVERSAL.SEQUENCE
636636

637637
@classmethod
638-
def enc(cls, _ll, size_len=0):
638+
def enc(cls, _ll, size_len=None):
639639
# type: (Union[bytes, List[BERcodec_Object[Any]]], Optional[int]) -> bytes
640640
if isinstance(_ll, bytes):
641641
ll = _ll
642642
else:
643643
ll = b"".join(x.enc(cls.codec) for x in _ll)
644-
if not size_len:
644+
# None = apply conf; explicit 0 keeps short-form lengths.
645+
if size_len is None:
645646
size_len = conf.ASN1_default_long_size
646647
return chb(int(cls.tag)) + BER_len_enc(len(ll), size=size_len) + ll
647648

scapy/asn1fields.py

Lines changed: 50 additions & 96 deletions
Original file line numberDiff line numberDiff line change
@@ -526,38 +526,31 @@ def get_fields_list(self):
526526
return reduce(lambda x, y: x + y.get_fields_list(),
527527
self.seq, [])
528528

529-
def _dissect_sequence_children(self, pkt, s):
530-
# type: (Any, bytes) -> bytes
531-
if len(s) == 0:
532-
for obj in self.seq:
533-
obj.set_val(pkt, None)
534-
return s
535-
for obj in self.seq:
536-
try:
537-
s = obj.dissect(pkt, s)
538-
except ASN1F_badsequence:
539-
break
540-
return s
541-
542-
def _m2i_ber(self, pkt, s):
543-
# type: (Any, bytes) -> Tuple[Any, bytes]
544-
s = self._apply_tagging_dec(s, pkt, _fname=pkt.name)
545-
codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)
546-
i, s, remain = codec.check_type_check_len(s)
547-
s = self._dissect_sequence_children(pkt, s)
548-
if len(s) > 0:
549-
raise BER_Decoding_Error("unexpected remainder", remaining=s)
550-
return [], remain
551-
552529
def m2i(self, pkt, s):
553530
# type: (Any, bytes) -> Tuple[Any, bytes]
554531
"""
555532
ASN1F_SEQUENCE behaves transparently, with nested ASN1_objects being
556-
dissected one by one. m2i returns an empty list (along with the proper
557-
remainder). It is discarded by dissect() and should not be missed
558-
elsewhere.
533+
dissected one by one. Because we use obj.dissect (see loop below)
534+
instead of obj.m2i (as we trust dissect to do the appropriate set_vals)
535+
we do not directly retrieve the list of nested objects.
536+
Thus m2i returns an empty list (along with the proper remainder).
537+
It is discarded by dissect() and should not be missed elsewhere.
559538
"""
560-
return self._m2i_ber(pkt, s)
539+
s = self._apply_tagging_dec(s, pkt, _fname=pkt.name)
540+
codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)
541+
i, s, remain = codec.check_type_check_len(s)
542+
if len(s) == 0:
543+
for obj in self.seq:
544+
obj.set_val(pkt, None)
545+
else:
546+
for obj in self.seq:
547+
try:
548+
s = obj.dissect(pkt, s)
549+
except ASN1F_badsequence:
550+
break
551+
if len(s) > 0:
552+
raise BER_Decoding_Error("unexpected remainder", remaining=s)
553+
return [], remain
561554

562555
def dissect(self, pkt, s):
563556
# type: (Any, bytes) -> bytes
@@ -655,7 +648,8 @@ def build(self, pkt):
655648
elif self.holds_packets:
656649
s = b"".join(bytes(i) for i in val)
657650
else:
658-
# Use i2m so element implicit/explicit tags match m2i()/fld.m2i()
651+
# BER: element fields may carry implicit/explicit tags; i2m
652+
# matches m2i()/fld.m2i(). (Packet elements use bytes() above.)
659653
s = b"".join(self.fld.i2m(pkt, i) for i in val)
660654
return self.i2m(pkt, s)
661655

@@ -741,11 +735,6 @@ def i2repr(self, pkt, x):
741735
# type: (ASN1_Packet, Any) -> str
742736
return self._field.i2repr(pkt, x)
743737

744-
def set_absent(self, pkt):
745-
# type: (ASN1_Packet) -> None
746-
# Used by codecs that track optionality explicitly (e.g. PER).
747-
self._field.set_val(pkt, None)
748-
749738

750739
class ASN1F_omit(ASN1F_field[None, None]):
751740
"""
@@ -788,66 +777,30 @@ def __init__(self, name, default, *args, **kwargs):
788777
self.default = default
789778
self.current_choice = None
790779
self.choices = {} # type: Dict[int, _CHOICE_T]
791-
self.choice_order = [] # type: List[int]
792-
self.choice_list = [] # type: List[_CHOICE_T]
793780
self.pktchoices = {}
794781
for p in args:
795782
if hasattr(p, "ASN1_root"):
796783
p = cast('ASN1_Packet', p)
797784
# should be ASN1_Packet
798785
if hasattr(p.ASN1_root, "choices"):
799786
root = cast(ASN1F_CHOICE, p.ASN1_root)
800-
for k in root.choice_order:
801-
self._register_choice(k, root.choices[k])
787+
for k, v in root.choices.items():
788+
# ASN1F_CHOICE recursion
789+
self.choices[k] = v
802790
else:
803-
self._register_choice(p.ASN1_root.network_tag, p)
791+
self.choices[p.ASN1_root.network_tag] = p
804792
elif hasattr(p, "ASN1_tag"):
805793
if isinstance(p, type):
806794
# should be ASN1F_field class
807-
self._register_choice(int(p.ASN1_tag), p)
795+
self.choices[int(p.ASN1_tag)] = p
808796
else:
809797
# should be ASN1F_field instance
810-
self._register_choice(p.network_tag, p)
798+
self.choices[p.network_tag] = p
811799
if hasattr(p, "cls"):
812800
self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501
813801
else:
814802
raise ASN1_Error("ASN1F_CHOICE: no tag found for one field")
815803

816-
def _register_choice(self, tag, choice):
817-
# type: (int, _CHOICE_T) -> None
818-
self.choices[tag] = choice
819-
self.choice_order.append(tag)
820-
self.choice_list.append(choice)
821-
822-
def _dissect_choice_payload(self, pkt, choice, payload):
823-
# type: (ASN1_Packet, _CHOICE_T, bytes) -> Tuple[ASN1_Object[Any], bytes]
824-
if hasattr(choice, "ASN1_root"):
825-
return self.extract_packet(choice, payload, _underlayer=pkt) # type: ignore
826-
if isinstance(choice, type):
827-
return choice(self.name, b"").m2i(pkt, payload)
828-
return choice.m2i(pkt, payload)
829-
830-
def _m2i_ber(self, pkt, s):
831-
# type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes]
832-
s = self._apply_tagging_dec(s, pkt)
833-
tag, _ = BER_id_dec(s)
834-
return self._m2i_tagged(pkt, tag, s)
835-
836-
def _m2i_tagged(self, pkt, tag, payload):
837-
# type: (ASN1_Packet, int, bytes) -> Tuple[ASN1_Object[Any], bytes]
838-
if tag in self.choices:
839-
choice = self.choices[tag]
840-
elif self.flexible_tag:
841-
choice = ASN1F_field
842-
else:
843-
raise ASN1_Error(
844-
"ASN1F_CHOICE: unexpected field in '%s' "
845-
"(tag %s not in possible tags %s)" % (
846-
self.name, tag, list(self.choices.keys())
847-
)
848-
)
849-
return self._dissect_choice_payload(pkt, choice, payload)
850-
851804
def m2i(self, pkt, s):
852805
# type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes]
853806
"""
@@ -856,27 +809,28 @@ def m2i(self, pkt, s):
856809
"""
857810
if len(s) == 0:
858811
raise ASN1_Error("ASN1F_CHOICE: got empty string")
859-
return self._m2i_ber(pkt, s)
860-
861-
def _choice_index_for(self, x):
862-
# type: (Any) -> Optional[int]
863-
for index, choice in enumerate(self.choice_list):
864-
if isinstance(choice, type) and hasattr(choice, "ASN1_root"):
865-
if isinstance(x, choice):
866-
return index
867-
elif hasattr(choice, "ASN1_tag"):
868-
if isinstance(x, ASN1_Object) and x.tag == choice.ASN1_tag:
869-
return index
870-
return None
871-
872-
def _choice_for_index(self, index):
873-
# type: (int) -> _CHOICE_T
874-
return self.choice_list[index]
875-
876-
def _choice_tag_for(self, x):
877-
# type: (Any) -> Optional[int]
878-
index = self._choice_index_for(x)
879-
return None if index is None else self.choice_order[index]
812+
s = self._apply_tagging_dec(s, pkt)
813+
tag, _ = BER_id_dec(s)
814+
if tag in self.choices:
815+
choice = self.choices[tag]
816+
else:
817+
if self.flexible_tag:
818+
choice = ASN1F_field
819+
else:
820+
raise ASN1_Error(
821+
"ASN1F_CHOICE: unexpected field in '%s' "
822+
"(tag %s not in possible tags %s)" % (
823+
self.name, tag, list(self.choices.keys())
824+
)
825+
)
826+
if hasattr(choice, "ASN1_root"):
827+
# we don't want to import ASN1_Packet in this module...
828+
return self.extract_packet(choice, s, _underlayer=pkt) # type: ignore
829+
elif isinstance(choice, type):
830+
return choice(self.name, b"").m2i(pkt, s)
831+
else:
832+
# XXX check properly if this is an ASN1F_PACKET
833+
return choice.m2i(pkt, s)
880834

881835
def i2m(self, pkt, x):
882836
# type: (ASN1_Packet, Any) -> bytes

test/scapy/layers/ber_codec.py

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -231,6 +231,10 @@ def check_ber_sequence_and_set():
231231
try:
232232
long_seq = BERcodec_SEQUENCE.enc(payload)
233233
assert long_seq.startswith(b"0\x84")
234+
# Explicit size_len=0 must keep short-form lengths even when the
235+
# LDAP-style default long size is set.
236+
short_seq = BERcodec_SEQUENCE.enc(payload, size_len=0)
237+
assert short_seq[0] == 0x30 and short_seq[1] == len(payload)
234238
finally:
235239
conf.ASN1_default_long_size = 0
236240

0 commit comments

Comments
 (0)