-
Notifications
You must be signed in to change notification settings - Fork 15
/
__init__.py
348 lines (261 loc) · 9.36 KB
/
__init__.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
# The contents of this file are subject to the BitTorrent Open Source License
# Version 1.1 (the License). You may not copy or use this file, in either
# source code or executable form, except in compliance with the License. You
# may obtain a copy of the License at http://www.bittorrent.com/license/.
#
# Software distributed under the License is distributed on an AS IS basis,
# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
# for the specific language governing rights and limitations under the
# License.
# Written by Petru Paler
"""bencode.py - bencode encoder + decoder."""
from bencode.BTL import BTFailure
from bencode.exceptions import BencodeDecodeError
from collections import deque
import sys
try:
from typing import Dict, List, Tuple, Deque, Union, TextIO, BinaryIO, Any
except ImportError:
Dict = List = Tuple = Deque = Union = TextIO = BinaryIO = Any = None
try:
from collections import OrderedDict
except ImportError:
OrderedDict = None
try:
import pathlib
except ImportError:
pathlib = None
__all__ = (
'BTFailure',
'BencodeDecodeError',
'bencode',
'bdecode',
'bread',
'bwrite',
'encode',
'decode'
)
def decode_int(x, f):
# type: (bytes, int) -> Tuple[int, int]
f += 1
newf = x.index(b'e', f)
n = int(x[f:newf])
if x[f:f + 1] == b'-':
if x[f + 1:f + 2] == b'0':
raise ValueError
elif x[f:f + 1] == b'0' and newf != f + 1:
raise ValueError
return n, newf + 1
def decode_string(x, f, try_decode_utf8=True, force_decode_utf8=False):
# type: (bytes, int, bool, bool) -> Tuple[bytes, int]
"""Decode torrent bencoded 'string' in x starting at f.
An attempt is made to convert the string to a python string from utf-8.
However, both string and non-string binary data is intermixed in the
torrent bencoding standard. So we have to guess whether the byte
sequence is a string or just binary data. We make this guess by trying
to decode (from utf-8), and if that fails, assuming it is binary data.
There are some instances where the data SHOULD be a string though.
You can check enforce this by setting force_decode_utf8 to True. If the
decoding from utf-8 fails, an UnidcodeDecodeError is raised. Similarly,
if you know it should not be a string, you can skip the decoding
attempt by setting try_decode_utf8=False.
"""
colon = x.index(b':', f)
n = int(x[f:colon])
if x[f:f + 1] == b'0' and colon != f + 1:
raise ValueError
colon += 1
s = x[colon:colon + n]
if try_decode_utf8:
try:
return s.decode('utf-8'), colon + n
except UnicodeDecodeError:
if force_decode_utf8:
raise
return bytes(s), colon + n
def decode_list(x, f):
# type: (bytes, int) -> Tuple[List, int]
r, f = [], f + 1
while x[f:f + 1] != b'e':
v, f = decode_func[x[f:f + 1]](x, f)
r.append(v)
return r, f + 1
def decode_dict_py26(x, f):
# type: (bytes, int) -> Tuple[Dict[str, Any], int]
r, f = {}, f + 1
while x[f] != 'e':
k, f = decode_string(x, f)
r[k], f = decode_func[x[f]](x, f)
return r, f + 1
def decode_dict(x, f, force_sort=True):
# type: (bytes, int, bool) -> Tuple[OrderedDict[str, Any], int]
"""Decode bencoded data to an OrderedDict.
The BitTorrent standard states that:
Keys must be strings and appear in sorted order (sorted as raw
strings, not alphanumerics)
- http://www.bittorrent.org/beps/bep_0003.html
Therefore, this function will force the keys to be strings (decoded
from utf-8), and by default the keys are (re)sorted after reading.
Set force_sort to False to keep the order of the dictionary as
represented in x, as many other encoders and decoders do not force this
property.
"""
r, f = OrderedDict(), f + 1
while x[f:f + 1] != b'e':
k, f = decode_string(x, f, force_decode_utf8=True)
r[k], f = decode_func[x[f:f + 1]](x, f)
if force_sort:
r = OrderedDict(sorted(r.items()))
return r, f + 1
# noinspection PyDictCreation
decode_func = {}
decode_func[b'l'] = decode_list
decode_func[b'i'] = decode_int
decode_func[b'0'] = decode_string
decode_func[b'1'] = decode_string
decode_func[b'2'] = decode_string
decode_func[b'3'] = decode_string
decode_func[b'4'] = decode_string
decode_func[b'5'] = decode_string
decode_func[b'6'] = decode_string
decode_func[b'7'] = decode_string
decode_func[b'8'] = decode_string
decode_func[b'9'] = decode_string
if sys.version_info[0] == 2 and sys.version_info[1] == 6:
decode_func[b'd'] = decode_dict_py26
else:
decode_func[b'd'] = decode_dict
def bdecode(value):
# type: (bytes) -> Union[Tuple, List, OrderedDict, bool, int, str, bytes]
"""
Decode bencode formatted byte string ``value``.
:param value: Bencode formatted string
:type value: bytes
:return: Decoded value
:rtype: object
"""
try:
r, l = decode_func[value[0:1]](value, 0)
except (IndexError, KeyError, TypeError, ValueError):
raise BencodeDecodeError("not a valid bencoded string")
if l != len(value):
raise BencodeDecodeError("invalid bencoded value (data after valid prefix)")
return r
class Bencached(object):
__slots__ = ['bencoded']
def __init__(self, s):
self.bencoded = s
def encode_bencached(x, r):
# type: (Bencached, Deque[bytes]) -> None
r.append(x.bencoded)
def encode_int(x, r):
# type: (int, Deque[bytes]) -> None
r.extend((b'i', str(x).encode('utf-8'), b'e'))
def encode_bool(x, r):
# type: (bool, Deque[bytes]) -> None
if x:
encode_int(1, r)
else:
encode_int(0, r)
def encode_bytes(x, r):
# type: (bytes, Deque[bytes]) -> None
r.extend((str(len(x)).encode('utf-8'), b':', x))
def encode_string(x, r):
# type: (str, Deque[bytes]) -> None
try:
s = x.encode('utf-8')
except UnicodeDecodeError:
encode_bytes(x, r)
return
r.extend((str(len(s)).encode('utf-8'), b':', s))
def encode_list(x, r):
# type: (List, Deque[bytes]) -> None
r.append(b'l')
for i in x:
encode_func[type(i)](i, r)
r.append(b'e')
def encode_dict(x, r):
# type: (Dict, Deque[bytes]) -> None
r.append(b'd')
ilist = list(x.items())
ilist.sort()
for k, v in ilist:
k = k.encode('utf-8')
r.extend((str(len(k)).encode('utf-8'), b':', k))
encode_func[type(v)](v, r)
r.append(b'e')
# noinspection PyDictCreation
encode_func = {}
encode_func[Bencached] = encode_bencached
if sys.version_info[0] == 2:
from types import DictType, IntType, ListType, LongType, StringType, TupleType, UnicodeType
encode_func[DictType] = encode_dict
encode_func[IntType] = encode_int
encode_func[ListType] = encode_list
encode_func[LongType] = encode_int
encode_func[StringType] = encode_string
encode_func[TupleType] = encode_list
encode_func[UnicodeType] = encode_string
if OrderedDict is not None:
encode_func[OrderedDict] = encode_dict
try:
from types import BooleanType
encode_func[BooleanType] = encode_bool
except ImportError:
pass
else:
encode_func[OrderedDict] = encode_dict
encode_func[bool] = encode_bool
encode_func[dict] = encode_dict
encode_func[int] = encode_int
encode_func[list] = encode_list
encode_func[str] = encode_string
encode_func[tuple] = encode_list
encode_func[bytes] = encode_bytes
def bencode(value):
# type: (Union[Tuple, List, OrderedDict, Dict, bool, int, str, bytes]) -> bytes
"""
Encode ``value`` into the bencode format.
:param value: Value
:type value: object
:return: Bencode formatted string
:rtype: str
"""
r = deque() # makes more sense for something with lots of appends
# Encode provided value
encode_func[type(value)](value, r)
# Join parts
return b''.join(r)
# Method proxies (for compatibility with other libraries)
decode = bdecode
encode = bencode
def bread(fd):
# type: (Union[bytes, str, pathlib.Path, pathlib.PurePath, TextIO, BinaryIO]) -> bytes
"""Return bdecoded data from filename, file, or file-like object.
if fd is a bytes/string or pathlib.Path-like object, it is opened and
read, otherwise .read() is used. if read() not available, exception
raised.
"""
if isinstance(fd, (bytes, str)):
with open(fd, 'rb') as fd:
return bdecode(fd.read())
elif pathlib is not None and isinstance(fd, (pathlib.Path, pathlib.PurePath)):
with open(str(fd), 'rb') as fd:
return bdecode(fd.read())
else:
return bdecode(fd.read())
def bwrite(data, fd):
# type: (Union[Tuple, List, OrderedDict, Dict, bool, int, str, bytes], Union[bytes, str, pathlib.Path, pathlib.PurePath, TextIO, BinaryIO]) -> None
"""Write data in bencoded form to filename, file, or file-like object.
if fd is bytes/string or pathlib.Path-like object, it is opened and
written to, otherwise .write() is used. if write() is not available,
exception raised.
"""
if isinstance(fd, (bytes, str)):
with open(fd, 'wb') as fd:
fd.write(bencode(data))
elif pathlib is not None and isinstance(fd, (pathlib.Path, pathlib.PurePath)):
with open(str(fd), 'wb') as fd:
fd.write(bencode(data))
else:
fd.write(bencode(data))