root / userdata / asn1.py @ 4264d385
History | View | Annotate | Download (6.6 kB)
1 |
# -*- coding: ascii -*-
|
---|---|
2 |
#
|
3 |
# Util/asn1.py : Minimal support for ASN.1 DER binary encoding.
|
4 |
#
|
5 |
# ===================================================================
|
6 |
# The contents of this file are dedicated to the public domain. To
|
7 |
# the extent that dedication to the public domain is not available,
|
8 |
# everyone is granted a worldwide, perpetual, royalty-free,
|
9 |
# non-exclusive license to exercise all rights associated with the
|
10 |
# contents of this file for any purpose whatsoever.
|
11 |
# No rights are reserved.
|
12 |
#
|
13 |
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
14 |
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
|
15 |
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
16 |
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
|
17 |
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
18 |
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
19 |
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
20 |
# SOFTWARE.
|
21 |
# ===================================================================
|
22 |
|
23 |
from Crypto.Util.number import long_to_bytes, bytes_to_long |
24 |
import sys |
25 |
|
26 |
if sys.version_info[0] == 2: |
27 |
def b(s): |
28 |
return s
|
29 |
def bchr(s): |
30 |
return chr(s) |
31 |
def bstr(s): |
32 |
return str(s) |
33 |
def bord(s): |
34 |
return ord(s) |
35 |
else:
|
36 |
def b(s): |
37 |
return s.encode("latin-1") # utf-8 would cause some side-effects we don't want |
38 |
def bchr(s): |
39 |
return bytes([s]) |
40 |
def bstr(s): |
41 |
if isinstance(s,str): |
42 |
return bytes(s,"latin-1") |
43 |
else:
|
44 |
return bytes(s) |
45 |
def bord(s): |
46 |
return s
|
47 |
|
48 |
__all__ = [ 'DerObject', 'DerInteger', 'DerSequence' ] |
49 |
|
50 |
class DerObject: |
51 |
typeTags = { 'SEQUENCE':b('\x30'), 'BIT STRING':b('\x03'), 'INTEGER':b('\x02') } |
52 |
|
53 |
def __init__(self, ASN1Type=None): |
54 |
self.typeTag = self.typeTags.get(ASN1Type, ASN1Type) |
55 |
self.payload = b('') |
56 |
|
57 |
def _lengthOctets(self, payloadLen): |
58 |
'''
|
59 |
Return an octet string that is suitable for the BER/DER
|
60 |
length element if the relevant payload is of the given
|
61 |
size (in bytes).
|
62 |
'''
|
63 |
if payloadLen>127: |
64 |
encoding = long_to_bytes(payloadLen) |
65 |
return bchr(len(encoding)+128) + encoding |
66 |
return bchr(payloadLen)
|
67 |
|
68 |
def encode(self): |
69 |
return self.typeTag + self._lengthOctets(len(self.payload)) + self.payload |
70 |
|
71 |
def _decodeLen(self, idx, str): |
72 |
'''
|
73 |
Given a string and an index to a DER LV,
|
74 |
this function returns a tuple with the length of V
|
75 |
and an index to the first byte of it.
|
76 |
'''
|
77 |
length = bord(str[idx])
|
78 |
if length<=127: |
79 |
return (length,idx+1) |
80 |
else:
|
81 |
payloadLength = bytes_to_long(str[idx+1:idx+1+(length & 0x7F)]) |
82 |
if payloadLength<=127: |
83 |
raise ValueError("Not a DER length tag.") |
84 |
return (payloadLength, idx+1+(length & 0x7F)) |
85 |
|
86 |
def decode(self, input, noLeftOvers=0): |
87 |
try:
|
88 |
self.typeTag = input[0] |
89 |
if (bord(self.typeTag) & 0x1F)==0x1F: |
90 |
raise ValueError("Unsupported DER tag") |
91 |
(length,idx) = self._decodeLen(1,input) |
92 |
if noLeftOvers and len(input) != (idx+length): |
93 |
raise ValueError("Not a DER structure") |
94 |
self.payload = input[idx:idx+length] |
95 |
except IndexError: |
96 |
raise ValueError("Not a valid DER SEQUENCE.") |
97 |
return idx+length
|
98 |
|
99 |
class DerInteger(DerObject): |
100 |
def __init__(self, value = 0): |
101 |
DerObject.__init__(self, 'INTEGER') |
102 |
self.value = value
|
103 |
|
104 |
def encode(self): |
105 |
self.payload = long_to_bytes(self.value) |
106 |
if bord(self.payload[0])>127: |
107 |
self.payload = b('\x00') + self.payload |
108 |
return DerObject.encode(self) |
109 |
|
110 |
def decode(self, input, noLeftOvers=0): |
111 |
tlvLength = DerObject.decode(self, input,noLeftOvers) |
112 |
if bord(self.payload[0])>127: |
113 |
raise ValueError ("Negative INTEGER.") |
114 |
self.value = bytes_to_long(self.payload) |
115 |
return tlvLength
|
116 |
|
117 |
class DerSequence(DerObject): |
118 |
def __init__(self): |
119 |
DerObject.__init__(self, 'SEQUENCE') |
120 |
self._seq = []
|
121 |
def __delitem__(self, n): |
122 |
del self._seq[n] |
123 |
def __getitem__(self, n): |
124 |
return self._seq[n] |
125 |
def __setitem__(self, key, value): |
126 |
self._seq[key] = value
|
127 |
if sys.version_info[0] == 2: |
128 |
def __setslice__(self,i,j,sequence): |
129 |
self._seq[i:j] = sequence
|
130 |
def __delslice__(self,i,j): |
131 |
del self._seq[i:j] |
132 |
def __getslice__(self, i, j): |
133 |
return self._seq[max(0, i):max(0, j)] |
134 |
def __len__(self): |
135 |
return len(self._seq) |
136 |
def append(self, item): |
137 |
return self._seq.append(item) |
138 |
|
139 |
def hasOnlyInts(self): |
140 |
if not self._seq: return 0 |
141 |
test = 0
|
142 |
for item in self._seq: |
143 |
try:
|
144 |
test += item |
145 |
except TypeError: |
146 |
return 0 |
147 |
return 1 |
148 |
|
149 |
def encode(self): |
150 |
'''
|
151 |
Return the DER encoding for the ASN.1 SEQUENCE containing
|
152 |
the non-negative integers and longs added to this object.
|
153 |
'''
|
154 |
self.payload = b('') |
155 |
for item in self._seq: |
156 |
try:
|
157 |
self.payload += item
|
158 |
except:
|
159 |
try:
|
160 |
self.payload += DerInteger(item).encode()
|
161 |
except:
|
162 |
raise ValueError("Trying to DER encode an unknown object") |
163 |
return DerObject.encode(self) |
164 |
|
165 |
def decode(self, input,noLeftOvers=0): |
166 |
'''
|
167 |
This function decodes the given string into a sequence of
|
168 |
ASN.1 objects. Yet, we only know about unsigned INTEGERs.
|
169 |
Any other type is stored as its rough TLV. In the latter
|
170 |
case, the correctectness of the TLV is not checked.
|
171 |
'''
|
172 |
self._seq = []
|
173 |
try:
|
174 |
tlvLength = DerObject.decode(self, input,noLeftOvers) |
175 |
if self.typeTag!=self.typeTags['SEQUENCE'][0]: |
176 |
raise ValueError("Not a DER SEQUENCE.") |
177 |
# Scan one TLV at once
|
178 |
idx = 0
|
179 |
while idx<len(self.payload): |
180 |
typeTag = self.payload[idx]
|
181 |
if typeTag==self.typeTags['INTEGER'][0]: |
182 |
newInteger = DerInteger() |
183 |
idx += newInteger.decode(self.payload[idx:])
|
184 |
self._seq.append(newInteger.value)
|
185 |
else:
|
186 |
itemLen,itemIdx = self._decodeLen(idx+1,self.payload) |
187 |
self._seq.append(self.payload[idx:itemIdx+itemLen]) |
188 |
idx = itemIdx + itemLen |
189 |
except IndexError: |
190 |
raise ValueError("Not a valid DER SEQUENCE.") |
191 |
return tlvLength
|