split platform

This commit is contained in:
j 2016-02-06 15:06:57 +05:30
commit 8c9b09577d
2261 changed files with 676163 additions and 0 deletions

View file

@ -0,0 +1,8 @@
import sys
# http://www.python.org/dev/peps/pep-0396/
__version__ = '0.1.9'
if sys.version_info[:2] < (2, 4):
raise RuntimeError('PyASN1 requires Python 2.4 or later')

View file

@ -0,0 +1 @@
# This file is necessary to make this directory a package.

View file

@ -0,0 +1 @@
# This file is necessary to make this directory a package.

View file

@ -0,0 +1,841 @@
# BER decoder
from pyasn1.type import tag, univ, char, useful, tagmap
from pyasn1.codec.ber import eoo
from pyasn1.compat.octets import oct2int, isOctetsType
from pyasn1 import debug, error
class AbstractDecoder:
protoComponent = None
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
raise error.PyAsn1Error('Decoder not implemented for %s' % (tagSet,))
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
raise error.PyAsn1Error('Indefinite length mode decoder not implemented for %s' % (tagSet,))
class AbstractSimpleDecoder(AbstractDecoder):
tagFormats = (tag.tagFormatSimple,)
def _createComponent(self, asn1Spec, tagSet, value=None):
if tagSet[0][1] not in self.tagFormats:
raise error.PyAsn1Error('Invalid tag format %s for %s' % (tagSet[0], self.protoComponent.prettyPrintType()))
if asn1Spec is None:
return self.protoComponent.clone(value, tagSet)
elif value is None:
return asn1Spec
else:
return asn1Spec.clone(value)
class AbstractConstructedDecoder(AbstractDecoder):
tagFormats = (tag.tagFormatConstructed,)
def _createComponent(self, asn1Spec, tagSet, value=None):
if tagSet[0][1] not in self.tagFormats:
raise error.PyAsn1Error('Invalid tag format %s for %s' % (tagSet[0], self.protoComponent.prettyPrintType()))
if asn1Spec is None:
return self.protoComponent.clone(tagSet)
else:
return asn1Spec.clone()
class ExplicitTagDecoder(AbstractSimpleDecoder):
protoComponent = univ.Any('')
tagFormats = (tag.tagFormatConstructed,)
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
if substrateFun:
return substrateFun(
self._createComponent(asn1Spec, tagSet, ''),
substrate, length
)
head, tail = substrate[:length], substrate[length:]
value, _ = decodeFun(head, asn1Spec, tagSet, length)
return value, tail
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
if substrateFun:
return substrateFun(
self._createComponent(asn1Spec, tagSet, ''),
substrate, length
)
value, substrate = decodeFun(substrate, asn1Spec, tagSet, length)
terminator, substrate = decodeFun(substrate, allowEoo=True)
if eoo.endOfOctets.isSameTypeWith(terminator) and \
terminator == eoo.endOfOctets:
return value, substrate
else:
raise error.PyAsn1Error('Missing end-of-octets terminator')
explicitTagDecoder = ExplicitTagDecoder()
class IntegerDecoder(AbstractSimpleDecoder):
protoComponent = univ.Integer(0)
precomputedValues = {
'\x00': 0,
'\x01': 1,
'\x02': 2,
'\x03': 3,
'\x04': 4,
'\x05': 5,
'\x06': 6,
'\x07': 7,
'\x08': 8,
'\x09': 9,
'\xff': -1,
'\xfe': -2,
'\xfd': -3,
'\xfc': -4,
'\xfb': -5
}
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
if not head:
return self._createComponent(asn1Spec, tagSet, 0), tail
if head in self.precomputedValues:
value = self.precomputedValues[head]
else:
firstOctet = oct2int(head[0])
if firstOctet & 0x80:
value = -1
else:
value = 0
for octet in head:
value = value << 8 | oct2int(octet)
return self._createComponent(asn1Spec, tagSet, value), tail
class BooleanDecoder(IntegerDecoder):
protoComponent = univ.Boolean(0)
def _createComponent(self, asn1Spec, tagSet, value=None):
return IntegerDecoder._createComponent(self, asn1Spec, tagSet, value and 1 or 0)
class BitStringDecoder(AbstractSimpleDecoder):
protoComponent = univ.BitString(())
tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed)
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
if tagSet[0][1] == tag.tagFormatSimple: # XXX what tag to check?
if not head:
raise error.PyAsn1Error('Empty substrate')
trailingBits = oct2int(head[0])
if trailingBits > 7:
raise error.PyAsn1Error(
'Trailing bits overflow %s' % trailingBits
)
head = head[1:]
lsb = p = 0; l = len(head)-1; b = []
while p <= l:
if p == l:
lsb = trailingBits
j = 7
o = oct2int(head[p])
while j >= lsb:
b.append((o>>j)&0x01)
j = j - 1
p = p + 1
return self._createComponent(asn1Spec, tagSet, b), tail
r = self._createComponent(asn1Spec, tagSet, ())
if substrateFun:
return substrateFun(r, substrate, length)
while head:
component, head = decodeFun(head, self.protoComponent)
r = r + component
return r, tail
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
r = self._createComponent(asn1Spec, tagSet, '')
if substrateFun:
return substrateFun(r, substrate, length)
while substrate:
component, substrate = decodeFun(substrate, self.protoComponent,
allowEoo=True)
if eoo.endOfOctets.isSameTypeWith(component) and \
component == eoo.endOfOctets:
break
r = r + component
else:
raise error.SubstrateUnderrunError(
'No EOO seen before substrate ends'
)
return r, substrate
class OctetStringDecoder(AbstractSimpleDecoder):
protoComponent = univ.OctetString('')
tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed)
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
if tagSet[0][1] == tag.tagFormatSimple: # XXX what tag to check?
return self._createComponent(asn1Spec, tagSet, head), tail
r = self._createComponent(asn1Spec, tagSet, '')
if substrateFun:
return substrateFun(r, substrate, length)
while head:
component, head = decodeFun(head, self.protoComponent)
r = r + component
return r, tail
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
r = self._createComponent(asn1Spec, tagSet, '')
if substrateFun:
return substrateFun(r, substrate, length)
while substrate:
component, substrate = decodeFun(substrate, self.protoComponent,
allowEoo=True)
if eoo.endOfOctets.isSameTypeWith(component) and \
component == eoo.endOfOctets:
break
r = r + component
else:
raise error.SubstrateUnderrunError(
'No EOO seen before substrate ends'
)
return r, substrate
class NullDecoder(AbstractSimpleDecoder):
protoComponent = univ.Null('')
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
r = self._createComponent(asn1Spec, tagSet)
if head:
raise error.PyAsn1Error('Unexpected %d-octet substrate for Null' % length)
return r, tail
class ObjectIdentifierDecoder(AbstractSimpleDecoder):
protoComponent = univ.ObjectIdentifier(())
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
if not head:
raise error.PyAsn1Error('Empty substrate')
oid = ()
index = 0
substrateLen = len(head)
while index < substrateLen:
subId = oct2int(head[index])
index += 1
if subId < 128:
oid = oid + (subId,)
elif subId > 128:
# Construct subid from a number of octets
nextSubId = subId
subId = 0
while nextSubId >= 128:
subId = (subId << 7) + (nextSubId & 0x7F)
if index >= substrateLen:
raise error.SubstrateUnderrunError(
'Short substrate for sub-OID past %s' % (oid,)
)
nextSubId = oct2int(head[index])
index += 1
oid = oid + ((subId << 7) + nextSubId,)
elif subId == 128:
# ASN.1 spec forbids leading zeros (0x80) in OID
# encoding, tolerating it opens a vulnerability. See
# http://www.cosic.esat.kuleuven.be/publications/article-1432.pdf
# page 7
raise error.PyAsn1Error('Invalid octet 0x80 in OID encoding')
# Decode two leading arcs
if 0 <= oid[0] <= 39:
oid = (0,) + oid
elif 40 <= oid[0] <= 79:
oid = (1, oid[0]-40) + oid[1:]
elif oid[0] >= 80:
oid = (2, oid[0]-80) + oid[1:]
else:
raise error.PyAsn1Error('Malformed first OID octet: %s' % head[0])
return self._createComponent(asn1Spec, tagSet, oid), tail
class RealDecoder(AbstractSimpleDecoder):
protoComponent = univ.Real()
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
if not head:
return self._createComponent(asn1Spec, tagSet, 0.0), tail
fo = oct2int(head[0]); head = head[1:]
if fo & 0x80: # binary encoding
if not head:
raise error.PyAsn1Error("Incomplete floating-point value")
n = (fo & 0x03) + 1
if n == 4:
n = oct2int(head[0])
head = head[1:]
eo, head = head[:n], head[n:]
if not eo or not head:
raise error.PyAsn1Error('Real exponent screwed')
e = oct2int(eo[0]) & 0x80 and -1 or 0
while eo: # exponent
e <<= 8
e |= oct2int(eo[0])
eo = eo[1:]
b = fo >> 4 & 0x03 # base bits
if b > 2:
raise error.PyAsn1Error('Illegal Real base')
if b == 1: # encbase = 8
e *= 3
elif b == 2: # encbase = 16
e *= 4
p = 0
while head: # value
p <<= 8
p |= oct2int(head[0])
head = head[1:]
if fo & 0x40: # sign bit
p = -p
sf = fo >> 2 & 0x03 # scale bits
p *= 2**sf
value = (p, 2, e)
elif fo & 0x40: # infinite value
value = fo & 0x01 and '-inf' or 'inf'
elif fo & 0xc0 == 0: # character encoding
if not head:
raise error.PyAsn1Error("Incomplete floating-point value")
try:
if fo & 0x3 == 0x1: # NR1
value = (int(head), 10, 0)
elif fo & 0x3 == 0x2: # NR2
value = float(head)
elif fo & 0x3 == 0x3: # NR3
value = float(head)
else:
raise error.SubstrateUnderrunError(
'Unknown NR (tag %s)' % fo
)
except ValueError:
raise error.SubstrateUnderrunError(
'Bad character Real syntax'
)
else:
raise error.SubstrateUnderrunError(
'Unknown encoding (tag %s)' % fo
)
return self._createComponent(asn1Spec, tagSet, value), tail
class SequenceDecoder(AbstractConstructedDecoder):
protoComponent = univ.Sequence()
def _getComponentTagMap(self, r, idx):
try:
return r.getComponentTagMapNearPosition(idx)
except error.PyAsn1Error:
return
def _getComponentPositionByType(self, r, t, idx):
return r.getComponentPositionNearType(t, idx)
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
r = self._createComponent(asn1Spec, tagSet)
idx = 0
if substrateFun:
return substrateFun(r, substrate, length)
while head:
asn1Spec = self._getComponentTagMap(r, idx)
component, head = decodeFun(head, asn1Spec)
idx = self._getComponentPositionByType(
r, component.getEffectiveTagSet(), idx
)
r.setComponentByPosition(idx, component, asn1Spec is None)
idx = idx + 1
r.setDefaultComponents()
r.verifySizeSpec()
return r, tail
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
r = self._createComponent(asn1Spec, tagSet)
if substrateFun:
return substrateFun(r, substrate, length)
idx = 0
while substrate:
asn1Spec = self._getComponentTagMap(r, idx)
component, substrate = decodeFun(substrate, asn1Spec, allowEoo=True)
if eoo.endOfOctets.isSameTypeWith(component) and \
component == eoo.endOfOctets:
break
idx = self._getComponentPositionByType(
r, component.getEffectiveTagSet(), idx
)
r.setComponentByPosition(idx, component, asn1Spec is None)
idx = idx + 1
else:
raise error.SubstrateUnderrunError(
'No EOO seen before substrate ends'
)
r.setDefaultComponents()
r.verifySizeSpec()
return r, substrate
class SequenceOfDecoder(AbstractConstructedDecoder):
protoComponent = univ.SequenceOf()
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
r = self._createComponent(asn1Spec, tagSet)
if substrateFun:
return substrateFun(r, substrate, length)
asn1Spec = r.getComponentType()
idx = 0
while head:
component, head = decodeFun(head, asn1Spec)
r.setComponentByPosition(idx, component, asn1Spec is None)
idx = idx + 1
r.verifySizeSpec()
return r, tail
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
r = self._createComponent(asn1Spec, tagSet)
if substrateFun:
return substrateFun(r, substrate, length)
asn1Spec = r.getComponentType()
idx = 0
while substrate:
component, substrate = decodeFun(substrate, asn1Spec, allowEoo=True)
if eoo.endOfOctets.isSameTypeWith(component) and \
component == eoo.endOfOctets:
break
r.setComponentByPosition(idx, component, asn1Spec is None)
idx = idx + 1
else:
raise error.SubstrateUnderrunError(
'No EOO seen before substrate ends'
)
r.verifySizeSpec()
return r, substrate
class SetDecoder(SequenceDecoder):
protoComponent = univ.Set()
def _getComponentTagMap(self, r, idx):
return r.getComponentTagMap()
def _getComponentPositionByType(self, r, t, idx):
nextIdx = r.getComponentPositionByType(t)
if nextIdx is None:
return idx
else:
return nextIdx
class SetOfDecoder(SequenceOfDecoder):
protoComponent = univ.SetOf()
class ChoiceDecoder(AbstractConstructedDecoder):
protoComponent = univ.Choice()
tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed)
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
r = self._createComponent(asn1Spec, tagSet)
if substrateFun:
return substrateFun(r, substrate, length)
if r.getTagSet() == tagSet: # explicitly tagged Choice
component, head = decodeFun(
head, r.getComponentTagMap()
)
else:
component, head = decodeFun(
head, r.getComponentTagMap(), tagSet, length, state
)
if isinstance(component, univ.Choice):
effectiveTagSet = component.getEffectiveTagSet()
else:
effectiveTagSet = component.getTagSet()
r.setComponentByType(effectiveTagSet, component, 0, asn1Spec is None)
return r, tail
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
r = self._createComponent(asn1Spec, tagSet)
if substrateFun:
return substrateFun(r, substrate, length)
if r.getTagSet() == tagSet: # explicitly tagged Choice
component, substrate = decodeFun(substrate, r.getComponentTagMap())
# eat up EOO marker
eooMarker, substrate = decodeFun(substrate, allowEoo=True)
if not eoo.endOfOctets.isSameTypeWith(eooMarker) or \
eooMarker != eoo.endOfOctets:
raise error.PyAsn1Error('No EOO seen before substrate ends')
else:
component, substrate= decodeFun(
substrate, r.getComponentTagMap(), tagSet, length, state
)
if isinstance(component, univ.Choice):
effectiveTagSet = component.getEffectiveTagSet()
else:
effectiveTagSet = component.getTagSet()
r.setComponentByType(effectiveTagSet, component, 0, asn1Spec is None)
return r, substrate
class AnyDecoder(AbstractSimpleDecoder):
protoComponent = univ.Any()
tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed)
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
if asn1Spec is None or \
asn1Spec is not None and tagSet != asn1Spec.getTagSet():
# untagged Any container, recover inner header substrate
length = length + len(fullSubstrate) - len(substrate)
substrate = fullSubstrate
if substrateFun:
return substrateFun(self._createComponent(asn1Spec, tagSet),
substrate, length)
head, tail = substrate[:length], substrate[length:]
return self._createComponent(asn1Spec, tagSet, value=head), tail
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
length, state, decodeFun, substrateFun):
if asn1Spec is not None and tagSet == asn1Spec.getTagSet():
# tagged Any type -- consume header substrate
header = ''
else:
# untagged Any, recover header substrate
header = fullSubstrate[:-len(substrate)]
r = self._createComponent(asn1Spec, tagSet, header)
# Any components do not inherit initial tag
asn1Spec = self.protoComponent
if substrateFun:
return substrateFun(r, substrate, length)
while substrate:
component, substrate = decodeFun(substrate, asn1Spec, allowEoo=True)
if eoo.endOfOctets.isSameTypeWith(component) and \
component == eoo.endOfOctets:
break
r = r + component
else:
raise error.SubstrateUnderrunError(
'No EOO seen before substrate ends'
)
return r, substrate
# character string types
class UTF8StringDecoder(OctetStringDecoder):
protoComponent = char.UTF8String()
class NumericStringDecoder(OctetStringDecoder):
protoComponent = char.NumericString()
class PrintableStringDecoder(OctetStringDecoder):
protoComponent = char.PrintableString()
class TeletexStringDecoder(OctetStringDecoder):
protoComponent = char.TeletexString()
class VideotexStringDecoder(OctetStringDecoder):
protoComponent = char.VideotexString()
class IA5StringDecoder(OctetStringDecoder):
protoComponent = char.IA5String()
class GraphicStringDecoder(OctetStringDecoder):
protoComponent = char.GraphicString()
class VisibleStringDecoder(OctetStringDecoder):
protoComponent = char.VisibleString()
class GeneralStringDecoder(OctetStringDecoder):
protoComponent = char.GeneralString()
class UniversalStringDecoder(OctetStringDecoder):
protoComponent = char.UniversalString()
class BMPStringDecoder(OctetStringDecoder):
protoComponent = char.BMPString()
# "useful" types
class ObjectDescriptorDecoder(OctetStringDecoder):
protoComponent = useful.ObjectDescriptor()
class GeneralizedTimeDecoder(OctetStringDecoder):
protoComponent = useful.GeneralizedTime()
class UTCTimeDecoder(OctetStringDecoder):
protoComponent = useful.UTCTime()
tagMap = {
univ.Integer.tagSet: IntegerDecoder(),
univ.Boolean.tagSet: BooleanDecoder(),
univ.BitString.tagSet: BitStringDecoder(),
univ.OctetString.tagSet: OctetStringDecoder(),
univ.Null.tagSet: NullDecoder(),
univ.ObjectIdentifier.tagSet: ObjectIdentifierDecoder(),
univ.Enumerated.tagSet: IntegerDecoder(),
univ.Real.tagSet: RealDecoder(),
univ.Sequence.tagSet: SequenceDecoder(), # conflicts with SequenceOf
univ.Set.tagSet: SetDecoder(), # conflicts with SetOf
univ.Choice.tagSet: ChoiceDecoder(), # conflicts with Any
# character string types
char.UTF8String.tagSet: UTF8StringDecoder(),
char.NumericString.tagSet: NumericStringDecoder(),
char.PrintableString.tagSet: PrintableStringDecoder(),
char.TeletexString.tagSet: TeletexStringDecoder(),
char.VideotexString.tagSet: VideotexStringDecoder(),
char.IA5String.tagSet: IA5StringDecoder(),
char.GraphicString.tagSet: GraphicStringDecoder(),
char.VisibleString.tagSet: VisibleStringDecoder(),
char.GeneralString.tagSet: GeneralStringDecoder(),
char.UniversalString.tagSet: UniversalStringDecoder(),
char.BMPString.tagSet: BMPStringDecoder(),
# useful types
useful.ObjectDescriptor.tagSet: ObjectDescriptorDecoder(),
useful.GeneralizedTime.tagSet: GeneralizedTimeDecoder(),
useful.UTCTime.tagSet: UTCTimeDecoder()
}
# Type-to-codec map for ambiguous ASN.1 types
typeMap = {
univ.Set.typeId: SetDecoder(),
univ.SetOf.typeId: SetOfDecoder(),
univ.Sequence.typeId: SequenceDecoder(),
univ.SequenceOf.typeId: SequenceOfDecoder(),
univ.Choice.typeId: ChoiceDecoder(),
univ.Any.typeId: AnyDecoder()
}
( stDecodeTag, stDecodeLength, stGetValueDecoder, stGetValueDecoderByAsn1Spec,
stGetValueDecoderByTag, stTryAsExplicitTag, stDecodeValue,
stDumpRawValue, stErrorCondition, stStop ) = [x for x in range(10)]
class Decoder:
defaultErrorState = stErrorCondition
# defaultErrorState = stDumpRawValue
defaultRawDecoder = AnyDecoder()
supportIndefLength = True
def __init__(self, tagMap, typeMap={}):
self.__tagMap = tagMap
self.__typeMap = typeMap
# Tag & TagSet objects caches
self.__tagCache = {}
self.__tagSetCache = {}
def __call__(self, substrate, asn1Spec=None, tagSet=None,
length=None, state=stDecodeTag, recursiveFlag=1,
substrateFun=None, allowEoo=False):
if debug.logger & debug.flagDecoder:
debug.logger('decoder called at scope %s with state %d, working with up to %d octets of substrate: %s' % (debug.scope, state, len(substrate), debug.hexdump(substrate)))
fullSubstrate = substrate
while state != stStop:
if state == stDecodeTag:
if not substrate:
raise error.SubstrateUnderrunError(
'Short octet stream on tag decoding'
)
if not isOctetsType(substrate) and \
not isinstance(substrate, univ.OctetString):
raise error.PyAsn1Error('Bad octet stream type')
# Decode tag
firstOctet = substrate[0]
substrate = substrate[1:]
if firstOctet in self.__tagCache:
lastTag = self.__tagCache[firstOctet]
else:
t = oct2int(firstOctet)
# Look for end-of-octets sentinel
if t == 0:
if substrate and oct2int(substrate[0]) == 0:
if allowEoo and self.supportIndefLength:
debug.logger and debug.logger & debug.flagDecoder and debug.logger('end-of-octets sentinel found')
value, substrate = eoo.endOfOctets, substrate[1:]
state = stStop
continue
else:
raise error.PyAsn1Error('Unexpected end-of-contents sentinel')
else:
raise error.PyAsn1Error('Zero tag encountered')
tagClass = t&0xC0
tagFormat = t&0x20
tagId = t&0x1F
if tagId == 0x1F:
tagId = 0
while 1:
if not substrate:
raise error.SubstrateUnderrunError(
'Short octet stream on long tag decoding'
)
t = oct2int(substrate[0])
tagId = tagId << 7 | (t&0x7F)
substrate = substrate[1:]
if not t&0x80:
break
lastTag = tag.Tag(
tagClass=tagClass, tagFormat=tagFormat, tagId=tagId
)
if tagId < 31:
# cache short tags
self.__tagCache[firstOctet] = lastTag
if tagSet is None:
if firstOctet in self.__tagSetCache:
tagSet = self.__tagSetCache[firstOctet]
else:
# base tag not recovered
tagSet = tag.TagSet((), lastTag)
if firstOctet in self.__tagCache:
self.__tagSetCache[firstOctet] = tagSet
else:
tagSet = lastTag + tagSet
state = stDecodeLength
debug.logger and debug.logger & debug.flagDecoder and debug.logger('tag decoded into %s, decoding length' % tagSet)
if state == stDecodeLength:
# Decode length
if not substrate:
raise error.SubstrateUnderrunError(
'Short octet stream on length decoding'
)
firstOctet = oct2int(substrate[0])
if firstOctet == 128:
size = 1
length = -1
elif firstOctet < 128:
length, size = firstOctet, 1
else:
size = firstOctet & 0x7F
# encoded in size bytes
length = 0
lengthString = substrate[1:size+1]
# missing check on maximum size, which shouldn't be a
# problem, we can handle more than is possible
if len(lengthString) != size:
raise error.SubstrateUnderrunError(
'%s<%s at %s' %
(size, len(lengthString), tagSet)
)
for char in lengthString:
length = (length << 8) | oct2int(char)
size = size + 1
substrate = substrate[size:]
if length != -1 and len(substrate) < length:
raise error.SubstrateUnderrunError(
'%d-octet short' % (length - len(substrate))
)
if length == -1 and not self.supportIndefLength:
error.PyAsn1Error('Indefinite length encoding not supported by this codec')
state = stGetValueDecoder
debug.logger and debug.logger & debug.flagDecoder and debug.logger('value length decoded into %d, payload substrate is: %s' % (length, debug.hexdump(length == -1 and substrate or substrate[:length])))
if state == stGetValueDecoder:
if asn1Spec is None:
state = stGetValueDecoderByTag
else:
state = stGetValueDecoderByAsn1Spec
#
# There're two ways of creating subtypes in ASN.1 what influences
# decoder operation. These methods are:
# 1) Either base types used in or no IMPLICIT tagging has been
# applied on subtyping.
# 2) Subtype syntax drops base type information (by means of
# IMPLICIT tagging.
# The first case allows for complete tag recovery from substrate
# while the second one requires original ASN.1 type spec for
# decoding.
#
# In either case a set of tags (tagSet) is coming from substrate
# in an incremental, tag-by-tag fashion (this is the case of
# EXPLICIT tag which is most basic). Outermost tag comes first
# from the wire.
#
if state == stGetValueDecoderByTag:
if tagSet in self.__tagMap:
concreteDecoder = self.__tagMap[tagSet]
else:
concreteDecoder = None
if concreteDecoder:
state = stDecodeValue
else:
_k = tagSet[:1]
if _k in self.__tagMap:
concreteDecoder = self.__tagMap[_k]
else:
concreteDecoder = None
if concreteDecoder:
state = stDecodeValue
else:
state = stTryAsExplicitTag
if debug.logger and debug.logger & debug.flagDecoder:
debug.logger('codec %s chosen by a built-in type, decoding %s' % (concreteDecoder and concreteDecoder.__class__.__name__ or "<none>", state == stDecodeValue and 'value' or 'as explicit tag'))
debug.scope.push(concreteDecoder is None and '?' or concreteDecoder.protoComponent.__class__.__name__)
if state == stGetValueDecoderByAsn1Spec:
if isinstance(asn1Spec, (dict, tagmap.TagMap)):
if tagSet in asn1Spec:
__chosenSpec = asn1Spec[tagSet]
else:
__chosenSpec = None
if debug.logger and debug.logger & debug.flagDecoder:
debug.logger('candidate ASN.1 spec is a map of:')
for t, v in asn1Spec.getPosMap().items():
debug.logger(' %s -> %s' % (t, v.__class__.__name__))
if asn1Spec.getNegMap():
debug.logger('but neither of: ')
for t, v in asn1Spec.getNegMap().items():
debug.logger(' %s -> %s' % (t, v.__class__.__name__))
debug.logger('new candidate ASN.1 spec is %s, chosen by %s' % (__chosenSpec is None and '<none>' or __chosenSpec.prettyPrintType(), tagSet))
else:
__chosenSpec = asn1Spec
debug.logger and debug.logger & debug.flagDecoder and debug.logger('candidate ASN.1 spec is %s' % asn1Spec.__class__.__name__)
if __chosenSpec is not None and (
tagSet == __chosenSpec.getTagSet() or \
tagSet in __chosenSpec.getTagMap()
):
# use base type for codec lookup to recover untagged types
baseTagSet = __chosenSpec.baseTagSet
if __chosenSpec.typeId is not None and \
__chosenSpec.typeId in self.__typeMap:
# ambiguous type
concreteDecoder = self.__typeMap[__chosenSpec.typeId]
debug.logger and debug.logger & debug.flagDecoder and debug.logger('value decoder chosen for an ambiguous type by type ID %s' % (__chosenSpec.typeId,))
elif baseTagSet in self.__tagMap:
# base type or tagged subtype
concreteDecoder = self.__tagMap[baseTagSet]
debug.logger and debug.logger & debug.flagDecoder and debug.logger('value decoder chosen by base %s' % (baseTagSet,))
else:
concreteDecoder = None
if concreteDecoder:
asn1Spec = __chosenSpec
state = stDecodeValue
else:
state = stTryAsExplicitTag
else:
concreteDecoder = None
state = stTryAsExplicitTag
if debug.logger and debug.logger & debug.flagDecoder:
debug.logger('codec %s chosen by ASN.1 spec, decoding %s' % (state == stDecodeValue and concreteDecoder.__class__.__name__ or "<none>", state == stDecodeValue and 'value' or 'as explicit tag'))
debug.scope.push(__chosenSpec is None and '?' or __chosenSpec.__class__.__name__)
if state == stTryAsExplicitTag:
if tagSet and \
tagSet[0][1] == tag.tagFormatConstructed and \
tagSet[0][0] != tag.tagClassUniversal:
# Assume explicit tagging
concreteDecoder = explicitTagDecoder
state = stDecodeValue
else:
concreteDecoder = None
state = self.defaultErrorState
debug.logger and debug.logger & debug.flagDecoder and debug.logger('codec %s chosen, decoding %s' % (concreteDecoder and concreteDecoder.__class__.__name__ or "<none>", state == stDecodeValue and 'value' or 'as failure'))
if state == stDumpRawValue:
concreteDecoder = self.defaultRawDecoder
debug.logger and debug.logger & debug.flagDecoder and debug.logger('codec %s chosen, decoding value' % concreteDecoder.__class__.__name__)
state = stDecodeValue
if state == stDecodeValue:
if recursiveFlag == 0 and not substrateFun: # legacy
substrateFun = lambda a,b,c: (a,b[:c])
if length == -1: # indef length
value, substrate = concreteDecoder.indefLenValueDecoder(
fullSubstrate, substrate, asn1Spec, tagSet, length,
stGetValueDecoder, self, substrateFun
)
else:
value, substrate = concreteDecoder.valueDecoder(
fullSubstrate, substrate, asn1Spec, tagSet, length,
stGetValueDecoder, self, substrateFun
)
state = stStop
debug.logger and debug.logger & debug.flagDecoder and debug.logger('codec %s yields type %s, value:\n%s\n...remaining substrate is: %s' % (concreteDecoder.__class__.__name__, value.__class__.__name__, value.prettyPrint(), substrate and debug.hexdump(substrate) or '<none>'))
if state == stErrorCondition:
raise error.PyAsn1Error(
'%s not in asn1Spec: %s' % (tagSet, asn1Spec)
)
if debug.logger and debug.logger & debug.flagDecoder:
debug.scope.pop()
debug.logger('decoder left scope %s, call completed' % debug.scope)
return value, substrate
decode = Decoder(tagMap, typeMap)
# XXX
# non-recursive decoding; return position rather than substrate

