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

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> 

6 

7""" 

8Classes that implement ASN.1 data structures. 

9""" 

10 

11import copy 

12 

13from functools import reduce 

14 

15from scapy.asn1.asn1 import ( 

16 ASN1_BIT_STRING, 

17 ASN1_BOOLEAN, 

18 ASN1_Class, 

19 ASN1_Class_UNIVERSAL, 

20 ASN1_Error, 

21 ASN1_INTEGER, 

22 ASN1_NULL, 

23 ASN1_OID, 

24 ASN1_Object, 

25 ASN1_STRING, 

26) 

27from scapy.asn1.ber import ( 

28 BER_Decoding_Error, 

29 BER_id_dec, 

30 BER_tagging_dec, 

31 BER_tagging_enc, 

32) 

33from scapy.base_classes import BasePacket 

34from scapy.volatile import ( 

35 GeneralizedTime, 

36 RandChoice, 

37 RandInt, 

38 RandNum, 

39 RandOID, 

40 RandString, 

41 RandField, 

42) 

43 

44from scapy import packet 

45 

46from typing import ( 

47 Any, 

48 AnyStr, 

49 Callable, 

50 Dict, 

51 Generic, 

52 List, 

53 Optional, 

54 Tuple, 

55 Type, 

56 TypeVar, 

57 Union, 

58 cast, 

59 TYPE_CHECKING, 

60) 

61 

62if TYPE_CHECKING: 

63 from scapy.asn1packet import ASN1_Packet 

64 

65 

66class ASN1F_badsequence(Exception): 

67 pass 

68 

69 

70class ASN1F_element(object): 

71 pass 

72 

73 

74########################## 

75# Basic ASN1 Field # 

76########################## 

77 

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

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

80 

81 

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

83 holds_packets = 0 

84 islist = 0 

85 ASN1_tag = ASN1_Class_UNIVERSAL.ANY 

86 context = ASN1_Class_UNIVERSAL # type: Type[ASN1_Class] 

87 

88 def __init__(self, 

89 name, # type: str 

90 default, # type: Optional[_A] 

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

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

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

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

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

96 ): 

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

98 if context is not None: 

99 self.context = context 

100 self.name = name 

101 if default is None: 

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

103 elif isinstance(default, ASN1_NULL): 

104 self.default = default # type: ignore 

105 else: 

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

107 self.size_len = size_len 

108 self.flexible_tag = flexible_tag 

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

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

111 raise ASN1_Error(err_msg) 

112 self.implicit_tag = implicit_tag and int(implicit_tag) 

113 self.explicit_tag = explicit_tag and int(explicit_tag) 

114 # network_tag gets useful for ASN1F_CHOICE 

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

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

117 

118 def register_owner(self, cls): 

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

120 self.owners.append(cls) 

121 

122 def i2repr(self, pkt, x): 

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

124 return repr(x) 

125 

126 def i2h(self, pkt, x): 

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

128 return x 

129 

130 def m2i(self, pkt, s): 

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

132 """ 

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

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

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

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

137 exceptions needed for ASN1F_optional processing. 

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

139 

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

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

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

143 """ 

144 diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, 

145 implicit_tag=self.implicit_tag, 

146 explicit_tag=self.explicit_tag, 

147 safe=self.flexible_tag, 

148 _fname=self.name) 

149 if diff_tag is not None: 

150 # this implies that flexible_tag was True 

151 if self.implicit_tag is not None: 

152 self.implicit_tag = diff_tag 

153 elif self.explicit_tag is not None: 

154 self.explicit_tag = diff_tag 

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

156 if self.flexible_tag: 

157 return codec.safedec(s, context=self.context) # type: ignore 

158 else: 

159 return codec.dec(s, context=self.context) # type: ignore 

160 

161 def i2m(self, pkt, x): 

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

163 if x is None: 

164 return b"" 

165 if isinstance(x, ASN1_Object): 

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

167 x.tag == ASN1_Class_UNIVERSAL.RAW or 

168 x.tag == ASN1_Class_UNIVERSAL.ERROR or 

169 self.ASN1_tag == x.tag): 

170 s = x.enc(pkt.ASN1_codec) 

171 else: 

172 raise ASN1_Error("Encoding Error: got %r instead of an %r for field [%s]" % (x, self.ASN1_tag, self.name)) # noqa: E501 

173 else: 

174 s = self.ASN1_tag.get_codec(pkt.ASN1_codec).enc(x, size_len=self.size_len) 

