Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/asn1fields.py: 74%

1# SPDX-License-Identifier: GPL-2.0-only

2# This file is part of Scapy

3# See https://scapy.net/ for more information

4# Copyright (C) Philippe Biondi <phil@secdev.org>

5# Acknowledgment: Maxence Tury <maxence.tury@ssi.gouv.fr>

7"""

8Classes that implement ASN.1 data structures.

9"""

11import copy

13from functools import reduce

15from scapy.asn1.asn1 import (

16 ASN1_BIT_STRING,

17 ASN1_BOOLEAN,

18 ASN1_Class,

19 ASN1_Class_UNIVERSAL,

20 ASN1_Decoding_Error,

21 ASN1_Error,

22 ASN1_INTEGER,

23 ASN1_NULL,

24 ASN1_OID,

25 ASN1_Object,

26 ASN1_STRING,

27)

28from scapy.asn1.ber import (

29 BER_Decoding_Error,

30 BER_id_dec,

31)

32from scapy.base_classes import BasePacket

33from scapy.volatile import (

34 GeneralizedTime,

35 RandChoice,

36 RandInt,

37 RandNum,

38 RandOID,

39 RandString,

40 RandField,

41)

43from scapy import packet

45from typing import (

46 Any,

47 AnyStr,

48 Callable,

49 Dict,

50 Generic,

51 List,

52 Optional,

53 Tuple,

54 Type,

55 TypeVar,

56 Union,

57 cast,

58 TYPE_CHECKING,

59)

61if TYPE_CHECKING:

62 from scapy.asn1packet import ASN1_Packet

65class ASN1F_badsequence(Exception):

66 pass

69class ASN1F_element(object):

70 pass

73##########################

74# Basic ASN1 Field #

75##########################

77_I = TypeVar('_I') # Internal storage

78_A = TypeVar('_A') # ASN.1 object

81class ASN1F_field(ASN1F_element, Generic[_I, _A]):

82 holds_packets = 0

83 islist = 0

84 ASN1_tag = ASN1_Class_UNIVERSAL.ANY

85 context = ASN1_Class_UNIVERSAL # type: Type[ASN1_Class]

87 def __init__(self,

88 name, # type: str

89 default, # type: Optional[_A]

90 context=None, # type: Optional[Type[ASN1_Class]]

91 implicit_tag=None, # type: Optional[int]

92 explicit_tag=None, # type: Optional[int]

93 flexible_tag=False, # type: Optional[bool]

94 size_len=None, # type: Optional[int]

95 ):

96 # type: (...) -> None

97 if context is not None:

98 self.context = context

99 self.name = name

100 if default is None:

101 self.default = default # type: Optional[_A]

102 elif isinstance(default, ASN1_NULL):

103 self.default = default # type: ignore

104 else:

105 self.default = self.ASN1_tag.asn1_object(default) # type: ignore

106 self.size_len = size_len

107 self.flexible_tag = flexible_tag

108 if (implicit_tag is not None) and (explicit_tag is not None):

109 err_msg = "field cannot be both implicitly and explicitly tagged"

110 raise ASN1_Error(err_msg)

111 self.implicit_tag = implicit_tag and int(implicit_tag)

112 self.explicit_tag = explicit_tag and int(explicit_tag)

113 # network_tag gets useful for ASN1F_CHOICE

114 self.network_tag = int(implicit_tag or explicit_tag or self.ASN1_tag)

115 self.owners = [] # type: List[Type[ASN1_Packet]]

116

117 def register_owner(self, cls):

118 # type: (Type[ASN1_Packet]) -> None

119 self.owners.append(cls)

120

121 def _apply_diff_tag(self, diff_tag):

122 # type: (Optional[int]) -> None

123 # this implies that flexible_tag was True

124 if diff_tag is not None:

125 if self.implicit_tag is not None:

126 self.implicit_tag = diff_tag

127 elif self.explicit_tag is not None:

128 self.explicit_tag = diff_tag

129

130 def _tagging_dec(self, pkt, s, **kwargs):

131 # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes]

132 # Codec provides tagging_*; OER implements real tags, UPER/PER use

133 # identity helpers (no BER-style tagging).

134 return pkt.ASN1_codec.tagging_dec(s, **kwargs) # type: ignore

135

136 def _tagging_enc(self, pkt, s, **kwargs):

137 # type: (ASN1_Packet, bytes, **Any) -> bytes

138 return pkt.ASN1_codec.tagging_enc(s, **kwargs) # type: ignore

139

140 def _apply_tagging_dec(self, s, pkt, hidden_tag=None, **kwargs):

141 # type: (bytes, ASN1_Packet, Optional[Any], **Any) -> bytes

142 # Always pass the field tags; callers may override hidden_tag (PACKET)

143 # or add decode metadata such as _fname.

144 if hidden_tag is None:

145 hidden_tag = self.ASN1_tag

146 diff_tag, s = self._tagging_dec(

147 pkt, s,

148 hidden_tag=hidden_tag,

149 implicit_tag=self.implicit_tag,

150 explicit_tag=self.explicit_tag,

151 safe=self.flexible_tag,

152 **kwargs,

153 )

154 self._apply_diff_tag(diff_tag)

