magic.py

# Found on a russian zope mailing list, and modified to fix bugs in parsing
# the magic file and string making
# -- Daniel Berlin <dberlin@dberlin.org>
import sys, struct, time, re, exceptions, pprint, stat, os, pwd, grp

_mew = 0

# _magic='/tmp/magic'
# _magic='/usr/share/magic.mime'
_magic='/usr/share/magic.mime'
mime = 1

_ldate_adjust = lambda x: time.mktime( time.gmtime(x) )

BUFFER_SIZE = 1024 * 128 # 128K should be enough...

class MagicError(exceptions.Exception): pass

def _handle(fmt='@x',adj=None): return fmt, struct.calcsize(fmt), adj

KnownTypes = {
        # 'byte':_handle('@b'),
        'byte':_handle('@B'),
        'ubyte':_handle('@B'),

        'string':('s',0,None),
        'pstring':_handle('p'),

#       'short':_handle('@h'),
#       'beshort':_handle('>h'),
#       'leshort':_handle('<h'),
        'short':_handle('@H'),
        'beshort':_handle('>H'),
        'leshort':_handle('<H'),
        'ushort':_handle('@H'),
        'ubeshort':_handle('>H'),
        'uleshort':_handle('<H'),

        'long':_handle('@l'),
        'belong':_handle('>l'),
        'lelong':_handle('<l'),
        'ulong':_handle('@L'),
        'ubelong':_handle('>L'),
        'ulelong':_handle('<L'),

        'date':_handle('=l'),
        'bedate':_handle('>l'),
        'ledate':_handle('<l'),
        'ldate':_handle('=l',_ldate_adjust),
        'beldate':_handle('>l',_ldate_adjust),
        'leldate':_handle('<l',_ldate_adjust),
}

_mew_cnt = 0
def mew(x):
    global _mew_cnt
    if _mew :
        if x=='.' :
            _mew_cnt += 1
            if _mew_cnt % 64 == 0 : sys.stderr.write( '\n' )
            sys.stderr.write( '.' )
        else:
            sys.stderr.write( '\b'+x )

def has_format(s):
    n = 0
    l = None
    for c in s :
        if c == '%' :
            if l == '%' : n -= 1
            else        : n += 1
        l = c
    return n

def read_asciiz(file,size=None,pos=None):
    s = []
    if pos :
        mew('s')
        file.seek( pos, 0 )
    mew('z')
    if size is not None :
        s = [file.read( size ).split('\0')[0]]
    else:
        while 1 :
            c = file.read(1)
            if (not c) or (ord(c)==0) or (c=='\n') : break
            s.append (c)
    mew('Z')
    return ''.join(s)

def a2i(v,base=0):
    if v[-1:] in 'lL' : v = v[:-1]
    return int( v, base )

_cmap = {
        '\\' : '\\',
        '0' : '\0',
}
for c in range(ord('a'),ord('z')+1) :
    try               : e = eval('"\\%c"' % chr(c))
    except ValueError : pass
    else              : _cmap[chr(c)] = e
else:
    del c
    del e

def make_string(s):
    return eval( '"'+s.replace('"','\\"')+'"')

class MagicTestError(MagicError): pass

class MagicTest:
    def __init__(self,offset,mtype,test,message,line=None,level=None):
        self.line, self.level = line, level
        self.mtype = mtype
        self.mtest = test
        self.subtests = []
        self.mask = None
        self.smod = None
        self.nmod = None
        self.offset, self.type, self.test, self.message = \
                        offset,mtype,test,message
        if self.mtype == 'true' : return # XXX hack to enable level skips
        if test[-1:]=='\\' and test[-2:]!='\\\\' :
            self.test += 'n' # looks like someone wanted EOL to match?
        if mtype[:6]=='string' :
            if '/' in mtype : # for strings
                self.type, self.smod = \
                                        mtype[:mtype.find('/')], mtype[mtype.find('/')+1:]
        else:
            for nm in '&+-' :
                if nm in mtype : # for integer-based
                    self.nmod, self.type, self.mask = (
                            nm,
                            mtype[:mtype.find(nm)],
                            # convert mask to int, autodetect base
                            int( mtype[mtype.find(nm)+1:], 0 )
                    )
                    break
        self.struct, self.size, self.cast = KnownTypes[ self.type ]
    def __str__(self):
        return '%s %s %s %s' % (
                self.offset, self.mtype, self.mtest, self.message
        )
    def __repr__(self):
        return 'MagicTest(%s,%s,%s,%s,line=%s,level=%s,subtests=\n%s%s)' % (
                `self.offset`, `self.mtype`, `self.mtest`, `self.message`,
                `self.line`, `self.level`,
                '\t'*self.level, pprint.pformat(self.subtests)
        )
    def run(self,file):
        result = ''
        do_close = 0
        try:
            if type(file) == type('x') :
                file = open( file, 'r', BUFFER_SIZE )
                do_close = 1