175 return BER_tagging_enc(s, 

176 implicit_tag=self.implicit_tag, 

177 explicit_tag=self.explicit_tag) 

178 

179 def any2i(self, pkt, x): 

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

181 return cast(_I, x) 

182 

183 def extract_packet(self, 

184 cls, # type: Type[ASN1_Packet] 

185 s, # type: bytes 

186 _underlayer=None # type: Optional[ASN1_Packet] 

187 ): 

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

189 try: 

190 c = cls(s, _underlayer=_underlayer) 

191 except ASN1F_badsequence: 

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

193 cpad = c.getlayer(packet.Raw) 

194 s = b"" 

195 if cpad is not None: 

196 s = cpad.load 

197 if cpad.underlayer: 

198 del cpad.underlayer.payload 

199 return c, s 

200 

201 def build(self, pkt): 

202 # type: (ASN1_Packet) -> bytes 

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

204 

205 def dissect(self, pkt, s): 

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

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

208 self.set_val(pkt, v) 

209 return s 

210 

211 def do_copy(self, x): 

212 # type: (Any) -> Any 

213 if isinstance(x, list): 

214 x = x[:] 

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

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

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

218 return x 

219 if hasattr(x, "copy"): 

220 return x.copy() 

221 return x 

222 

223 def set_val(self, pkt, val): 

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

225 setattr(pkt, self.name, val) 

226 

227 def is_empty(self, pkt): 

228 # type: (ASN1_Packet) -> bool 

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

230 

231 def get_fields_list(self): 

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

233 return [self] 

234 

235 def __str__(self): 

236 # type: () -> str 

237 return repr(self) 

238 

239 def randval(self): 

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

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

242 

243 def copy(self): 

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

245 return copy.copy(self) 

246 

247 

248############################ 

249# Simple ASN1 Fields # 

250############################ 

251 

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

253 ASN1_tag = ASN1_Class_UNIVERSAL.BOOLEAN 

254 

255 def randval(self): 

256 # type: () -> RandChoice 

257 return RandChoice(True, False) 

258 

259 

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

261 ASN1_tag = ASN1_Class_UNIVERSAL.INTEGER 

262 

263 def randval(self): 

264 # type: () -> RandNum 

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

266 

267 

268class ASN1F_enum_INTEGER(ASN1F_INTEGER): 

269 def __init__(self, 

270 name, # type: str 

271 default, # type: ASN1_INTEGER 

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

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

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

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

276 ): 

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

278 super(ASN1F_enum_INTEGER, self).__init__( 

279 name, default, context=context, 

280 implicit_tag=implicit_tag, 

281 explicit_tag=explicit_tag 

282 ) 

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

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

285 if isinstance(enum, list): 

286 keys = range(len(enum)) 

287 else: 

288 keys = list(enum) 

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

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

291 for k in keys: 

292 i2s[k] = enum[k] 

293 s2i[enum[k]] = k 

294 

295 def i2m(self, 

296 pkt, # type: ASN1_Packet 

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

298 ): 

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

300 if not isinstance(s, str): 

301 vs = s 

302 else: 

303 vs = self.s2i[s] 

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

305 

306 def i2repr(self, 

307 pkt, # type: ASN1_Packet 

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

309 ): 

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

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

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

313 if r: 

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

315 return repr(x) 

316 

317 

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

319 ASN1_tag = ASN1_Class_UNIVERSAL.BIT_STRING 

320 

321 def __init__(self, 

322 name, # type: str 

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

324 default_readable=True, # type: bool 

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

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

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

328 ): 

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

330 super(ASN1F_BIT_STRING, self).__init__( 

331 name, None, context=context, 

332 implicit_tag=implicit_tag, 

333 explicit_tag=explicit_tag 

334 ) 

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

336 self.default = ASN1_BIT_STRING(default, 

337 readable=default_readable) 

338 else: 

339 self.default = default 

340 

341 def randval(self): 

342 # type: () -> RandString 

343 return RandString(RandNum(0, 1000)) 

344 

345 

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

347 ASN1_tag = ASN1_Class_UNIVERSAL.STRING 

348 

349 def randval(self): 

350 # type: () -> RandString 

351 return RandString(RandNum(0, 1000)) 

352 

353 

354class ASN1F_NULL(ASN1F_INTEGER): 

355 ASN1_tag = ASN1_Class_UNIVERSAL.NULL 

356 

357 

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

359 ASN1_tag = ASN1_Class_UNIVERSAL.OID 

360 

361 def randval(self): 

362 # type: () -> RandOID 

363 return RandOID() 