155 return s

156

157 def _codec_kwargs(self, pkt):

158 # type: (ASN1_Packet) -> Dict[str, Any]

159 # OER/UPER need extra constraints (oer_unsigned, uper_min/max, …) on

160 # every enc/dec call; override this instead of hardcoding BER size_len.

161 return {"size_len": self.size_len}

162

163 def _use_object_enc(self, pkt, item):

164 # type: (ASN1_Packet, ASN1_Object[Any]) -> bool

165 # BER/LDAP: item.enc() when size_len is unset. UPER must override to

166 # False so constrained integers go through codec.enc(**kwargs).

167 return self.size_len is None

168

169 def _encode_item(self, pkt, item):

170 # type: (ASN1_Packet, Any) -> bytes

171 """Encode a field value with codec kwargs, without field tagging."""

172 if item is None:

173 return b""

174 if isinstance(item, ASN1_Object):

175 if (self.ASN1_tag == ASN1_Class_UNIVERSAL.ANY or

176 item.tag == ASN1_Class_UNIVERSAL.RAW or

177 item.tag == ASN1_Class_UNIVERSAL.ERROR):

178 return item.enc(pkt.ASN1_codec)

179 if self.ASN1_tag != item.tag:

180 raise ASN1_Error(

181 "Encoding Error: got %r instead of an %r for field [%s]" %

182 (item, self.ASN1_tag, self.name)

183 )

184 if self._use_object_enc(pkt, item):

185 return item.enc(pkt.ASN1_codec)

186 item = item.val

187 elif hasattr(item, "self_build"):

188 # Packet values (e.g. ASN1F_STRING_PacketField) must still go through

189 # the BER type codec so the universal tag/length are applied.

190 item = item.self_build()

191 codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)

192 return codec.enc(item, **self._codec_kwargs(pkt))

193

194 def i2repr(self, pkt, x):

195 # type: (ASN1_Packet, _I) -> str

196 return repr(x)

197

198 def i2h(self, pkt, x):

199 # type: (ASN1_Packet, _I) -> Any

200 return x

201

202 def m2i(self, pkt, s):

203 # type: (ASN1_Packet, bytes) -> Tuple[_A, bytes]

204 """

205 The good thing about safedec is that it may still decode ASN1

206 even if there is a mismatch between the expected tag (self.ASN1_tag)

207 and the actual tag; the decoded ASN1 object will simply be put

208 into an ASN1_BADTAG object. However, safedec prevents the raising of

209 exceptions needed for ASN1F_optional processing.

210 Thus we use 'flexible_tag', which should be False with ASN1F_optional.

211

212 Regarding other fields, we might need to know whether encoding went

213 as expected or not. Noticeably, input methods from cert.py expect

214 certain exceptions to be raised. Hence default flexible_tag is False.

215 """

216 s = self._apply_tagging_dec(s, pkt, _fname=self.name)

217 codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)

218 dec = codec.safedec if self.flexible_tag else codec.dec

219 return dec(s, context=self.context, **self._codec_kwargs(pkt)) # type: ignore

220

221 def i2m(self, pkt, x):

222 # type: (ASN1_Packet, Union[bytes, _I, _A]) -> bytes

223 if x is None:

224 return b""

225 s = self._encode_item(pkt, x)

226 return self._tagging_enc(

227 pkt, s,

228 implicit_tag=self.implicit_tag,

229 explicit_tag=self.explicit_tag,

230 )

231

232 def any2i(self, pkt, x):

233 # type: (ASN1_Packet, Any) -> _I

234 return cast(_I, x)

235

236 def extract_packet(self,

237 cls, # type: Type[ASN1_Packet]

238 s, # type: bytes

239 _underlayer=None # type: Optional[ASN1_Packet]

240 ):

241 # type: (...) -> Tuple[ASN1_Packet, bytes]

242 try:

243 c = cls(s, _underlayer=_underlayer)

244 except ASN1F_badsequence:

245 c = packet.Raw(s, _underlayer=_underlayer) # type: ignore

246 cpad = c.getlayer(packet.Raw)

247 s = b""

248 if cpad is not None:

249 s = cpad.load

250 if cpad.underlayer:

251 del cpad.underlayer.payload

252 return c, s

253

254 def build(self, pkt):

255 # type: (ASN1_Packet) -> bytes

256 return self.i2m(pkt, getattr(pkt, self.name))

257

258 def dissect(self, pkt, s):

259 # type: (ASN1_Packet, bytes) -> bytes

260 v, s = self.m2i(pkt, s)

261 self.set_val(pkt, v)

262 return s

263

264 def do_copy(self, x):

265 # type: (Any) -> Any

266 if isinstance(x, list):

267 x = x[:]

268 for i in range(len(x)):

269 if isinstance(x[i], BasePacket):

270 x[i] = x[i].copy()

271 return x

272 if hasattr(x, "copy"):

273 return x.copy()

274 return x

275

276 def set_val(self, pkt, val):

277 # type: (ASN1_Packet, Any) -> None

278 setattr(pkt, self.name, val)

279

280 def is_empty(self, pkt):

281 # type: (ASN1_Packet) -> bool

282 return getattr(pkt, self.name) is None

283