#                       else:
#                               saved_pos = file.tell()
            if self.mtype != 'true' :
                data = self.read(file)
                last = file.tell()
            else:
                data = last = None
            if self.check( data ) :
                result = self.message+' '
                if has_format( result ) : result %= data
                for test in self.subtests :
                    m = test.run(file)
                    if m is not None : result += m
                return make_string( result )
        finally:
            if do_close :
                file.close()
#                       else:
#                               file.seek( saved_pos, 0 )
    def get_mod_and_value(self):
        if self.type[-6:] == 'string' :
            # "something like\tthis\n"
            if self.test[0] in '=<>' :
                mod, value = self.test[0], make_string( self.test[1:] )
            else:
                mod, value = '=', make_string( self.test )
        else:
            if self.test[0] in '=<>&^' :
                mod, value = self.test[0], a2i(self.test[1:])
            elif self.test[0] == 'x':
                mod = self.test[0]
                value = 0
            else:
                mod, value = '=', a2i(self.test)
        return mod, value
    def read(self,file):
        mew( 's' )
        file.seek( self.offset(file), 0 ) # SEEK_SET
        mew( 'r' )
        try:
            data = rdata = None
            # XXX self.size might be 0 here...
            if self.size == 0 :
                # this is an ASCIIZ string...
                size = None
                if self.test != '>\\0' : # magic's hack for string read...
                    value = self.get_mod_and_value()[1]
                    size = (value=='\0') and None or len(value)
                rdata = data = read_asciiz( file, size=size )
            else:
                rdata = file.read( self.size )
                if not rdata or (len(rdata)!=self.size) : return None
                data = struct.unpack( self.struct, rdata )[0] # XXX hack??
        except:
            print >>sys.stderr, self
            print >>sys.stderr, '@%s struct=%s size=%d rdata=%s' % (
                    self.offset, `self.struct`, self.size,`rdata`)
            raise
        mew( 'R' )
        if self.cast : data = self.cast( data )
        if self.mask :
            try:
                if   self.nmod == '&' : data &= self.mask
                elif self.nmod == '+' : data += self.mask
                elif self.nmod == '-' : data -= self.mask
                else: raise MagicTestError(self.nmod)
            except:
                print >>sys.stderr,'data=%s nmod=%s mask=%s' % (
                        `data`, `self.nmod`, `self.mask`
                )
                raise
        return data
    def check(self,data):
        mew('.')
        if self.mtype == 'true' :
            return '' # not None !
        mod, value = self.get_mod_and_value()
        if self.type[-6:] == 'string' :
            # "something like\tthis\n"
            if self.smod :
                xdata = data
                if 'b' in self.smod : # all blanks are optional
                    xdata = ''.join( data.split() )
                    value = ''.join( value.split() )
                if 'c' in self.smod : # all blanks are optional
                    xdata = xdata.upper()
                    value = value.upper()
            # if 'B' in self.smod : # compact blanks
            ### XXX sorry, i don't understand this :-(
            #       data = ' '.join( data.split() )
            #       if ' ' not in data : return None
            else:
                xdata = data
        try:
            if   mod == '=' : result = data == value
            elif mod == '<' : result = data < value
            elif mod == '>' : result = data > value
            elif mod == '&' : result = data & value
            elif mod == '^' : result = (data & (~value)) == 0
            elif mod == 'x' : result = 1
            else            : raise MagicTestError(self.test)
            if result :
                zdata, zval = `data`, `value`
                if self.mtype[-6:]!='string' :
                    try: zdata, zval = hex(data), hex(value)
                    except: zdata, zval = `data`, `value`
                if 0 : print >>sys.stderr, '%s @%s %s:%s %s %s => %s (%s)' % (
                        '>'*self.level, self.offset,
                        zdata, self.mtype, `mod`, zval, `result`,
                        self.message
                )
            return result
        except:
            print >>sys.stderr,'mtype=%s data=%s mod=%s value=%s' % (
                    `self.mtype`, `data`, `mod`, `value`
            )
            raise
    def add(self,mt):
        if not isinstance(mt,MagicTest) :
            raise MagicTestError((mt,'incorrect subtest type %s'%(type(mt),)))
        if mt.level == self.level+1 :
            self.subtests.append( mt )
        elif self.subtests :
            self.subtests[-1].add( mt )
        elif mt.level > self.level+1 :
            # it's possible to get level 3 just after level 1 !!! :-(
            level = self.level + 1
            while level < mt.level :
                xmt = MagicTest(None,'true','x','',line=self.line,level=level)
                self.add( xmt )
                level += 1
            else:
                self.add( mt ) # retry...
        else:
            raise MagicTestError((mt,'incorrect subtest level %s'%(`mt.level`,)))
    def last_test(self):
        return self.subtests[-1]
