Python code coverage for Lib/binhex.py

#	count	content
1	n/a	"""Macintosh binhex compression/decompression.
2	n/a
3	n/a	easy interface:
4	n/a	binhex(inputfilename, outputfilename)
5	n/a	hexbin(inputfilename, outputfilename)
6	n/a	"""
7	n/a
8	n/a	#
9	n/a	# Jack Jansen, CWI, August 1995.
10	n/a	#
11	n/a	# The module is supposed to be as compatible as possible. Especially the
12	n/a	# easy interface should work "as expected" on any platform.
13	n/a	# XXXX Note: currently, textfiles appear in mac-form on all platforms.
14	n/a	# We seem to lack a simple character-translate in python.
15	n/a	# (we should probably use ISO-Latin-1 on all but the mac platform).
16	n/a	# XXXX The simple routines are too simple: they expect to hold the complete
17	n/a	# files in-core. Should be fixed.
18	n/a	# XXXX It would be nice to handle AppleDouble format on unix
19	n/a	# (for servers serving macs).
20	n/a	# XXXX I don't understand what happens when you get 0x90 times the same byte on
21	n/a	# input. The resulting code (xx 90 90) would appear to be interpreted as an
22	n/a	# escaped value of 0x90. All coders I've seen appear to ignore this nicety...
23	n/a	#
24	n/a	import io
25	n/a	import os
26	n/a	import struct
27	n/a	import binascii
28	n/a
29	n/a	__all__ = ["binhex","hexbin","Error"]
30	n/a
31	n/a	class Error(Exception):
32	n/a	pass
33	n/a
34	n/a	# States (what have we written)
35	n/a	_DID_HEADER = 0
36	n/a	_DID_DATA = 1
37	n/a
38	n/a	# Various constants
39	n/a	REASONABLY_LARGE = 32768 # Minimal amount we pass the rle-coder
40	n/a	LINELEN = 64
41	n/a	RUNCHAR = b"\x90"
42	n/a
43	n/a	#
44	n/a	# This code is no longer byte-order dependent
45	n/a
46	n/a
47	n/a	class FInfo:
48	n/a	def __init__(self):
49	n/a	self.Type = '????'
50	n/a	self.Creator = '????'
51	n/a	self.Flags = 0
52	n/a
53	n/a	def getfileinfo(name):
54	n/a	finfo = FInfo()
55	n/a	with io.open(name, 'rb') as fp:
56	n/a	# Quick check for textfile
57	n/a	data = fp.read(512)
58	n/a	if 0 not in data:
59	n/a	finfo.Type = 'TEXT'
60	n/a	fp.seek(0, 2)
61	n/a	dsize = fp.tell()
62	n/a	dir, file = os.path.split(name)
63	n/a	file = file.replace(':', '-', 1)
64	n/a	return file, finfo, dsize, 0
65	n/a
66	n/a	class openrsrc:
67	n/a	def __init__(self, *args):
68	n/a	pass
69	n/a
70	n/a	def read(self, *args):
71	n/a	return b''
72	n/a
73	n/a	def write(self, *args):
74	n/a	pass
75	n/a
76	n/a	def close(self):
77	n/a	pass
78	n/a
79	n/a	class _Hqxcoderengine:
80	n/a	"""Write data to the coder in 3-byte chunks"""
81	n/a
82	n/a	def __init__(self, ofp):
83	n/a	self.ofp = ofp
84	n/a	self.data = b''
85	n/a	self.hqxdata = b''
86	n/a	self.linelen = LINELEN - 1
87	n/a
88	n/a	def write(self, data):
89	n/a	self.data = self.data + data
90	n/a	datalen = len(self.data)
91	n/a	todo = (datalen // 3) * 3
92	n/a	data = self.data[:todo]
93	n/a	self.data = self.data[todo:]
94	n/a	if not data:
95	n/a	return
96	n/a	self.hqxdata = self.hqxdata + binascii.b2a_hqx(data)
97	n/a	self._flush(0)
98	n/a
99	n/a	def _flush(self, force):
100	n/a	first = 0
101	n/a	while first <= len(self.hqxdata) - self.linelen:
102	n/a	last = first + self.linelen
103	n/a	self.ofp.write(self.hqxdata[first:last] + b'\n')
104	n/a	self.linelen = LINELEN
105	n/a	first = last
106	n/a	self.hqxdata = self.hqxdata[first:]
107	n/a	if force:
108	n/a	self.ofp.write(self.hqxdata + b':\n')
109	n/a
110	n/a	def close(self):
111	n/a	if self.data:
112	n/a	self.hqxdata = self.hqxdata + binascii.b2a_hqx(self.data)
113	n/a	self._flush(1)
114	n/a	self.ofp.close()
115	n/a	del self.ofp
116	n/a
117	n/a	class _Rlecoderengine:
118	n/a	"""Write data to the RLE-coder in suitably large chunks"""
119	n/a
120	n/a	def __init__(self, ofp):
121	n/a	self.ofp = ofp
122	n/a	self.data = b''
123	n/a
124	n/a	def write(self, data):
125	n/a	self.data = self.data + data
126	n/a	if len(self.data) < REASONABLY_LARGE:
127	n/a	return
128	n/a	rledata = binascii.rlecode_hqx(self.data)
129	n/a	self.ofp.write(rledata)
130	n/a	self.data = b''
131	n/a
132	n/a	def close(self):
133	n/a	if self.data:
134	n/a	rledata = binascii.rlecode_hqx(self.data)
135	n/a	self.ofp.write(rledata)
136	n/a	self.ofp.close()
137	n/a	del self.ofp
138	n/a
139	n/a	class BinHex:
140	n/a	def __init__(self, name_finfo_dlen_rlen, ofp):
141	n/a	name, finfo, dlen, rlen = name_finfo_dlen_rlen
142	n/a	close_on_error = False
143	n/a	if isinstance(ofp, str):
144	n/a	ofname = ofp
145	n/a	ofp = io.open(ofname, 'wb')
146	n/a	close_on_error = True
147	n/a	try:
148	n/a	ofp.write(b'(This file must be converted with BinHex 4.0)\r\r:')
149	n/a	hqxer = _Hqxcoderengine(ofp)
150	n/a	self.ofp = _Rlecoderengine(hqxer)
151	n/a	self.crc = 0
152	n/a	if finfo is None:
153	n/a	finfo = FInfo()
154	n/a	self.dlen = dlen
155	n/a	self.rlen = rlen
156	n/a	self._writeinfo(name, finfo)
157	n/a	self.state = _DID_HEADER
158	n/a	except:
159	n/a	if close_on_error:
160	n/a	ofp.close()
161	n/a	raise
162	n/a
163	n/a	def _writeinfo(self, name, finfo):
164	n/a	nl = len(name)
165	n/a	if nl > 63:
166	n/a	raise Error('Filename too long')
167	n/a	d = bytes([nl]) + name.encode("latin-1") + b'\0'
168	n/a	tp, cr = finfo.Type, finfo.Creator
169	n/a	if isinstance(tp, str):
170	n/a	tp = tp.encode("latin-1")
171	n/a	if isinstance(cr, str):
172	n/a	cr = cr.encode("latin-1")
173	n/a	d2 = tp + cr
174	n/a
175	n/a	# Force all structs to be packed with big-endian
176	n/a	d3 = struct.pack('>h', finfo.Flags)
177	n/a	d4 = struct.pack('>ii', self.dlen, self.rlen)
178	n/a	info = d + d2 + d3 + d4
179	n/a	self._write(info)
180	n/a	self._writecrc()
181	n/a
182	n/a	def _write(self, data):
183	n/a	self.crc = binascii.crc_hqx(data, self.crc)
184	n/a	self.ofp.write(data)
185	n/a
186	n/a	def _writecrc(self):
187	n/a	# XXXX Should this be here??
188	n/a	# self.crc = binascii.crc_hqx('\0\0', self.crc)
189	n/a	if self.crc < 0:
190	n/a	fmt = '>h'
191	n/a	else:
192	n/a	fmt = '>H'
193	n/a	self.ofp.write(struct.pack(fmt, self.crc))
194	n/a	self.crc = 0
195	n/a
196	n/a	def write(self, data):
197	n/a	if self.state != _DID_HEADER:
198	n/a	raise Error('Writing data at the wrong time')
199	n/a	self.dlen = self.dlen - len(data)
200	n/a	self._write(data)
201	n/a
202	n/a	def close_data(self):
203	n/a	if self.dlen != 0:
204	n/a	raise Error('Incorrect data size, diff=%r' % (self.rlen,))
205	n/a	self._writecrc()
206	n/a	self.state = _DID_DATA
207	n/a
208	n/a	def write_rsrc(self, data):
209	n/a	if self.state < _DID_DATA:
210	n/a	self.close_data()
211	n/a	if self.state != _DID_DATA:
212	n/a	raise Error('Writing resource data at the wrong time')
213	n/a	self.rlen = self.rlen - len(data)
214	n/a	self._write(data)
215	n/a
216	n/a	def close(self):
217	n/a	if self.state is None:
218	n/a	return
219	n/a	try:
220	n/a	if self.state < _DID_DATA:
221	n/a	self.close_data()
222	n/a	if self.state != _DID_DATA:
223	n/a	raise Error('Close at the wrong time')
224	n/a	if self.rlen != 0:
225	n/a	raise Error("Incorrect resource-datasize, diff=%r" % (self.rlen,))
226	n/a	self._writecrc()
227	n/a	finally:
228	n/a	self.state = None
229	n/a	ofp = self.ofp
230	n/a	del self.ofp
231	n/a	ofp.close()
232	n/a
233	n/a	def binhex(inp, out):
234	n/a	"""binhex(infilename, outfilename): create binhex-encoded copy of a file"""
235	n/a	finfo = getfileinfo(inp)
236	n/a	ofp = BinHex(finfo, out)
237	n/a
238	n/a	with io.open(inp, 'rb') as ifp:
239	n/a	# XXXX Do textfile translation on non-mac systems
240	n/a	while True:
241	n/a	d = ifp.read(128000)
242	n/a	if not d: break
243	n/a	ofp.write(d)
244	n/a	ofp.close_data()
245	n/a
246	n/a	ifp = openrsrc(inp, 'rb')
247	n/a	while True:
248	n/a	d = ifp.read(128000)
249	n/a	if not d: break
250	n/a	ofp.write_rsrc(d)
251	n/a	ofp.close()
252	n/a	ifp.close()
253	n/a
254	n/a	class _Hqxdecoderengine:
255	n/a	"""Read data via the decoder in 4-byte chunks"""
256	n/a
257	n/a	def __init__(self, ifp):
258	n/a	self.ifp = ifp
259	n/a	self.eof = 0
260	n/a
261	n/a	def read(self, totalwtd):
262	n/a	"""Read at least wtd bytes (or until EOF)"""
263	n/a	decdata = b''
264	n/a	wtd = totalwtd
265	n/a	#
266	n/a	# The loop here is convoluted, since we don't really now how
267	n/a	# much to decode: there may be newlines in the incoming data.
268	n/a	while wtd > 0:
269	n/a	if self.eof: return decdata
270	n/a	wtd = ((wtd + 2) // 3) * 4
271	n/a	data = self.ifp.read(wtd)
272	n/a	#
273	n/a	# Next problem: there may not be a complete number of
274	n/a	# bytes in what we pass to a2b. Solve by yet another
275	n/a	# loop.
276	n/a	#
277	n/a	while True:
278	n/a	try:
279	n/a	decdatacur, self.eof = binascii.a2b_hqx(data)
280	n/a	break
281	n/a	except binascii.Incomplete:
282	n/a	pass
283	n/a	newdata = self.ifp.read(1)
284	n/a	if not newdata:
285	n/a	raise Error('Premature EOF on binhex file')
286	n/a	data = data + newdata
287	n/a	decdata = decdata + decdatacur
288	n/a	wtd = totalwtd - len(decdata)
289	n/a	if not decdata and not self.eof:
290	n/a	raise Error('Premature EOF on binhex file')
291	n/a	return decdata
292	n/a
293	n/a	def close(self):
294	n/a	self.ifp.close()
295	n/a
296	n/a	class _Rledecoderengine:
297	n/a	"""Read data via the RLE-coder"""
298	n/a
299	n/a	def __init__(self, ifp):
300	n/a	self.ifp = ifp
301	n/a	self.pre_buffer = b''
302	n/a	self.post_buffer = b''
303	n/a	self.eof = 0
304	n/a
305	n/a	def read(self, wtd):
306	n/a	if wtd > len(self.post_buffer):
307	n/a	self._fill(wtd - len(self.post_buffer))
308	n/a	rv = self.post_buffer[:wtd]
309	n/a	self.post_buffer = self.post_buffer[wtd:]
310	n/a	return rv
311	n/a
312	n/a	def _fill(self, wtd):
313	n/a	self.pre_buffer = self.pre_buffer + self.ifp.read(wtd + 4)
314	n/a	if self.ifp.eof:
315	n/a	self.post_buffer = self.post_buffer + \
316	n/a	binascii.rledecode_hqx(self.pre_buffer)
317	n/a	self.pre_buffer = b''
318	n/a	return
319	n/a
320	n/a	#
321	n/a	# Obfuscated code ahead. We have to take care that we don't
322	n/a	# end up with an orphaned RUNCHAR later on. So, we keep a couple
323	n/a	# of bytes in the buffer, depending on what the end of
324	n/a	# the buffer looks like:
325	n/a	# '\220\0\220' - Keep 3 bytes: repeated \220 (escaped as \220\0)
326	n/a	# '?\220' - Keep 2 bytes: repeated something-else
327	n/a	# '\220\0' - Escaped \220: Keep 2 bytes.
328	n/a	# '?\220?' - Complete repeat sequence: decode all
329	n/a	# otherwise: keep 1 byte.
330	n/a	#
331	n/a	mark = len(self.pre_buffer)
332	n/a	if self.pre_buffer[-3:] == RUNCHAR + b'\0' + RUNCHAR:
333	n/a	mark = mark - 3
334	n/a	elif self.pre_buffer[-1:] == RUNCHAR:
335	n/a	mark = mark - 2
336	n/a	elif self.pre_buffer[-2:] == RUNCHAR + b'\0':
337	n/a	mark = mark - 2
338	n/a	elif self.pre_buffer[-2:-1] == RUNCHAR:
339	n/a	pass # Decode all
340	n/a	else:
341	n/a	mark = mark - 1
342	n/a
343	n/a	self.post_buffer = self.post_buffer + \
344	n/a	binascii.rledecode_hqx(self.pre_buffer[:mark])
345	n/a	self.pre_buffer = self.pre_buffer[mark:]
346	n/a
347	n/a	def close(self):
348	n/a	self.ifp.close()
349	n/a
350	n/a	class HexBin:
351	n/a	def __init__(self, ifp):
352	n/a	if isinstance(ifp, str):
353	n/a	ifp = io.open(ifp, 'rb')
354	n/a	#
355	n/a	# Find initial colon.
356	n/a	#
357	n/a	while True:
358	n/a	ch = ifp.read(1)
359	n/a	if not ch:
360	n/a	raise Error("No binhex data found")
361	n/a	# Cater for \r\n terminated lines (which show up as \n\r, hence
362	n/a	# all lines start with \r)
363	n/a	if ch == b'\r':
364	n/a	continue
365	n/a	if ch == b':':
366	n/a	break
367	n/a
368	n/a	hqxifp = _Hqxdecoderengine(ifp)
369	n/a	self.ifp = _Rledecoderengine(hqxifp)
370	n/a	self.crc = 0
371	n/a	self._readheader()
372	n/a
373	n/a	def _read(self, len):
374	n/a	data = self.ifp.read(len)
375	n/a	self.crc = binascii.crc_hqx(data, self.crc)
376	n/a	return data
377	n/a
378	n/a	def _checkcrc(self):
379	n/a	filecrc = struct.unpack('>h', self.ifp.read(2))[0] & 0xffff
380	n/a	#self.crc = binascii.crc_hqx('\0\0', self.crc)
381	n/a	# XXXX Is this needed??
382	n/a	self.crc = self.crc & 0xffff
383	n/a	if filecrc != self.crc:
384	n/a	raise Error('CRC error, computed %x, read %x'
385	n/a	% (self.crc, filecrc))
386	n/a	self.crc = 0
387	n/a
388	n/a	def _readheader(self):
389	n/a	len = self._read(1)
390	n/a	fname = self._read(ord(len))
391	n/a	rest = self._read(1 + 4 + 4 + 2 + 4 + 4)
392	n/a	self._checkcrc()
393	n/a
394	n/a	type = rest[1:5]
395	n/a	creator = rest[5:9]
396	n/a	flags = struct.unpack('>h', rest[9:11])[0]
397	n/a	self.dlen = struct.unpack('>l', rest[11:15])[0]
398	n/a	self.rlen = struct.unpack('>l', rest[15:19])[0]
399	n/a
400	n/a	self.FName = fname
401	n/a	self.FInfo = FInfo()
402	n/a	self.FInfo.Creator = creator
403	n/a	self.FInfo.Type = type
404	n/a	self.FInfo.Flags = flags
405	n/a
406	n/a	self.state = _DID_HEADER
407	n/a
408	n/a	def read(self, *n):
409	n/a	if self.state != _DID_HEADER:
410	n/a	raise Error('Read data at wrong time')
411	n/a	if n:
412	n/a	n = n[0]
413	n/a	n = min(n, self.dlen)
414	n/a	else:
415	n/a	n = self.dlen
416	n/a	rv = b''
417	n/a	while len(rv) < n:
418	n/a	rv = rv + self._read(n-len(rv))
419	n/a	self.dlen = self.dlen - n
420	n/a	return rv
421	n/a
422	n/a	def close_data(self):
423	n/a	if self.state != _DID_HEADER:
424	n/a	raise Error('close_data at wrong time')
425	n/a	if self.dlen:
426	n/a	dummy = self._read(self.dlen)
427	n/a	self._checkcrc()
428	n/a	self.state = _DID_DATA
429	n/a
430	n/a	def read_rsrc(self, *n):
431	n/a	if self.state == _DID_HEADER:
432	n/a	self.close_data()
433	n/a	if self.state != _DID_DATA:
434	n/a	raise Error('Read resource data at wrong time')
435	n/a	if n:
436	n/a	n = n[0]
437	n/a	n = min(n, self.rlen)
438	n/a	else:
439	n/a	n = self.rlen
440	n/a	self.rlen = self.rlen - n
441	n/a	return self._read(n)
442	n/a
443	n/a	def close(self):
444	n/a	if self.state is None:
445	n/a	return
446	n/a	try:
447	n/a	if self.rlen:
448	n/a	dummy = self.read_rsrc(self.rlen)
449	n/a	self._checkcrc()
450	n/a	finally:
451	n/a	self.state = None
452	n/a	self.ifp.close()
453	n/a
454	n/a	def hexbin(inp, out):
455	n/a	"""hexbin(infilename, outfilename) - Decode binhexed file"""
456	n/a	ifp = HexBin(inp)
457	n/a	finfo = ifp.FInfo
458	n/a	if not out:
459	n/a	out = ifp.FName
460	n/a
461	n/a	with io.open(out, 'wb') as ofp:
462	n/a	# XXXX Do translation on non-mac systems
463	n/a	while True:
464	n/a	d = ifp.read(128000)
465	n/a	if not d: break
466	n/a	ofp.write(d)
467	n/a	ifp.close_data()
468	n/a
469	n/a	d = ifp.read_rsrc(128000)
470	n/a	if d:
471	n/a	ofp = openrsrc(out, 'wb')
472	n/a	ofp.write(d)
473	n/a	while True:
474	n/a	d = ifp.read_rsrc(128000)
475	n/a	if not d: break
476	n/a	ofp.write(d)
477	n/a	ofp.close()
478	n/a
479	n/a	ifp.close()