284 def get_fields_list(self):

285 # type: () -> List[ASN1F_field[Any, Any]]

286 return [self]

287

288 def __str__(self):

289 # type: () -> str

290 return repr(self)

291

292 def randval(self):

293 # type: () -> RandField[_I]

294 return cast(RandField[_I], RandInt())

295

296 def copy(self):

297 # type: () -> ASN1F_field[_I, _A]

298 return copy.copy(self)

299

300

301############################

302# Simple ASN1 Fields #

303############################

304

305class ASN1F_BOOLEAN(ASN1F_field[bool, ASN1_BOOLEAN]):

306 ASN1_tag = ASN1_Class_UNIVERSAL.BOOLEAN

307

308 def randval(self):

309 # type: () -> RandChoice

310 return RandChoice(True, False)

311

312

313class ASN1F_INTEGER(ASN1F_field[int, ASN1_INTEGER]):

314 ASN1_tag = ASN1_Class_UNIVERSAL.INTEGER

315

316 def randval(self):

317 # type: () -> RandNum

318 return RandNum(-2**64, 2**64 - 1)

319

320

321class ASN1F_enum_INTEGER(ASN1F_INTEGER):

322 def __init__(self,

323 name, # type: str

324 default, # type: ASN1_INTEGER

325 enum, # type: Dict[int, str]

326 context=None, # type: Optional[Any]

327 implicit_tag=None, # type: Optional[Any]

328 explicit_tag=None, # type: Optional[Any]

329 ):

330 # type: (...) -> None

331 super(ASN1F_enum_INTEGER, self).__init__(

332 name, default, context=context,

333 implicit_tag=implicit_tag,

334 explicit_tag=explicit_tag

335 )

336 i2s = self.i2s = {} # type: Dict[int, str]

337 s2i = self.s2i = {} # type: Dict[str, int]

338 if isinstance(enum, list):

339 keys = range(len(enum))

340 else:

341 keys = list(enum)

342 if any(isinstance(x, str) for x in keys):

343 i2s, s2i = s2i, i2s # type: ignore

344 for k in keys:

345 i2s[k] = enum[k]

346 s2i[enum[k]] = k

347

348 def i2m(self,

349 pkt, # type: ASN1_Packet

350 s, # type: Union[bytes, str, int, ASN1_INTEGER]

351 ):

352 # type: (...) -> bytes

353 if not isinstance(s, str):

354 vs = s

355 else:

356 vs = self.s2i[s]

357 return super(ASN1F_enum_INTEGER, self).i2m(pkt, vs)

358

359 def i2repr(self,

360 pkt, # type: ASN1_Packet

361 x, # type: Union[str, int]

362 ):

363 # type: (...) -> str

364 if x is not None and isinstance(x, ASN1_INTEGER):

365 r = self.i2s.get(x.val)

366 if r:

367 return "'%s' %s" % (r, repr(x))

368 return repr(x)

369

370

371class ASN1F_BIT_STRING(ASN1F_field[str, ASN1_BIT_STRING]):

372 ASN1_tag = ASN1_Class_UNIVERSAL.BIT_STRING

373

374 def __init__(self,

375 name, # type: str

376 default, # type: Optional[Union[ASN1_BIT_STRING, AnyStr]]

377 default_readable=True, # type: bool

378 context=None, # type: Optional[Any]

379 implicit_tag=None, # type: Optional[int]

380 explicit_tag=None, # type: Optional[int]

381 ):

382 # type: (...) -> None

383 super(ASN1F_BIT_STRING, self).__init__(

384 name, None, context=context,

385 implicit_tag=implicit_tag,

386 explicit_tag=explicit_tag

387 )

388 if isinstance(default, (bytes, str)):

389 self.default = ASN1_BIT_STRING(default,

390 readable=default_readable)

391 else:

392 self.default = default

393

394 def randval(self):

395 # type: () -> RandString

396 return RandString(RandNum(0, 1000))

397

398

399class ASN1F_STRING(ASN1F_field[str, ASN1_STRING]):

400 ASN1_tag = ASN1_Class_UNIVERSAL.STRING

401

402 def randval(self):

403 # type: () -> RandString

404 return RandString(RandNum(0, 1000))

405

406

407class ASN1F_NULL(ASN1F_INTEGER):

408 ASN1_tag = ASN1_Class_UNIVERSAL.NULL

409

410

411class ASN1F_OID(ASN1F_field[str, ASN1_OID]):

412 ASN1_tag = ASN1_Class_UNIVERSAL.OID

413

414 def randval(self):

415 # type: () -> RandOID

416 return RandOID()

417

418

419class ASN1F_ENUMERATED(ASN1F_enum_INTEGER):

420 ASN1_tag = ASN1_Class_UNIVERSAL.ENUMERATED

421

422

423class ASN1F_UTF8_STRING(ASN1F_STRING):

424 ASN1_tag = ASN1_Class_UNIVERSAL.UTF8_STRING

425

426

427class ASN1F_NUMERIC_STRING(ASN1F_STRING):

428 ASN1_tag = ASN1_Class_UNIVERSAL.NUMERIC_STRING

429

430

431class ASN1F_PRINTABLE_STRING(ASN1F_STRING):