View file

@ -0,0 +1,433 @@
# BER encoder
from pyasn1.type import base, tag, univ, char, useful
from pyasn1.codec.ber import eoo
from pyasn1.compat.octets import int2oct, oct2int, ints2octs, null, str2octs
from pyasn1 import debug, error
class Error(Exception): pass
class AbstractItemEncoder:
supportIndefLenMode = 1
def encodeTag(self, t, isConstructed):
tagClass, tagFormat, tagId = t.asTuple() # this is a hotspot
v = tagClass | tagFormat
if isConstructed:
v = v|tag.tagFormatConstructed
if tagId < 31:
return int2oct(v|tagId)
else:
s = int2oct(tagId&0x7f)
tagId = tagId >> 7
while tagId:
s = int2oct(0x80|(tagId&0x7f)) + s
tagId = tagId >> 7
return int2oct(v|0x1F) + s
def encodeLength(self, length, defMode):
if not defMode and self.supportIndefLenMode:
return int2oct(0x80)
if length < 0x80:
return int2oct(length)
else:
substrate = null
while length:
substrate = int2oct(length&0xff) + substrate
length = length >> 8
substrateLen = len(substrate)
if substrateLen > 126:
raise Error('Length octets overflow (%d)' % substrateLen)
return int2oct(0x80 | substrateLen) + substrate
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
raise Error('Not implemented')
def _encodeEndOfOctets(self, encodeFun, defMode):
if defMode or not self.supportIndefLenMode:
return null
else:
return encodeFun(eoo.endOfOctets, defMode)
def encode(self, encodeFun, value, defMode, maxChunkSize):
substrate, isConstructed = self.encodeValue(
encodeFun, value, defMode, maxChunkSize
)
tagSet = value.getTagSet()
if tagSet:
if not isConstructed: # primitive form implies definite mode
defMode = 1
return self.encodeTag(
tagSet[-1], isConstructed
) + self.encodeLength(
len(substrate), defMode
) + substrate + self._encodeEndOfOctets(encodeFun, defMode)
else:
return substrate # untagged value
class EndOfOctetsEncoder(AbstractItemEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
return null, 0
class ExplicitlyTaggedItemEncoder(AbstractItemEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
if isinstance(value, base.AbstractConstructedAsn1Item):
value = value.clone(tagSet=value.getTagSet()[:-1],
cloneValueFlag=1)
else:
value = value.clone(tagSet=value.getTagSet()[:-1])
return encodeFun(value, defMode, maxChunkSize), 1
explicitlyTaggedItemEncoder = ExplicitlyTaggedItemEncoder()
class BooleanEncoder(AbstractItemEncoder):
supportIndefLenMode = 0
_true = ints2octs((1,))
_false = ints2octs((0,))
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
return value and self._true or self._false, 0
class IntegerEncoder(AbstractItemEncoder):
supportIndefLenMode = 0
supportCompactZero = False
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
if value == 0: # shortcut for zero value
if self.supportCompactZero:
# this seems to be a correct way for encoding zeros
return null, 0
else:
# this seems to be a widespread way for encoding zeros
return ints2octs((0,)), 0
octets = []
value = int(value) # to save on ops on asn1 type
while 1:
octets.insert(0, value & 0xff)
if value == 0 or value == -1:
break
value = value >> 8
if value == 0 and octets[0] & 0x80:
octets.insert(0, 0)
while len(octets) > 1 and \
(octets[0] == 0 and octets[1] & 0x80 == 0 or \
octets[0] == 0xff and octets[1] & 0x80 != 0):
del octets[0]
return ints2octs(octets), 0
class BitStringEncoder(AbstractItemEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
if not maxChunkSize or len(value) <= maxChunkSize*8:
out_len = (len(value) + 7) // 8
out_list = out_len * [0]
j = 7
i = -1
for val in value:
j += 1
if j == 8:
i += 1
j = 0
out_list[i] = out_list[i] | val << (7-j)
return int2oct(7-j) + ints2octs(out_list), 0
else:
pos = 0; substrate = null
while 1:
# count in octets
v = value.clone(value[pos*8:pos*8+maxChunkSize*8])
if not v:
break
substrate = substrate + encodeFun(v, defMode, maxChunkSize)
pos = pos + maxChunkSize
return substrate, 1
class OctetStringEncoder(AbstractItemEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
if not maxChunkSize or len(value) <= maxChunkSize:
return value.asOctets(), 0
else:
pos = 0; substrate = null
while 1:
v = value.clone(value[pos:pos+maxChunkSize])
if not v:
break
substrate = substrate + encodeFun(v, defMode, maxChunkSize)
pos = pos + maxChunkSize
return substrate, 1
class NullEncoder(AbstractItemEncoder):
supportIndefLenMode = 0
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
return null, 0
class ObjectIdentifierEncoder(AbstractItemEncoder):
supportIndefLenMode = 0
precomputedValues = {
(1, 3, 6, 1, 2): (43, 6, 1, 2),
(1, 3, 6, 1, 4): (43, 6, 1, 4)
}
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
oid = value.asTuple()
if oid[:5] in self.precomputedValues:
octets = self.precomputedValues[oid[:5]]
oid = oid[5:]
else:
if len(oid) < 2:
raise error.PyAsn1Error('Short OID %s' % (value,))
octets = ()
# Build the first twos
if oid[0] == 0 and 0 <= oid[1] <= 39:
oid = (oid[1],) + oid[2:]
elif oid[0] == 1 and 0 <= oid[1] <= 39:
oid = (oid[1] + 40,) + oid[2:]
elif oid[0] == 2:
oid = (oid[1] + 80,) + oid[2:]
else:
raise error.PyAsn1Error(
'Impossible initial arcs %s at %s' % (oid[:2], value)
)
# Cycle through subIds
for subId in oid:
if subId > -1 and subId < 128:
# Optimize for the common case
octets = octets + (subId & 0x7f,)
elif subId < 0:
raise error.PyAsn1Error(
'Negative OID arc %s at %s' % (subId, value)
)
else:
# Pack large Sub-Object IDs
res = (subId & 0x7f,)
subId = subId >> 7
while subId > 0:
res = (0x80 | (subId & 0x7f),) + res
subId = subId >> 7
# Add packed Sub-Object ID to resulted Object ID
octets += res
return ints2octs(octets), 0
class RealEncoder(AbstractItemEncoder):
supportIndefLenMode = 0
binEncBase = 2 # set to None to choose encoding base automatically
def _dropFloatingPoint(self, m, encbase, e):
ms, es = 1, 1
if m < 0:
ms = -1 # mantissa sign
if e < 0:
es = -1 # exponenta sign
m *= ms
if encbase == 8:
m = m*2**(abs(e) % 3 * es)
e = abs(e) // 3 * es
elif encbase == 16:
m = m*2**(abs(e) % 4 * es)
e = abs(e) // 4 * es
while 1:
if int(m) != m:
m *= encbase
e -= 1
continue
break
return ms, int(m), encbase, e
def _chooseEncBase(self, value):
m, b, e = value
base = [2, 8, 16]
if value.binEncBase in base:
return self._dropFloatingPoint(m, value.binEncBase, e)
elif self.binEncBase in base:
return self._dropFloatingPoint(m, self.binEncBase, e)
# auto choosing base 2/8/16
mantissa = [m, m, m]
exponenta = [e, e, e]
encbase = 2
e = float('inf')
for i in range(3):
sign, mantissa[i], base[i], exponenta[i] = \
self._dropFloatingPoint(mantissa[i], base[i], exponenta[i])
if abs(exponenta[i]) < abs(e) or \
(abs(exponenta[i]) == abs(e) and mantissa[i] < m):
e = exponenta[i]
m = int(mantissa[i])
encbase = base[i]
return sign, m, encbase, e
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
if value.isPlusInfinity():
return int2oct(0x40), 0
if value.isMinusInfinity():
return int2oct(0x41), 0
m, b, e = value
if not m:
return null, 0
if b == 10:
return str2octs('\x03%dE%s%d' % (m, e == 0 and '+' or '', e)), 0
elif b == 2:
fo = 0x80 # binary encoding
ms, m, encbase, e = self._chooseEncBase(value)
if ms < 0: # mantissa sign
fo = fo | 0x40 # sign bit
# exponenta & mantissa normalization
if encbase == 2:
while m & 0x1 == 0:
m >>= 1
e += 1
elif encbase == 8:
while m & 0x7 == 0:
m >>= 3
e += 1
fo |= 0x10
else: # encbase = 16
while m & 0xf == 0:
m >>= 4
e += 1
fo |= 0x20
sf = 0 # scale factor
while m & 0x1 == 0:
m >>= 1
sf += 1
if sf > 3:
raise error.PyAsn1Error('Scale factor overflow') # bug if raised
fo |= sf << 2
eo = null
if e == 0 or e == -1:
eo = int2oct(e&0xff)
else:
while e not in (0, -1):
eo = int2oct(e&0xff) + eo
e >>= 8
if e == 0 and eo and oct2int(eo[0]) & 0x80:
eo = int2oct(0) + eo
if e == -1 and eo and not (oct2int(eo[0]) & 0x80):
eo = int2oct(0xff) + eo
n = len(eo)
if n > 0xff:
raise error.PyAsn1Error('Real exponent overflow')
if n == 1:
pass
elif n == 2:
fo |= 1
elif n == 3:
fo |= 2
else:
fo |= 3
eo = int2oct(n&0xff) + eo
po = null
while m:
po = int2oct(m&0xff) + po
m >>= 8
substrate = int2oct(fo) + eo + po
return substrate, 0
else:
raise error.PyAsn1Error('Prohibited Real base %s' % b)
class SequenceEncoder(AbstractItemEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
value.setDefaultComponents()
value.verifySizeSpec()
substrate = null; idx = len(value)
while idx > 0:
idx = idx - 1
if value[idx] is None: # Optional component
continue
component = value.getDefaultComponentByPosition(idx)
if component is not None and component == value[idx]:
continue
substrate = encodeFun(
value[idx], defMode, maxChunkSize
) + substrate
return substrate, 1
class SequenceOfEncoder(AbstractItemEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
value.verifySizeSpec()
substrate = null; idx = len(value)
while idx > 0:
idx = idx - 1
substrate = encodeFun(
value[idx], defMode, maxChunkSize
) + substrate
return substrate, 1
class ChoiceEncoder(AbstractItemEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
return encodeFun(value.getComponent(), defMode, maxChunkSize), 1
class AnyEncoder(OctetStringEncoder):
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
return value.asOctets(), defMode == 0
tagMap = {
eoo.endOfOctets.tagSet: EndOfOctetsEncoder(),
univ.Boolean.tagSet: BooleanEncoder(),
univ.Integer.tagSet: IntegerEncoder(),
univ.BitString.tagSet: BitStringEncoder(),
univ.OctetString.tagSet: OctetStringEncoder(),
univ.Null.tagSet: NullEncoder(),
univ.ObjectIdentifier.tagSet: ObjectIdentifierEncoder(),
univ.Enumerated.tagSet: IntegerEncoder(),
univ.Real.tagSet: RealEncoder(),
# Sequence & Set have same tags as SequenceOf & SetOf
univ.SequenceOf.tagSet: SequenceOfEncoder(),
univ.SetOf.tagSet: SequenceOfEncoder(),
univ.Choice.tagSet: ChoiceEncoder(),
# character string types
char.UTF8String.tagSet: OctetStringEncoder(),
char.NumericString.tagSet: OctetStringEncoder(),
char.PrintableString.tagSet: OctetStringEncoder(),
char.TeletexString.tagSet: OctetStringEncoder(),
char.VideotexString.tagSet: OctetStringEncoder(),
char.IA5String.tagSet: OctetStringEncoder(),
char.GraphicString.tagSet: OctetStringEncoder(),
char.VisibleString.tagSet: OctetStringEncoder(),
char.GeneralString.tagSet: OctetStringEncoder(),
char.UniversalString.tagSet: OctetStringEncoder(),
char.BMPString.tagSet: OctetStringEncoder(),
# useful types
useful.ObjectDescriptor.tagSet: OctetStringEncoder(),
useful.GeneralizedTime.tagSet: OctetStringEncoder(),
useful.UTCTime.tagSet: OctetStringEncoder()
}
# Type-to-codec map for ambiguous ASN.1 types
typeMap = {
univ.Set.typeId: SequenceEncoder(),
univ.SetOf.typeId: SequenceOfEncoder(),
univ.Sequence.typeId: SequenceEncoder(),
univ.SequenceOf.typeId: SequenceOfEncoder(),
univ.Choice.typeId: ChoiceEncoder(),
univ.Any.typeId: AnyEncoder()
}
class Encoder:
supportIndefLength = True
def __init__(self, tagMap, typeMap={}):
self.__tagMap = tagMap
self.__typeMap = typeMap
def __call__(self, value, defMode=True, maxChunkSize=0):
if not defMode and not self.supportIndefLength:
raise error.PyAsn1Error('Indefinite length encoding not supported by this codec')
debug.logger & debug.flagEncoder and debug.logger('encoder called in %sdef mode, chunk size %s for type %s, value:\n%s' % (not defMode and 'in' or '', maxChunkSize, value.prettyPrintType(), value.prettyPrint()))
tagSet = value.getTagSet()
if len(tagSet) > 1:
concreteEncoder = explicitlyTaggedItemEncoder
else:
if value.typeId is not None and value.typeId in self.__typeMap:
concreteEncoder = self.__typeMap[value.typeId]
elif tagSet in self.__tagMap:
concreteEncoder = self.__tagMap[tagSet]
else:
tagSet = value.baseTagSet
if tagSet in self.__tagMap:
concreteEncoder = self.__tagMap[tagSet]
else:
raise Error('No encoder for %s' % (value,))
debug.logger & debug.flagEncoder and debug.logger('using value codec %s chosen by %s' % (concreteEncoder.__class__.__name__, tagSet))
substrate = concreteEncoder.encode(
self, value, defMode, maxChunkSize
)
debug.logger & debug.flagEncoder and debug.logger('built %s octets of substrate: %s\nencoder completed' % (len(substrate), debug.hexdump(substrate)))
return substrate
encode = Encoder(tagMap, typeMap)

View file

@ -0,0 +1,8 @@
from pyasn1.type import base, tag
class EndOfOctets(base.AbstractSimpleAsn1Item):
defaultValue = 0
tagSet = tag.initTagSet(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 0x00)
)
endOfOctets = EndOfOctets()

View file

@ -0,0 +1 @@
# This file is necessary to make this directory a package.

View file

@ -0,0 +1,35 @@
# CER decoder
from pyasn1.type import univ
from pyasn1.codec.ber import decoder
from pyasn1.compat.octets import oct2int
from pyasn1 import error
class BooleanDecoder(decoder.AbstractSimpleDecoder):
protoComponent = univ.Boolean(0)
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
state, decodeFun, substrateFun):
head, tail = substrate[:length], substrate[length:]
if not head or length != 1:
raise error.PyAsn1Error('Not single-octet Boolean payload')
byte = oct2int(head[0])
# CER/DER specifies encoding of TRUE as 0xFF and FALSE as 0x0, while
# BER allows any non-zero value as TRUE; cf. sections 8.2.2. and 11.1
# in http://www.itu.int/ITU-T/studygroups/com17/languages/X.690-0207.pdf
if byte == 0xff:
value = 1
elif byte == 0x00:
value = 0
else:
raise error.PyAsn1Error('Unexpected Boolean payload: %s' % byte)
return self._createComponent(asn1Spec, tagSet, value), tail
tagMap = decoder.tagMap.copy()
tagMap.update({
univ.Boolean.tagSet: BooleanDecoder()
})
typeMap = decoder.typeMap
class Decoder(decoder.Decoder): pass
decode = Decoder(tagMap, decoder.typeMap)

View file

@ -0,0 +1,130 @@
# CER encoder
from pyasn1.type import univ
from pyasn1.type import useful
from pyasn1.codec.ber import encoder
from pyasn1.compat.octets import int2oct, str2octs, null
from pyasn1 import error
class BooleanEncoder(encoder.IntegerEncoder):
def encodeValue(self, encodeFun, client, defMode, maxChunkSize):
if client == 0:
substrate = int2oct(0)
else:
substrate = int2oct(255)
return substrate, 0
class BitStringEncoder(encoder.BitStringEncoder):
def encodeValue(self, encodeFun, client, defMode, maxChunkSize):
return encoder.BitStringEncoder.encodeValue(
self, encodeFun, client, defMode, 1000
)
class OctetStringEncoder(encoder.OctetStringEncoder):
def encodeValue(self, encodeFun, client, defMode, maxChunkSize):
return encoder.OctetStringEncoder.encodeValue(
self, encodeFun, client, defMode, 1000
)
class RealEncoder(encoder.RealEncoder):
def _chooseEncBase(self, value):
m, b, e = value
return self._dropFloatingPoint(m, b, e)
# specialized GeneralStringEncoder here
class GeneralizedTimeEncoder(OctetStringEncoder):
zchar = str2octs('Z')
pluschar = str2octs('+')
minuschar = str2octs('-')
zero = str2octs('0')
def encodeValue(self, encodeFun, client, defMode, maxChunkSize):
octets = client.asOctets()
# This breaks too many existing data items
# if '.' not in octets:
# raise error.PyAsn1Error('Format must include fraction of second: %r' % octets)
if len(octets) < 15:
raise error.PyAsn1Error('Bad UTC time length: %r' % octets)
if self.pluschar in octets or self.minuschar in octets:
raise error.PyAsn1Error('Must be UTC time: %r' % octets)
if octets[-1] != self.zchar[0]:
raise error.PyAsn1Error('Missing timezone specifier: %r' % octets)
return encoder.OctetStringEncoder.encodeValue(
self, encodeFun, client, defMode, 1000
)
class UTCTimeEncoder(encoder.OctetStringEncoder):
zchar = str2octs('Z')
pluschar = str2octs('+')
minuschar = str2octs('-')
def encodeValue(self, encodeFun, client, defMode, maxChunkSize):
octets = client.asOctets()
if self.pluschar in octets or self.minuschar in octets:
raise error.PyAsn1Error('Must be UTC time: %r' % octets)
if octets and octets[-1] != self.zchar[0]:
client = client.clone(octets + self.zchar)
if len(client) != 13:
raise error.PyAsn1Error('Bad UTC time length: %r' % client)
return encoder.OctetStringEncoder.encodeValue(
self, encodeFun, client, defMode, 1000
)
class SetOfEncoder(encoder.SequenceOfEncoder):
def encodeValue(self, encodeFun, client, defMode, maxChunkSize):
if isinstance(client, univ.SequenceAndSetBase):
client.setDefaultComponents()
client.verifySizeSpec()
substrate = null; idx = len(client)
# This is certainly a hack but how else do I distinguish SetOf
# from Set if they have the same tags&constraints?
if isinstance(client, univ.SequenceAndSetBase):
# Set
comps = []
while idx > 0:
idx = idx - 1
if client[idx] is None: # Optional component
continue
if client.getDefaultComponentByPosition(idx) == client[idx]:
continue
comps.append(client[idx])
comps.sort(key=lambda x: isinstance(x, univ.Choice) and \
x.getMinTagSet() or x.getTagSet())
for c in comps:
substrate += encodeFun(c, defMode, maxChunkSize)
else:
# SetOf
compSubs = []
while idx > 0:
idx = idx - 1
compSubs.append(
encodeFun(client[idx], defMode, maxChunkSize)
)
compSubs.sort() # perhaps padding's not needed
substrate = null
for compSub in compSubs:
substrate += compSub
return substrate, 1
tagMap = encoder.tagMap.copy()
tagMap.update({
univ.Boolean.tagSet: BooleanEncoder(),
univ.BitString.tagSet: BitStringEncoder(),
univ.OctetString.tagSet: OctetStringEncoder(),
univ.Real.tagSet: RealEncoder(),
useful.GeneralizedTime.tagSet: GeneralizedTimeEncoder(),
useful.UTCTime.tagSet: UTCTimeEncoder(),
univ.SetOf().tagSet: SetOfEncoder() # conflcts with Set
})
typeMap = encoder.typeMap.copy()
typeMap.update({
univ.Set.typeId: SetOfEncoder(),
univ.SetOf.typeId: SetOfEncoder()
})
class Encoder(encoder.Encoder):
def __call__(self, client, defMode=False, maxChunkSize=0):
return encoder.Encoder.__call__(self, client, defMode, maxChunkSize)
encode = Encoder(tagMap, typeMap)
# EncoderFactory queries class instance and builds a map of tags -> encoders

View file

@ -0,0 +1 @@
# This file is necessary to make this directory a package.

View file

@ -0,0 +1,9 @@
# DER decoder
from pyasn1.codec.cer import decoder
tagMap = decoder.tagMap
typeMap = decoder.typeMap
class Decoder(decoder.Decoder):
supportIndefLength = False
decode = Decoder(tagMap, typeMap)

View file

@ -0,0 +1,32 @@
# DER encoder
from pyasn1.type import univ
from pyasn1.codec.cer import encoder
from pyasn1 import error
class SetOfEncoder(encoder.SetOfEncoder):
def _cmpSetComponents(self, c1, c2):
tagSet1 = isinstance(c1, univ.Choice) and \
c1.getEffectiveTagSet() or c1.getTagSet()
tagSet2 = isinstance(c2, univ.Choice) and \
c2.getEffectiveTagSet() or c2.getTagSet()
return cmp(tagSet1, tagSet2)
tagMap = encoder.tagMap.copy()
tagMap.update({
# Overload CER encoders with BER ones (a bit hackerish XXX)
univ.BitString.tagSet: encoder.encoder.BitStringEncoder(),
univ.OctetString.tagSet: encoder.encoder.OctetStringEncoder(),
# Set & SetOf have same tags
univ.SetOf().tagSet: SetOfEncoder()
})
typeMap = encoder.typeMap
class Encoder(encoder.Encoder):
supportIndefLength = False
def __call__(self, client, defMode=True, maxChunkSize=0):
if not defMode:
raise error.PyAsn1Error('DER forbids indefinite length mode')
return encoder.Encoder.__call__(self, client, defMode, maxChunkSize)
encode = Encoder(tagMap, typeMap)

View file

@ -0,0 +1 @@
# This file is necessary to make this directory a package.

View file

@ -0,0 +1,10 @@
from sys import version_info
if version_info[0:2] < (2, 6):
def bin(x):
if x <= 1:
return '0b'+str(x)
else:
return bin(x>>1) + str(x&1)
else:
bin = bin

View file

@ -0,0 +1,22 @@
from sys import version_info
if version_info[0] <= 2:
int2oct = chr
ints2octs = lambda s: ''.join([ int2oct(x) for x in s ])
null = ''
oct2int = ord
octs2ints = lambda s: [ oct2int(x) for x in s ]
str2octs = lambda x: x
octs2str = lambda x: x
isOctetsType = lambda s: isinstance(s, str)
isStringType = lambda s: isinstance(s, (str, unicode))
else:
ints2octs = bytes
int2oct = lambda x: ints2octs((x,))
null = ints2octs()
oct2int = lambda x: x
octs2ints = lambda s: [ x for x in s ]
str2octs = lambda x: x.encode()
octs2str = lambda x: x.decode()
isOctetsType = lambda s: isinstance(s, bytes)
isStringType = lambda s: isinstance(s, str)

View file

@ -0,0 +1,110 @@
import time
import logging
from pyasn1.compat.octets import octs2ints
from pyasn1 import error
from pyasn1 import __version__
flagNone = 0x0000
flagEncoder = 0x0001
flagDecoder = 0x0002
flagAll = 0xffff
flagMap = {
'encoder': flagEncoder,
'decoder': flagDecoder,
'all': flagAll
}
class Printer:
def __init__(self, logger=None, handler=None, formatter=None):
if logger is None:
logger = logging.getLogger('pyasn1')
logger.setLevel(logging.DEBUG)
if handler is None:
handler = logging.StreamHandler()
if formatter is None:
formatter = logging.Formatter('%(asctime)s %(name)s: %(message)s')
handler.setFormatter(formatter)
handler.setLevel(logging.DEBUG)
logger.addHandler(handler)
self.__logger = logger
def __call__(self, msg): self.__logger.debug(msg)
def __str__(self): return '<python built-in logging>'
if hasattr(logging, 'NullHandler'):
NullHandler = logging.NullHandler
else:
# Python 2.6 and older
class NullHandler(logging.Handler):
def emit(self, record):
pass
class Debug:
defaultPrinter = None
def __init__(self, *flags, **options):
self._flags = flagNone
if options.get('printer') is not None:
self._printer = options.get('printer')
elif self.defaultPrinter is not None:
self._printer = self.defaultPrinter
if 'loggerName' in options:
# route our logs to parent logger
self._printer = Printer(
logger=logging.getLogger(options['loggerName']),
handler=NullHandler()
)
else:
self._printer = Printer()
self('running pyasn1 version %s' % __version__)
for f in flags:
inverse = f and f[0] in ('!', '~')
if inverse:
f = f[1:]
try:
if inverse:
self._flags &= ~flagMap[f]
else:
self._flags |= flagMap[f]
except KeyError:
raise error.PyAsn1Error('bad debug flag %s' % f)
self('debug category \'%s\' %s' % (f, inverse and 'disabled' or 'enabled'))
def __str__(self):
return 'logger %s, flags %x' % (self._printer, self._flags)
def __call__(self, msg):
self._printer(msg)
def __and__(self, flag):
return self._flags & flag
def __rand__(self, flag):
return flag & self._flags
logger = 0
def setLogger(l):
global logger
logger = l
def hexdump(octets):
return ' '.join(
[ '%s%.2X' % (n%16 == 0 and ('\n%.5d: ' % n) or '', x)
for n,x in zip(range(len(octets)), octs2ints(octets)) ]
)
class Scope:
def __init__(self):
self._list = []
def __str__(self): return '.'.join(self._list)
def push(self, token):
self._list.append(token)
def pop(self):
return self._list.pop()
scope = Scope()

View file

@ -0,0 +1,3 @@
class PyAsn1Error(Exception): pass
class ValueConstraintError(PyAsn1Error): pass
class SubstrateUnderrunError(PyAsn1Error): pass

View file

@ -0,0 +1 @@
# This file is necessary to make this directory a package.

View file

@ -0,0 +1,278 @@
# Base classes for ASN.1 types
import sys
from pyasn1.type import constraint, tagmap, tag
from pyasn1 import error
class Asn1Item: pass
class Asn1ItemBase(Asn1Item):
# Set of tags for this ASN.1 type
tagSet = tag.TagSet()
# A list of constraint.Constraint instances for checking values
subtypeSpec = constraint.ConstraintsIntersection()
# Used for ambiguous ASN.1 types identification
typeId = None
def __init__(self, tagSet=None, subtypeSpec=None):
if tagSet is None:
self._tagSet = self.tagSet
else:
self._tagSet = tagSet
if subtypeSpec is None:
self._subtypeSpec = self.subtypeSpec
else:
self._subtypeSpec = subtypeSpec
def _verifySubtypeSpec(self, value, idx=None):
try:
self._subtypeSpec(value, idx)
except error.PyAsn1Error:
c, i, t = sys.exc_info()
raise c('%s at %s' % (i, self.__class__.__name__))
def getSubtypeSpec(self): return self._subtypeSpec
def getTagSet(self): return self._tagSet
def getEffectiveTagSet(self): return self._tagSet # used by untagged types
def getTagMap(self): return tagmap.TagMap({self._tagSet: self})
def isSameTypeWith(self, other, matchTags=True, matchConstraints=True):
return self is other or \
(not matchTags or \
self._tagSet == other.getTagSet()) and \
(not matchConstraints or \
self._subtypeSpec==other.getSubtypeSpec())
def isSuperTypeOf(self, other, matchTags=True, matchConstraints=True):
"""Returns true if argument is a ASN1 subtype of ourselves"""
return (not matchTags or \
self._tagSet.isSuperTagSetOf(other.getTagSet())) and \
(not matchConstraints or \
(self._subtypeSpec.isSuperTypeOf(other.getSubtypeSpec())))
class NoValue:
def __getattr__(self, attr):
raise error.PyAsn1Error('No value for %s()' % attr)
def __getitem__(self, i):
raise error.PyAsn1Error('No value')
def __repr__(self): return '%s()' % self.__class__.__name__
noValue = NoValue()
# Base class for "simple" ASN.1 objects. These are immutable.
class AbstractSimpleAsn1Item(Asn1ItemBase):
defaultValue = noValue
def __init__(self, value=None, tagSet=None, subtypeSpec=None):
Asn1ItemBase.__init__(self, tagSet, subtypeSpec)
if value is None or value is noValue:
value = self.defaultValue
if value is None or value is noValue:
self.__hashedValue = value = noValue
else:
value = self.prettyIn(value)
self._verifySubtypeSpec(value)
self.__hashedValue = hash(value)
self._value = value
self._len = None
def __repr__(self):
r = []
if self._value is not self.defaultValue:
r.append(self.prettyOut(self._value))
if self._tagSet is not self.tagSet:
r.append('tagSet=%r' % (self._tagSet,))
if self._subtypeSpec is not self.subtypeSpec:
r.append('subtypeSpec=%r' % (self._subtypeSpec,))
return '%s(%s)' % (self.__class__.__name__, ', '.join(r))
def __str__(self): return str(self._value)
def __eq__(self, other):
return self is other and True or self._value == other
def __ne__(self, other): return self._value != other
def __lt__(self, other): return self._value < other
def __le__(self, other): return self._value <= other
def __gt__(self, other): return self._value > other
def __ge__(self, other): return self._value >= other
if sys.version_info[0] <= 2:
def __nonzero__(self): return bool(self._value)
else:
def __bool__(self): return bool(self._value)
def __hash__(self):
return self.__hashedValue is noValue and hash(noValue) or self.__hashedValue
def hasValue(self):
return not isinstance(self._value, NoValue)
def clone(self, value=None, tagSet=None, subtypeSpec=None):
if value is None and tagSet is None and subtypeSpec is None:
return self
if value is None:
value = self._value
if tagSet is None:
tagSet = self._tagSet
if subtypeSpec is None:
subtypeSpec = self._subtypeSpec
return self.__class__(value, tagSet, subtypeSpec)
def subtype(self, value=None, implicitTag=None, explicitTag=None,
subtypeSpec=None):
if value is None:
value = self._value
if implicitTag is not None:
tagSet = self._tagSet.tagImplicitly(implicitTag)
elif explicitTag is not None:
tagSet = self._tagSet.tagExplicitly(explicitTag)
else:
tagSet = self._tagSet
if subtypeSpec is None:
subtypeSpec = self._subtypeSpec
else:
subtypeSpec = subtypeSpec + self._subtypeSpec
return self.__class__(value, tagSet, subtypeSpec)
def prettyIn(self, value): return value
def prettyOut(self, value): return str(value)
def prettyPrint(self, scope=0):
if self.hasValue():
return self.prettyOut(self._value)
else:
return '<no value>'
# XXX Compatibility stub
def prettyPrinter(self, scope=0): return self.prettyPrint(scope)
def prettyPrintType(self, scope=0):
return '%s -> %s' % (self.getTagSet(), self.__class__.__name__)
#
# Constructed types:
# * There are five of them: Sequence, SequenceOf/SetOf, Set and Choice
# * ASN1 types and values are represened by Python class instances
# * Value initialization is made for defaulted components only
# * Primary method of component addressing is by-position. Data model for base
# type is Python sequence. Additional type-specific addressing methods
# may be implemented for particular types.
# * SequenceOf and SetOf types do not implement any additional methods
# * Sequence, Set and Choice types also implement by-identifier addressing
# * Sequence, Set and Choice types also implement by-asn1-type (tag) addressing
# * Sequence and Set types may include optional and defaulted
# components
# * Constructed types hold a reference to component types used for value
# verification and ordering.
# * Component type is a scalar type for SequenceOf/SetOf types and a list
# of types for Sequence/Set/Choice.
#
class AbstractConstructedAsn1Item(Asn1ItemBase):
componentType = None
sizeSpec = constraint.ConstraintsIntersection()
def __init__(self, componentType=None, tagSet=None,
subtypeSpec=None, sizeSpec=None):
Asn1ItemBase.__init__(self, tagSet, subtypeSpec)
if componentType is None:
self._componentType = self.componentType
else:
self._componentType = componentType
if sizeSpec is None:
self._sizeSpec = self.sizeSpec
else:
self._sizeSpec = sizeSpec
self._componentValues = []
self._componentValuesSet = 0
def __repr__(self):
r = []
if self._componentType is not self.componentType:
r.append('componentType=%r' % (self._componentType,))
if self._tagSet is not self.tagSet:
r.append('tagSet=%r' % (self._tagSet,))
if self._subtypeSpec is not self.subtypeSpec:
r.append('subtypeSpec=%r' % (self._subtypeSpec,))
r = '%s(%s)' % (self.__class__.__name__, ', '.join(r))
if self._componentValues:
r += '.setComponents(%s)' % ', '.join([repr(x) for x in self._componentValues])
return r
def __eq__(self, other):
return self is other and True or self._componentValues == other
def __ne__(self, other): return self._componentValues != other
def __lt__(self, other): return self._componentValues < other
def __le__(self, other): return self._componentValues <= other
def __gt__(self, other): return self._componentValues > other
def __ge__(self, other): return self._componentValues >= other
if sys.version_info[0] <= 2:
def __nonzero__(self): return bool(self._componentValues)
else:
def __bool__(self): return bool(self._componentValues)
def getComponentTagMap(self):
raise error.PyAsn1Error('Method not implemented')
def _cloneComponentValues(self, myClone, cloneValueFlag): pass
def clone(self, tagSet=None, subtypeSpec=None, sizeSpec=None,
cloneValueFlag=None):
if tagSet is None:
tagSet = self._tagSet
if subtypeSpec is None:
subtypeSpec = self._subtypeSpec
if sizeSpec is None:
sizeSpec = self._sizeSpec
r = self.__class__(self._componentType, tagSet, subtypeSpec, sizeSpec)
if cloneValueFlag:
self._cloneComponentValues(r, cloneValueFlag)
return r
def subtype(self, implicitTag=None, explicitTag=None, subtypeSpec=None,
sizeSpec=None, cloneValueFlag=None):
if implicitTag is not None:
tagSet = self._tagSet.tagImplicitly(implicitTag)
elif explicitTag is not None:
tagSet = self._tagSet.tagExplicitly(explicitTag)
else:
tagSet = self._tagSet
if subtypeSpec is None:
subtypeSpec = self._subtypeSpec
else:
subtypeSpec = subtypeSpec + self._subtypeSpec
if sizeSpec is None:
sizeSpec = self._sizeSpec
else:
sizeSpec = sizeSpec + self._sizeSpec
r = self.__class__(self._componentType, tagSet, subtypeSpec, sizeSpec)
if cloneValueFlag:
self._cloneComponentValues(r, cloneValueFlag)
return r
def _verifyComponent(self, idx, value): pass
def verifySizeSpec(self): self._sizeSpec(self)
def getComponentByPosition(self, idx):
raise error.PyAsn1Error('Method not implemented')
def setComponentByPosition(self, idx, value, verifyConstraints=True):
raise error.PyAsn1Error('Method not implemented')
def setComponents(self, *args, **kwargs):
for idx in range(len(args)):
self[idx] = args[idx]
for k in kwargs:
self[k] = kwargs[k]
return self
def getComponentType(self): return self._componentType
def setDefaultComponents(self): pass
def __getitem__(self, idx): return self.getComponentByPosition(idx)
def __setitem__(self, idx, value): self.setComponentByPosition(idx, value)
def __len__(self): return len(self._componentValues)
def clear(self):
self._componentValues = []
self._componentValuesSet = 0

View file

@ -0,0 +1,64 @@
# ASN.1 "character string" types
from pyasn1.type import univ, tag
class NumericString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 18)
)
class PrintableString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 19)
)
class TeletexString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 20)
)
class T61String(TeletexString): pass
class VideotexString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 21)
)
class IA5String(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 22)
)
class GraphicString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 25)
)
class VisibleString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 26)
)
class ISO646String(VisibleString): pass
class GeneralString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 27)
)
class UniversalString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 28)
)
encoding = "utf-32-be"
class BMPString(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 30)
)
encoding = "utf-16-be"
class UTF8String(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 12)
)
encoding = "utf-8"