364 

365 

366class ASN1F_ENUMERATED(ASN1F_enum_INTEGER): 

367 ASN1_tag = ASN1_Class_UNIVERSAL.ENUMERATED 

368 

369 

370class ASN1F_UTF8_STRING(ASN1F_STRING): 

371 ASN1_tag = ASN1_Class_UNIVERSAL.UTF8_STRING 

372 

373 

374class ASN1F_NUMERIC_STRING(ASN1F_STRING): 

375 ASN1_tag = ASN1_Class_UNIVERSAL.NUMERIC_STRING 

376 

377 

378class ASN1F_PRINTABLE_STRING(ASN1F_STRING): 

379 ASN1_tag = ASN1_Class_UNIVERSAL.PRINTABLE_STRING 

380 

381 

382class ASN1F_T61_STRING(ASN1F_STRING): 

383 ASN1_tag = ASN1_Class_UNIVERSAL.T61_STRING 

384 

385 

386class ASN1F_VIDEOTEX_STRING(ASN1F_STRING): 

387 ASN1_tag = ASN1_Class_UNIVERSAL.VIDEOTEX_STRING 

388 

389 

390class ASN1F_IA5_STRING(ASN1F_STRING): 

391 ASN1_tag = ASN1_Class_UNIVERSAL.IA5_STRING 

392 

393 

394class ASN1F_GENERAL_STRING(ASN1F_STRING): 

395 ASN1_tag = ASN1_Class_UNIVERSAL.GENERAL_STRING 

396 

397 

398class ASN1F_UTC_TIME(ASN1F_STRING): 

399 ASN1_tag = ASN1_Class_UNIVERSAL.UTC_TIME 

400 

401 def randval(self): # type: ignore 

402 # type: () -> GeneralizedTime 

403 return GeneralizedTime() 

404 

405 

406class ASN1F_GENERALIZED_TIME(ASN1F_STRING): 

407 ASN1_tag = ASN1_Class_UNIVERSAL.GENERALIZED_TIME 

408 

409 def randval(self): # type: ignore 

410 # type: () -> GeneralizedTime 

411 return GeneralizedTime() 

412 

413 

414class ASN1F_ISO646_STRING(ASN1F_STRING): 

415 ASN1_tag = ASN1_Class_UNIVERSAL.ISO646_STRING 

416 

417 

418class ASN1F_UNIVERSAL_STRING(ASN1F_STRING): 

419 ASN1_tag = ASN1_Class_UNIVERSAL.UNIVERSAL_STRING 

420 

421 

422class ASN1F_BMP_STRING(ASN1F_STRING): 

423 ASN1_tag = ASN1_Class_UNIVERSAL.BMP_STRING 

424 

425 

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

427 # Here is how you could decode a SEQUENCE 

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

429 # class PrivSeq(ASN1_Packet): 

430 # ASN1_codec = ASN1_Codecs.BER 

431 # ASN1_root = ASN1F_SEQUENCE( 

432 # <asn1 field #0>, 

433 # ... 

434 # <asn1 field #N>, 

435 # explicit_tag=0, 

436 # flexible_tag=True) 

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

438 ASN1_tag = ASN1_Class_UNIVERSAL.SEQUENCE 

439 holds_packets = 1 

440 

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

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

443 name = "dummy_seq_name" 

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

445 super(ASN1F_SEQUENCE, self).__init__( 

446 name, default, **kwargs 

447 ) 

448 self.seq = seq 

449 self.islist = len(seq) > 1 

450 

451 def __repr__(self): 

452 # type: () -> str 

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

454 

455 def is_empty(self, pkt): 

456 # type: (ASN1_Packet) -> bool 

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

458 

459 def get_fields_list(self): 

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

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

462 self.seq, []) 

463 

464 def m2i(self, pkt, s): 

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

466 """ 

467 ASN1F_SEQUENCE behaves transparently, with nested ASN1_objects being 

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

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

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

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

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

473 """ 

474 diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, 

475 implicit_tag=self.implicit_tag, 

476 explicit_tag=self.explicit_tag, 

477 safe=self.flexible_tag, 

478 _fname=pkt.name) 

479 if diff_tag is not None: 

480 if self.implicit_tag is not None: 

481 self.implicit_tag = diff_tag 

482 elif self.explicit_tag is not None: 

483 self.explicit_tag = diff_tag 

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

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

486 if len(s) == 0: 

487 for obj in self.seq: 

488 obj.set_val(pkt, None) 

489 else: 

490 for obj in self.seq: 

491 try: 