432 ASN1_tag = ASN1_Class_UNIVERSAL.PRINTABLE_STRING

433

434

435class ASN1F_T61_STRING(ASN1F_STRING):

436 ASN1_tag = ASN1_Class_UNIVERSAL.T61_STRING

437

438

439class ASN1F_VIDEOTEX_STRING(ASN1F_STRING):

440 ASN1_tag = ASN1_Class_UNIVERSAL.VIDEOTEX_STRING

441

442

443class ASN1F_IA5_STRING(ASN1F_STRING):

444 ASN1_tag = ASN1_Class_UNIVERSAL.IA5_STRING

445

446

447class ASN1F_GENERAL_STRING(ASN1F_STRING):

448 ASN1_tag = ASN1_Class_UNIVERSAL.GENERAL_STRING

449

450

451class ASN1F_UTC_TIME(ASN1F_STRING):

452 ASN1_tag = ASN1_Class_UNIVERSAL.UTC_TIME

453

454 def randval(self): # type: ignore

455 # type: () -> GeneralizedTime

456 return GeneralizedTime()

457

458

459class ASN1F_GENERALIZED_TIME(ASN1F_STRING):

460 ASN1_tag = ASN1_Class_UNIVERSAL.GENERALIZED_TIME

461

462 def randval(self): # type: ignore

463 # type: () -> GeneralizedTime

464 return GeneralizedTime()

465

466

467class ASN1F_ISO646_STRING(ASN1F_STRING):

468 ASN1_tag = ASN1_Class_UNIVERSAL.ISO646_STRING

469

470

471class ASN1F_UNIVERSAL_STRING(ASN1F_STRING):

472 ASN1_tag = ASN1_Class_UNIVERSAL.UNIVERSAL_STRING

473

474

475class ASN1F_BMP_STRING(ASN1F_STRING):

476 ASN1_tag = ASN1_Class_UNIVERSAL.BMP_STRING

477

478

479class ASN1F_SEQUENCE(ASN1F_field[List[Any], List[Any]]):

480 # Here is how you could decode a SEQUENCE

481 # with an unknown, private high-tag prefix :

482 # class PrivSeq(ASN1_Packet):

483 # ASN1_codec = ASN1_Codecs.BER

484 # ASN1_root = ASN1F_SEQUENCE(

485 # <asn1 field #0>,

486 # ...

487 # <asn1 field #N>,

488 # explicit_tag=0,

489 # flexible_tag=True)

490 # Because we use flexible_tag, the value of the explicit_tag does not matter. # noqa: E501

491 ASN1_tag = ASN1_Class_UNIVERSAL.SEQUENCE

492 holds_packets = 1

493

494 def __init__(self, *seq, **kwargs):

495 # type: (*Any, **Any) -> None

496 name = "dummy_seq_name"

497 default = [field.default for field in seq]

498 super(ASN1F_SEQUENCE, self).__init__(

499 name, default, **kwargs

500 )

501 self.seq = seq

502 self.islist = len(seq) > 1

503

504 def __repr__(self):

505 # type: () -> str

506 return "<%s%r>" % (self.__class__.__name__, self.seq)

507

508 def is_empty(self, pkt):

509 # type: (ASN1_Packet) -> bool

510 return all(f.is_empty(pkt) for f in self.seq)

511

512 def get_fields_list(self):

513 # type: () -> List[ASN1F_field[Any, Any]]

514 return reduce(lambda x, y: x + y.get_fields_list(),

515 self.seq, [])

516

517 def m2i(self, pkt, s):

518 # type: (Any, bytes) -> Tuple[Any, bytes]

519 """

520 ASN1F_SEQUENCE behaves transparently, with nested ASN1_objects being

521 dissected one by one. Because we use obj.dissect (see loop below)

522 instead of obj.m2i (as we trust dissect to do the appropriate set_vals)

523 we do not directly retrieve the list of nested objects.

524 Thus m2i returns an empty list (along with the proper remainder).

525 It is discarded by dissect() and should not be missed elsewhere.

526 """

527 s = self._apply_tagging_dec(s, pkt, _fname=pkt.name)

528 codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)

529 i, s, remain = codec.check_type_check_len(s)

530 if len(s) == 0:

531 for obj in self.seq:

532 obj.set_val(pkt, None)

533 else:

534 for obj in self.seq:

535 try:

536 s = obj.dissect(pkt, s)

537 except ASN1F_badsequence:

538 break

539 if len(s) > 0:

540 raise BER_Decoding_Error(

541 "unexpected remainder in %s" % pkt.name,

542 remaining=s,

543 )

544 return [], remain

545

546 def dissect(self, pkt, s):

547 # type: (Any, bytes) -> bytes

548 _, x = self.m2i(pkt, s)

549 return x

550

551 def build(self, pkt):

552 # type: (ASN1_Packet) -> bytes

553 s = reduce(lambda x, y: x + y.build(pkt),

554 self.seq, b"")

555 return super(ASN1F_SEQUENCE, self).i2m(pkt, s)

556

557

558class ASN1F_SET(ASN1F_SEQUENCE):

559 ASN1_tag = ASN1_Class_UNIVERSAL.SET

560

561

