PS3 Data Recovery

ps3rec.py

import os
import struct
import sys
import zlib
import argparse
import ctypes


class FileCarver(object):
    def __init__(self, f, offset):
        self.length = 0
        self.extension = ""
        self.name = None
        self._offset = offset
        self._file = f

    def seek(self, offset, whence=0):
        if whence != 1:
            offset += self._offset
        self._file.seek(offset, whence)

    def tell(self):
        return self._file.tell() - self._offset

    def test(self):
        raise NotImplementedError("FileCarver tester not implemented!")

    def parse(self):
        raise NotImplementedError("FileCarver parser not implemented!")

    def read(self, size):
        return self._file.read(size)

    def u8be(self):
        return struct.unpack(">B", self._file.read(1))[0]

    def u16be(self):
        return struct.unpack(">H", self._file.read(2))[0]

    def u32be(self):
        return struct.unpack(">L", self._file.read(4))[0]

    def u64be(self):
        return struct.unpack(">Q", self._file.read(8))[0]

    def floatbe(self):
        return struct.unpack(">f", self._file.read(4))[0]

    def doublebe(self):
        return struct.unpack(">d", self._file.read(8))[0]

    def u8le(self):
        return struct.unpack("<B", self._file.read(1))[0]

    def u16le(self):
        return struct.unpack("<H", self._file.read(2))[0]

    def u32le(self):
        return struct.unpack("<L", self._file.read(4))[0]

    def u64le(self):
        return struct.unpack("<Q", self._file.read(8))[0]

    def floatle(self):
        return struct.unpack("<f", self._file.read(4))[0]

    def doublele(self):
        return struct.unpack("<d", self._file.read(8))[0]

    def read_cstring(self):
        s = []
        while True:
            c = self.read(1)
            if c == chr(0):
                return "".join(s)
            s.append(c)

    def get_file_name(self):
        return "{:016X}{}".format(self._offset, self.extension)

    def recover(self, path):
        file_name = self.get_file_name()
        whole_path = path + '/' + file_name
        with open(whole_path, 'wb') as f:
            if self.length != 0:
                self.seek(0)
                
                bufsize = 0x100000
                remains = self.length
                while remains > 0:
                    read = min(remains, bufsize)
                    buf = self._file.read(read)
                    remains -= read
                    f.write(buf)

    def __str__(self):
        return "{} at {:x} of length {:#x}".format(self.__class__.__name__,
                                                  self._offset,
                                                  self.length)


class SelfCarver(FileCarver):
    def test(self):
        magic = self.read(4)
        return magic == b"SCE\0"

    def parse(self):
        self.seek(0x10)
        self.length = self.u64be() + self.u64be()
        self.extension = ".self"


class ElfCarver(FileCarver):
    def test(self):
        magic = self.read(4)
        return magic == b"\x7FELF"

    def parse(self):
        self.extension = ".elf"


class PUPCarver(FileCarver):
    def test(self):
        magic = self.read(8)
        return magic == b"SCEUF\0\0"

    def parse(self):
        self.extension = ".pup"


class SFOCarver(FileCarver):
    def test(self):
        magic = self.read(4)
        return magic == b"\0PSF"

    def parse(self):
        self.extension = ".sfo"
        self.seek(8)
        key_table_start = self.u32le()
        data_table_start = self.u32le()
        num_ents = self.u32le()

        last_data_offset = 0
        last_data_size = 0
        for x in range(num_ents):
            self.seek(4, 1)
            data_size = self.u32le()
            self.seek(4, 1)
            data_offset = self.u32le()
            if data_offset > last_data_offset:
                last_data_size = data_size
                last_data_offset = data_offset
        self.length = data_table_start + last_data_offset + last_data_size


class RIFFCarver(FileCarver):
    def test(self):
        magic = self.read(4)
        return magic == b"RIFF"

    def parse(self):
        self.extension = ".riff"
        self.seek(4)
        self.length = self.u32le()


