|
1 | 1 | import binascii |
2 | 2 | from datetime import datetime |
3 | 3 | from datetime import timedelta |
| 4 | +from functools import wraps |
4 | 5 |
|
5 | 6 | from ..errors import Error |
6 | 7 | from ..errors import EncodeError as _EncodeError |
|
9 | 10 | from .. import compat |
10 | 11 |
|
11 | 12 |
|
12 | | -class EncodeError(_EncodeError): |
13 | | - """General ASN.1 encode error. |
14 | | -
|
| 13 | +class BaseType(object): |
| 14 | + """ |
| 15 | + Base Type class containing common functionality between all codecs |
15 | 16 | """ |
16 | 17 |
|
17 | | - def __init__(self, message): |
18 | | - super(EncodeError, self).__init__() |
19 | | - self.message = message |
20 | | - self.location = [] |
| 18 | + def __init__(self, name, type_name): |
| 19 | + self.name = name |
| 20 | + self.type_name = type_name |
| 21 | + self.optional = False |
| 22 | + self.default = None |
21 | 23 |
|
22 | | - def __str__(self): |
23 | | - if self.location: |
24 | | - return "{}: {}".format(': '.join(self.location[::-1]), |
25 | | - self.message) |
26 | | - else: |
27 | | - return self.message |
| 24 | + def set_default(self, value): |
| 25 | + self.default = value |
28 | 26 |
|
| 27 | + def get_default(self): |
| 28 | + return self.default |
29 | 29 |
|
30 | | -class DecodeError(_DecodeError): |
31 | | - """General ASN.1 decode error with error location in the message. |
| 30 | + def has_default(self): |
| 31 | + return self.default is not None |
32 | 32 |
|
33 | | - """ |
| 33 | + def is_default(self, value): |
| 34 | + return value == self.default |
34 | 35 |
|
35 | | - def __init__(self, message): |
36 | | - super(DecodeError, self).__init__() |
| 36 | + def encode(self, *args, **kwargs): |
| 37 | + raise NotImplementedError('To be implemented by subclasses.') |
| 38 | + |
| 39 | + def decode(self, *args, **kwargs): |
| 40 | + raise NotImplementedError('To be implemented by subclasses.') |
| 41 | + |
| 42 | + |
| 43 | +class ErrorWithLocation(Exception): |
| 44 | + """ |
| 45 | + Mixin for Error classes which have location list |
| 46 | + """ |
| 47 | + def __init__(self, message, location=None): |
37 | 48 | self.message = message |
38 | | - self.location = [] |
| 49 | + self.location = [location] if location else [] |
| 50 | + |
| 51 | + def add_location(self, element_name): |
| 52 | + self.location.append(element_name) |
39 | 53 |
|
40 | 54 | def __str__(self): |
41 | 55 | if self.location: |
42 | | - return "{}: {}".format(': '.join(self.location[::-1]), |
| 56 | + return "{}: {}".format('.'.join(self.location[::-1]), |
43 | 57 | self.message) |
44 | 58 | else: |
45 | 59 | return self.message |
46 | 60 |
|
47 | 61 |
|
48 | | -class ConstraintsError(_ConstraintsError): |
49 | | - """General ASN.1 constraints error with error location in the message. |
| 62 | +class EncodeError(ErrorWithLocation, _EncodeError): |
| 63 | + """General ASN.1 encode error. |
50 | 64 |
|
51 | 65 | """ |
| 66 | + pass |
52 | 67 |
|
53 | | - def __init__(self, message): |
54 | | - super(ConstraintsError, self).__init__() |
55 | | - self.message = message |
56 | | - self.location = [] |
| 68 | + |
| 69 | +class DecodeError(ErrorWithLocation, _DecodeError): |
| 70 | + """ |
| 71 | + General ASN.1 decode error with error location in the message. |
| 72 | + """ |
| 73 | + |
| 74 | + def __init__(self, message, offset=None, location=None): |
| 75 | + """ |
| 76 | +
|
| 77 | + :param str message: Message for error |
| 78 | + :param int offset: Data offset at which error occurred. Can be bits or bytes depending on codec |
| 79 | + :param str location: Name of element in which error occured |
| 80 | + """ |
| 81 | + super(DecodeError, self).__init__(message, location=location) |
| 82 | + self.offset = offset |
57 | 83 |
|
58 | 84 | def __str__(self): |
59 | 85 | if self.location: |
60 | | - return "{}: {}".format(': '.join(self.location[::-1]), |
61 | | - self.message) |
| 86 | + _str = "{}: {}".format('.'.join(self.location[::-1]), self.message) |
62 | 87 | else: |
63 | | - return self.message |
| 88 | + _str = self.message |
| 89 | + if self.offset is not None: |
| 90 | + _str += self.get_offset_message() |
| 91 | + return _str |
| 92 | + |
| 93 | + def get_offset_message(self): |
| 94 | + """ |
| 95 | + Get offset details to add to error message |
| 96 | + :return: |
| 97 | + """ |
| 98 | + return ' (At offset: {})'.format(self.offset) |
| 99 | + |
| 100 | + |
| 101 | +class ConstraintsError(ErrorWithLocation, _ConstraintsError): |
| 102 | + """ |
| 103 | + General ASN.1 constraints error with error location in the message. |
| 104 | + """ |
| 105 | + pass |
64 | 106 |
|
65 | 107 |
|
66 | 108 | class DecodeTagError(DecodeError): |
67 | 109 | """ASN.1 tag decode error. |
68 | 110 |
|
69 | 111 | """ |
70 | 112 |
|
71 | | - def __init__(self, type_name, expected_tag, actual_tag, offset): |
72 | | - message = "Expected {} with tag '{}' at offset {}, but got '{}'.".format( |
| 113 | + def __init__(self, type_name, expected_tag, actual_tag, offset=None, location=None): |
| 114 | + message = "Expected {} with tag '{}', but got '{}'.".format( |
73 | 115 | type_name, |
74 | 116 | binascii.hexlify(expected_tag).decode('ascii'), |
75 | | - offset, |
76 | 117 | binascii.hexlify(actual_tag).decode('ascii')) |
77 | | - super(DecodeTagError, self).__init__(message) |
| 118 | + super(DecodeTagError, self).__init__(message, offset=offset, location=location) |
| 119 | + |
| 120 | + self.expected_tag = expected_tag |
| 121 | + self.actual_tag = actual_tag |
| 122 | + self.type_name = type_name |
78 | 123 |
|
79 | 124 |
|
80 | 125 | class DecodeContentsLengthError(DecodeError): |
81 | 126 | """ASN.1 contents length decode error. |
82 | 127 |
|
83 | 128 | """ |
84 | 129 |
|
85 | | - def __init__(self, length, offset, contents_max): |
86 | | - message = ('Expected at least {} contents byte(s) at offset {}, ' |
87 | | - 'but got {}.').format(length, |
88 | | - offset, |
89 | | - contents_max - offset) |
90 | | - super(DecodeContentsLengthError, self).__init__(message) |
| 130 | + def __init__(self, length, offset, contents_max, location=None): |
| 131 | + message = 'Expected at least {} contents byte(s), but got {}.'.format(length, contents_max - offset) |
| 132 | + super(DecodeContentsLengthError, self).__init__(message, offset=offset, location=location) |
91 | 133 |
|
92 | 134 | self.length = length |
93 | | - self.offset = offset |
94 | 135 | self.contents_max = contents_max |
95 | 136 |
|
96 | 137 |
|
97 | 138 | class OutOfDataError(DecodeError): |
98 | 139 |
|
99 | | - def __init__(self, offset): |
| 140 | + def __init__(self, offset_bits, location=None): |
100 | 141 | super(OutOfDataError, self).__init__( |
101 | | - 'out of data at bit offset {} ({}.{} bytes)'.format( |
102 | | - offset, |
103 | | - *divmod(offset, 8))) |
| 142 | + 'out of data', offset=offset_bits, location=location) |
| 143 | + |
| 144 | + def get_offset_message(self): |
| 145 | + """ |
| 146 | + Get offset details to add to error message |
| 147 | + :return: |
| 148 | + """ |
| 149 | + return ' (At bit offset: {})'.format(self.offset) |
| 150 | + |
| 151 | + |
| 152 | +def add_error_location(method): |
| 153 | + """ |
| 154 | + Method decorator which catches ErrorWithLocation subclasses and adds element name to location |
| 155 | + If decorator is applied to parent Type class method, this functionality can be disabled on a per-child |
| 156 | + Type basis by setting no_error_location=True |
| 157 | + :param method: |
| 158 | + :return: |
| 159 | + """ |
| 160 | + @wraps(method) |
| 161 | + def new_method(self, *args, **kwargs): |
| 162 | + try: |
| 163 | + return method(self, *args, **kwargs) |
| 164 | + except ErrorWithLocation as e: |
| 165 | + # Don't add name if it is blank (for SEQUENCE OF, SET OF etc) |
| 166 | + if self.name and not getattr(self, 'no_error_location', False): |
| 167 | + e.add_location(self.name) |
| 168 | + raise e |
| 169 | + return new_method |
104 | 170 |
|
105 | 171 |
|
106 | 172 | def _generalized_time_to_datetime(string): |
|
0 commit comments