562_SEQ_T = Union[

563 'ASN1_Packet',

564 Type[ASN1F_field[Any, Any]],

565 'ASN1F_PACKET',

566 ASN1F_field[Any, Any],

567]

568

569

570class ASN1F_SEQUENCE_OF(ASN1F_field[List[_SEQ_T],

571 List[ASN1_Object[Any]]]):

572 """

573 Two types are allowed as cls: ASN1_Packet, ASN1F_field

574 """

575 ASN1_tag = ASN1_Class_UNIVERSAL.SEQUENCE

576 islist = 1

577

578 def __init__(self,

579 name, # type: str

580 default, # type: Any

581 cls, # type: _SEQ_T

582 context=None, # type: Optional[Any]

583 implicit_tag=None, # type: Optional[Any]

584 explicit_tag=None, # type: Optional[Any]

585 ):

586 # type: (...) -> None

587 if isinstance(cls, type) and issubclass(cls, ASN1F_field) or \

588 isinstance(cls, ASN1F_field):

589 if isinstance(cls, type):

590 self.fld = cls(name, b"")

591 else:

592 self.fld = cls

593 self._extract_packet = lambda s, pkt: self.fld.m2i(pkt, s)

594 self.holds_packets = 0

595 elif hasattr(cls, "ASN1_root") or callable(cls):

596 self.cls = cast("Type[ASN1_Packet]", cls)

597 self._extract_packet = lambda s, pkt: self.extract_packet(

598 self.cls, s, _underlayer=pkt)

599 self.holds_packets = 1

600 else:

601 raise ValueError("cls should be an ASN1_Packet or ASN1_field")

602 super(ASN1F_SEQUENCE_OF, self).__init__(

603 name, None, context=context,

604 implicit_tag=implicit_tag, explicit_tag=explicit_tag

605 )

606 self.default = default

607

608 def is_empty(self,

609 pkt, # type: ASN1_Packet

610 ):

611 # type: (...) -> bool

612 return ASN1F_field.is_empty(self, pkt)

613

614 def m2i(self,

615 pkt, # type: ASN1_Packet

616 s, # type: bytes

617 ):

618 # type: (...) -> Tuple[List[Any], bytes]

619 s = self._apply_tagging_dec(s, pkt)

620 codec = self.ASN1_tag.get_codec(pkt.ASN1_codec)

621 i, s, remain = codec.check_type_check_len(s)

622 lst = []

623 while s:

624 c, s = self._extract_packet(s, pkt) # type: ignore

625 if c:

626 lst.append(c)

627 if len(s) > 0:

628 raise BER_Decoding_Error(

629 "unexpected remainder in %s" % pkt.name,

630 remaining=s,

631 )

632 return lst, remain

633

634 def build(self, pkt):

635 # type: (ASN1_Packet) -> bytes

636 val = getattr(pkt, self.name)

637 if isinstance(val, ASN1_Object) and \

638 val.tag == ASN1_Class_UNIVERSAL.RAW:

639 s = cast(Union[List[_SEQ_T], bytes], val)

640 elif val is None:

641 s = b""

642 elif self.holds_packets:

643 s = b"".join(bytes(i) for i in val)

644 else:

645 # BER: element fields may carry implicit/explicit tags; i2m

646 # matches m2i()/fld.m2i(). (Packet elements use bytes() above.)

647 s = b"".join(self.fld.i2m(pkt, i) for i in val)

648 return self.i2m(pkt, s)

649

650 def i2repr(self, pkt, x):

651 # type: (ASN1_Packet, _I) -> str

652 if self.holds_packets:

653 return super(ASN1F_SEQUENCE_OF, self).i2repr(pkt, x) # type: ignore

654 elif x is None:

655 return "[]"

656 else:

657 return "[%s]" % ", ".join(

658 self.fld.i2repr(pkt, x) for x in x # type: ignore

659 )

660

661 def randval(self):

662 # type: () -> Any

663 if self.holds_packets:

664 return packet.fuzz(self.cls())

665 else:

666 return self.fld.randval()

667

668 def __repr__(self):

669 # type: () -> str

670 return "<%s %s>" % (self.__class__.__name__, self.name)

671

672

673class ASN1F_SET_OF(ASN1F_SEQUENCE_OF):

674 ASN1_tag = ASN1_Class_UNIVERSAL.SET

675

676

677class ASN1F_IPADDRESS(ASN1F_STRING):

678 ASN1_tag = ASN1_Class_UNIVERSAL.IPADDRESS

679

680

681class ASN1F_TIME_TICKS(ASN1F_INTEGER):

682 ASN1_tag = ASN1_Class_UNIVERSAL.TIME_TICKS

683

684

685#############################

686# Complex ASN1 Fields #

687#############################

688

689class ASN1F_optional(ASN1F_element):

690 """

691 ASN.1 field that is optional.

692 """

693 def __init__(self, field):

694 # type: (ASN1F_field[Any, Any]) -> None

695 field.flexible_tag = False

696 self._field = field

697

698 def __getattr__(self, attr):

699 # type: (str) -> Optional[Any]

700 return getattr(self._field, attr)

701

702 def m2i(self, pkt, s):

703 # type: (ASN1_Packet, bytes) -> Tuple[Any, bytes]