class TRPCarver(FileCarver):
    SIZEOF_HEADER = 0x40

    def test(self):
        magic = self.u32be()
        return magic == 0xDCA24D00

    def parse(self):
        self.extension = ".trp"
        self.seek(8)
        self.length = self.u64be() + TRPCarver.SIZEOF_HEADER

class PNGCarver(FileCarver):
    def test(self):
        magic = self.read(8)
        if magic == b'\x89\x50\x4E\x47\x0D\x0A\x1A\x0A':
            return True
        return False

    def parse(self):
        self.extension = ".png"
        self.seek(8)
        while True:
            length = self.u32be()
            ctype = self.read(4)
            cdata = self.read(length)
            ccrc = self.u32be()
            crc = zlib.crc32(ctype + cdata) & 0xffffffff
            if crc != ccrc:
                print("ERROR: Invalid CRC!"),
                break
            if ctype == "IEND":
                break
        self.length = self.tell()

class BIKCarver(FileCarver):
    def test(self):
        magic = self.read(3)
        return magic == b'BIK'

    def parse(self):
        self.extension = ".bik"
        self.seek(4)
        self.length = self.u32le() + 8

class NPDCarver(FileCarver):
    def test(self):
        magic = self.read(4)
        return magic == b'NPD\0'

    def parse(self):
        self.seek(0x90)
        npdtype = self.u8be()
        if (npdtype & 0xf) == 0x0:
            self.extension = ".edat"
        elif (npdtype & 0xf) == 0x1:
            self.extension = ".sdat"
        else:
            self.extension = ".npd"
        self.seek(0x98)
        datalen = self.u64be()


all_carvers = [a_carver for a_carver in FileCarver.__subclasses__()]


class FileScanner:
    def __init__(self, inpath, outpath, carvers, interval=0x200):
        self.file = open(inpath, "rb")
        self.outpath = outpath
        self.carvers = carvers
        self.interval = interval
        self.carvings = []

    def run(self, start=0, end=0, interval=0x800):
        if end == 0:
            self.file.seek(0, 2)
            end = self.file.tell()
            self.file.seek(0, 0)

        for offset in range(start, end, interval):
            for carver in self.carvers:
                tester = carver(self.file, offset)
                self.file.seek(offset)
                if tester.test():
                    self.file.seek(offset)
                    print("Parsing {}...".format(tester.__class__.__name__)),
                    tester.parse()
                    self.carvings.append(tester)
                    print(str(tester))

            # Print progress
            if (offset & 0xfffffff) == 0:
                print('{:016x}'.format(offset))

    def dump(self):
        if not os.path.exists(self.outpath):
            os.makedirs(self.outpath)

        for carving in self.carvings:
            carving.recover(self.outpath)

    def close(self):
        self.file.close()

        
"""
struct    direct {
    u_int32_t d_ino;        /* inode number of entry */
    u_int16_t d_reclen;        /* length of this record */
    u_int8_t  d_type;         /* file type, see below */
    u_int8_t  d_namlen;        /* length of string in d_name */
    char      d_name[MAXNAMLEN + 1];/* name with length <= MAXNAMLEN */
};
"""
class Direct(ctypes.BigEndianStructure):
    _fields_ = [
        ("ino", ctypes.c_uint32),       # 0x00
        ("reclen", ctypes.c_uint16),    # 0x04
        ("type", ctypes.c_uint8),       # 0x06
        ("namlen", ctypes.c_uint8)      # 0x07
    ]
      
        