#end class MagicTest

class OffsetError(MagicError): pass

class Offset:
    pos_format = {'b':'<B','B':'>B','s':'<H','S':'>H','l':'<I','L':'>I',}
    pattern0 = re.compile(r'''    # mere offset
                ^
                &?                                          # possible ampersand
                (       0                                       # just zero
                |       [1-9]{1,1}[0-9]*        # decimal
                |       0[0-7]+                         # octal
                |       0x[0-9a-f]+                     # hex
                )
                $
                ''', re.X|re.I
    )
    pattern1 = re.compile(r'''    # indirect offset
                ^\(
                (?P<base>&?0                  # just zero
                        |&?[1-9]{1,1}[0-9]* # decimal
                        |&?0[0-7]*          # octal
                        |&?0x[0-9A-F]+      # hex
                )
                (?P<type>
                        \.         # this dot might be alone
                        [BSL]? # one of this chars in either case
                )?
                (?P<sign>
                        [-+]{0,1}
                )?
                (?P<off>0              # just zero
                        |[1-9]{1,1}[0-9]*  # decimal
                        |0[0-7]*           # octal
                        |0x[0-9a-f]+       # hex
                )?
                \)$''', re.X|re.I
    )
    def __init__(self,s):
        self.source = s
        self.value  = None
        self.relative = 0
        self.base = self.type = self.sign = self.offs = None
        m = Offset.pattern0.match( s )
        if m : # just a number
            if s[0] == '&' :
                self.relative, self.value = 1, int( s[1:], 0 )
            else:
                self.value = int( s, 0 )
            return
        m = Offset.pattern1.match( s )
        if m : # real indirect offset
            try:
                self.base = m.group('base')
                if self.base[0] == '&' :
                    self.relative, self.base = 1, int( self.base[1:], 0 )
                else:
                    self.base = int( self.base, 0 )
                if m.group('type') : self.type = m.group('type')[1:]
                self.sign = m.group('sign')
                if m.group('off') : self.offs = int( m.group('off'), 0 )
                if self.sign == '-' : self.offs = 0 - self.offs
            except:
                print >>sys.stderr, '$$', m.groupdict()
                raise
            return
        raise OffsetError(`s`)
    def __call__(self,file=None):
        if self.value is not None : return self.value
        pos = file.tell()
        try:
            if not self.relative : file.seek( self.offset, 0 )
            frmt = Offset.pos_format.get( self.type, 'I' )
            size = struct.calcsize( frmt )
            data = struct.unpack( frmt, file.read( size ) )
            if self.offs : data += self.offs
            return data
        finally:
            file.seek( pos, 0 )
    def __str__(self): return self.source
    def __repr__(self): return 'Offset(%s)' % `self.source`
#end class Offset

class MagicFileError(MagicError): pass