704 try:

705 return self._field.m2i(pkt, s)

706 except (ASN1_Error, ASN1F_badsequence, ASN1_Decoding_Error):

707 # ASN1_Error may be raised by ASN1F_CHOICE

708 return None, s

709

710 def dissect(self, pkt, s):

711 # type: (ASN1_Packet, bytes) -> bytes

712 try:

713 return self._field.dissect(pkt, s)

714 except (ASN1_Error, ASN1F_badsequence, ASN1_Decoding_Error):

715 self._field.set_val(pkt, None)

716 return s

717

718 def build(self, pkt):

719 # type: (ASN1_Packet) -> bytes

720 if self._field.is_empty(pkt):

721 return b""

722 return self._field.build(pkt)

723

724 def any2i(self, pkt, x):

725 # type: (ASN1_Packet, Any) -> Any

726 return self._field.any2i(pkt, x)

727

728 def i2repr(self, pkt, x):

729 # type: (ASN1_Packet, Any) -> str

730 return self._field.i2repr(pkt, x)

731

732

733class ASN1F_omit(ASN1F_field[None, None]):

734 """

735 ASN.1 field that is not specified. This is simply omitted on the network.

736 This is different from ASN1F_NULL which has a network representation.

737 """

738 def m2i(self, pkt, s):

739 # type: (ASN1_Packet, bytes) -> Tuple[None, bytes]

740 return None, s

741

742 def i2m(self, pkt, x):

743 # type: (ASN1_Packet, Optional[bytes]) -> bytes

744 return b""

745

746

747_CHOICE_T = Union['ASN1_Packet', Type[ASN1F_field[Any, Any]], 'ASN1F_PACKET']

748

749

750class ASN1F_CHOICE(ASN1F_field[_CHOICE_T, ASN1_Object[Any]]):

751 """

752 Multiple types are allowed: ASN1_Packet, ASN1F_field and ASN1F_PACKET(),

753 See layers/x509.py for examples.

754 Other ASN1F_field instances than ASN1F_PACKET instances must not be used.

755 """

756 holds_packets = 1

757 ASN1_tag = ASN1_Class_UNIVERSAL.ANY

758

759 def __init__(self, name, default, *args, **kwargs):

760 # type: (str, Any, *_CHOICE_T, **Any) -> None

761 if "implicit_tag" in kwargs:

762 err_msg = "ASN1F_CHOICE has been called with an implicit_tag"

763 raise ASN1_Error(err_msg)

764 self.implicit_tag = None

765 for kwarg in ["context", "explicit_tag"]:

766 setattr(self, kwarg, kwargs.get(kwarg))

767 super(ASN1F_CHOICE, self).__init__(

768 name, None, context=self.context,

769 explicit_tag=self.explicit_tag

770 )

771 self.default = default

772 self.current_choice = None

773 self.choices = {} # type: Dict[int, _CHOICE_T]

774 self.pktchoices = {}

775 for p in args:

776 if hasattr(p, "ASN1_root"):

777 p = cast('ASN1_Packet', p)

778 # should be ASN1_Packet

779 if hasattr(p.ASN1_root, "choices"):

780 root = cast(ASN1F_CHOICE, p.ASN1_root)

781 for k, v in root.choices.items():

782 # ASN1F_CHOICE recursion

783 self.choices[k] = v

784 else:

785 self.choices[p.ASN1_root.network_tag] = p

786 elif hasattr(p, "ASN1_tag"):

787 if isinstance(p, type):

788 # should be ASN1F_field class

789 self.choices[int(p.ASN1_tag)] = p

790 else:

791 # should be ASN1F_PACKET instance

792 self.choices[p.network_tag] = p

793 self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501

794 else:

795 raise ASN1_Error("ASN1F_CHOICE: no tag found for one field")

796

797 def m2i(self, pkt, s):

798 # type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes]

799 """

800 First we have to retrieve the appropriate choice.

801 Then we extract the field/packet, according to this choice.

802 """

803 if len(s) == 0:

804 raise ASN1_Error("ASN1F_CHOICE: got empty string")

805 s = self._apply_tagging_dec(s, pkt)

806 tag, _ = BER_id_dec(s)

807 if tag in self.choices:

808 choice = self.choices[tag]

809 else:

810 if self.flexible_tag:

811 choice = ASN1F_field

812 else:

813 raise ASN1_Error(

814 "ASN1F_CHOICE: unexpected field in '%s' "

815 "(tag %s not in possible tags %s)" % (

816 self.name, tag, list(self.choices.keys())

817 )

818 )

819 if hasattr(choice, "ASN1_root"):

820 # we don't want to import ASN1_Packet in this module...

821 return self.extract_packet(choice, s, _underlayer=pkt) # type: ignore

822 elif isinstance(choice, type):

823 return choice(self.name, b"").m2i(pkt, s)

824 else:

825 # XXX check properly if this is an ASN1F_PACKET

826 return choice.m2i(pkt, s)

827

828 def i2m(self, pkt, x):

829 # type: (ASN1_Packet, Any) -> bytes

830 if x is None:

831 s = b""

832 else:

833 # Use the packet codec for ASN1_Object values; bytes(x) would