492 s = obj.dissect(pkt, s) 

493 except ASN1F_badsequence: 

494 break 

495 if len(s) > 0: 

496 raise BER_Decoding_Error( 

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

498 remaining=s, 

499 ) 

500 return [], remain 

501 

502 def dissect(self, pkt, s): 

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

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

505 return x 

506 

507 def build(self, pkt): 

508 # type: (ASN1_Packet) -> bytes 

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

510 self.seq, b"") 

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

512 

513 

514class ASN1F_SET(ASN1F_SEQUENCE): 

515 ASN1_tag = ASN1_Class_UNIVERSAL.SET 

516 

517 

518_SEQ_T = Union[ 

519 'ASN1_Packet', 

520 Type[ASN1F_field[Any, Any]], 

521 'ASN1F_PACKET', 

522 ASN1F_field[Any, Any], 

523] 

524 

525 

526class ASN1F_SEQUENCE_OF(ASN1F_field[List[_SEQ_T], 

527 List[ASN1_Object[Any]]]): 

528 """ 

529 Two types are allowed as cls: ASN1_Packet, ASN1F_field 

530 """ 

531 ASN1_tag = ASN1_Class_UNIVERSAL.SEQUENCE 

532 islist = 1 

533 

534 def __init__(self, 

535 name, # type: str 

536 default, # type: Any 

537 cls, # type: _SEQ_T 

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

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

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

541 ): 

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

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

544 isinstance(cls, ASN1F_field): 

545 if isinstance(cls, type): 

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

547 else: 

548 self.fld = cls 

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

550 self.holds_packets = 0 

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

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

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

554 self.cls, s, _underlayer=pkt) 

555 self.holds_packets = 1 

556 else: 

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

558 super(ASN1F_SEQUENCE_OF, self).__init__( 

559 name, None, context=context, 

560 implicit_tag=implicit_tag, explicit_tag=explicit_tag 

561 ) 

562 self.default = default 

563 

564 def is_empty(self, 

565 pkt, # type: ASN1_Packet 

566 ): 

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

568 return ASN1F_field.is_empty(self, pkt) 

569 

570 def m2i(self, 

571 pkt, # type: ASN1_Packet 

572 s, # type: bytes 

573 ): 

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

575 diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, 

576 implicit_tag=self.implicit_tag, 

577 explicit_tag=self.explicit_tag, 

578 safe=self.flexible_tag) 

579 if diff_tag is not None: 

580 if self.implicit_tag is not None: 

581 self.implicit_tag = diff_tag 

582 elif self.explicit_tag is not None: 

583 self.explicit_tag = diff_tag 

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

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

586 lst = [] 

587 while s: 

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

589 if c: 

590 lst.append(c) 

591 if len(s) > 0: 

592 raise BER_Decoding_Error( 

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

594 remaining=s, 

595 ) 

596 return lst, remain 

597 

598 def build(self, pkt): 

599 # type: (ASN1_Packet) -> bytes 

600 val = getattr(pkt, self.name) 

601 if isinstance(val, ASN1_Object) and \ 

602 val.tag == ASN1_Class_UNIVERSAL.RAW: 

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

604 elif val is None: 

605 s = b"" 

606 else: 

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

608 return self.i2m(pkt, s) 

609 

610 def i2repr(self, pkt, x): 

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

612 if self.holds_packets: 

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

614 elif x is None: 

615 return "[]" 

616 else: 

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

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

619 ) 

620 

621 def randval(self): 

622 # type: () -> Any 

623 if self.holds_packets: 

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

625 else: 

626 return self.fld.randval() 

627 

628 def __repr__(self): 

629 # type: () -> str 

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

631 

632 

633class ASN1F_SET_OF(ASN1F_SEQUENCE_OF): 

634 ASN1_tag = ASN1_Class_UNIVERSAL.SET 

635 

636 

637class ASN1F_IPADDRESS(ASN1F_STRING): 

638 ASN1_tag = ASN1_Class_UNIVERSAL.IPADDRESS 

639 

640 

641class ASN1F_TIME_TICKS(ASN1F_INTEGER): 

642 ASN1_tag = ASN1_Class_UNIVERSAL.TIME_TICKS 

643 

644 

645############################# 

646# Complex ASN1 Fields # 

647############################# 

648 

649class ASN1F_optional(ASN1F_element): 

650 """ 

651 ASN.1 field that is optional. 

652 """ 

653 def __init__(self, field): 

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

655 field.flexible_tag = False 

656 self._field = field 

657 