class MagicFile:
    def __init__(self,filename=_magic):
        self.file = None
        self.tests = []
        self.total_tests = 0
        self.load( filename )
        self.ack_tests = None
        self.nak_tests = None
    def __del__(self):
        self.close()
    def load(self,filename=None):
        self.open( filename )
        self.parse()
        self.close()
    def open(self,filename=None):
        self.close()
        if filename is not None :
            self.filename = filename
        self.file = open( self.filename, 'r', BUFFER_SIZE )
    def close(self):
        if self.file :
            self.file.close()
            self.file = None
    def parse(self):
        line_no = 0
        for line in self.file.xreadlines() :
            line_no += 1
            if not line or line[0]=='#' : continue
            line = line.lstrip().rstrip('\r\n')
            if not line or line[0]=='#' : continue
            try:
                x = self.parse_line( line )
                if x is None :
                    print >>sys.stderr, '#[%04d]#'%line_no, line
                    continue
            except:
                print >>sys.stderr, '###[%04d]###'%line_no, line
                raise
            self.total_tests += 1
            level, offset, mtype, test, message = x
            new_test = MagicTest(offset,mtype,test,message,
                    line=line_no,level=level)
            try:
                if level == 0 :
                    self.tests.append( new_test )
                else:
                    self.tests[-1].add( new_test )
            except:
                if 1 :
                    print >>sys.stderr, 'total tests=%s' % (
                            `self.total_tests`,
                    )
                    print >>sys.stderr, 'level=%s' % (
                            `level`,
                    )
                    print >>sys.stderr, 'tests=%s' % (
                            pprint.pformat(self.tests),
                    )
                raise
        else:
            while self.tests[-1].level > 0 :
                self.tests.pop()
    def parse_line(self,line):
        # print >>sys.stderr, 'line=[%s]' % line
        if (not line) or line[0]=='#' : return None
        level = 0
        offset = mtype = test = message = ''
        mask = None
        # get optional level (count leading '>')
        while line and line[0]=='>' :
            line, level = line[1:], level+1
        # get offset
        while line and not line[0].isspace() :
            offset, line = offset+line[0], line[1:]
        try:
            offset = Offset(offset)
        except:
            print >>sys.stderr, 'line=[%s]' % line
            raise
        # skip spaces
        line = line.lstrip()
        # get type
        c = None
        while line :
            last_c, c, line = c, line[0], line[1:]
            if last_c!='\\' and c.isspace() :
                break # unescaped space - end of field
            else:
                mtype += c
                if last_c == '\\' :
                    c = None # don't fuck my brain with sequential backslashes
        # skip spaces
        line = line.lstrip()
        # get test
        c = None
        while line :
            last_c, c, line = c, line[0], line[1:]
            if last_c!='\\' and c.isspace() :
                break # unescaped space - end of field
            else:
                test += c
                if last_c == '\\' :
                    c = None # don't fuck my brain with sequential backslashes
        # skip spaces
        line = line.lstrip()
        # get message
        message = line
        if mime and line.find("\t") != -1:
            message=line[0:line.find("\t")]
        #
        # print '>>', level, offset, mtype, test, message
        return level, offset, mtype, test, message
    def detect(self,file):
        self.ack_tests = 0
        self.nak_tests = 0
        answers = []
        for test in self.tests :
            message = test.run( file )
            if message :
                self.ack_tests += 1
                answers.append( message )
            else:
                self.nak_tests += 1
        if answers :
            return '; '.join( answers )
#end class MagicFile

def username(uid):
    try:
        return pwd.getpwuid( uid )[0]
    except:
        return '#%s'%uid

def groupname(gid):
    try:
        return grp.getgrgid( gid )[0]
    except:
        return '#%s'%gid

def get_file_type(fname,follow):
    t = None
    if not follow :
        try:
            st = os.lstat( fname ) # stat that entry, don't follow links!
        except os.error, why :
            pass
        else:
            if stat.S_ISLNK(st[stat.ST_MODE]) :
                t = 'symbolic link'
                try:
                    lnk = os.readlink( fname )
                except:
                    t += ' (unreadable)'
                else:
                    t += ' to '+lnk
    if t is None :
        try:
            st = os.stat( fname )
        except os.error, why :
            return "can't stat `%s' (%s)." % (why.filename,why.strerror)

    dmaj, dmin = (st.st_rdev>>8)&0x0FF, st.st_rdev&0x0FF

    if 0 : pass
    elif stat.S_ISSOCK(st.st_mode) : t = 'socket'
    elif stat.S_ISLNK (st.st_mode) : t = follow and 'symbolic link' or t
    elif stat.S_ISREG (st.st_mode) : t = 'file'
    elif stat.S_ISBLK (st.st_mode) : t = 'block special (%d/%d)'%(dmaj,dmin)
    elif stat.S_ISDIR (st.st_mode) : t = 'directory'
    elif stat.S_ISCHR (st.st_mode) : t = 'character special (%d/%d)'%(dmaj,dmin)
    elif stat.S_ISFIFO(st.st_mode) : t = 'pipe'
    else: t = '<unknown>'

    if st.st_mode & stat.S_ISUID :
        t = 'setuid(%d=%s) %s'%(st.st_uid,username(st.st_uid),t)
    if st.st_mode & stat.S_ISGID :
        t = 'setgid(%d=%s) %s'%(st.st_gid,groupname(st.st_gid),t)
    if st.st_mode & stat.S_ISVTX :
        t = 'sticky '+t

    return t

