encoder.py 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331
  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 warnings
  8. from pyasn1 import error
  9. from pyasn1.codec.ber import encoder
  10. from pyasn1.type import univ
  11. from pyasn1.type import useful
  12. __all__ = ['Encoder', 'encode']
  13. class BooleanEncoder(encoder.IntegerEncoder):
  14. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  15. if value == 0:
  16. substrate = (0,)
  17. else:
  18. substrate = (255,)
  19. return substrate, False, False
  20. class RealEncoder(encoder.RealEncoder):
  21. def _chooseEncBase(self, value):
  22. m, b, e = value
  23. return self._dropFloatingPoint(m, b, e)
  24. # specialized GeneralStringEncoder here
  25. class TimeEncoderMixIn(object):
  26. Z_CHAR = ord('Z')
  27. PLUS_CHAR = ord('+')
  28. MINUS_CHAR = ord('-')
  29. COMMA_CHAR = ord(',')
  30. DOT_CHAR = ord('.')
  31. ZERO_CHAR = ord('0')
  32. MIN_LENGTH = 12
  33. MAX_LENGTH = 19
  34. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  35. # CER encoding constraints:
  36. # - minutes are mandatory, seconds are optional
  37. # - sub-seconds must NOT be zero / no meaningless zeros
  38. # - no hanging fraction dot
  39. # - time in UTC (Z)
  40. # - only dot is allowed for fractions
  41. if asn1Spec is not None:
  42. value = asn1Spec.clone(value)
  43. numbers = value.asNumbers()
  44. if self.PLUS_CHAR in numbers or self.MINUS_CHAR in numbers:
  45. raise error.PyAsn1Error('Must be UTC time: %r' % value)
  46. if numbers[-1] != self.Z_CHAR:
  47. raise error.PyAsn1Error('Missing "Z" time zone specifier: %r' % value)
  48. if self.COMMA_CHAR in numbers:
  49. raise error.PyAsn1Error('Comma in fractions disallowed: %r' % value)
  50. if self.DOT_CHAR in numbers:
  51. isModified = False
  52. numbers = list(numbers)
  53. searchIndex = min(numbers.index(self.DOT_CHAR) + 4, len(numbers) - 1)
  54. while numbers[searchIndex] != self.DOT_CHAR:
  55. if numbers[searchIndex] == self.ZERO_CHAR:
  56. del numbers[searchIndex]
  57. isModified = True
  58. searchIndex -= 1
  59. searchIndex += 1
  60. if searchIndex < len(numbers):
  61. if numbers[searchIndex] == self.Z_CHAR:
  62. # drop hanging comma
  63. del numbers[searchIndex - 1]
  64. isModified = True
  65. if isModified:
  66. value = value.clone(numbers)
  67. if not self.MIN_LENGTH < len(numbers) < self.MAX_LENGTH:
  68. raise error.PyAsn1Error('Length constraint violated: %r' % value)
  69. options.update(maxChunkSize=1000)
  70. return encoder.OctetStringEncoder.encodeValue(
  71. self, value, asn1Spec, encodeFun, **options
  72. )
  73. class GeneralizedTimeEncoder(TimeEncoderMixIn, encoder.OctetStringEncoder):
  74. MIN_LENGTH = 12
  75. MAX_LENGTH = 20
  76. class UTCTimeEncoder(TimeEncoderMixIn, encoder.OctetStringEncoder):
  77. MIN_LENGTH = 10
  78. MAX_LENGTH = 14
  79. class SetOfEncoder(encoder.SequenceOfEncoder):
  80. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  81. chunks = self._encodeComponents(
  82. value, asn1Spec, encodeFun, **options)
  83. # sort by serialised and padded components
  84. if len(chunks) > 1:
  85. zero = b'\x00'
  86. maxLen = max(map(len, chunks))
  87. paddedChunks = [
  88. (x.ljust(maxLen, zero), x) for x in chunks
  89. ]
  90. paddedChunks.sort(key=lambda x: x[0])
  91. chunks = [x[1] for x in paddedChunks]
  92. return b''.join(chunks), True, True
  93. class SequenceOfEncoder(encoder.SequenceOfEncoder):
  94. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  95. if options.get('ifNotEmpty', False) and not len(value):
  96. return b'', True, True
  97. chunks = self._encodeComponents(
  98. value, asn1Spec, encodeFun, **options)
  99. return b''.join(chunks), True, True
  100. class SetEncoder(encoder.SequenceEncoder):
  101. @staticmethod
  102. def _componentSortKey(componentAndType):
  103. """Sort SET components by tag
  104. Sort regardless of the Choice value (static sort)
  105. """
  106. component, asn1Spec = componentAndType
  107. if asn1Spec is None:
  108. asn1Spec = component
  109. if asn1Spec.typeId == univ.Choice.typeId and not asn1Spec.tagSet:
  110. if asn1Spec.tagSet:
  111. return asn1Spec.tagSet
  112. else:
  113. return asn1Spec.componentType.minTagSet
  114. else:
  115. return asn1Spec.tagSet
  116. def encodeValue(self, value, asn1Spec, encodeFun, **options):
  117. substrate = b''
  118. comps = []
  119. compsMap = {}
  120. if asn1Spec is None:
  121. # instance of ASN.1 schema
  122. inconsistency = value.isInconsistent
  123. if inconsistency:
  124. raise error.PyAsn1Error(
  125. f"ASN.1 object {value.__class__.__name__} is inconsistent")
  126. namedTypes = value.componentType
  127. for idx, component in enumerate(value.values()):
  128. if namedTypes:
  129. namedType = namedTypes[idx]
  130. if namedType.isOptional and not component.isValue:
  131. continue
  132. if namedType.isDefaulted and component == namedType.asn1Object:
  133. continue
  134. compsMap[id(component)] = namedType
  135. else:
  136. compsMap[id(component)] = None
  137. comps.append((component, asn1Spec))
  138. else:
  139. # bare Python value + ASN.1 schema
  140. for idx, namedType in enumerate(asn1Spec.componentType.namedTypes):
  141. try:
  142. component = value[namedType.name]
  143. except KeyError:
  144. raise error.PyAsn1Error('Component name "%s" not found in %r' % (namedType.name, value))
  145. if namedType.isOptional and namedType.name not in value:
  146. continue
  147. if namedType.isDefaulted and component == namedType.asn1Object:
  148. continue
  149. compsMap[id(component)] = namedType
  150. comps.append((component, asn1Spec[idx]))
  151. for comp, compType in sorted(comps, key=self._componentSortKey):
  152. namedType = compsMap[id(comp)]
  153. if namedType:
  154. options.update(ifNotEmpty=namedType.isOptional)
  155. chunk = encodeFun(comp, compType, **options)
  156. # wrap open type blob if needed
  157. if namedType and namedType.openType:
  158. wrapType = namedType.asn1Object
  159. if wrapType.tagSet and not wrapType.isSameTypeWith(comp):
  160. chunk = encodeFun(chunk, wrapType, **options)
  161. substrate += chunk
  162. return substrate, True, True
  163. class SequenceEncoder(encoder.SequenceEncoder):
  164. omitEmptyOptionals = True
  165. TAG_MAP = encoder.TAG_MAP.copy()
  166. TAG_MAP.update({
  167. univ.Boolean.tagSet: BooleanEncoder(),
  168. univ.Real.tagSet: RealEncoder(),
  169. useful.GeneralizedTime.tagSet: GeneralizedTimeEncoder(),
  170. useful.UTCTime.tagSet: UTCTimeEncoder(),
  171. # Sequence & Set have same tags as SequenceOf & SetOf
  172. univ.SetOf.tagSet: SetOfEncoder(),
  173. univ.Sequence.typeId: SequenceEncoder()
  174. })
  175. TYPE_MAP = encoder.TYPE_MAP.copy()
  176. TYPE_MAP.update({
  177. univ.Boolean.typeId: BooleanEncoder(),
  178. univ.Real.typeId: RealEncoder(),
  179. useful.GeneralizedTime.typeId: GeneralizedTimeEncoder(),
  180. useful.UTCTime.typeId: UTCTimeEncoder(),
  181. # Sequence & Set have same tags as SequenceOf & SetOf
  182. univ.Set.typeId: SetEncoder(),
  183. univ.SetOf.typeId: SetOfEncoder(),
  184. univ.Sequence.typeId: SequenceEncoder(),
  185. univ.SequenceOf.typeId: SequenceOfEncoder()
  186. })
  187. class SingleItemEncoder(encoder.SingleItemEncoder):
  188. fixedDefLengthMode = False
  189. fixedChunkSize = 1000
  190. TAG_MAP = TAG_MAP
  191. TYPE_MAP = TYPE_MAP
  192. class Encoder(encoder.Encoder):
  193. SINGLE_ITEM_ENCODER = SingleItemEncoder
  194. #: Turns ASN.1 object into CER octet stream.
  195. #:
  196. #: Takes any ASN.1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative)
  197. #: walks all its components recursively and produces a CER octet stream.
  198. #:
  199. #: Parameters
  200. #: ----------
  201. #: value: either a Python or pyasn1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative)
  202. #: A Python or pyasn1 object to encode. If Python object is given, `asnSpec`
  203. #: parameter is required to guide the encoding process.
  204. #:
  205. #: Keyword Args
  206. #: ------------
  207. #: asn1Spec:
  208. #: Optional ASN.1 schema or value object e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative
  209. #:
  210. #: Returns
  211. #: -------
  212. #: : :py:class:`bytes`
  213. #: Given ASN.1 object encoded into BER octet-stream
  214. #:
  215. #: Raises
  216. #: ------
  217. #: ~pyasn1.error.PyAsn1Error
  218. #: On encoding errors
  219. #:
  220. #: Examples
  221. #: --------
  222. #: Encode Python value into CER with ASN.1 schema
  223. #:
  224. #: .. code-block:: pycon
  225. #:
  226. #: >>> seq = SequenceOf(componentType=Integer())
  227. #: >>> encode([1, 2, 3], asn1Spec=seq)
  228. #: b'0\x80\x02\x01\x01\x02\x01\x02\x02\x01\x03\x00\x00'
  229. #:
  230. #: Encode ASN.1 value object into CER
  231. #:
  232. #: .. code-block:: pycon
  233. #:
  234. #: >>> seq = SequenceOf(componentType=Integer())
  235. #: >>> seq.extend([1, 2, 3])
  236. #: >>> encode(seq)
  237. #: b'0\x80\x02\x01\x01\x02\x01\x02\x02\x01\x03\x00\x00'
  238. #:
  239. encode = Encoder()
  240. # EncoderFactory queries class instance and builds a map of tags -> encoders
  241. def __getattr__(attr: str):
  242. if newAttr := {"tagMap": "TAG_MAP", "typeMap": "TYPE_MAP"}.get(attr):
  243. warnings.warn(f"{attr} is deprecated. Please use {newAttr} instead.", DeprecationWarning)
  244. return globals()[newAttr]
  245. raise AttributeError(attr)