834 # follow conf.ASN1_default_codec instead.

835 if isinstance(x, ASN1_Object):

836 s = x.enc(pkt.ASN1_codec)

837 else:

838 s = bytes(x)

839 if hash(type(x)) in self.pktchoices:

840 imp, exp = self.pktchoices[hash(type(x))]

841 s = self._tagging_enc(

842 pkt, s,

843 implicit_tag=imp,

844 explicit_tag=exp,

845 )

846 return self._tagging_enc(pkt, s, explicit_tag=self.explicit_tag)

847

848 def randval(self):

849 # type: () -> RandChoice

850 randchoices = []

851 for p in self.choices.values():

852 if hasattr(p, "ASN1_root"):

853 # should be ASN1_Packet class

854 randchoices.append(packet.fuzz(p())) # type: ignore

855 elif hasattr(p, "ASN1_tag"):

856 if isinstance(p, type):

857 # should be (basic) ASN1F_field class

858 randchoices.append(p("dummy", None).randval())

859 else:

860 # should be ASN1F_PACKET instance

861 randchoices.append(p.randval())

862 return RandChoice(*randchoices)

863

864

865class ASN1F_PACKET(ASN1F_field['ASN1_Packet', Optional['ASN1_Packet']]):

866 holds_packets = 1

867

868 def __init__(self,

869 name, # type: str

870 default, # type: Optional[ASN1_Packet]

871 cls, # type: Type[ASN1_Packet]

872 context=None, # type: Optional[Any]

873 implicit_tag=None, # type: Optional[int]

874 explicit_tag=None, # type: Optional[int]

875 next_cls_cb=None, # type: Optional[Callable[[ASN1_Packet], Type[ASN1_Packet]]] # noqa: E501

876 ):

877 # type: (...) -> None

878 self.cls = cls

879 self.next_cls_cb = next_cls_cb

880 super(ASN1F_PACKET, self).__init__(

881 name, None, context=context,

882 implicit_tag=implicit_tag, explicit_tag=explicit_tag

883 )

884 if implicit_tag is None and explicit_tag is None and cls is not None:

885 if cls.ASN1_root.ASN1_tag == ASN1_Class_UNIVERSAL.SEQUENCE:

886 self.network_tag = 16 | 0x20 # 16 + CONSTRUCTED

887 self.default = default

888

889 def m2i(self, pkt, s):

890 # type: (ASN1_Packet, bytes) -> Tuple[Any, bytes]

891 if self.next_cls_cb:

892 cls = self.next_cls_cb(pkt) or self.cls

893 else:

894 cls = self.cls

895 if not hasattr(cls, "ASN1_root"):

896 # A normal Packet (!= ASN1)

897 return self.extract_packet(cls, s, _underlayer=pkt)

898 s = self._apply_tagging_dec(

899 s, pkt,

900 hidden_tag=cls.ASN1_root.ASN1_tag, # noqa: E501

901 _fname=self.name,

902 )

903 if not s:

904 return None, s

905 return self.extract_packet(cls, s, _underlayer=pkt)

906

907 def i2m(self,

908 pkt, # type: ASN1_Packet

909 x # type: Union[bytes, ASN1_Packet, None, ASN1_Object[Optional[ASN1_Packet]]] # noqa: E501

910 ):

911 # type: (...) -> bytes

912 if x is None:

913 s = b""

914 elif isinstance(x, bytes):

915 s = x

916 elif isinstance(x, ASN1_Object):

917 if x.val:

918 s = bytes(x.val)

919 else:

920 s = b""

921 else:

922 s = bytes(x)

923 if not hasattr(x, "ASN1_root"):

924 # A normal Packet (!= ASN1)

925 return s

926 return self._tagging_enc(

927 pkt, s,

928 implicit_tag=self.implicit_tag,

929 explicit_tag=self.explicit_tag,

930 )

931

932 def any2i(self,

933 pkt, # type: ASN1_Packet

934 x # type: Union[bytes, ASN1_Packet, None, ASN1_Object[Optional[ASN1_Packet]]] # noqa: E501

935 ):

936 # type: (...) -> 'ASN1_Packet'

937 if hasattr(x, "add_underlayer"):

938 x.add_underlayer(pkt) # type: ignore

939 return super(ASN1F_PACKET, self).any2i(pkt, x)

940

941 def randval(self): # type: ignore

942 # type: () -> ASN1_Packet

943 return packet.fuzz(self.cls())

944

945

946class ASN1F_BIT_STRING_ENCAPS(ASN1F_BIT_STRING):

947 """

948 We may emulate simple string encapsulation with explicit_tag=0x04,

949 but we need a specific class for bit strings because of unused bits, etc.

950 """

951 ASN1_tag = ASN1_Class_UNIVERSAL.BIT_STRING

952

953 def __init__(self,

954 name, # type: str

955 default, # type: Optional[ASN1_Packet]

956 cls, # type: Type[ASN1_Packet]

957 context=None, # type: Optional[Any]

958 implicit_tag=None, # type: Optional[int]

959 explicit_tag=None, # type: Optional[int]

960 ):

961 # type: (...) -> None

962 self.cls = cls