View file

@ -0,0 +1,200 @@
#
# ASN.1 subtype constraints classes.
#
# Constraints are relatively rare, but every ASN1 object
# is doing checks all the time for whether they have any
# constraints and whether they are applicable to the object.
#
# What we're going to do is define objects/functions that
# can be called unconditionally if they are present, and that
# are simply not present if there are no constraints.
#
# Original concept and code by Mike C. Fletcher.
#
import sys
from pyasn1.type import error
class AbstractConstraint:
"""Abstract base-class for constraint objects
Constraints should be stored in a simple sequence in the
namespace of their client Asn1Item sub-classes.
"""
def __init__(self, *values):
self._valueMap = {}
self._setValues(values)
self.__hashedValues = None
def __call__(self, value, idx=None):
try:
self._testValue(value, idx)
except error.ValueConstraintError:
raise error.ValueConstraintError(
'%s failed at: \"%s\"' % (self, sys.exc_info()[1])
)
def __repr__(self):
return '%s(%s)' % (
self.__class__.__name__,
', '.join([repr(x) for x in self._values])
)
def __eq__(self, other):
return self is other and True or self._values == other
def __ne__(self, other): return self._values != other
def __lt__(self, other): return self._values < other
def __le__(self, other): return self._values <= other
def __gt__(self, other): return self._values > other
def __ge__(self, other): return self._values >= other
if sys.version_info[0] <= 2:
def __nonzero__(self): return bool(self._values)
else:
def __bool__(self): return bool(self._values)
def __hash__(self):
if self.__hashedValues is None:
self.__hashedValues = hash((self.__class__.__name__, self._values))
return self.__hashedValues
def _setValues(self, values): self._values = values
def _testValue(self, value, idx):
raise error.ValueConstraintError(value)
# Constraints derivation logic
def getValueMap(self): return self._valueMap
def isSuperTypeOf(self, otherConstraint):
return self in otherConstraint.getValueMap() or \
otherConstraint is self or otherConstraint == self
def isSubTypeOf(self, otherConstraint):
return otherConstraint in self._valueMap or \
otherConstraint is self or otherConstraint == self
class SingleValueConstraint(AbstractConstraint):
"""Value must be part of defined values constraint"""
def _testValue(self, value, idx):
# XXX index vals for performance?
if value not in self._values:
raise error.ValueConstraintError(value)
class ContainedSubtypeConstraint(AbstractConstraint):
"""Value must satisfy all of defined set of constraints"""
def _testValue(self, value, idx):
for c in self._values:
c(value, idx)
class ValueRangeConstraint(AbstractConstraint):
"""Value must be within start and stop values (inclusive)"""
def _testValue(self, value, idx):
if value < self.start or value > self.stop:
raise error.ValueConstraintError(value)
def _setValues(self, values):
if len(values) != 2:
raise error.PyAsn1Error(
'%s: bad constraint values' % (self.__class__.__name__,)
)
self.start, self.stop = values
if self.start > self.stop:
raise error.PyAsn1Error(
'%s: screwed constraint values (start > stop): %s > %s' % (
self.__class__.__name__,
self.start, self.stop
)
)
AbstractConstraint._setValues(self, values)
class ValueSizeConstraint(ValueRangeConstraint):
"""len(value) must be within start and stop values (inclusive)"""
def _testValue(self, value, idx):
l = len(value)
if l < self.start or l > self.stop:
raise error.ValueConstraintError(value)
class PermittedAlphabetConstraint(SingleValueConstraint):
def _setValues(self, values):
self._values = ()
for v in values:
self._values = self._values + tuple(v)
def _testValue(self, value, idx):
for v in value:
if v not in self._values:
raise error.ValueConstraintError(value)
# This is a bit kludgy, meaning two op modes within a single constraing
class InnerTypeConstraint(AbstractConstraint):
"""Value must satisfy type and presense constraints"""
def _testValue(self, value, idx):
if self.__singleTypeConstraint:
self.__singleTypeConstraint(value)
elif self.__multipleTypeConstraint:
if idx not in self.__multipleTypeConstraint:
raise error.ValueConstraintError(value)
constraint, status = self.__multipleTypeConstraint[idx]
if status == 'ABSENT': # XXX presense is not checked!
raise error.ValueConstraintError(value)
constraint(value)
def _setValues(self, values):
self.__multipleTypeConstraint = {}
self.__singleTypeConstraint = None
for v in values:
if isinstance(v, tuple):
self.__multipleTypeConstraint[v[0]] = v[1], v[2]
else:
self.__singleTypeConstraint = v
AbstractConstraint._setValues(self, values)
# Boolean ops on constraints
class ConstraintsExclusion(AbstractConstraint):
"""Value must not fit the single constraint"""
def _testValue(self, value, idx):
try:
self._values[0](value, idx)
except error.ValueConstraintError:
return
else:
raise error.ValueConstraintError(value)
def _setValues(self, values):
if len(values) != 1:
raise error.PyAsn1Error('Single constraint expected')
AbstractConstraint._setValues(self, values)
class AbstractConstraintSet(AbstractConstraint):
"""Value must not satisfy the single constraint"""
def __getitem__(self, idx): return self._values[idx]
def __add__(self, value): return self.__class__(self, value)
def __radd__(self, value): return self.__class__(self, value)
def __len__(self): return len(self._values)
# Constraints inclusion in sets
def _setValues(self, values):
self._values = values
for v in values:
self._valueMap[v] = 1
self._valueMap.update(v.getValueMap())
class ConstraintsIntersection(AbstractConstraintSet):
"""Value must satisfy all constraints"""
def _testValue(self, value, idx):
for v in self._values:
v(value, idx)
class ConstraintsUnion(AbstractConstraintSet):
"""Value must satisfy at least one constraint"""
def _testValue(self, value, idx):
for v in self._values:
try:
v(value, idx)
except error.ValueConstraintError:
pass
else:
return
raise error.ValueConstraintError(
'all of %s failed for \"%s\"' % (self._values, value)
)
# XXX
# add tests for type check

