Lib/binhex.py

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