encoder.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954
  1. #
  2. # This file is part of pyasn1 software.
  3. #
  4. # Copyright (c) 2005-2020, Ilya Etingof <etingof@gmail.com>
  5. # License: https://pyasn1.readthedocs.io/en/latest/license.html
  6. #
  7. import sys
  8. import warnings
  9. from pyasn1 import debug
  10. from pyasn1 import error
  11. from pyasn1.codec.ber import eoo
  12. from pyasn1.compat import _MISSING
  13. from pyasn1.compat.integer import to_bytes
  14. from pyasn1.type import char
  15. from pyasn1.type import tag
  16. from pyasn1.type import univ
  17. from pyasn1.type import useful
  18. __all__ = ['Encoder', 'encode']
  19. LOG = debug.registerLoggee(__name__, flags=debug.DEBUG_ENCODER)
  20. class AbstractItemEncoder(object):
  21. supportIndefLenMode = True
  22. # An outcome of otherwise legit call `encodeFun(eoo.endOfOctets)`
  23. eooIntegerSubstrate = (0, 0)
  24. eooOctetsSubstrate = bytes(eooIntegerSubstrate)
  25. # noinspection PyMethodMayBeStatic
  26. def encodeTag(self, singleTag, isConstructed):
  27. tagClass, tagFormat, tagId = singleTag
  28. encodedTag = tagClass | tagFormat
  29. if isConstructed:
  30. encodedTag |= tag.tagFormatConstructed
  31. if tagId < 31:
  32. return encodedTag | tagId,
  33. else:
  34. substrate = tagId & 0x7f,
  35. tagId >>= 7
  36. while tagId:
  37. substrate = (0x80 | (tagId & 0x7f),) + substrate
  38. tagId >>= 7
  39. return (encodedTag | 0x1F,) + substrate
  40. def encodeLength(self, length, defMode):
  41. if not defMode and self.supportIndefLenMode:
  42. return (0x80,)
  43. if length < 0x80:
  44. return length,
  45. else:
  46. substrate = ()
  47. while length:
  48. substrate = (length & 0xff,) + substrate
  49. length >>= 8
  50. substrateLen = len(substrate)
  51. if substrateLen > 126:
  52. raise error.PyAsn1Error('Length octets overflow (%d)' % substrateLen)
  53. return (0x80 | substrateLen,) + substrate
  54. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  55. raise error.PyAsn1Error('Not implemented')
  56. def encode(self, value, asn1Spec=None, encodeFun=None, **options):
  57. if asn1Spec is None:
  58. tagSet = value.tagSet
  59. else:
  60. tagSet = asn1Spec.tagSet
  61. # untagged item?
  62. if not tagSet:
  63. substrate, isConstructed, isOctets = self.encodeValue(
  64. value, asn1Spec, encodeFun, **options
  65. )
  66. return substrate
  67. defMode = options.get('defMode', True)
  68. substrate = b''
  69. for idx, singleTag in enumerate(tagSet.superTags):
  70. defModeOverride = defMode
  71. # base tag?
  72. if not idx:
  73. try:
  74. substrate, isConstructed, isOctets = self.encodeValue(
  75. value, asn1Spec, encodeFun, **options
  76. )
  77. except error.PyAsn1Error as exc:
  78. raise error.PyAsn1Error(
  79. 'Error encoding %r: %s' % (value, exc))
  80. if LOG:
  81. LOG('encoded %svalue %s into %s' % (
  82. isConstructed and 'constructed ' or '', value, substrate
  83. ))
  84. if not substrate and isConstructed and options.get('ifNotEmpty', False):
  85. return substrate
  86. if not isConstructed:
  87. defModeOverride = True
  88. if LOG:
  89. LOG('overridden encoding mode into definitive for primitive type')
  90. header = self.encodeTag(singleTag, isConstructed)
  91. if LOG:
  92. LOG('encoded %stag %s into %s' % (
  93. isConstructed and 'constructed ' or '',
  94. singleTag, debug.hexdump(bytes(header))))
  95. header += self.encodeLength(len(substrate), defModeOverride)
  96. if LOG:
  97. LOG('encoded %s octets (tag + payload) into %s' % (
  98. len(substrate), debug.hexdump(bytes(header))))
  99. if isOctets:
  100. substrate = bytes(header) + substrate
  101. if not defModeOverride:
  102. substrate += self.eooOctetsSubstrate
  103. else:
  104. substrate = header + substrate
  105. if not defModeOverride:
  106. substrate += self.eooIntegerSubstrate
  107. if not isOctets:
  108. substrate = bytes(substrate)
  109. return substrate
  110. class EndOfOctetsEncoder(AbstractItemEncoder):
  111. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  112. return b'', False, True
  113. class BooleanEncoder(AbstractItemEncoder):
  114. supportIndefLenMode = False
  115. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  116. return value and (1,) or (0,), False, False
  117. class IntegerEncoder(AbstractItemEncoder):
  118. supportIndefLenMode = False
  119. supportCompactZero = False
  120. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  121. if value == 0:
  122. if LOG:
  123. LOG('encoding %spayload for zero INTEGER' % (
  124. self.supportCompactZero and 'no ' or ''
  125. ))
  126. # de-facto way to encode zero
  127. if self.supportCompactZero:
  128. return (), False, False
  129. else:
  130. return (0,), False, False
  131. return to_bytes(int(value), signed=True), False, True
  132. class BitStringEncoder(AbstractItemEncoder):
  133. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  134. if asn1Spec is not None:
  135. # TODO: try to avoid ASN.1 schema instantiation
  136. value = asn1Spec.clone(value)
  137. valueLength = len(value)
  138. if valueLength % 8:
  139. alignedValue = value << (8 - valueLength % 8)
  140. else:
  141. alignedValue = value
  142. maxChunkSize = options.get('maxChunkSize', 0)
  143. if not maxChunkSize or len(alignedValue) <= maxChunkSize * 8:
  144. substrate = alignedValue.asOctets()
  145. return bytes((len(substrate) * 8 - valueLength,)) + substrate, False, True
  146. if LOG:
  147. LOG('encoding into up to %s-octet chunks' % maxChunkSize)
  148. baseTag = value.tagSet.baseTag
  149. # strip off explicit tags
  150. if baseTag:
  151. tagSet = tag.TagSet(baseTag, baseTag)
  152. else:
  153. tagSet = tag.TagSet()
  154. alignedValue = alignedValue.clone(tagSet=tagSet)
  155. stop = 0
  156. substrate = b''
  157. while stop < valueLength:
  158. start = stop
  159. stop = min(start + maxChunkSize * 8, valueLength)
  160. substrate += encodeFun(alignedValue[start:stop], asn1Spec, **options)
  161. return substrate, True, True
  162. class OctetStringEncoder(AbstractItemEncoder):
  163. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  164. if asn1Spec is None:
  165. substrate = value.asOctets()
  166. elif not isinstance(value, bytes):
  167. substrate = asn1Spec.clone(value).asOctets()
  168. else:
  169. substrate = value
  170. maxChunkSize = options.get('maxChunkSize', 0)
  171. if not maxChunkSize or len(substrate) <= maxChunkSize:
  172. return substrate, False, True
  173. if LOG:
  174. LOG('encoding into up to %s-octet chunks' % maxChunkSize)
  175. # strip off explicit tags for inner chunks
  176. if asn1Spec is None:
  177. baseTag = value.tagSet.baseTag
  178. # strip off explicit tags
  179. if baseTag:
  180. tagSet = tag.TagSet(baseTag, baseTag)
  181. else:
  182. tagSet = tag.TagSet()
  183. asn1Spec = value.clone(tagSet=tagSet)
  184. elif not isinstance(value, bytes):
  185. baseTag = asn1Spec.tagSet.baseTag
  186. # strip off explicit tags
  187. if baseTag:
  188. tagSet = tag.TagSet(baseTag, baseTag)
  189. else:
  190. tagSet = tag.TagSet()
  191. asn1Spec = asn1Spec.clone(tagSet=tagSet)
  192. pos = 0
  193. substrate = b''
  194. while True:
  195. chunk = value[pos:pos + maxChunkSize]
  196. if not chunk:
  197. break
  198. substrate += encodeFun(chunk, asn1Spec, **options)
  199. pos += maxChunkSize
  200. return substrate, True, True
  201. class NullEncoder(AbstractItemEncoder):
  202. supportIndefLenMode = False
  203. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  204. return b'', False, True
  205. class ObjectIdentifierEncoder(AbstractItemEncoder):
  206. supportIndefLenMode = False
  207. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  208. if asn1Spec is not None:
  209. value = asn1Spec.clone(value)
  210. oid = value.asTuple()
  211. # Build the first pair
  212. try:
  213. first = oid[0]
  214. second = oid[1]
  215. except IndexError:
  216. raise error.PyAsn1Error('Short OID %s' % (value,))
  217. if 0 <= second <= 39:
  218. if first == 1:
  219. oid = (second + 40,) + oid[2:]
  220. elif first == 0:
  221. oid = (second,) + oid[2:]
  222. elif first == 2:
  223. oid = (second + 80,) + oid[2:]
  224. else:
  225. raise error.PyAsn1Error('Impossible first/second arcs at %s' % (value,))
  226. elif first == 2:
  227. oid = (second + 80,) + oid[2:]
  228. else:
  229. raise error.PyAsn1Error('Impossible first/second arcs at %s' % (value,))
  230. octets = ()
  231. # Cycle through subIds
  232. for subOid in oid:
  233. if 0 <= subOid <= 127:
  234. # Optimize for the common case
  235. octets += (subOid,)
  236. elif subOid > 127:
  237. # Pack large Sub-Object IDs
  238. res = (subOid & 0x7f,)
  239. subOid >>= 7
  240. while subOid:
  241. res = (0x80 | (subOid & 0x7f),) + res
  242. subOid >>= 7
  243. # Add packed Sub-Object ID to resulted Object ID
  244. octets += res
  245. else:
  246. raise error.PyAsn1Error('Negative OID arc %s at %s' % (subOid, value))
  247. return octets, False, False
  248. class RelativeOIDEncoder(AbstractItemEncoder):
  249. supportIndefLenMode = False
  250. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  251. if asn1Spec is not None:
  252. value = asn1Spec.clone(value)
  253. octets = ()
  254. # Cycle through subIds
  255. for subOid in value.asTuple():
  256. if 0 <= subOid <= 127:
  257. # Optimize for the common case
  258. octets += (subOid,)
  259. elif subOid > 127:
  260. # Pack large Sub-Object IDs
  261. res = (subOid & 0x7f,)
  262. subOid >>= 7
  263. while subOid:
  264. res = (0x80 | (subOid & 0x7f),) + res
  265. subOid >>= 7
  266. # Add packed Sub-Object ID to resulted RELATIVE-OID
  267. octets += res
  268. else:
  269. raise error.PyAsn1Error('Negative RELATIVE-OID arc %s at %s' % (subOid, value))
  270. return octets, False, False
  271. class RealEncoder(AbstractItemEncoder):
  272. supportIndefLenMode = False
  273. binEncBase = 2 # set to None to choose encoding base automatically
  274. @staticmethod
  275. def _dropFloatingPoint(m, encbase, e):
  276. ms, es = 1, 1
  277. if m < 0:
  278. ms = -1 # mantissa sign
  279. if e < 0:
  280. es = -1 # exponent sign
  281. m *= ms
  282. if encbase == 8:
  283. m *= 2 ** (abs(e) % 3 * es)
  284. e = abs(e) // 3 * es
  285. elif encbase == 16:
  286. m *= 2 ** (abs(e) % 4 * es)
  287. e = abs(e) // 4 * es
  288. while True:
  289. if int(m) != m:
  290. m *= encbase
  291. e -= 1
  292. continue
  293. break
  294. return ms, int(m), encbase, e
  295. def _chooseEncBase(self, value):
  296. m, b, e = value
  297. encBase = [2, 8, 16]
  298. if value.binEncBase in encBase:
  299. return self._dropFloatingPoint(m, value.binEncBase, e)
  300. elif self.binEncBase in encBase:
  301. return self._dropFloatingPoint(m, self.binEncBase, e)
  302. # auto choosing base 2/8/16
  303. mantissa = [m, m, m]
  304. exponent = [e, e, e]
  305. sign = 1
  306. encbase = 2
  307. e = float('inf')
  308. for i in range(3):
  309. (sign,
  310. mantissa[i],
  311. encBase[i],
  312. exponent[i]) = self._dropFloatingPoint(mantissa[i], encBase[i], exponent[i])
  313. if abs(exponent[i]) < abs(e) or (abs(exponent[i]) == abs(e) and mantissa[i] < m):
  314. e = exponent[i]
  315. m = int(mantissa[i])
  316. encbase = encBase[i]
  317. if LOG:
  318. LOG('automatically chosen REAL encoding base %s, sign %s, mantissa %s, '
  319. 'exponent %s' % (encbase, sign, m, e))
  320. return sign, m, encbase, e
  321. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  322. if asn1Spec is not None:
  323. value = asn1Spec.clone(value)
  324. if value.isPlusInf:
  325. return (0x40,), False, False
  326. if value.isMinusInf:
  327. return (0x41,), False, False
  328. m, b, e = value
  329. if not m:
  330. return b'', False, True
  331. if b == 10:
  332. if LOG:
  333. LOG('encoding REAL into character form')
  334. return b'\x03%dE%s%d' % (m, e == 0 and b'+' or b'', e), False, True
  335. elif b == 2:
  336. fo = 0x80 # binary encoding
  337. ms, m, encbase, e = self._chooseEncBase(value)
  338. if ms < 0: # mantissa sign
  339. fo |= 0x40 # sign bit
  340. # exponent & mantissa normalization
  341. if encbase == 2:
  342. while m & 0x1 == 0:
  343. m >>= 1
  344. e += 1
  345. elif encbase == 8:
  346. while m & 0x7 == 0:
  347. m >>= 3
  348. e += 1
  349. fo |= 0x10
  350. else: # encbase = 16
  351. while m & 0xf == 0:
  352. m >>= 4
  353. e += 1
  354. fo |= 0x20
  355. sf = 0 # scale factor
  356. while m & 0x1 == 0:
  357. m >>= 1
  358. sf += 1
  359. if sf > 3:
  360. raise error.PyAsn1Error('Scale factor overflow') # bug if raised
  361. fo |= sf << 2
  362. eo = b''
  363. if e == 0 or e == -1:
  364. eo = bytes((e & 0xff,))
  365. else:
  366. while e not in (0, -1):
  367. eo = bytes((e & 0xff,)) + eo
  368. e >>= 8
  369. if e == 0 and eo and eo[0] & 0x80:
  370. eo = bytes((0,)) + eo
  371. if e == -1 and eo and not (eo[0] & 0x80):
  372. eo = bytes((0xff,)) + eo
  373. n = len(eo)
  374. if n > 0xff:
  375. raise error.PyAsn1Error('Real exponent overflow')
  376. if n == 1:
  377. pass
  378. elif n == 2:
  379. fo |= 1
  380. elif n == 3:
  381. fo |= 2
  382. else:
  383. fo |= 3
  384. eo = bytes((n & 0xff,)) + eo
  385. po = b''
  386. while m:
  387. po = bytes((m & 0xff,)) + po
  388. m >>= 8
  389. substrate = bytes((fo,)) + eo + po
  390. return substrate, False, True
  391. else:
  392. raise error.PyAsn1Error('Prohibited Real base %s' % b)
  393. class SequenceEncoder(AbstractItemEncoder):
  394. omitEmptyOptionals = False
  395. # TODO: handling three flavors of input is too much -- split over codecs
  396. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  397. substrate = b''
  398. omitEmptyOptionals = options.get(
  399. 'omitEmptyOptionals', self.omitEmptyOptionals)
  400. if LOG:
  401. LOG('%sencoding empty OPTIONAL components' % (
  402. omitEmptyOptionals and 'not ' or ''))
  403. if asn1Spec is None:
  404. # instance of ASN.1 schema
  405. inconsistency = value.isInconsistent
  406. if inconsistency:
  407. raise error.PyAsn1Error(
  408. f"ASN.1 object {value.__class__.__name__} is inconsistent")
  409. namedTypes = value.componentType
  410. for idx, component in enumerate(value.values()):
  411. if namedTypes:
  412. namedType = namedTypes[idx]
  413. if namedType.isOptional and not component.isValue:
  414. if LOG:
  415. LOG('not encoding OPTIONAL component %r' % (namedType,))
  416. continue
  417. if namedType.isDefaulted and component == namedType.asn1Object:
  418. if LOG:
  419. LOG('not encoding DEFAULT component %r' % (namedType,))
  420. continue
  421. if omitEmptyOptionals:
  422. options.update(ifNotEmpty=namedType.isOptional)
  423. # wrap open type blob if needed
  424. if namedTypes and namedType.openType:
  425. wrapType = namedType.asn1Object
  426. if wrapType.typeId in (
  427. univ.SetOf.typeId, univ.SequenceOf.typeId):
  428. substrate += encodeFun(
  429. component, asn1Spec,
  430. **dict(options, wrapType=wrapType.componentType))
  431. else:
  432. chunk = encodeFun(component, asn1Spec, **options)
  433. if wrapType.isSameTypeWith(component):
  434. substrate += chunk
  435. else:
  436. substrate += encodeFun(chunk, wrapType, **options)
  437. if LOG:
  438. LOG('wrapped with wrap type %r' % (wrapType,))
  439. else:
  440. substrate += encodeFun(component, asn1Spec, **options)
  441. else:
  442. # bare Python value + ASN.1 schema
  443. for idx, namedType in enumerate(asn1Spec.componentType.namedTypes):
  444. try:
  445. component = value[namedType.name]
  446. except KeyError:
  447. raise error.PyAsn1Error('Component name "%s" not found in %r' % (
  448. namedType.name, value))
  449. if namedType.isOptional and namedType.name not in value:
  450. if LOG:
  451. LOG('not encoding OPTIONAL component %r' % (namedType,))
  452. continue
  453. if namedType.isDefaulted and component == namedType.asn1Object:
  454. if LOG:
  455. LOG('not encoding DEFAULT component %r' % (namedType,))
  456. continue
  457. if omitEmptyOptionals:
  458. options.update(ifNotEmpty=namedType.isOptional)
  459. componentSpec = namedType.asn1Object
  460. # wrap open type blob if needed
  461. if namedType.openType:
  462. if componentSpec.typeId in (
  463. univ.SetOf.typeId, univ.SequenceOf.typeId):
  464. substrate += encodeFun(
  465. component, componentSpec,
  466. **dict(options, wrapType=componentSpec.componentType))
  467. else:
  468. chunk = encodeFun(component, componentSpec, **options)
  469. if componentSpec.isSameTypeWith(component):
  470. substrate += chunk
  471. else:
  472. substrate += encodeFun(chunk, componentSpec, **options)
  473. if LOG:
  474. LOG('wrapped with wrap type %r' % (componentSpec,))
  475. else:
  476. substrate += encodeFun(component, componentSpec, **options)
  477. return substrate, True, True
  478. class SequenceOfEncoder(AbstractItemEncoder):
  479. def _encodeComponents(self, value, asn1Spec, encodeFun, **options):
  480. if asn1Spec is None:
  481. inconsistency = value.isInconsistent
  482. if inconsistency:
  483. raise error.PyAsn1Error(
  484. f"ASN.1 object {value.__class__.__name__} is inconsistent")
  485. else:
  486. asn1Spec = asn1Spec.componentType
  487. chunks = []
  488. wrapType = options.pop('wrapType', None)
  489. for idx, component in enumerate(value):
  490. chunk = encodeFun(component, asn1Spec, **options)
  491. if (wrapType is not None and
  492. not wrapType.isSameTypeWith(component)):
  493. # wrap encoded value with wrapper container (e.g. ANY)
  494. chunk = encodeFun(chunk, wrapType, **options)
  495. if LOG:
  496. LOG('wrapped with wrap type %r' % (wrapType,))
  497. chunks.append(chunk)
  498. return chunks
  499. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  500. chunks = self._encodeComponents(
  501. value, asn1Spec, encodeFun, **options)
  502. return b''.join(chunks), True, True
  503. class ChoiceEncoder(AbstractItemEncoder):
  504. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  505. if asn1Spec is None:
  506. component = value.getComponent()
  507. else:
  508. names = [namedType.name for namedType in asn1Spec.componentType.namedTypes
  509. if namedType.name in value]
  510. if len(names) != 1:
  511. raise error.PyAsn1Error('%s components for Choice at %r' % (len(names) and 'Multiple ' or 'None ', value))
  512. name = names[0]
  513. component = value[name]
  514. asn1Spec = asn1Spec[name]
  515. return encodeFun(component, asn1Spec, **options), True, True
  516. class AnyEncoder(OctetStringEncoder):
  517. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  518. if asn1Spec is None:
  519. value = value.asOctets()
  520. elif not isinstance(value, bytes):
  521. value = asn1Spec.clone(value).asOctets()
  522. return value, not options.get('defMode', True), True
  523. TAG_MAP = {
  524. eoo.endOfOctets.tagSet: EndOfOctetsEncoder(),
  525. univ.Boolean.tagSet: BooleanEncoder(),
  526. univ.Integer.tagSet: IntegerEncoder(),
  527. univ.BitString.tagSet: BitStringEncoder(),
  528. univ.OctetString.tagSet: OctetStringEncoder(),
  529. univ.Null.tagSet: NullEncoder(),
  530. univ.ObjectIdentifier.tagSet: ObjectIdentifierEncoder(),
  531. univ.RelativeOID.tagSet: RelativeOIDEncoder(),
  532. univ.Enumerated.tagSet: IntegerEncoder(),
  533. univ.Real.tagSet: RealEncoder(),
  534. # Sequence & Set have same tags as SequenceOf & SetOf
  535. univ.SequenceOf.tagSet: SequenceOfEncoder(),
  536. univ.SetOf.tagSet: SequenceOfEncoder(),
  537. univ.Choice.tagSet: ChoiceEncoder(),
  538. # character string types
  539. char.UTF8String.tagSet: OctetStringEncoder(),
  540. char.NumericString.tagSet: OctetStringEncoder(),
  541. char.PrintableString.tagSet: OctetStringEncoder(),
  542. char.TeletexString.tagSet: OctetStringEncoder(),
  543. char.VideotexString.tagSet: OctetStringEncoder(),
  544. char.IA5String.tagSet: OctetStringEncoder(),
  545. char.GraphicString.tagSet: OctetStringEncoder(),
  546. char.VisibleString.tagSet: OctetStringEncoder(),
  547. char.GeneralString.tagSet: OctetStringEncoder(),
  548. char.UniversalString.tagSet: OctetStringEncoder(),
  549. char.BMPString.tagSet: OctetStringEncoder(),
  550. # useful types
  551. useful.ObjectDescriptor.tagSet: OctetStringEncoder(),
  552. useful.GeneralizedTime.tagSet: OctetStringEncoder(),
  553. useful.UTCTime.tagSet: OctetStringEncoder()
  554. }
  555. # Put in ambiguous & non-ambiguous types for faster codec lookup
  556. TYPE_MAP = {
  557. univ.Boolean.typeId: BooleanEncoder(),
  558. univ.Integer.typeId: IntegerEncoder(),
  559. univ.BitString.typeId: BitStringEncoder(),
  560. univ.OctetString.typeId: OctetStringEncoder(),
  561. univ.Null.typeId: NullEncoder(),
  562. univ.ObjectIdentifier.typeId: ObjectIdentifierEncoder(),
  563. univ.RelativeOID.typeId: RelativeOIDEncoder(),
  564. univ.Enumerated.typeId: IntegerEncoder(),
  565. univ.Real.typeId: RealEncoder(),
  566. # Sequence & Set have same tags as SequenceOf & SetOf
  567. univ.Set.typeId: SequenceEncoder(),
  568. univ.SetOf.typeId: SequenceOfEncoder(),
  569. univ.Sequence.typeId: SequenceEncoder(),
  570. univ.SequenceOf.typeId: SequenceOfEncoder(),
  571. univ.Choice.typeId: ChoiceEncoder(),
  572. univ.Any.typeId: AnyEncoder(),
  573. # character string types
  574. char.UTF8String.typeId: OctetStringEncoder(),
  575. char.NumericString.typeId: OctetStringEncoder(),
  576. char.PrintableString.typeId: OctetStringEncoder(),
  577. char.TeletexString.typeId: OctetStringEncoder(),
  578. char.VideotexString.typeId: OctetStringEncoder(),
  579. char.IA5String.typeId: OctetStringEncoder(),
  580. char.GraphicString.typeId: OctetStringEncoder(),
  581. char.VisibleString.typeId: OctetStringEncoder(),
  582. char.GeneralString.typeId: OctetStringEncoder(),
  583. char.UniversalString.typeId: OctetStringEncoder(),
  584. char.BMPString.typeId: OctetStringEncoder(),
  585. # useful types
  586. useful.ObjectDescriptor.typeId: OctetStringEncoder(),
  587. useful.GeneralizedTime.typeId: OctetStringEncoder(),
  588. useful.UTCTime.typeId: OctetStringEncoder()
  589. }
  590. class SingleItemEncoder(object):
  591. fixedDefLengthMode = None
  592. fixedChunkSize = None
  593. TAG_MAP = TAG_MAP
  594. TYPE_MAP = TYPE_MAP
  595. def __init__(self, tagMap=_MISSING, typeMap=_MISSING, **ignored):
  596. self._tagMap = tagMap if tagMap is not _MISSING else self.TAG_MAP
  597. self._typeMap = typeMap if typeMap is not _MISSING else self.TYPE_MAP
  598. def __call__(self, value, asn1Spec=None, **options):
  599. try:
  600. if asn1Spec is None:
  601. typeId = value.typeId
  602. else:
  603. typeId = asn1Spec.typeId
  604. except AttributeError:
  605. raise error.PyAsn1Error('Value %r is not ASN.1 type instance '
  606. 'and "asn1Spec" not given' % (value,))
  607. if LOG:
  608. LOG('encoder called in %sdef mode, chunk size %s for type %s, '
  609. 'value:\n%s' % (not options.get('defMode', True) and 'in' or '',
  610. options.get('maxChunkSize', 0),
  611. asn1Spec is None and value.prettyPrintType() or
  612. asn1Spec.prettyPrintType(), value))
  613. if self.fixedDefLengthMode is not None:
  614. options.update(defMode=self.fixedDefLengthMode)
  615. if self.fixedChunkSize is not None:
  616. options.update(maxChunkSize=self.fixedChunkSize)
  617. try:
  618. concreteEncoder = self._typeMap[typeId]
  619. if LOG:
  620. LOG('using value codec %s chosen by type ID '
  621. '%s' % (concreteEncoder.__class__.__name__, typeId))
  622. except KeyError:
  623. if asn1Spec is None:
  624. tagSet = value.tagSet
  625. else:
  626. tagSet = asn1Spec.tagSet
  627. # use base type for codec lookup to recover untagged types
  628. baseTagSet = tag.TagSet(tagSet.baseTag, tagSet.baseTag)
  629. try:
  630. concreteEncoder = self._tagMap[baseTagSet]
  631. except KeyError:
  632. raise error.PyAsn1Error('No encoder for %r (%s)' % (value, tagSet))
  633. if LOG:
  634. LOG('using value codec %s chosen by tagSet '
  635. '%s' % (concreteEncoder.__class__.__name__, tagSet))
  636. substrate = concreteEncoder.encode(value, asn1Spec, self, **options)
  637. if LOG:
  638. LOG('codec %s built %s octets of substrate: %s\nencoder '
  639. 'completed' % (concreteEncoder, len(substrate),
  640. debug.hexdump(substrate)))
  641. return substrate
  642. class Encoder(object):
  643. SINGLE_ITEM_ENCODER = SingleItemEncoder
  644. def __init__(self, tagMap=_MISSING, typeMap=_MISSING, **options):
  645. self._singleItemEncoder = self.SINGLE_ITEM_ENCODER(
  646. tagMap=tagMap, typeMap=typeMap, **options
  647. )
  648. def __call__(self, pyObject, asn1Spec=None, **options):
  649. return self._singleItemEncoder(
  650. pyObject, asn1Spec=asn1Spec, **options)
  651. #: Turns ASN.1 object into BER octet stream.
  652. #:
  653. #: Takes any ASN.1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative)
  654. #: walks all its components recursively and produces a BER octet stream.
  655. #:
  656. #: Parameters
  657. #: ----------
  658. #: value: either a Python or pyasn1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative)
  659. #: A Python or pyasn1 object to encode. If Python object is given, `asnSpec`
  660. #: parameter is required to guide the encoding process.
  661. #:
  662. #: Keyword Args
  663. #: ------------
  664. #: asn1Spec:
  665. #: Optional ASN.1 schema or value object e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative
  666. #:
  667. #: defMode: :py:class:`bool`
  668. #: If :obj:`False`, produces indefinite length encoding
  669. #:
  670. #: maxChunkSize: :py:class:`int`
  671. #: Maximum chunk size in chunked encoding mode (0 denotes unlimited chunk size)
  672. #:
  673. #: Returns
  674. #: -------
  675. #: : :py:class:`bytes`
  676. #: Given ASN.1 object encoded into BER octetstream
  677. #:
  678. #: Raises
  679. #: ------
  680. #: ~pyasn1.error.PyAsn1Error
  681. #: On encoding errors
  682. #:
  683. #: Examples
  684. #: --------
  685. #: Encode Python value into BER with ASN.1 schema
  686. #:
  687. #: .. code-block:: pycon
  688. #:
  689. #: >>> seq = SequenceOf(componentType=Integer())
  690. #: >>> encode([1, 2, 3], asn1Spec=seq)
  691. #: b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03'
  692. #:
  693. #: Encode ASN.1 value object into BER
  694. #:
  695. #: .. code-block:: pycon
  696. #:
  697. #: >>> seq = SequenceOf(componentType=Integer())
  698. #: >>> seq.extend([1, 2, 3])
  699. #: >>> encode(seq)
  700. #: b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03'
  701. #:
  702. encode = Encoder()
  703. def __getattr__(attr: str):
  704. if newAttr := {"tagMap": "TAG_MAP", "typeMap": "TYPE_MAP"}.get(attr):
  705. warnings.warn(f"{attr} is deprecated. Please use {newAttr} instead.", DeprecationWarning)
  706. return globals()[newAttr]
  707. raise AttributeError(attr)