View file

@ -0,0 +1,3 @@
from pyasn1.error import PyAsn1Error
class ValueConstraintError(PyAsn1Error): pass

View file

@ -0,0 +1,149 @@
# NamedType specification for constructed types
import sys
from pyasn1.type import tagmap
from pyasn1 import error
class NamedType:
isOptional = 0
isDefaulted = 0
def __init__(self, name, t):
self.__name = name; self.__type = t
def __repr__(self): return '%s(%r, %r)' % (
self.__class__.__name__, self.__name, self.__type
)
def __eq__(self, other): return tuple(self) == tuple(other)
def __ne__(self, other): return tuple(self) != tuple(other)
def __lt__(self, other): return tuple(self) < tuple(other)
def __le__(self, other): return tuple(self) <= tuple(other)
def __gt__(self, other): return tuple(self) > tuple(other)
def __ge__(self, other): return tuple(self) >= tuple(other)
def __hash__(self): return hash(tuple(self))
def getType(self): return self.__type
def getName(self): return self.__name
def __getitem__(self, idx):
if idx == 0: return self.__name
if idx == 1: return self.__type
raise IndexError()
class OptionalNamedType(NamedType):
isOptional = 1
class DefaultedNamedType(NamedType):
isDefaulted = 1
class NamedTypes:
def __init__(self, *namedTypes):
self.__namedTypes = namedTypes
self.__namedTypesLen = len(self.__namedTypes)
self.__minTagSet = None
self.__tagToPosIdx = {}; self.__nameToPosIdx = {}
self.__tagMap = { False: None, True: None }
self.__ambigiousTypes = {}
def __repr__(self):
return '%s(%s)' % (
self.__class__.__name__,
', '.join([ repr(x) for x in self.__namedTypes ])
)
def __eq__(self, other): return tuple(self) == tuple(other)
def __ne__(self, other): return tuple(self) != tuple(other)
def __lt__(self, other): return tuple(self) < tuple(other)
def __le__(self, other): return tuple(self) <= tuple(other)
def __gt__(self, other): return tuple(self) > tuple(other)
def __ge__(self, other): return tuple(self) >= tuple(other)
def __hash__(self): return hash(tuple(self))
def __getitem__(self, idx): return self.__namedTypes[idx]
if sys.version_info[0] <= 2:
def __nonzero__(self): return bool(self.__namedTypesLen)
else:
def __bool__(self): return bool(self.__namedTypesLen)
def __len__(self): return self.__namedTypesLen
def clone(self): return self.__class__(*self.__namedTypes)
def getTypeByPosition(self, idx):
if idx < 0 or idx >= self.__namedTypesLen:
raise error.PyAsn1Error('Type position out of range')
else:
return self.__namedTypes[idx].getType()
def getPositionByType(self, tagSet):
if not self.__tagToPosIdx:
idx = self.__namedTypesLen
while idx > 0:
idx = idx - 1
tagMap = self.__namedTypes[idx].getType().getTagMap()
for t in tagMap.getPosMap():
if t in self.__tagToPosIdx:
raise error.PyAsn1Error('Duplicate type %s' % (t,))
self.__tagToPosIdx[t] = idx
try:
return self.__tagToPosIdx[tagSet]
except KeyError:
raise error.PyAsn1Error('Type %s not found' % (tagSet,))
def getNameByPosition(self, idx):
try:
return self.__namedTypes[idx].getName()
except IndexError:
raise error.PyAsn1Error('Type position out of range')
def getPositionByName(self, name):
if not self.__nameToPosIdx:
idx = self.__namedTypesLen
while idx > 0:
idx = idx - 1
n = self.__namedTypes[idx].getName()
if n in self.__nameToPosIdx:
raise error.PyAsn1Error('Duplicate name %s' % (n,))
self.__nameToPosIdx[n] = idx
try:
return self.__nameToPosIdx[name]
except KeyError:
raise error.PyAsn1Error('Name %s not found' % (name,))
def __buildAmbigiousTagMap(self):
ambigiousTypes = ()
idx = self.__namedTypesLen
while idx > 0:
idx = idx - 1
t = self.__namedTypes[idx]
if t.isOptional or t.isDefaulted:
ambigiousTypes = (t, ) + ambigiousTypes
else:
ambigiousTypes = (t, )
self.__ambigiousTypes[idx] = NamedTypes(*ambigiousTypes)
def getTagMapNearPosition(self, idx):
if not self.__ambigiousTypes: self.__buildAmbigiousTagMap()
try:
return self.__ambigiousTypes[idx].getTagMap()
except KeyError:
raise error.PyAsn1Error('Type position out of range')
def getPositionNearType(self, tagSet, idx):
if not self.__ambigiousTypes: self.__buildAmbigiousTagMap()
try:
return idx+self.__ambigiousTypes[idx].getPositionByType(tagSet)
except KeyError:
raise error.PyAsn1Error('Type position out of range')
def genMinTagSet(self):
if self.__minTagSet is None:
for t in self.__namedTypes:
__type = t.getType()
tagSet = getattr(__type,'getMinTagSet',__type.getTagSet)()
if self.__minTagSet is None or tagSet < self.__minTagSet:
self.__minTagSet = tagSet
return self.__minTagSet
def getTagMap(self, uniq=False):
if self.__tagMap[uniq] is None:
tagMap = tagmap.TagMap()
for nt in self.__namedTypes:
tagMap = tagMap.clone(
nt.getType(), nt.getType().getTagMap(), uniq
)
self.__tagMap[uniq] = tagMap
return self.__tagMap[uniq]