963 super(ASN1F_BIT_STRING_ENCAPS, self).__init__( # type: ignore

964 name,

965 default and bytes(default),

966 context=context,

967 implicit_tag=implicit_tag,

968 explicit_tag=explicit_tag

969 )

970

971 def m2i(self, pkt, s): # type: ignore

972 # type: (ASN1_Packet, bytes) -> Tuple[Optional[ASN1_Packet], bytes]

973 bit_string, remain = super(ASN1F_BIT_STRING_ENCAPS, self).m2i(pkt, s)

974 if len(bit_string.val) % 8 != 0:

975 raise BER_Decoding_Error("wrong bit string", remaining=s)

976 if bit_string.val_readable:

977 p, s = self.extract_packet(self.cls, bit_string.val_readable,

978 _underlayer=pkt)

979 else:

980 return None, bit_string.val_readable

981 if len(s) > 0:

982 raise BER_Decoding_Error(

983 "unexpected remainder in %s" % pkt.name,

984 remaining=s,

985 )

986 return p, remain

987

988 def i2m(self, pkt, x): # type: ignore

989 # type: (ASN1_Packet, Optional[ASN1_BIT_STRING]) -> bytes

990 if not isinstance(x, ASN1_BIT_STRING):

991 x = ASN1_BIT_STRING(

992 b"" if x is None else bytes(x), # type: ignore

993 readable=True,

994 )

995 return super(ASN1F_BIT_STRING_ENCAPS, self).i2m(pkt, x)

996

997

998class ASN1F_FLAGS(ASN1F_BIT_STRING):

999 def __init__(self,

1000 name, # type: str

1001 default, # type: Optional[str]

1002 mapping, # type: List[str]

1003 context=None, # type: Optional[Any]

1004 implicit_tag=None, # type: Optional[int]

1005 explicit_tag=None, # type: Optional[Any]

1006 ):

1007 # type: (...) -> None

1008 self.mapping = mapping

1009 super(ASN1F_FLAGS, self).__init__(

1010 name, default,

1011 default_readable=False,

1012 context=context,

1013 implicit_tag=implicit_tag,

1014 explicit_tag=explicit_tag

1015 )

1016

1017 def any2i(self, pkt, x):

1018 # type: (ASN1_Packet, Any) -> str

1019 if isinstance(x, str):

1020 if any(y not in ["0", "1"] for y in x):

1021 # resolve the flags

1022 value = ["0"] * len(self.mapping)

1023 for i in x.split("+"):

1024 value[self.mapping.index(i)] = "1"

1025 x = "".join(value)

1026 x = ASN1_BIT_STRING(x)

1027 return super(ASN1F_FLAGS, self).any2i(pkt, x)

1028

1029 def get_flags(self, pkt):

1030 # type: (ASN1_Packet) -> List[str]

1031 fbytes = getattr(pkt, self.name).val

1032 return [self.mapping[i] for i, positional in enumerate(fbytes)

1033 if positional == '1' and i < len(self.mapping)]

1034

1035 def i2repr(self, pkt, x):

1036 # type: (ASN1_Packet, Any) -> str

1037 if x is not None:

1038 pretty_s = ", ".join(self.get_flags(pkt))

1039 return pretty_s + " " + repr(x)

1040 return repr(x)

1041

1042

1043class ASN1F_STRING_PacketField(ASN1F_STRING):

1044 """

1045 ASN1F_STRING that holds packets.

1046 """

1047 holds_packets = 1

1048

1049 def i2m(self, pkt, val):

1050 # type: (ASN1_Packet, Any) -> bytes

1051 if hasattr(val, "ASN1_root"):

1052 val = ASN1_STRING(bytes(val))

1053 return super(ASN1F_STRING_PacketField, self).i2m(pkt, val)

1054

1055 def any2i(self, pkt, x):

1056 # type: (ASN1_Packet, Any) -> Any

1057 if hasattr(x, "add_underlayer"):

1058 x.add_underlayer(pkt)

1059 return super(ASN1F_STRING_PacketField, self).any2i(pkt, x)

1060

1061

1062class ASN1F_STRING_ENCAPS(ASN1F_STRING_PacketField):

1063 """

1064 ASN1F_STRING that encapsulates a single ASN1 packet.

1065 """

1066

1067 def __init__(self,

1068 name, # type: str

1069 default, # type: Optional[ASN1_Packet]

1070 cls, # type: Type[ASN1_Packet]

1071 context=None, # type: Optional[Any]

1072 implicit_tag=None, # type: Optional[int]

1073 explicit_tag=None, # type: Optional[int]

1074 ):

1075 # type: (...) -> None

1076 self.cls = cls

1077 super(ASN1F_STRING_ENCAPS, self).__init__(

1078 name,

1079 default and bytes(default), # type: ignore

1080 context=context,

1081 implicit_tag=implicit_tag,

1082 explicit_tag=explicit_tag

1083 )

1084

1085 def m2i(self, pkt, s): # type: ignore

1086 # type: (ASN1_Packet, bytes) -> Tuple[ASN1_Packet, bytes]

1087 val = super(ASN1F_STRING_ENCAPS, self).m2i(pkt, s)

1088 return self.cls(val[0].val, _underlayer=pkt), val[1]