Update libs
This commit is contained in:
1 parent
2d05aafc04
commit
88b71c8116
157 files changed
+15686
-2017
No files matched your search
@@ -0,0 +1 @@
|
||||
majorVersionId = '1'
|
||||
Whitespace-only changes.
Whitespace-only changes.
@@ -0,0 +1,746 @@
|
||||
# BER decoder
|
||||
from pyasn1.type import tag, base, univ, char, useful, tagmap
|
||||
from pyasn1.codec.ber import eoo
|
||||
from pyasn1.compat.octets import oct2int, octs2ints
|
||||
from pyasn1 import error
|
||||
|
||||
class AbstractDecoder:
|
||||
protoComponent = None
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
raise error.PyAsn1Error('Decoder not implemented for %s' % tagSet)
|
||||
|
||||
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
raise error.PyAsn1Error('Indefinite length mode decoder not implemented for %s' % tagSet)
|
||||
|
||||
class AbstractSimpleDecoder(AbstractDecoder):
|
||||
def _createComponent(self, asn1Spec, tagSet, value=None):
|
||||
if asn1Spec is None:
|
||||
return self.protoComponent.clone(value, tagSet)
|
||||
elif value is None:
|
||||
return asn1Spec
|
||||
else:
|
||||
return asn1Spec.clone(value)
|
||||
|
||||
class AbstractConstructedDecoder(AbstractDecoder):
|
||||
def _createComponent(self, asn1Spec, tagSet, value=None):
|
||||
if asn1Spec is None:
|
||||
return self.protoComponent.clone(tagSet)
|
||||
else:
|
||||
return asn1Spec.clone()
|
||||
|
||||
class EndOfOctetsDecoder(AbstractSimpleDecoder):
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
return eoo.endOfOctets, substrate[:length]
|
||||
|
||||
class ExplicitTagDecoder(AbstractSimpleDecoder):
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
return decodeFun(substrate[:length], asn1Spec, tagSet, length)
|
||||
|
||||
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
value, substrate = decodeFun(substrate, asn1Spec, tagSet, length)
|
||||
terminator, substrate = decodeFun(substrate)
|
||||
if 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 _valueFilter(self, value):
|
||||
try:
|
||||
return int(value)
|
||||
except OverflowError:
|
||||
return value
|
||||
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
|
||||
state, decodeFun):
|
||||
substrate = substrate[:length]
|
||||
if not substrate:
|
||||
raise error.PyAsn1Error('Empty substrate')
|
||||
if substrate in self.precomputedValues:
|
||||
value = self.precomputedValues[substrate]
|
||||
else:
|
||||
firstOctet = oct2int(substrate[0])
|
||||
if firstOctet & 0x80:
|
||||
value = -1
|
||||
else:
|
||||
value = 0
|
||||
for octet in substrate:
|
||||
value = value << 8 | oct2int(octet)
|
||||
value = self._valueFilter(value)
|
||||
return self._createComponent(asn1Spec, tagSet, value), substrate
|
||||
|
||||
class BooleanDecoder(IntegerDecoder):
|
||||
protoComponent = univ.Boolean(0)
|
||||
def _valueFilter(self, value):
|
||||
if value:
|
||||
return 1
|
||||
else:
|
||||
return 0
|
||||
|
||||
class BitStringDecoder(AbstractSimpleDecoder):
|
||||
protoComponent = univ.BitString(())
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
|
||||
state, decodeFun):
|
||||
substrate = substrate[:length]
|
||||
if tagSet[0][1] == tag.tagFormatSimple: # XXX what tag to check?
|
||||
if not substrate:
|
||||
raise error.PyAsn1Error('Missing initial octet')
|
||||
trailingBits = oct2int(substrate[0])
|
||||
if trailingBits > 7:
|
||||
raise error.PyAsn1Error(
|
||||
'Trailing bits overflow %s' % trailingBits
|
||||
)
|
||||
substrate = substrate[1:]
|
||||
lsb = p = 0; l = len(substrate)-1; b = ()
|
||||
while p <= l:
|
||||
if p == l:
|
||||
lsb = trailingBits
|
||||
j = 7
|
||||
o = oct2int(substrate[p])
|
||||
while j >= lsb:
|
||||
b = b + ((o>>j)&0x01,)
|
||||
j = j - 1
|
||||
p = p + 1
|
||||
return self._createComponent(asn1Spec, tagSet, b), ''
|
||||
r = self._createComponent(asn1Spec, tagSet, ())
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
component, substrate = decodeFun(substrate)
|
||||
r = r + component
|
||||
return r, substrate
|
||||
|
||||
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
r = self._createComponent(asn1Spec, tagSet, '')
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
component, substrate = decodeFun(substrate)
|
||||
if 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('')
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
|
||||
state, decodeFun):
|
||||
substrate = substrate[:length]
|
||||
if tagSet[0][1] == tag.tagFormatSimple: # XXX what tag to check?
|
||||
return self._createComponent(asn1Spec, tagSet, substrate), ''
|
||||
r = self._createComponent(asn1Spec, tagSet, '')
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
component, substrate = decodeFun(substrate)
|
||||
r = r + component
|
||||
return r, substrate
|
||||
|
||||
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
r = self._createComponent(asn1Spec, tagSet, '')
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
component, substrate = decodeFun(substrate)
|
||||
if 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):
|
||||
substrate = substrate[:length]
|
||||
r = self._createComponent(asn1Spec, tagSet)
|
||||
if substrate:
|
||||
raise error.PyAsn1Error('Unexpected substrate for Null')
|
||||
return r, substrate
|
||||
|
||||
class ObjectIdentifierDecoder(AbstractSimpleDecoder):
|
||||
protoComponent = univ.ObjectIdentifier(())
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length,
|
||||
state, decodeFun):
|
||||
substrate = substrate[:length]
|
||||
if not substrate:
|
||||
raise error.PyAsn1Error('Empty substrate')
|
||||
oid = (); index = 0
|
||||
# Get the first subid
|
||||
subId = oct2int(substrate[index])
|
||||
oid = oid + divmod(subId, 40)
|
||||
|
||||
index = index + 1
|
||||
substrateLen = len(substrate)
|
||||
|
||||
while index < substrateLen:
|
||||
subId = oct2int(substrate[index])
|
||||
if subId < 128:
|
||||
oid = oid + (subId,)
|
||||
index = index + 1
|
||||
else:
|
||||
# Construct subid from a number of octets
|
||||
nextSubId = subId
|
||||
subId = 0
|
||||
while nextSubId >= 128 and index < substrateLen:
|
||||
subId = (subId << 7) + (nextSubId & 0x7F)
|
||||
index = index + 1
|
||||
nextSubId = oct2int(substrate[index])
|
||||
if index == substrateLen:
|
||||
raise error.SubstrateUnderrunError(
|
||||
'Short substrate for OID %s' % oid
|
||||
)
|
||||
subId = (subId << 7) + nextSubId
|
||||
oid = oid + (subId,)
|
||||
index = index + 1
|
||||
return self._createComponent(asn1Spec, tagSet, oid), substrate[index:]
|
||||
|
||||
class RealDecoder(AbstractSimpleDecoder):
|
||||
protoComponent = univ.Real()
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
substrate = substrate[:length]
|
||||
if not length:
|
||||
raise error.SubstrateUnderrunError('Short substrate for Real')
|
||||
fo = oct2int(substrate[0]); substrate = substrate[1:]
|
||||
if fo & 0x40: # infinite value
|
||||
value = fo & 0x01 and '-inf' or 'inf'
|
||||
elif fo & 0x80: # binary enoding
|
||||
if fo & 0x11 == 0:
|
||||
n = 1
|
||||
elif fo & 0x01:
|
||||
n = 2
|
||||
elif fo & 0x02:
|
||||
n = 3
|
||||
else:
|
||||
n = oct2int(substrate[0])
|
||||
eo, substrate = substrate[:n], substrate[n:]
|
||||
if not eo or not substrate:
|
||||
raise error.PyAsn1Error('Real exponent screwed')
|
||||
e = 0
|
||||
while eo: # exponent
|
||||
e <<= 8
|
||||
e |= oct2int(eo[0])
|
||||
eo = eo[1:]
|
||||
p = 0
|
||||
while substrate: # value
|
||||
p <<= 8
|
||||
p |= oct2int(substrate[0])
|
||||
substrate = substrate[1:]
|
||||
if fo & 0x40: # sign bit
|
||||
p = -p
|
||||
value = (p, 2, e)
|
||||
elif fo & 0xc0 == 0: # character encoding
|
||||
try:
|
||||
if fo & 0x3 == 0x1: # NR1
|
||||
value = (int(substrate), 10, 0)
|
||||
elif fo & 0x3 == 0x2: # NR2
|
||||
value = float(substrate)
|
||||
elif fo & 0x3 == 0x3: # NR3
|
||||
value = float(substrate)
|
||||
else:
|
||||
raise error.SubstrateUnderrunError(
|
||||
'Unknown NR (tag %s)' % fo
|
||||
)
|
||||
except ValueError:
|
||||
raise error.SubstrateUnderrunError(
|
||||
'Bad character Real syntax'
|
||||
)
|
||||
elif fo & 0xc0 == 0x40: # special real value
|
||||
pass
|
||||
else:
|
||||
raise error.SubstrateUnderrunError(
|
||||
'Unknown encoding (tag %s)' % fo
|
||||
)
|
||||
return self._createComponent(asn1Spec, tagSet, value), substrate
|
||||
|
||||
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):
|
||||
substrate = substrate[:length]
|
||||
r = self._createComponent(asn1Spec, tagSet)
|
||||
idx = 0
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
asn1Spec = self._getComponentTagMap(r, idx)
|
||||
component, substrate = decodeFun(
|
||||
substrate, asn1Spec
|
||||
)
|
||||
idx = self._getComponentPositionByType(
|
||||
r, component.getEffectiveTagSet(), idx
|
||||
)
|
||||
r.setComponentByPosition(idx, component, asn1Spec is None)
|
||||
idx = idx + 1
|
||||
r.setDefaultComponents()
|
||||
r.verifySizeSpec()
|
||||
return r, substrate
|
||||
|
||||
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
r = self._createComponent(asn1Spec, tagSet)
|
||||
idx = 0
|
||||
while substrate:
|
||||
asn1Spec = self._getComponentTagMap(r, idx)
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
component, substrate = decodeFun(substrate, asn1Spec)
|
||||
if 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):
|
||||
substrate = substrate[:length]
|
||||
r = self._createComponent(asn1Spec, tagSet)
|
||||
asn1Spec = r.getComponentType()
|
||||
idx = 0
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
component, substrate = decodeFun(
|
||||
substrate, asn1Spec
|
||||
)
|
||||
r.setComponentByPosition(idx, component, asn1Spec is None)
|
||||
idx = idx + 1
|
||||
r.verifySizeSpec()
|
||||
return r, substrate
|
||||
|
||||
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
r = self._createComponent(asn1Spec, tagSet)
|
||||
asn1Spec = r.getComponentType()
|
||||
idx = 0
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
component, substrate = decodeFun(substrate, asn1Spec)
|
||||
if 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()
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
substrate = substrate[:length]
|
||||
r = self._createComponent(asn1Spec, tagSet)
|
||||
if not decodeFun:
|
||||
return r, substrate
|
||||
if r.getTagSet() == tagSet: # explicitly tagged Choice
|
||||
component, substrate = decodeFun(
|
||||
substrate, r.getComponentTagMap()
|
||||
)
|
||||
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
|
||||
|
||||
indefLenValueDecoder = valueDecoder
|
||||
|
||||
class AnyDecoder(AbstractSimpleDecoder):
|
||||
protoComponent = univ.Any()
|
||||
def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
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
|
||||
substrate = substrate[:length]
|
||||
return self._createComponent(asn1Spec, tagSet, value=substrate), ''
|
||||
|
||||
def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet,
|
||||
length, state, decodeFun):
|
||||
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 not decodeFun:
|
||||
return r, substrate
|
||||
while substrate:
|
||||
component, substrate = decodeFun(substrate, asn1Spec)
|
||||
if 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 GeneralizedTimeDecoder(OctetStringDecoder):
|
||||
protoComponent = useful.GeneralizedTime()
|
||||
class UTCTimeDecoder(OctetStringDecoder):
|
||||
protoComponent = useful.UTCTime()
|
||||
|
||||
tagMap = {
|
||||
eoo.endOfOctets.tagSet: EndOfOctetsDecoder(),
|
||||
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.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()
|
||||
def __init__(self, tagMap, typeMap={}):
|
||||
self.__tagMap = tagMap
|
||||
self.__typeMap = typeMap
|
||||
self.__endOfOctetsTagSet = eoo.endOfOctets.getTagSet()
|
||||
# Tag & TagSet objects caches
|
||||
self.__tagCache = {}
|
||||
self.__tagSetCache = {}
|
||||
|
||||
def __call__(self, substrate, asn1Spec=None, tagSet=None,
|
||||
length=None, state=stDecodeTag, recursiveFlag=1):
|
||||
fullSubstrate = substrate
|
||||
while state != stStop:
|
||||
if state == stDecodeTag:
|
||||
# Decode tag
|
||||
if not substrate:
|
||||
raise error.SubstrateUnderrunError(
|
||||
'Short octet stream on tag decoding'
|
||||
)
|
||||
|
||||
firstOctet = substrate[0]
|
||||
substrate = substrate[1:]
|
||||
if firstOctet in self.__tagCache:
|
||||
lastTag = self.__tagCache[firstOctet]
|
||||
else:
|
||||
t = oct2int(firstOctet)
|
||||
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
|
||||
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
|
||||
state = stGetValueDecoder
|
||||
substrate = substrate[size:]
|
||||
if length != -1 and len(substrate) < length:
|
||||
raise error.SubstrateUnderrunError(
|
||||
'%d-octet short' % (length - len(substrate))
|
||||
)
|
||||
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 state == stGetValueDecoderByAsn1Spec:
|
||||
if isinstance(asn1Spec, (dict, tagmap.TagMap)):
|
||||
if tagSet in asn1Spec:
|
||||
__chosenSpec = asn1Spec[tagSet]
|
||||
else:
|
||||
__chosenSpec = None
|
||||
else:
|
||||
__chosenSpec = asn1Spec
|
||||
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]
|
||||
elif baseTagSet in self.__tagMap:
|
||||
# base type or tagged subtype
|
||||
concreteDecoder = self.__tagMap[baseTagSet]
|
||||
else:
|
||||
concreteDecoder = None
|
||||
if concreteDecoder:
|
||||
asn1Spec = __chosenSpec
|
||||
state = stDecodeValue
|
||||
else:
|
||||
state = stTryAsExplicitTag
|
||||
elif tagSet == self.__endOfOctetsTagSet:
|
||||
concreteDecoder = self.__tagMap[tagSet]
|
||||
state = stDecodeValue
|
||||
else:
|
||||
state = stTryAsExplicitTag
|
||||
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:
|
||||
state = self.defaultErrorState
|
||||
if state == stDumpRawValue:
|
||||
concreteDecoder = self.defaultRawDecoder
|
||||
state = stDecodeValue
|
||||
if state == stDecodeValue:
|
||||
if recursiveFlag:
|
||||
decodeFun = self
|
||||
else:
|
||||
decodeFun = None
|
||||
if length == -1: # indef length
|
||||
value, substrate = concreteDecoder.indefLenValueDecoder(
|
||||
fullSubstrate, substrate, asn1Spec, tagSet, length,
|
||||
stGetValueDecoder, decodeFun
|
||||
)
|
||||
else:
|
||||
value, _substrate = concreteDecoder.valueDecoder(
|
||||
fullSubstrate, substrate, asn1Spec, tagSet, length,
|
||||
stGetValueDecoder, decodeFun
|
||||
)
|
||||
if recursiveFlag:
|
||||
substrate = substrate[length:]
|
||||
else:
|
||||
substrate = _substrate
|
||||
state = stStop
|
||||
if state == stErrorCondition:
|
||||
raise error.PyAsn1Error(
|
||||
'%r not in asn1Spec: %r' % (tagSet, asn1Spec)
|
||||
)
|
||||
return value, substrate
|
||||
|
||||
decode = Decoder(tagMap, typeMap)
|
||||
|
||||
# XXX
|
||||
# non-recursive decoding; return position rather than substrate
|
||||
@@ -0,0 +1,334 @@
|
||||
# BER encoder
|
||||
from pyasn1.type import base, tag, univ, char, useful
|
||||
from pyasn1.codec.ber import eoo
|
||||
from pyasn1.compat.octets import int2oct, ints2octs, null, str2octs
|
||||
from pyasn1 import 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 IntegerEncoder(AbstractItemEncoder):
|
||||
supportIndefLenMode = 0
|
||||
def encodeValue(self, encodeFun, value, defMode, maxChunkSize):
|
||||
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:
|
||||
r = {}; l = len(value); p = 0; j = 7
|
||||
while p < l:
|
||||
i, j = divmod(p, 8)
|
||||
r[i] = r.get(i,0) | value[p]<<(7-j)
|
||||
p = p + 1
|
||||
keys = list(r); keys.sort()
|
||||
return int2oct(7-j) + ints2octs([r[k] for k in keys]), 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]]
|
||||
index = 5
|
||||
else:
|
||||
if len(oid) < 2:
|
||||
raise error.PyAsn1Error('Short OID %s' % value)
|
||||
|
||||
# Build the first twos
|
||||
index = 0
|
||||
subid = oid[index] * 40
|
||||
subid = subid + oid[index+1]
|
||||
if subid < 0 or subid > 0xff:
|
||||
raise error.PyAsn1Error(
|
||||
'Initial sub-ID overflow %s in OID %s' % (oid[index:], value)
|
||||
)
|
||||
octets = (subid,)
|
||||
index = index + 2
|
||||
|
||||
# Cycle through subids
|
||||
for subid in oid[index:]:
|
||||
if subid > -1 and subid < 128:
|
||||
# Optimize for the common case
|
||||
octets = octets + (subid & 0x7f,)
|
||||
elif subid < 0 or subid > 0xFFFFFFFF:
|
||||
raise error.PyAsn1Error(
|
||||
'SubId overflow %s in %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):
|
||||
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 enoding
|
||||
if m < 0:
|
||||
fo = fo | 0x40 # sign bit
|
||||
m = -m
|
||||
while int(m) != m: # drop floating point
|
||||
m *= 2
|
||||
e -= 1
|
||||
while m & 0x1 == 0: # mantissa normalization
|
||||
m >>= 1
|
||||
e += 1
|
||||
eo = null
|
||||
while e:
|
||||
eo = int2oct(e&0xff) + eo
|
||||
e >>= 8
|
||||
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+1) + 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: IntegerEncoder(),
|
||||
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.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:
|
||||
def __init__(self, tagMap, typeMap={}):
|
||||
self.__tagMap = tagMap
|
||||
self.__typeMap = typeMap
|
||||
|
||||
def __call__(self, value, defMode=1, maxChunkSize=0):
|
||||
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:
|
||||
baseTagSet = value.baseTagSet
|
||||
if baseTagSet in self.__tagMap:
|
||||
concreteEncoder = self.__tagMap[baseTagSet]
|
||||
else:
|
||||
raise Error('No encoder for %s' % value)
|
||||
return concreteEncoder.encode(
|
||||
self, value, defMode, maxChunkSize
|
||||
)
|
||||
|
||||
encode = Encoder(tagMap, typeMap)
|
||||
@@ -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()
|
||||
Whitespace-only changes.
@@ -0,0 +1,32 @@
|
||||
# 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):
|
||||
substrate = substrate[:length]
|
||||
if not substrate:
|
||||
raise error.PyAsn1Error('Empty substrate')
|
||||
byte = oct2int(substrate[0])
|
||||
if byte == 0xff:
|
||||
value = 1
|
||||
elif byte == 0x00:
|
||||
value = 0
|
||||
else:
|
||||
raise error.PyAsn1Error('Boolean CER violation: %s' % byte)
|
||||
return self._createComponent(asn1Spec, tagSet, value), substrate[1:]
|
||||
|
||||
tagMap = decoder.tagMap.copy()
|
||||
tagMap.update({
|
||||
univ.Boolean.tagSet: BooleanDecoder(),
|
||||
})
|
||||
|
||||
typeMap = decoder.typeMap
|
||||
|
||||
class Decoder(decoder.Decoder): pass
|
||||
|
||||
decode = Decoder(tagMap, decoder.typeMap)
|
||||
@@ -0,0 +1,87 @@
|
||||
# CER encoder
|
||||
from pyasn1.type import univ
|
||||
from pyasn1.codec.ber import encoder
|
||||
from pyasn1.compat.octets import int2oct, null
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
# specialized RealEncoder here
|
||||
# specialized GeneralStringEncoder here
|
||||
# specialized GeneralizedTimeEncoder here
|
||||
# specialized UTCTimeEncoder here
|
||||
|
||||
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.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=0, 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
|
||||
Whitespace-only changes.
@@ -0,0 +1,5 @@
|
||||
# DER decoder
|
||||
from pyasn1.type import univ
|
||||
from pyasn1.codec.cer import decoder
|
||||
|
||||
decode = decoder.Decoder(decoder.tagMap, decoder.typeMap)
|
||||
@@ -0,0 +1,28 @@
|
||||
# DER encoder
|
||||
from pyasn1.type import univ
|
||||
from pyasn1.codec.cer import encoder
|
||||
|
||||
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 encodrs 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):
|
||||
def __call__(self, client, defMode=1, maxChunkSize=0):
|
||||
return encoder.Encoder.__call__(self, client, defMode, maxChunkSize)
|
||||
|
||||
encode = Encoder(tagMap, typeMap)
|
||||
Whitespace-only changes.
@@ -0,0 +1,18 @@
|
||||
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
|
||||
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()
|
||||
@@ -0,0 +1,3 @@
|
||||
class PyAsn1Error(Exception): pass
|
||||
class ValueConstraintError(PyAsn1Error): pass
|
||||
class SubstrateUnderrunError(PyAsn1Error): pass
|
||||
Whitespace-only changes.
@@ -0,0 +1,244 @@
|
||||
# Base classes for ASN.1 types
|
||||
import sys
|
||||
from pyasn1.type import constraint, tagmap
|
||||
from pyasn1 import error
|
||||
|
||||
class Asn1Item: pass
|
||||
|
||||
class Asn1ItemBase(Asn1Item):
|
||||
# Set of tags for this ASN.1 type
|
||||
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):
|
||||
return self is other or \
|
||||
self._tagSet == other.getTagSet() and \
|
||||
self._subtypeSpec == other.getSubtypeSpec()
|
||||
def isSuperTypeOf(self, other):
|
||||
"""Returns true if argument is a ASN1 subtype of ourselves"""
|
||||
return self._tagSet.isSuperTagSetOf(other.getTagSet()) and \
|
||||
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')
|
||||
|
||||
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):
|
||||
if self._value is noValue:
|
||||
return self.__class__.__name__ + '()'
|
||||
else:
|
||||
return self.__class__.__name__ + '(%s)' % (self.prettyOut(self._value),)
|
||||
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
|
||||
|
||||
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): return self.prettyOut(self._value)
|
||||
# XXX Compatibility stub
|
||||
def prettyPrinter(self, scope=0): return self.prettyPrint(scope)
|
||||
|
||||
#
|
||||
# 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 = self.__class__.__name__ + '()'
|
||||
for idx in range(len(self._componentValues)):
|
||||
if self._componentValues[idx] is None:
|
||||
continue
|
||||
r = r + '.setComponentByPosition(%s, %r)' % (
|
||||
idx, self._componentValues[idx]
|
||||
)
|
||||
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 getComponentType(self): return self._componentType
|
||||
|
||||
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
|
||||
|
||||
def setDefaultComponents(self): pass
|
||||
@@ -0,0 +1,61 @@
|
||||
# ASN.1 "character string" types
|
||||
from pyasn1.type import univ, tag
|
||||
|
||||
class UTF8String(univ.OctetString):
|
||||
tagSet = univ.OctetString.tagSet.tagImplicitly(
|
||||
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 12)
|
||||
)
|
||||
encoding = "utf-8"
|
||||
|
||||
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 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 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"
|
||||
@@ -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
|
||||
@@ -0,0 +1,3 @@
|
||||
from pyasn1.error import PyAsn1Error
|
||||
|
||||
class ValueConstraintError(PyAsn1Error): pass
|
||||
@@ -0,0 +1,132 @@
|
||||
# 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(%s, %s)' % (
|
||||
self.__class__.__name__, self.__name, self.__type
|
||||
)
|
||||
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):
|
||||
r = '%s(' % self.__class__.__name__
|
||||
for n in self.__namedTypes:
|
||||
r = r + '%r, ' % (n,)
|
||||
return r + ')'
|
||||
|
||||
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 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]
|
||||
@@ -0,0 +1,46 @@
|
||||
# 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' % name)
|
||||
self.valToNameIdx[val] = name
|
||||
self.namedValues = self.namedValues + ((name, val),)
|
||||
automaticVal = automaticVal + 1
|
||||
def __str__(self): return str(self.namedValues)
|
||||
|
||||
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?
|
||||
@@ -0,0 +1,122 @@
|
||||
# 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 __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 __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)
|
||||
@@ -0,0 +1,52 @@
|
||||
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 = '%r/%r' % (self.__posMap, self.__negMap)
|
||||
if self.__defType is not None:
|
||||
s = s + '/%r' % (self.__defType,)
|
||||
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
@@ -0,0 +1,12 @@
|
||||
# ASN.1 "useful" types
|
||||
from pyasn1.type import char, tag
|
||||
|
||||
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)
|
||||
)
|
||||
Reference in new issue
Block a user