View file

@ -0,0 +1,58 @@
# ASN.1 named integers
from pyasn1 import error
__all__ = [ 'NamedValues' ]
class NamedValues:
def __init__(self, *namedValues):
self.nameToValIdx = {}; self.valToNameIdx = {}
self.namedValues = ()
automaticVal = 1
for namedValue in namedValues:
if isinstance(namedValue, tuple):
name, val = namedValue
else:
name = namedValue
val = automaticVal
if name in self.nameToValIdx:
raise error.PyAsn1Error('Duplicate name %s' % (name,))
self.nameToValIdx[name] = val
if val in self.valToNameIdx:
raise error.PyAsn1Error('Duplicate value %s=%s' % (name, val))
self.valToNameIdx[val] = name
self.namedValues = self.namedValues + ((name, val),)
automaticVal = automaticVal + 1
def __repr__(self):
return '%s(%s)' % (self.__class__.__name__, ', '.join([repr(x) for x in self.namedValues]))
def __str__(self): return str(self.namedValues)
def __eq__(self, other): return tuple(self) == tuple(other)
def __ne__(self, other): return tuple(self) != tuple(other)
def __lt__(self, other): return tuple(self) < tuple(other)
def __le__(self, other): return tuple(self) <= tuple(other)
def __gt__(self, other): return tuple(self) > tuple(other)
def __ge__(self, other): return tuple(self) >= tuple(other)
def __hash__(self): return hash(tuple(self))
def getName(self, value):
if value in self.valToNameIdx:
return self.valToNameIdx[value]
def getValue(self, name):
if name in self.nameToValIdx:
return self.nameToValIdx[name]
def __getitem__(self, i): return self.namedValues[i]
def __len__(self): return len(self.namedValues)
def __add__(self, namedValues):
return self.__class__(*self.namedValues + namedValues)
def __radd__(self, namedValues):
return self.__class__(*namedValues + tuple(self))
def clone(self, *namedValues):
return self.__class__(*tuple(self) + namedValues)
# XXX clone/subtype?