HELP = '''%s [options] [files...]

Options:

        -?, --help -- this help
        -m, --magic=<file> -- use this magic <file> instead of %s
        -f, --files=<namefile> -- read filenames for <namefile>
*       -C, --compile -- write "compiled" magic file
        -b, --brief -- don't prepend filenames to output lines
+       -c, --check -- check the magic file
        -i, --mime -- output MIME types
*       -k, --keep-going -- don't stop st the first match
        -n, --flush -- flush stdout after each line
        -v, --verson -- print version and exit
*       -z, --compressed -- try to look inside compressed files
        -L, --follow -- follow symlinks
        -s, --special -- don't skip special files

*       -- not implemented so far ;-)
+       -- implemented, but in another way...
'''

def main():
    import getopt
    global _magic
    try:
        brief = 0
        flush = 0
        follow= 0
        mime  = 0
        check = 0
        special=0
        try:
            opts, args = getopt.getopt(
                    sys.argv[1:],
                    '?m:f:CbciknvzLs',
                    (       'help',
                            'magic=',
                            'names=',
                            'compile',
                            'brief',
                            'check',
                            'mime',
                            'keep-going',
                            'flush',
                            'version',
                            'compressed',
                            'follow',
                            'special',
                    )
            )
        except getopt.error, why:
            print >>sys.stderr, sys.argv[0], why
            return 1
        else:
            files = None
            for o,v in opts :
                if o in ('-?','--help'):
                    print HELP % (
                            sys.argv[0],
                            _magic,
                    )
                    return 0
                elif o in ('-f','--files='):
                    files = v
                elif o in ('-m','--magic='):
                    _magic = v[:]
                elif o in ('-C','--compile'):
                    pass
                elif o in ('-b','--brief'):
                    brief = 1
                elif o in ('-c','--check'):
                    check = 1
                elif o in ('-i','--mime'):
                    mime = 1
                    if os.path.exists( _magic+'.mime' ) :
                        _magic += '.mime'
                        print >>sys.stderr,sys.argv[0]+':',\
                                                        "Using regular magic file `%s'" % _magic
                elif o in ('-k','--keep-going'):
                    pass
                elif o in ('-n','--flush'):
                    flush = 1
                elif o in ('-v','--version'):
                    print 'VERSION'
                    return 0
                elif o in ('-z','--compressed'):
                    pass
                elif o in ('-L','--follow'):
                    follow = 1
                elif o in ('-s','--special'):
                    special = 1
            else:
                if files :
                    files = map(lambda x: x.strip(), v.split(','))
                    if '-' in files and '-' in args :
                        error( 1, 'cannot use STDIN simultaneously for file list and data' )
                    for file in files :
                        for name in (
                                        (file=='-')
                                                and sys.stdin
                                                or open(file,'r',BUFFER_SIZE)
                        ).xreadlines():
                            name = name.strip()
                            if name not in args :
                                args.append( name )
        try:
            if check : print >>sys.stderr, 'Loading magic database...'
            t0 = time.time()
            m = MagicFile(_magic)
            t1 = time.time()
            if check :
                print >>sys.stderr, \
                                        m.total_tests, 'tests loaded', \
                                        'for', '%.2f' % (t1-t0), 'seconds'
                print >>sys.stderr, len(m.tests), 'tests at top level'
                return 0 # XXX "shortened" form ;-)

            mlen = max( map(len, args) )+1
            for arg in args :
                if not brief : print (arg + ':').ljust(mlen),
                ftype = get_file_type( arg, follow )
                if (special and ftype.find('special')>=0) \
                                or ftype[-4:] == 'file' :
                    t0 = time.time()
                    try:
                        t = m.detect( arg )
                    except (IOError,os.error), why:
                        t = "can't read `%s' (%s)" % (why.filename,why.strerror)
                    if ftype[-4:] == 'file' : t = ftype[:-4] + t
                    t1 = time.time()
                    print t and t or 'data'
                    if 0 : print \
                                                        '#\t%d tests ok, %d tests failed for %.2f seconds'%\
                                                        (m.ack_tests, m.nak_tests, t1-t0)
                else:
                    print mime and 'application/x-not-regular-file' or ftype
                if flush : sys.stdout.flush()
        # print >>sys.stderr, 'DONE'
        except:
            if check : return 1
            raise
        else:
            return 0
    finally:
        pass

if __name__ == '__main__' :
    sys.exit( main() )
# vim:ai
# EOF #