#!/usr/bin/env python3 # vim: set tabstop=4 shiftwidth=4 noexpandtab filetype=python: """cppo: Copy/catalog files from a ProDOS/DOS 3.3/ShrinkIt image/archive. copy all files: cppo [options] imagefile target_directory copy one file : cppo [options] imagefile /extract/path target_path catalog image : cppo -cat [options] imagefile options: -shk: ShrinkIt archive as source (also auto-enabled by filename). -ad : Netatalk-compatible AppleDouble metadata files and resource forks. -e : Nulib2-compatible filenames with type/auxtype and resource forks. -uc : Copy GS/OS mixed case filenames as uppercase. -pro: Adapt DOS 3.3 names to ProDOS and remove addr/len from file data. /extract/path examples: /FULL/PRODOS/PATH (ProDOS image source) "MY FILENAME" (DOS 3.3 image source) Dir:SubDir:FileName (ShrinkIt archive source) + after a file name indicates a GS/OS or Mac OS extended (forked) file. Wildcard matching (*) is not supported and images are not validated. ShrinkIt support requires Nulib2. cppo requires Python 2.6+ or 3.0+.""" # cppo by Ivan X, ivan@ivanx.com, ivanx.com/appleii # Does anyone want to rewrite/refactor this? It works, but it's a mess. import sys import os import time import datetime import shutil import errno import uuid import subprocess import tempfile from binascii import a2b_hex, b2a_hex class Globals: pass g = Globals() g.image_data = b'' g.out_data = bytearray(b'') g.ex_data = None g.activeDirBlock = None g.activeFileName = None g.activeFileSize = None g.activeFileBytesCopied = 0 g.resourceFork = 0 g.shk_hasrf = False g.PDOSPATH = [] g.PDOSPATH_INDEX = 0 g.PDOSPATH_SEGMENT = None g.DIRPATH = "" g.target_name = None g.target_dir = "" g.appledouble_dir = None g.image_file = None g.extract_file = None # runtime options g.use_appledouble = False # -ad (AppleDouble headers + resource forks) g.use_extended = False # -e (extended filenames + resource forks) g.catalog_only = False # -cat (catalog only, no extract) g.casefold_upper = False # -uc (GS/OS mixed case filenames extract as uppercase) g.src_shk = False # -shk (ShrinkIt archive source) g.prodos_names = False # -pro (adapt DOS 3.3 names to ProDOS) g.afpsync_msg = True # -s (sets False to suppress afpsync message at end) g.extract_in_place = False # -n (don't create parent dir for SHK, extract files in place) g.dos33 = False # (DOS 3.3 image source, selected automatically) # functions def pdosDateToUnixDate(arg1): # input: ProDOS date/time bit sequence string in format: # "yyyyyyymmmmddddd000hhhhh00mmmmmm" (ustr) # output: seconds since Unix epoch (1-Jan-1970), # or current date/time if no ProDOS date year = (binToDec(slyce(arg1,0,7)) + 1900) if year < 1940: year += 100 month = binToDec(slyce(arg1,7,4)) day = binToDec(slyce(arg1,11,5)) hour = binToDec(slyce(arg1,19,5)) minute = binToDec(slyce(arg1,26,6)) # print(year, month, day, hour, minute) td = (datetime.datetime(year, month, day, hour, minute) - datetime.datetime(1970,1,1)) unixDate_naive = (td.days*24*60*60 + td.seconds) td2 = (datetime.datetime.fromtimestamp(unixDate_naive) - datetime.datetime.utcfromtimestamp(unixDate_naive)) utcoffset = (td2.days*24*60*60 + td2.seconds) # print(unixDate_naive - utcoffset) return (unixDate_naive - utcoffset) # local time zone with DST def unixDateToADDate(arg1): # input: seconds since Unix epoch (1-Jan-1970 00:00:00 GMT) # output: seconds since Netatalk epoch (1-Jan-2000 00:00:00 GMT), # in hex-ustr (big endian) adDate = (arg1 - 946684800) if adDate < 0: adDate += 4294967296 # to get negative hex number adDateHex = to_hex(adDate).zfill(8).upper() # print(arg1, adDate, adDateHex) return adDateHex # cppo support functions: # arg1: directory block or [T,S] containing file entry, or shk file dir path # arg2: file index in overall directory (if applicable), or shk file name # returns byte position in disk image file def getStartPos(arg1, arg2): if g.dos33: return (ts(arg1) + (35 * (arg2 % 7)) + 11) else: # ProDOS return ( (arg1 * 512) + (39 * ((arg2 + (arg2 > 11)) % 13)) + (4 if arg2 > 11 else 43) ) def getStorageType(arg1, arg2): start = getStartPos(arg1, arg2) firstByte = readcharDec(g.image_data, start) return (int(firstByte != 255)*2 if g.dos33 else (firstByte//16)) def getFileName(arg1, arg2): start = getStartPos(arg1, arg2) if g.dos33: fileNameLo = bytearray() fileNameHi = readchars(g.image_data, start+3, 30) for b in fileNameHi: fileNameLo += to_bytes(to_dec(b)-128) fileName = bytes(fileNameLo).rstrip() else: # ProDOS firstByte = readcharDec(g.image_data, start) entryType = (firstByte//16) nameLength = (firstByte - entryType*16) fileName = readchars(g.image_data, start+1, nameLength) caseMask = getCaseMask(arg1, arg2) if caseMask and not g.casefold_upper: fileName = bytearray(fileName) for i in range(0, len(fileName)): if caseMask[i] == "1": fileName[i] = fileName[i].lower() fileName = bytes(fileName) return fileName def getCaseMask(arg1, arg2): start = getStartPos(arg1, arg2) caseMaskDec = (readcharDec(g.image_data, start+28) + readcharDec(g.image_data, start+29)*256) if caseMaskDec < 32768: return None else: return to_bin(caseMaskDec - 32768).zfill(15) def getFileType(arg1, arg2): if g.src_shk: return arg2.split('#')[1][0:2] start = getStartPos(arg1, arg2) if g.dos33: d33fileType = readcharDec(g.image_data, start+2) if (d33fileType & 127) == 4: return '06' # BIN elif (d33fileType & 127) == 1: return 'FA' # INT elif (d33fileType & 127) == 2: return 'FC' # BAS else: return '04' # TXT or other else: # ProDOS return readcharHex(g.image_data, start+16) def getAuxType(arg1, arg2): if g.src_shk: return arg2.split('#')[1][2:6] start = getStartPos(arg1, arg2) if g.dos33: fileType = getFileType(arg1, arg2) if fileType == '06': # BIN (B) # file address is in first two bytes of file data fileTSlist = [readcharDec(g.image_data, start+0), readcharDec(g.image_data, start+1)] fileStart = [readcharDec(g.image_data, ts(fileTSlist)+12), readcharDec(g.image_data, ts(fileTSlist)+13)] return (readcharHex(g.image_data, ts(fileStart)+1) + readcharHex(g.image_data, ts(fileStart)+0)) elif fileType == 'FC': # BAS (A) return '0801' elif fileType == 'FA': # INT (I) return '9600' else: # TXT (T) or other return '0000' else: # ProDOS return (readcharHex(g.image_data, start+32) + readcharHex(g.image_data, start+31)) def getKeyPointer(arg1, arg2): start = getStartPos(arg1, arg2) if g.dos33: return [readcharDec(g.image_data, start+0), readcharDec(g.image_data, start+1)] else: # ProDOS return (readcharDec(g.image_data, start+17) + readcharDec(g.image_data, start+18)*256) def getFileLength(arg1, arg2): start = getStartPos(arg1, arg2) if g.dos33: fileType = getFileType(arg1, arg2) fileTSlist = [readcharDec(g.image_data, start+0), readcharDec(g.image_data, start+1)] fileStart = [readcharDec(g.image_data, ts(fileTSlist)+12), readcharDec(g.image_data, ts(fileTSlist)+13)] if fileType == '06': # BIN (B) # file length is in second two bytes of file data return ((readcharDec(g.image_data, ts(fileStart)+2) + readcharDec(g.image_data, ts(fileStart)+3)*256) + 4) elif fileType == 'FC' or fileType == 'FA': # BAS (A) or INT (I) # file length is in first two bytes of file data return ((readcharDec(g.image_data, ts(fileStart)+0) + readcharDec(g.image_data, ts(fileStart)+1)*256) + 2) else: # TXT (T) or other # sadly, we have to walk the whole file # length is determined by sectors in TSlist, minus wherever # anything after the first zero in the last sector fileSize = 0 lastTSpair = None nextTSlistSector = fileTSlist endFound = False while not endFound: pos = ts(nextTSlistSector) for tsPos in range(12, 256, 2): if ts(readcharDec(g.image_data, pos+tsPos+0), readcharDec(g.image_data, pos+tsPos+1)) != 0: fileSize += 256 prevTSpair = [readcharDec(g.image_data, (pos+tsPos)+0), readcharDec(g.image_data, (pos+tsPos)+1)] else: lastTSpair = prevTSpair endFound = True break if not lastTSpair: nextTSlistSector = [readcharDec(g.image_data, pos+1), readcharDec(g.image_data, pos+2)] if nextTSlistSector[0]+nextTSlistSector[1] == 0: lastTSpair = prevTSpair endFound = True break fileSize -= 256 pos = ts(prevTSpair) # now find out where the file really ends by finding the last 00 for offset in range(255, -1, -1): #print("pos: " + to_hex(pos)) if readcharDec(g.image_data, pos+offset) != 0: fileSize += (offset + 1) break return fileSize else: # ProDOS return (readcharDec(g.image_data, start+21) + readcharDec(g.image_data, start+22)*256 + readcharDec(g.image_data, start+23)*65536) def getCreationDate(arg1, arg2): #outputs prodos creation date/time as Unix time # (seconds since Jan 1 1970 GMT) #or None if there is none if g.src_shk: return None elif g.dos33: return None else: # ProDOS start = getStartPos(arg1, arg2) pdosDate = (hexToBin(readcharHex(g.image_data, start+25)) + hexToBin(readcharHex(g.image_data, start+24)) + hexToBin(readcharHex(g.image_data, start+27)) + hexToBin(readcharHex(g.image_data, start+26))) try: rVal = pdosDateToUnixDate(pdosDate) except Exception: rVal = None return rVal def getModifiedDate(arg1, arg2): #outputs prodos modified date/time as Unix time # (seconds since Jan 1 1970 GMT) #or None if there is none if g.src_shk: modifiedDate = int(time.mktime(time.strptime(time.ctime( os.path.getmtime(os.path.join(arg1, arg2)))))) rVal = modifiedDate elif g.dos33: rVal = None else: # ProDOS start = getStartPos(arg1, arg2) pdosDate = (hexToBin(readcharHex(g.image_data, start+34)) + hexToBin(readcharHex(g.image_data, start+33)) + hexToBin(readcharHex(g.image_data, start+36)) + hexToBin(readcharHex(g.image_data, start+35))) try: rVal = pdosDateToUnixDate(pdosDate) except Exception: rVal = None return rVal def getVolumeName(): return getWorkingDirName(2) def getWorkingDirName(arg1, arg2=None): # arg1:block, arg2:casemask (optional) start = ( arg1 * 512 ) firstByte = readcharDec(g.image_data, start+4) entryType = (firstByte//16) nameLength = (firstByte - entryType*16) workingDirName = readchars(g.image_data, start+5, nameLength) if entryType == 15: # volume directory, get casemask from header caseMaskDec = (readcharDec(g.image_data, start+26) + readcharDec(g.image_data, start+27)*256) if caseMaskDec < 32768: caseMask = None else: caseMask = to_bin(caseMaskDec - 32768).zfill(15) else: # subdirectory, get casemask from arg2 (not available in header) caseMask = arg2 if caseMask and not g.casefold_upper: workingDirName = bytearray(workingDirName) for i in range(0, len(workingDirName)): if caseMask[i] == "1": workingDirName[i] = workingDirName[i].lower() workingDirName = bytes(workingDirName) return workingDirName def getDirEntryCount(arg1): if g.dos33: entryCount = 0 #nextSector = [readcharDec(g.image_data, ts(arg1)+1), # readcharDec(g.image_data, ts(arg1)+2)] nextSector = arg1 while True: top = ts(nextSector) pos = top+11 for e in range(0, 7): if readcharDec(g.image_data, pos+0) == 0: return entryCount # no more file entries else: if readcharDec(g.image_data, pos+0) != 255: entryCount += 1 # increment if not deleted file pos += 35 nextSector = [readcharDec(g.image_data, top+1), readcharDec(g.image_data, top+2)] if nextSector[0]+nextSector[1] == 0: # no more catalog sectors return entryCount else: # ProDOS start = arg1 * 512 return (readcharDec(g.image_data, start+37) + readcharDec(g.image_data, start+38)*256) def getDirNextChunkPointer(arg1): if g.dos33: start = ts(arg1) return [readcharDec(g.image_data, start+1), readcharDec(g.image_data, start+2)] else: # ProDOS start = arg1 * 512 return (readcharDec(g.image_data, start+2) + readcharDec(g.image_data, start+3)*256) def toProdosName(name): i=0 if name[0] == '.': # eliminate leading period name = name[1:] for c in name: if c != '.' and not c.isalnum(): name = name[:i] + '.' + name[i+1:] i+=1 name = name[:15] return name def ts(track, sector=None): # returns offset; track and sector can be dec, or hex-ustr # can also supply as [t,s] for convenience if sector == None: (track, sector) = track if isinstance(track, type("".encode("L1").decode("L1"))): # hex-ustr track = int(track, 16) if isinstance(sector, type("".encode("L1").decode("L1"))): # hex-ustr sector = int(sector, 16) return track*16*256 + sector*256 def sli(start, length=1, ext=None): """return a slice object from an offset and length""" return slice(start, start + length, ext) # --- main logic functions def copyFile(arg1, arg2): #arg1/arg2: # ProDOS : directory block / file index in overall directory # DOS 3.3 : [track, sector] / file index in overall VTOC # ShrinkIt: directory path / file name # copies file or dfork to g.out_data, rfork if any to g.ex_data g.activeFileBytesCopied = 0 if g.src_shk: with open(os.path.join(arg1, arg2), 'rb') as infile: g.out_data += infile.read() if g.shk_hasrf: print(" [data fork]") if g.use_extended or g.use_appledouble: print(" [resource fork]") if g.ex_data == None: g.ex_data = bytearray(b'') with open(os.path.join(arg1, (arg2 + "r")), 'rb') as infile: g.ex_data += infile.read() else: # ProDOS or DOS 3.3 storageType = getStorageType(arg1, arg2) keyPointer = getKeyPointer(arg1, arg2) fileLen = getFileLength(arg1, arg2) if storageType == 1: #seedling copyBlock(keyPointer, fileLen) elif storageType == 2: #sapling processIndexBlock(keyPointer) elif storageType == 3: #tree processMasterIndexBlock(keyPointer) elif storageType == 5: #extended (forked) processForkedFile(keyPointer) if g.prodos_names: # remove address/length data from DOS 3.3 file data if ProDOS target if getFileType(arg1, arg2) == '06': g.out_data = g.out_data[4:] elif (getFileType(arg1, arg2) == 'FA' or getFileType(arg1, arg2) == 'FC'): g.out_data = g.out_data[2:] def copyBlock(arg1, arg2): #arg1: block number or [t,s] to copy #arg2: bytes to write (should be 256 (DOS 3.3) or 512 (ProDOS), # unless final block with less) #print(arg1 + " " + arg2 + " " + g.activeFileBytesCopied) if arg1 == 0: outBytes = bytes(arg2) else: outBytes = slyce(g.image_data, (ts(arg1) if g.dos33 else arg1*512), arg2) if g.resourceFork > 0: if g.use_appledouble or g.use_extended: offset = (741 if g.use_appledouble else 0) if g.ex_data == None: g.ex_data = bytearray(b'') g.ex_data[ (g.activeFileBytesCopied + offset) : (g.activeFileBytesCopied + offset + arg2) ] = outBytes else: g.out_data[ g.activeFileBytesCopied : (g.activeFileBytesCopied + arg2) ] = outBytes g.activeFileBytesCopied += arg2 def processDir(arg1, arg2=None, arg3=None, arg4=None, arg5=None): # arg1: ProDOS directory block, or DOS 3.3 [track,sector] # for key block (with directory header): # arg2: casemask (optional), arg3:None, arg4:None, arg5:None # for secondary directory blocks (non-key block): # arg2/3/4/5: for non-key chunks: entryCount, entry#, # workingDirName, processedEntryCount entryCount = None e = None pe = None workingDirName = None if arg3: entryCount = arg2 e = arg3 workingDirName = arg4 pe = arg5 else: e = 0 pe = 0 entryCount = getDirEntryCount(arg1) if not g.dos33: workingDirName = getWorkingDirName(arg1, arg2).decode("L1") g.DIRPATH = (g.DIRPATH + "/" + workingDirName) if g.PDOSPATH_INDEX: if g.PDOSPATH_INDEX == 1: if ("/" + g.PDOSPATH_SEGMENT.lower()) != g.DIRPATH.lower(): print("ProDOS volume name does not match disk image.") quitNow(2) else: g.PDOSPATH_INDEX += 1 g.PDOSPATH_SEGMENT = g.PDOSPATH[g.PDOSPATH_INDEX] #else: print(g.DIRPATH) while pe < entryCount: if getStorageType(arg1, e) > 0: #print(pe, e, entryCount) processEntry(arg1, e) pe += 1 e += 1 if not (e + (0 if g.dos33 else (e>11)) ) % (7 if g.dos33 else 13): processDir(getDirNextChunkPointer(arg1), entryCount, e, workingDirName, pe) break def processEntry(arg1, arg2): # arg1=block number, [t,s] if g.dos33=True, or subdir name if g.src_shk=1 # arg2=index number of entry in directory, or file name if g.src_shk=1 #print(getFileName(arg1, arg2), getStorageType(arg1, arg2), # getFileType(arg1, arg2), getKeyPointer(arg1, arg2), # getFileLength(arg1, arg2), getAuxType(arg1, arg2), # getCreationDate(arg1, arg2), getModifiedDate(arg1, arg2)) eTargetName = None g.ex_data = None g.out_data = bytearray(b'') if g.src_shk: # ShrinkIt archive g.activeFileName = (arg2 if g.use_extended else arg2.split('#')[0]) if g.casefold_upper: g.activeFileName = g.activeFileName.upper() origFileName = g.activeFileName else: # ProDOS or DOS 3.3 image g.activeFileName = getFileName(arg1 ,arg2).decode("L1") origFileName = g.activeFileName if g.prodos_names: g.activeFileName = toProdosName(g.activeFileName) g.activeFileSize = getFileLength(arg1, arg2) if (not g.PDOSPATH_INDEX or g.activeFileName.upper() == g.PDOSPATH_SEGMENT.upper()): # if ProDOS directory, not file if not g.src_shk and getStorageType(arg1, arg2) == 13: if not g.PDOSPATH_INDEX: g.target_dir = g.target_dir + "/" + g.activeFileName g.appledouble_dir = g.target_dir + "/.AppleDouble" if not g.catalog_only or os.path.isdir(g.target_dir): makedirs(g.target_dir) if (not g.catalog_only and g.use_appledouble and not os.path.isdir(g.appledouble_dir)): makedirs(g.appledouble_dir) if g.PDOSPATH_SEGMENT: g.PDOSPATH_INDEX += 1 g.PDOSPATH_SEGMENT = g.PDOSPATH[g.PDOSPATH_INDEX] processDir(getKeyPointer(arg1, arg2), getCaseMask(arg1, arg2)) g.DIRPATH = g.DIRPATH.rsplit("/", 1)[0] if not g.PDOSPATH_INDEX: g.target_dir = g.target_dir.rsplit("/", 1)[0] g.appledouble_dir = (g.target_dir + "/.AppleDouble") else: # ProDOS or DOS 3.3 file either from image or ShrinkIt archive dirPrint = "" if g.DIRPATH: dirPrint = g.DIRPATH + "/" else: if g.src_shk: if "/".join(dirName.split('/')[3:]): dirPrint = ("/".join(dirName.split('/')[3:]) + "/") if (not g.extract_file or (os.path.basename(g.extract_file.lower()) == origFileName.split('#')[0].lower())): filePrint = g.activeFileName.split("#")[0] print(dirPrint + filePrint + ("+" if (g.shk_hasrf or (not g.src_shk and getStorageType(arg1, arg2) == 5)) else "") + ((" [" + origFileName + "] ") if (g.prodos_names and (origFileName != g.activeFileName)) else "")) if g.catalog_only: return if not g.target_name: g.target_name = g.activeFileName if g.use_extended: if g.src_shk: eTargetName = arg2 else: # ProDOS image eTargetName = (g.target_name + "#" + getFileType(arg1, arg2).lower() + getAuxType(arg1, arg2).lower()) # touch(g.target_dir + "/" + g.target_name) if g.use_appledouble: makeADfile() copyFile(arg1, arg2) saveName = (g.target_dir + "/" + (eTargetName if eTargetName else g.target_name)) saveFile(saveName, g.out_data) creationDate = getCreationDate(arg1, arg2) modifiedDate = getModifiedDate(arg1, arg2) if modifiedDate and not creationDate: creationDate = modifiedDate elif creationDate and not modifiedDate: modifiedDate = creationDate elif not creationDate and not modifiedDate: creationDate = (datetime.datetime.today() - datetime.datetime(1970,1,1)).days*24*60*60 modifiedDate = creationDate if g.use_appledouble: # AppleDouble # set dates ADfilePath = g.appledouble_dir + "/" + g.target_name writecharsHex(g.ex_data, 637, (unixDateToADDate(creationDate) + unixDateToADDate(modifiedDate))) writecharHex(g.ex_data, 645, "80") writecharHex(g.ex_data, 649, "80") #set type/creator writechars(g.ex_data, 653, b'p') writecharsHex(g.ex_data, 654, getFileType(arg1, arg2) + getAuxType(arg1, arg2)) writechars(g.ex_data, 657, b'pdos') saveFile(ADfilePath, g.ex_data) touch(saveName, modifiedDate) if g.use_extended: # extended name from ProDOS image if g.ex_data: saveFile((saveName + "r"), g.ex_data) touch((saveName + "r"), modifiedDate) if (g.PDOSPATH_SEGMENT or (g.extract_file and (g.extract_file.lower() == origFileName.lower()))): quitNow(0) g.target_name = None #else print(g.activeFileName + " doesn't match " + g.PDOSPATH_SEGMENT) def processForkedFile(arg1): # finder info except type/creator fInfoA_entryType = readcharDec(g.image_data, 9) fInfoB_entryType = readcharDec(g.image_data, 27) if fInfoA_entryType == 1: writechars(g.image_data, 661, readchars(g.image_data, 18, 8)) elif fInfoA_entryType == 2: writechars(g.image_data, 669, readchars(g.image_data, 10, 16)) if fInfoB_entryType == 1: writechars(g.image_data, 661, readchars(g.image_data, 36, 8)) elif fInfoB_entryType == 2: writechars(g.image_data, 669, readchars(g.image_data, 28, 16)) for f in (0, 256): g.resourceFork = f g.activeFileBytesCopied = 0 forkStart = (arg1 * 512) # start of Forked File key block # print("--" + forkStart) forkStorageType = readcharDec(g.image_data, forkStart+f+0) forkKeyPointer = (readcharDec(g.image_data, forkStart+f+1) + readcharDec(g.image_data, forkStart+f+2)*256) forkFileLen = (readcharDec(g.image_data, forkStart+f+5) + readcharDec(g.image_data, forkStart+f+6)*256 + readcharDec(g.image_data, forkStart+f+7)*256*256) g.activeFileSize = forkFileLen if g.resourceFork > 0: rsrcForkLenHex = (readcharHex(g.image_data, forkStart+f+7) + readcharHex(g.image_data, forkStart+f+6) + readcharHex(g.image_data, forkStart+f+5)) # print(">>>" + rsrcForkLenHex) if g.use_appledouble or g.use_extended: print(" [resource fork]") if g.use_appledouble: writecharsHex(g.ex_data, 35, rsrcForkLenHex) else: print(" [data fork]") if forkStorageType == 1: #seedling copyBlock(forkKeyPointer, forkFileLen) elif forkStorageType == 2: #sapling processIndexBlock(forkKeyPointer) elif forkStorageType == 3: #tree processMasterIndexBlock(forkKeyPointer) # print() g.resourceFork = 0 def processMasterIndexBlock(arg1): processIndexBlock(arg1, True) def processIndexBlock(arg1, arg2=False): #arg1: indexBlock, or [t,s] of track/sector list #arg2: if True, it's a Master Index Block pos = 12 if g.dos33 else 0 bytesRemaining = g.activeFileSize while g.activeFileBytesCopied < g.activeFileSize: if g.dos33: targetTS = [readcharDec(g.image_data, ts(arg1)+pos+0), readcharDec(g.image_data, ts(arg1)+pos+1)] #print(to_hex(targetTS[0]),to_hex(targetTS[1])) bytesRemaining = (g.activeFileSize - g.activeFileBytesCopied) bs = (bytesRemaining if bytesRemaining < 256 else 256) copyBlock(targetTS, bs) pos += 2 if pos > 255: # continue with next T/S list sector processIndexBlock([readcharDec(g.image_data, ts(arg1)+1), readcharDec(g.image_data, ts(arg1)+2)]) else: # ProDOS targetBlock = (readcharDec(g.image_data, arg1*512+pos) + readcharDec(g.image_data, arg1*512+(pos+256))*256) if arg2: processIndexBlock(targetBlock) else: bytesRemaining = (g.activeFileSize - g.activeFileBytesCopied) bs = (bytesRemaining if bytesRemaining < 512 else 512) copyBlock(targetBlock, bs) pos += 1 if pos > 255: break # go to next entry in Master Index Block (tree) def makeADfile(): if not g.use_appledouble: return touch(g.appledouble_dir + "/" + g.target_name) g.ex_data = bytearray(741) # ADv2 header writecharsHex(g.ex_data, hexToDec("00"), "0005160700020000") # number of entries writecharsHex(g.ex_data, hexToDec("18"), "000D") # Resource Fork writecharsHex(g.ex_data, hexToDec("1A"), "00000002000002E500000000") # Real Name writecharsHex(g.ex_data, hexToDec("26"), "00000003000000B600000000") # Comment writecharsHex(g.ex_data, hexToDec("32"), "00000004000001B500000000") # Dates Info writecharsHex(g.ex_data, hexToDec("3E"), "000000080000027D00000010") # Finder Info writecharsHex(g.ex_data, hexToDec("4A"), "000000090000028D00000020") # ProDOS file info writecharsHex(g.ex_data, hexToDec("56"), "0000000B000002C100000008") # AFP short name writecharsHex(g.ex_data, hexToDec("62"), "0000000D000002B500000000") # AFP File Info writecharsHex(g.ex_data, hexToDec("6E"), "0000000E000002B100000004") # AFP Directory ID writecharsHex(g.ex_data, hexToDec("7A"), "0000000F000002AD00000004") # dbd (second time) will create DEV, INO, SYN, SV~ def quitNow(exitcode=0): if (exitcode == 0 and g.afpsync_msg and g.use_appledouble and os.path.isdir("/usr/local/etc/netatalk")): print("File(s) have been copied to the target directory. " + "If the directory") print("is shared by Netatalk, please type 'afpsync' now.") if g.src_shk: # clean up for file in os.listdir('/tmp'): if file.startswith("cppo-"): shutil.rmtree('/tmp' + "/" + file) sys.exit(exitcode) def usage(exitcode=1): print(sys.modules[__name__].__doc__) quitNow(exitcode) def to_sys_name(name): if os.name == 'nt': if name[-1] == '.': name += '-' name = name.replace('./', '.-/') return name # --- ID bashbyter functions (adapted) def decToHex(arg1): # converts single-byte decimal value to hexadecimal equivalent # arg: decimal value from 0-255 # out: two-digit hex string from 00-FF #exit: 21=invalid arg if arg1<0 or arg1>255: sys.exit(21) return to_hex(arg1).upper() def hexToDec(arg1): # converts single-byte hexadecimal value to decimal equivalent # arg: two-digit hex value from 00-FF # out: decimal value #exit: 21=invalid arg if len(arg1) != 2: return 21 return to_dec(arg1) def hexToBin(arg1): # converts single-byte hexadecimal value to binary string # arg: two-digit hex value from 00-FF # out: binary string value #exit: 21=invalid arg if len(arg1) != 2: return 21 return to_bin(arg1).zfill(8) def binToDec(arg1): # converts single-byte binary string (8 bits) value to decimal # warning: no error checking # arg: binary string up to 8 bits # out: decimal value return to_dec([arg1]) def binToHex(arg1): # converts single-byte binary string (8 bits) value to hex # warning: no error checking # arg: binary string up to 8 bits # out: hex value return to_hex(arg1).upper() def charToDec(arg1): # converts single char (of type bytes) to corresponding decimal value # arg: one char (of type bytes) # out: decimal value from 0-255 #exit: 21=invalid arg if len(arg1) != 1: return 21 return to_dec(arg1) def charToHex(arg1): # converts single char (of type bytes) to corresponding hex value # arg: one char (of type bytes) # out: hexadecimal value from 00-FF #exit: 21=invalid arg if len(arg1) != 1: return 21 return to_hex(arg1).upper() def decToChar(arg1): # converts single-byte decimal value to equivalent char (of type bytes) # arg: decimal number from 0-255 # out: one character #exit: 21=invalid arg if arg1<0 or arg1>255: sys.exit(21) return to_bytes(arg1) def hexToChar(arg1): # converts single-byte hex value to corresponding char (of type bytes) # arg: two-digit hexadecimal number from 00-FF # out: one character #exit: 21=invalid arg if len(arg1) != 2: return 21 return to_bytes(arg1) def readchars(arg1, arg2=0, arg3=0): # read one or more characters from a bytes variable # arg1: bytes or bytearray variable # arg2: (optional) offset (# of bytes to skip before reading) # arg3: (optional) # of chars to read (default is to end of bytes var) # out: sequence of characters (bytes or bytearray) # exit: 21=invalid arg1, 22=invalid arg2, 23=invalid arg3 if not (isinstance(arg1, bytes) or isinstance(arg1, bytearray)): sys.exit(21) if arg2<0: sys.exit(22) if arg3<0: sys.exit(23) if arg3 == 0: arg3 = len(arg1) return slyce(arg1, arg2, arg3) def readcharDec(arg1, arg2=0): # read one character from bytes var & convert to equivalent dec value # arg1: bytes var # arg2: (optional) offset (# of bytes to skip before reading) # out: decimal value from 0-255 # exit: 21=invalid arg1, 22=invalid arg2 if not (isinstance(arg1, bytes) or isinstance(arg1, bytearray)): sys.exit(21) if arg2<0: sys.exit(22) return to_dec(slyce(arg1, arg2, 1)) def readcharHex(arg1, arg2=0): # read one character from bytes var & convert to corresponding hex value # arg1: bytes var # arg2: (optional) offset (# of bytes to skip before reading) # out: two-digit hex value from 00-FF # exit: 21=invalid arg1, 22=invalid arg2 if not (isinstance(arg1, bytes) or isinstance(arg1, bytearray)): sys.exit(21) if arg2<0: sys.exit(22) return to_hex(slyce(arg1, arg2, 1)) def writechars(arg1, arg2, arg3): # write one or more characters (bytes) to bytearray # arg1: bytearray variable # arg2: offset (# of bytes to skip before writing) # arg3: sequence of bytes (or bytearray) # out: nothing # exit: 21=invalid arg1, 22=invalid arg2, 23=invalid arg3 if not isinstance(arg1, bytearray): sys.exit(21) if arg2<0: sys.exit(22) if not (isinstance(arg3, bytes) or isinstance(arg3, bytearray)): sys.exit(23) arg1[arg2:arg2+len(arg3)] = arg3 def writecharDec(arg1, arg2, arg3): # write corresponding char of single-byte decimal value into bytearray # arg1: bytearray # arg2: offset (# of bytes to skip before writing) # arg3: decimal number from 0-255 # exit: 21=invalid arg1, 22=invalid arg2, 23=invalid arg3 # out: nothing if not isinstance(arg1, bytearray): sys.exit(21) if arg2<0: sys.exit(22) if not isnumber(arg3): sys.exit(23) arg1[arg2:arg2+1] = to_bytes(arg3) def writecharHex(arg1, arg2, arg3): # write corresponding character of single-byte hex value into bytearray # arg1: bytearray # arg2: offset (# of bytes to skip before writing) # arg3: two-digit hexadecimal number from 00-FF # out: nothing # exit: 21=invalid arg1, 22=invalid arg2, 23=invalid arg3 if not isinstance(arg1, bytearray): sys.exit(21) if arg2<0: sys.exit(22) if not isinstance(arg3, type("".encode("L1").decode("L1"))): sys.exit(23) arg1[arg2:arg2+1] = to_bytes(arg3) def writecharsHex(arg1, arg2, arg3): # write corresponding characters of hex values into bytearray # arg1: bytearray # arg2: offset (# of bytes to skip before writing) # arg3: string of two-digit hexadecimal numbers from 00-FF # out: nothing # exit: 21=invalid arg1, 22=invalid arg2, 23=invalid arg3 if not isinstance(arg1, bytearray): sys.exit(21) if arg2<0: sys.exit(22) if not isinstance(arg3, type("".encode("L1").decode("L1"))): sys.exit(23) arg1[arg2:arg2+len(to_bytes(arg3))] = to_bytes(arg3) #---- IvanX general purpose functions ----# def slyce(val, start_pos=0, length=1, reverse=False): """returns slice of object (but not a slice object) allows specifying length, and 3.x "bytes" consistency""" if start_pos < 0: the_slyce = val[start_pos:] else: the_slyce = val[start_pos:start_pos+length] return (the_slyce[::-1] if reverse else the_slyce) def to_hex(val): """convert bytes, decimal number, or [bin-ustr] to two-digit hex values unlike hex(), accepts bytes; has no leading 0x or trailing L""" if isinstance(val, list): # [bin-ustr] val = int(val[0], 2) if isinstance(val, bytes): # bytes return b2a_hex(val).decode("L1") elif isnumber(val): # hex returns str/bytes in P2, but str/unicode in P3, so # .encode().decode() always returns unicode in either if val < 0: print ("val: " + str(val)) return hex(val)[2:].encode("L1").decode("L1").split("L")[0] else: raise Exception("to_hex() requires bytes, int/long, or [bin-ustr]") def hex_slyce(val, start_pos=0, length=1, little_endian=False): """returns bytes slyce as hex-ustr""" return to_hex(slyce(val, start_pos, length, little_endian)) def dec_slyce(val, start_pos=0, length=1, little_endian=False): """returns bytes slyce converted to decimal int/long""" return to_dec(hex_slyce(val, start_pos, length, little_endian)) def bin_slyce(val, start_pos=0, length=1, little_endian=False): """returns bytes slyce converted to bin-ustr""" return to_bin(hex_slyce(val, start_pos, length, little_endian)) def to_dec(val): """convert bytes, hex-ustr or [bin-ustr] to decimal int/long""" if isinstance(val, list): # [bin-ustr] return int(val[0], 2) elif isinstance(val, bytes): # bytes return int(to_hex(val), 16) elif isinstance(val, type("".encode("L1").decode("L1"))): # hex-ustr return int(val, 16) elif isnumber(val): # int/long return val # so we can use a bytes[x] in P2 or P3 else: raise Exception("to_dec() requires bytes, hex-ustr or [bin-ustr]") def to_bin(val): """convert bytes, hex-ustr, or int/long to bin-ustr""" if isinstance(val, bytes): # bytes return (bin(to_dec(to_hex(val))))[2:].encode("L1").decode("L1") elif isinstance(val, type("".encode("L1").decode("L1"))): # hex-ustr return (bin(int(val, 16)))[2:].encode("L1").decode("L1") elif isnumber(val): # int/long return (bin(val))[2:].encode("L1").decode("L1") else: raise Exception("to_bin() requires bytes, hex-ustr, or int/long") def to_bytes(val): """converts hex-ustr, int/long, or [bin-ustr] to bytes""" if isinstance(val, list): # [bin-ustr] val = to_hex(val[0]) if isnumber(val): # int/long val = to_hex(val) if isinstance(val, type("".encode("L1").decode("L1"))): # hex-ustr return a2b_hex(bytes(val.encode("L1"))) # works on both P2 and P3 elif isinstance(val, bytes): # so we can use a bytes[x] in P2 or P3 return val else: raise Exception( "to_bytes() requires hex-ustr, int/long, or [bin-ustr]") def touch(filePath, modTime=None): # http://stackoverflow.com/questions/1158076/implement-touch-using-python # print(filePath) with open(to_sys_name(filePath), "ab"): os.utime(filePath, (None if (modTime is None) else (modTime, modTime))) def mkdir(dirPath): try: os.mkdir(to_sys_name(dirPath)) except FileExistsError: pass def makedirs(dirPath): try: os.makedirs(to_sys_name(dirPath)) except OSError as e: if e.errno != errno.EEXIST: raise def loadFile(filePath): with open(to_sys_name(filePath), "rb") as imageHandle: return imageHandle.read() def saveFile(filePath, fileData): with open(to_sys_name(filePath), "wb") as imageHandle: imageHandle.write(fileData) def isnumber(number): try: # make sure it's not a string len(number) return False except TypeError: pass try: int(number) except ValueError: return False return True #---- end IvanX general purpose functions ----# # --- start if __name__ == '__main__': args = sys.argv while True: # breaks when there are no more arguments starting with dash if len(args) == 1: usage() elif args[1][0] != '-': break elif args[1] == '-s': g.afpsync_msg = False args = args[1:] elif args[1] == '-n': g.extract_in_place = True args = args[1:] elif args[1] == '-uc': g.casefold_upper = True args = args[1:] elif args[1] == '-ad': g.use_appledouble = True g.prodos_names = True args = args[1:] elif args[1] == '-shk': g.src_shk = True args = args[1:] elif args[1] == '-pro': g.prodos_names = True args = args[1:] elif args[1] == '-e': g.use_extended = True g.prodos_names = True args = args[1:] elif args[1] == '-cat': g.catalog_only = True args = args[1:] else: usage() if g.use_appledouble and g.use_extended: usage() if g.catalog_only: if len(args) != 2: usage() else: if len(args) not in (3, 4): usage() g.image_file = args[1] if not os.path.isfile(g.image_file): print("Image/archive file \"" + g.image_file + "\" was not found.") quitNow(2) g.image_name = os.path.splitext(os.path.basename(g.image_file)) g.image_ext = g.image_name[1].lower() # automatically set ShrinkIt mode if extension suggests it if g.src_shk or g.image_ext in ('.shk', '.sdk', '.bxy'): if os.name == "nt": print("ShrinkIt archives cannot be extracted on Windows.") quitNow(2) else: try: with open(os.devnull, "w") as fnull: subprocess.call("nulib2", stdout = fnull, stderr = fnull) g.src_shk = True except Exception: print("Nulib2 is not available; not expanding ShrinkIt archive.") quitNow(2) if len(args) == 4: g.extract_file = args[2] if g.extract_file: targetPath = args[3] if os.path.isdir(targetPath): g.target_dir = targetPath elif targetPath.rsplit("/", 1) > 1: g.target_dir = targetPath.rsplit("/", 1)[0] g.target_name = targetPath.rsplit("/", 1)[1] if not os.path.isdir(g.target_dir): print("Target directory not found.") quitNow(2) else: if not g.catalog_only: if not os.path.isdir(args[2]): print("Target directory not found.") quitNow(2) if g.src_shk: g.prodos_names = False if not g.catalog_only: targetDir = (args[3] if g.extract_file else args[2]) unshkdir = ('/tmp' + "/cppo-" + str(uuid.uuid4())) makedirs(unshkdir) result = os.system("/bin/bash -c 'cd " + unshkdir + "; " + "result=$(nulib2 -xse " + os.path.abspath(g.image_file) + ((" " + args[2].replace('/', ':')) if g.extract_file else "") + " 2> /dev/null); " + "if [[ $result == \"Failed.\" ]]; then exit 3; " + "else if grep -q \"no records match\" <<< \"$result\"" + " > /dev/null; then exit 2; else exit 0; fi; fi'") if result == 512: print("File not found in ShrinkIt archive. Try cppo -cat to get the path,") print(" and omit any leading slash or colon.") quitNow(1) elif result != 0: print("ShrinkIt archive is invalid, or some other problem happened.") quitNow(1) if g.extract_file: g.extract_file = g.extract_file.replace(':', '/') extractPath = (unshkdir + "/" + g.extract_file) extractPathDir = os.path.dirname(extractPath) # move the extracted file to the root newunshkdir = ('/tmp' + "/cppo-" + str(uuid.uuid4())) makedirs(newunshkdir) for filename in os.listdir(extractPathDir): shutil.move(extractPathDir + "/" + filename, newunshkdir) shutil.rmtree(unshkdir) unshkdir = newunshkdir fileNames = [name for name in sorted(os.listdir(unshkdir)) if not name.startswith(".")] if g.extract_in_place: # extract in place from "-n" curDir = True elif (len(fileNames) == 1 and os.path.isdir(unshkdir + "/" + fileNames[0])): curDir = True # only one folder at top level, so extract in place volumeName = toProdosName(fileNames[0]) elif (len(fileNames) == 1 and # disk image, so extract in place fileNames[0][-1:] == "i"): curDir = True volumeName = toProdosName(fileNames[0].split("#")[0]) else: # extract in folder based on disk image name curDir = False volumeName = toProdosName(os.path.basename(g.image_file)) if volumeName[-4:].lower() in ('.shk', '.sdk', '.bxy'): volumeName = volumeName[:-4] if not g.catalog_only and not curDir and not g.extract_file: print("Extracting into " + volumeName) # recursively process unshrunk archive hierarchy for dirName, subdirList, fileList in os.walk(unshkdir): subdirList.sort() if not g.catalog_only: g.target_dir = (targetDir + ("" if curDir else ("/" + volumeName)) + ("/" if dirName.count('/') > 2 else "") + ("/".join(dirName.split('/')[3:]))) # chop tempdir if g.extract_file: # solo item, so don't put it in the tree g.target_dir = targetDir if g.casefold_upper: g.target_dir = g.target_dir.upper() g.appledouble_dir = (g.target_dir + "/.AppleDouble") makedirs(g.target_dir) if g.use_appledouble: makedirs(g.appledouble_dir) for fname in sorted(fileList): if fname[-1:] == "i": # disk image; rename to include suffix and correct type/auxtype imagePath = os.path.join(dirName, fname).split("#")[0] new_name = (imagePath + ("" if (imagePath.lower().endswith(".po") or imagePath.lower().endswith(".hdv")) else ".PO") + "#e00005") os.rename(os.path.join(dirName, fname), new_name) fname = os.path.basename(new_name) g.shk_hasrf = False rfork = False if (fname[-1:] == "r" and os.path.isfile(os.path.join(dirName, fname[:-1]))): rfork = True elif (os.path.isfile(os.path.join(dirName, (fname + "r")))): g.shk_hasrf = True if not rfork: processEntry(dirName, fname) shutil.rmtree(unshkdir, True) quitNow(0) # end script if SHK g.image_data = loadFile(g.image_file) # detect if image is 2mg and remove 64-byte header if so if g.image_ext in ('.2mg', '.2img'): g.image_data = g.image_data[64:] # handle 140K disk image if len(g.image_data) == 143360: #print("140K disk") prodos_disk = False fix_order = False # is it ProDOS? if to_hex(readchars(g.image_data, ts(0,0)+0, 4)) == '0138b003': #print("detected ProDOS by boot block") if readchars(g.image_data, ts(0,1)+3, 6) == b'PRODOS': prodos_disk = True #print("order OK (PO)") elif readchars(g.image_data, ts(0,14)+3, 6) == b'PRODOS': #print("order needs fixing (DO)") prodos_disk = True fix_order = True # is it DOS 3.3? else: #print("it's not ProDOS") if readcharDec(g.image_data, ts(17,0)+3) == 3: vtocT = readcharDec(g.image_data, ts(17,0)+1) vtocS = readcharDec(g.image_data, ts(17,0)+2) if vtocT<35 and vtocS<16: #print("it's DOS 3.3") g.dos33 = True # it's DOS 3.3; check sector order next if readcharDec(g.image_data, ts(17,14)+2) != 13: #print("order needs fixing (PO)") fix_order = True #else: print("order OK (DO)") # fall back on disk extension if weird boot block (e.g. AppleCommander) if not prodos_disk and not g.dos33: #print("format and ordering unknown, checking extension") if g.image_ext in ('.dsk', '.do'): fix_order = True #print("extension indicates DO, changing to PO") if fix_order: #print("fixing order") # for each track, # read each sector in the right sequence to make # valid ProDOS blocks (sector pairs) imageDataFixed = bytearray(143360) for t in range(0, 35): for s in [0, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 15]: writechars(imageDataFixed, ts(t,((15-s) if (s%15) else s)), readchars(g.image_data, ts(t,s), 256)) g.image_data = bytes(imageDataFixed) #print("saving fixed order file as outfile.dsk") #saveFile("outfile.dsk", g.image_data) #print("saved") if not prodos_disk and not g.dos33: print("Warning: Unable to determine disk format, assuming ProDOS.") # enforce leading slash if ProDOS if (not g.src_shk and not g.dos33 and g.extract_file and (args[2][0] not in ('/', ':'))): usage() if g.dos33: disk_name = (g.image_name[0] if g.image_ext in ('.dsk', '.do', '.po') else "".join(g.image_name)) if g.prodos_names: disk_name = toProdosName(disk_name) if not g.catalog_only: g.target_dir = (args[3] if g.extract_file else (args[2] + "/" + disk_name)) g.appledouble_dir = (g.target_dir + "/.AppleDouble") makedirs(g.target_dir) if g.use_appledouble: makedirs(g.appledouble_dir) if not g.extract_file: print("Extracting into " + disk_name) processDir([readcharDec(g.image_data, ts(17,0)+1), readcharDec(g.image_data, ts(17,0)+2)]) if g.extract_file: print("ProDOS file not found within image file.") quitNow(0) # below: ProDOS g.activeDirBlock = 0 g.activeFileName = "" g.activeFileSize = 0 g.activeFileBytesCopied = 0 g.resourceFork = 0 g.PDOSPATH_INDEX = 0 g.prodos_names = False if g.extract_file: g.PDOSPATH = g.extract_file.replace(':', '/').split('/') g.extract_file = None if not g.PDOSPATH[0]: g.PDOSPATH_INDEX += 1 g.PDOSPATH_SEGMENT = g.PDOSPATH[g.PDOSPATH_INDEX] g.appledouble_dir = (g.target_dir + "/.AppleDouble") if g.use_appledouble and not os.path.isdir(g.appledouble_dir): mkdir(g.appledouble_dir) processDir(2) print("ProDOS file not found within image file.") quitNow(2) else: if not g.catalog_only: # print(args[0], args[1], args[2]) g.target_dir = (args[2] + "/" + getVolumeName().decode("L1")) g.appledouble_dir = (g.target_dir + "/.AppleDouble") if not os.path.isdir(g.target_dir): makedirs(g.target_dir) if g.use_appledouble and not os.path.isdir(g.appledouble_dir): makedirs(g.appledouble_dir) processDir(2) if not g.catalog_only: quitNow(0)