View file

@ -0,0 +1,128 @@
# ASN.1 types tags
from operator import getitem
from pyasn1 import error
tagClassUniversal = 0x00
tagClassApplication = 0x40
tagClassContext = 0x80
tagClassPrivate = 0xC0
tagFormatSimple = 0x00
tagFormatConstructed = 0x20
tagCategoryImplicit = 0x01
tagCategoryExplicit = 0x02
tagCategoryUntagged = 0x04
class Tag:
def __init__(self, tagClass, tagFormat, tagId):
if tagId < 0:
raise error.PyAsn1Error(
'Negative tag ID (%s) not allowed' % (tagId,)
)
self.__tag = (tagClass, tagFormat, tagId)
self.uniq = (tagClass, tagId)
self.__hashedUniqTag = hash(self.uniq)
def __str__(self):
return '[%s:%s:%s]' % self.__tag
def __repr__(self):
return '%s(tagClass=%s, tagFormat=%s, tagId=%s)' % (
(self.__class__.__name__,) + self.__tag
)
# These is really a hotspot -- expose public "uniq" attribute to save on
# function calls
def __eq__(self, other): return self.uniq == other.uniq
def __ne__(self, other): return self.uniq != other.uniq
def __lt__(self, other): return self.uniq < other.uniq
def __le__(self, other): return self.uniq <= other.uniq
def __gt__(self, other): return self.uniq > other.uniq
def __ge__(self, other): return self.uniq >= other.uniq
def __hash__(self): return self.__hashedUniqTag
def __getitem__(self, idx): return self.__tag[idx]
def __and__(self, otherTag):
(tagClass, tagFormat, tagId) = otherTag
return self.__class__(
self.__tag&tagClass, self.__tag&tagFormat, self.__tag&tagId
)
def __or__(self, otherTag):
(tagClass, tagFormat, tagId) = otherTag
return self.__class__(
self.__tag[0]|tagClass,
self.__tag[1]|tagFormat,
self.__tag[2]|tagId
)
def asTuple(self): return self.__tag # __getitem__() is slow
class TagSet:
def __init__(self, baseTag=(), *superTags):
self.__baseTag = baseTag
self.__superTags = superTags
self.__hashedSuperTags = hash(superTags)
_uniq = ()
for t in superTags:
_uniq = _uniq + t.uniq
self.uniq = _uniq
self.__lenOfSuperTags = len(superTags)
def __str__(self):
return self.__superTags and '+'.join([str(x) for x in self.__superTags]) or '[untagged]'
def __repr__(self):
return '%s(%s)' % (
self.__class__.__name__,
'(), ' + ', '.join([repr(x) for x in self.__superTags])
)
def __add__(self, superTag):
return self.__class__(
self.__baseTag, *self.__superTags + (superTag,)
)
def __radd__(self, superTag):
return self.__class__(
self.__baseTag, *(superTag,) + self.__superTags
)
def tagExplicitly(self, superTag):
tagClass, tagFormat, tagId = superTag
if tagClass == tagClassUniversal:
raise error.PyAsn1Error(
'Can\'t tag with UNIVERSAL-class tag'
)
if tagFormat != tagFormatConstructed:
superTag = Tag(tagClass, tagFormatConstructed, tagId)
return self + superTag
def tagImplicitly(self, superTag):
tagClass, tagFormat, tagId = superTag
if self.__superTags:
superTag = Tag(tagClass, self.__superTags[-1][1], tagId)
return self[:-1] + superTag
def getBaseTag(self): return self.__baseTag
def __getitem__(self, idx):
if isinstance(idx, slice):
return self.__class__(
self.__baseTag, *getitem(self.__superTags, idx)
)
return self.__superTags[idx]
def __eq__(self, other): return self.uniq == other.uniq
def __ne__(self, other): return self.uniq != other.uniq
def __lt__(self, other): return self.uniq < other.uniq
def __le__(self, other): return self.uniq <= other.uniq
def __gt__(self, other): return self.uniq > other.uniq
def __ge__(self, other): return self.uniq >= other.uniq
def __hash__(self): return self.__hashedSuperTags
def __len__(self): return self.__lenOfSuperTags
def isSuperTagSetOf(self, tagSet):
if len(tagSet) < self.__lenOfSuperTags:
return
idx = self.__lenOfSuperTags - 1
while idx >= 0:
if self.__superTags[idx] != tagSet[idx]:
return
idx = idx - 1
return 1
def initTagSet(tag): return TagSet(tag, tag)