class DirectFinder(object):
    def __init__(self, inpath, start=None, end=None):
        self.file = open(inpath, "rb")
        self.log = open("log.txt", "w")
        self.file.seek(0, 2)
        length = self.file.tell()
        self.file.seek(0, 0)

        def align(offset):
            return (offset) & ~(0x800 - 1)

        if start is None:
            self.start = 0
        else:
            if start < 0 or start > length:
                raise ValueError(
                    "Invalid start parameter: must be between 0 and {}"
                    .format(length))
            self.start = align(start)

        if end is None:
            self.end = length
        else:
            if end < self.start or end > length:
                raise ValueError(
                    "Invalid end parameter: must be between {} and {}"
                    .format(start, length))
            self.end = align(end)
        

    def run(self):
        for offset in range(self.start, self.end, 0x800):
            #print "{:x}".format(offset)
            self.file.seek(offset)
            direct1 = Direct()
            self.file.readinto(direct1)
            name = self.file.read(direct1.namlen)
            if direct1.type == 0x4 \
                and direct1.namlen == 0x1 \
                    and name == b".":
                self.file.seek(offset + direct1.reclen)
                direct2 = Direct()
                self.file.readinto(direct2)
                name = self.file.read(direct2.namlen)
                if direct2.type == 0x4 \
                    and direct2.namlen == 0x2 \
                        and name == b"..":
                    self.print_directs(offset)
                
    def u32be(self):
        return struct.unpack(">L", self.file.read(4))[0]
    
    def u16be(self):
        return struct.unpack(">H", self.file.read(2))[0]
        
    def u8be(self):
        return struct.unpack(">B", self.file.read(1))[0]
                
    file_types = {
        0: "DT_UNKNOWN",
        1: "DT_FIFO",
        2: "DT_CHR",
        4: "DT_DIR",
        6: "DT_BLK",
        8: "DT_REG",
        10: "DT_LNK",
        12: "DT_SOCK",
        14: "DT_WHT"
    }
        
    def print_directs(self, offset):
        msg = "\n{:016X}".format(offset)
        print(msg)
        self.log.write(msg + "\n")
        while True:
            self.file.seek(offset)

            dino = self.u32be()
            dreclen = self.u16be()
            dtype = self.u8be()
            dnamlen = self.u8be()
            
            # Max file name length is 255
            if dreclen == 0 or dreclen >= 0x200 or dnamlen > 255:
                #print "End of direct stream"
                return

            if dtype not in (0,1,2,3,4,6,8,10,12,14):
                #print("Invalid file type!")
                return

            try:
                dname = self.file.read(dnamlen).decode('ascii')
            except:
                return

            msg = " +- Offset: {:016X}\n".format(offset)
            msg += " |  d_ino: {}\n".format(dino)
            msg += " |  d_reclen: {}\n".format(dreclen)
            msg += " |  d_type: {} ({})\n".format(dtype, self.file_types[dtype])
            msg += " |  d_namlen: {}\n".format(dnamlen)
            msg += " |  d_name: {}".format(dname)
            
            print(msg)
            self.log.write(msg + "\n")
            self.log.flush()
                      
            offset += dreclen


def run_meta_command(args):
    scanner = DirectFinder(args.input)
    scanner.run()


def run_carve_command(args):
    scanner = FileScanner(args.input, args.output, all_carvers)
    scanner.run()
    scanner.dump()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="PS3 Recovery Tool")

    subparser = parser.add_subparsers()
    subparser.required = True
    
    m_parser = subparser.add_parser('meta', help="Scan for metadata")
    m_parser.add_argument('input', help="Input file", type=str)
    m_parser.set_defaults(func=run_meta_command)

    c_parser = subparser.add_parser('carve', help="Run file carver")
    c_parser.add_argument('input', help="Input file", type=str)
    c_parser.add_argument('output', help="Output directory", type=str)
    c_parser.set_defaults(func=run_carve_command)

    args = parser.parse_args()
    args.func(args)

Usage

DIRENT scanner

Find dirent's (they contain file names) in an image.

python ps3rec.py meta <input image>

File carver

Find carvers based on their file formats.

python ps3rec.py carve <input image> <output directory>

Features to implement

• Pattern based carver. May require use of pattern matching strings (regex/wildcards).

class PatternCarver(FileCarver):
  start_pattern = "MAGIC"
  end_pattern = "EOF"

• Add find(pattern, max_distance) method to FileCarver
• Organize carvers into separate modules. This will require moving to an actual repository.
• Be able to select which carvers to use through use of some config file.
• Convert carvers to static classes to avoid excessive object creation.