658 def __getattr__(self, attr): 

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

660 return getattr(self._field, attr) 

661 

662 def m2i(self, pkt, s): 

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

664 try: 

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

666 except (ASN1_Error, ASN1F_badsequence, BER_Decoding_Error): 

667 # ASN1_Error may be raised by ASN1F_CHOICE 

668 return None, s 

669 

670 def dissect(self, pkt, s): 

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

672 try: 

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

674 except (ASN1_Error, ASN1F_badsequence, BER_Decoding_Error): 

675 self._field.set_val(pkt, None) 

676 return s 

677 

678 def build(self, pkt): 

679 # type: (ASN1_Packet) -> bytes 

680 if self._field.is_empty(pkt): 

681 return b"" 

682 return self._field.build(pkt) 

683 

684 def any2i(self, pkt, x): 

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

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

687 

688 def i2repr(self, pkt, x): 

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

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

691 

692 

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

694 """ 

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

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

697 """ 

698 def m2i(self, pkt, s): 

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

700 return None, s 

701 

702 def i2m(self, pkt, x): 

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

704 return b"" 

705 

706 

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

708 

709 

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

711 """ 

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

713 See layers/x509.py for examples. 

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

715 """ 

716 holds_packets = 1 

717 ASN1_tag = ASN1_Class_UNIVERSAL.ANY 

718 

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

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

721 if "implicit_tag" in kwargs: 

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

723 raise ASN1_Error(err_msg) 

724 self.implicit_tag = None 

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

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

727 super(ASN1F_CHOICE, self).__init__( 

728 name, None, context=self.context, 

729 explicit_tag=self.explicit_tag 

730 ) 

731 self.default = default 

732 self.current_choice = None 

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

734 self.pktchoices = {} 

735 for p in args: 

736 if hasattr(p, "ASN1_root"): 

737 p = cast('ASN1_Packet', p) 

738 # should be ASN1_Packet 

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

740 root = cast(ASN1F_CHOICE, p.ASN1_root) 

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

742 # ASN1F_CHOICE recursion 

743 self.choices[k] = v 

744 else: 

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

746 elif hasattr(p, "ASN1_tag"): 

747 if isinstance(p, type): 

748 # should be ASN1F_field class 

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

750 else: 

751 # should be ASN1F_field instance 

752 self.choices[p.network_tag] = p 

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

754 else: 

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

756 

757 def m2i(self, pkt, s): 

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

759 """ 

760 First we have to retrieve the appropriate choice. 

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

762 """ 

763 if len(s) == 0: 

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

765 _, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, 

766 explicit_tag=self.explicit_tag) 

767 tag, _ = BER_id_dec(s) 

768 if tag in self.choices: 

769 choice = self.choices[tag] 

770 else: 

771 if self.flexible_tag: 

772 choice = ASN1F_field 

773 else: 

774 raise ASN1_Error( 

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

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

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

778 ) 

779 ) 

780 if hasattr(choice, "ASN1_root"): 

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

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

783 elif isinstance(choice, type): 

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

785 else: 

786 # XXX check properly if this is an ASN1F_PACKET 

787 return choice.m2i(pkt, s) 

788 

789 def i2m(self, pkt, x): 

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

791 if x is None: 

792 s = b"" 

793 else: 

794 s = bytes(x) 

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

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

797 s = BER_tagging_enc(s, 

798 implicit_tag=imp, 

799 explicit_tag=exp) 

800 return BER_tagging_enc(s, explicit_tag=self.explicit_tag) 

801 

802 def randval(self): 

803 # type: () -> RandChoice 

804 randchoices = [] 

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

806 if hasattr(p, "ASN1_root"): 

807 # should be ASN1_Packet class 

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

809 elif hasattr(p, "ASN1_tag"): 

810 if isinstance(p, type): 

811 # should be (basic) ASN1F_field class 

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

813 else: 

814 # should be ASN1F_PACKET instance 

815 randchoices.append(p.randval()) 

816 return RandChoice(*randchoices) 

817 

818 

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

820 holds_packets = 1 

821 

822 def __init__(self, 

823 name, # type: str 

824 default, # type: Optional[ASN1_Packet] 

825 cls, # type: Type[ASN1_Packet] 

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

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

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

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

830 ): 

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

832 self.cls = cls 

833 self.next_cls_cb = next_cls_cb 

834 super(ASN1F_PACKET, self).__init__( 

835 name, None, context=context, 

836 implicit_tag=implicit_tag, explicit_tag=explicit_tag 

837 ) 

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

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