View file

@ -0,0 +1,66 @@
from pyasn1 import error
class TagMap:
def __init__(self, posMap={}, negMap={}, defType=None):
self.__posMap = posMap.copy()
self.__negMap = negMap.copy()
self.__defType = defType
def __contains__(self, tagSet):
return tagSet in self.__posMap or \
self.__defType is not None and tagSet not in self.__negMap
def __getitem__(self, tagSet):
if tagSet in self.__posMap:
return self.__posMap[tagSet]
elif tagSet in self.__negMap:
raise error.PyAsn1Error('Key in negative map')
elif self.__defType is not None:
return self.__defType
else:
raise KeyError()
def __repr__(self):
s = self.__class__.__name__ + '('
if self.__posMap:
s = s + 'posMap=%r, ' % (self.__posMap,)
if self.__negMap:
s = s + 'negMap=%r, ' % (self.__negMap,)
if self.__defType is not None:
s = s + 'defType=%r' % (self.__defType,)
return s + ')'
def __str__(self):
s = self.__class__.__name__ + ':\n'
if self.__posMap:
s = s + 'posMap:\n%s, ' % ',\n '.join([ x.prettyPrintType() for x in self.__posMap.values()])
if self.__negMap:
s = s + 'negMap:\n%s, ' % ',\n '.join([ x.prettyPrintType() for x in self.__negMap.values()])
if self.__defType is not None:
s = s + 'defType:\n%s, ' % self.__defType.prettyPrintType()
return s
def clone(self, parentType, tagMap, uniq=False):
if self.__defType is not None and tagMap.getDef() is not None:
raise error.PyAsn1Error('Duplicate default value at %s' % (self,))
if tagMap.getDef() is not None:
defType = tagMap.getDef()
else:
defType = self.__defType
posMap = self.__posMap.copy()
for k in tagMap.getPosMap():
if uniq and k in posMap:
raise error.PyAsn1Error('Duplicate positive key %s' % (k,))
posMap[k] = parentType
negMap = self.__negMap.copy()
negMap.update(tagMap.getNegMap())
return self.__class__(
posMap, negMap, defType,
)
def getPosMap(self): return self.__posMap.copy()
def getNegMap(self): return self.__negMap.copy()
def getDef(self): return self.__defType

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,17 @@
# ASN.1 "useful" types
from pyasn1.type import char, tag
class ObjectDescriptor(char.GraphicString):
tagSet = char.GraphicString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 7)
)
class GeneralizedTime(char.VisibleString):
tagSet = char.VisibleString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 24)
)
class UTCTime(char.VisibleString):
tagSet = char.VisibleString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 23)
)