suid_scan.py

#!/usr/bin/env python

__version__ = '1.1.1'

import argparse
import os
import shlex
import stat
import subprocess
import sys

# Sets the default hash function.
HASHER = '/usr/bin/openssl sha1'


def main(in_file=None, out_file=None, mail_to=None):
    # Get the inventory of everything on the filesystem and then sort it
    # alphabetically.
    filesystem_inventory = inventory_filesystem()
    
    # Cross-reference in_file with inventory, if applicable.
    if in_file:
        print("Cross-referencing found items against {}").format(in_file)
        filesystem_inventory = crossref_inventory(in_file, filesystem_inventory)
    
    # Sort the list for neatness.
    filesystem_inventory = sorted(filesystem_inventory, key=lambda entry: entry[0].lower())
    
    # Print the findings. Tabs are used for separation because they're parsable.
    print("Total dangerous items found: {}").format(len(filesystem_inventory))
    if out_file:
        with open(out_file, "w"):
            pass
    for entry in filesystem_inventory:
        line = '\t'.join(reversed(entry))
        if out_file:
            with open(out_file, "a") as f:
                f.write(line + "\n")
        else:
            print(line)
    if mail_to:
        send_mail(filesystem_inventory, mail_to)


def send_mail(inventory, mail_to):
    """
    Sends an email to the specified recipient containing the list of discovered
    items.
    
    :param inventory: a list of tuples of information
    :param mail_to: the address of the recipient
    """
    # First prep the information.
    body = "Total dangerous items found: {}\n\n".format(len(inventory))
    for entry in inventory:
        body += '\t'.join(reversed(entry))
        body += '\n'
    
    mail_command = [
        '/usr/bin/mail',
        '-s', 'SUID Scan Results',
        mail_to
    ]
    try:
        # Attempt to send the message.
        process = subprocess.Popen(mail_command, stdin=subprocess.PIPE)
        process.communicate(body)
    except Exception as e:
        print(e)
        print("Unable to send mail.")


def crossref_inventory(in_file, inventory):
    """
    Takes the contents of a previous run and cross-references them against the
    new findings, eliminating from the list anything that appears in both
    places.
    
    :param in_file: a file containing a list of files generated by this script
    :param inventory: a list of tuples of file information from this script
    :return: the inventory less the duplicated information
    """
    # Check the file path is valid.
    if not os.path.isfile(in_file):
        raise ValueError("Invalid input file given: {}").format(in_file)
    
    # Pull in all of the lines of the file.
    lines = []
    with open(in_file) as f:
        lines = f.read().splitlines()
    lines = [x for x in lines if x] # remove blanks
    file_inventory = []
    for line in lines:
        hash, mtime, path = line.split('\t')
        file_inventory.append((path, mtime, hash))
    
    ####
    # Iterate over the lines, and match to the items in inventory. Only those
    # items which don't match up will be added to the returned results.
    results = []
    # Find 'file_inventory' - 'inventory'
    for entry in file_inventory:
        should_add = True
        for match in inventory:
            if all(entry[x] == match[x] for x in xrange(3)):
                should_add = False
                break
        if should_add:
            results.append(entry)
    # Find 'inventory' - 'file_inventory'
    for entry in inventory:
        should_add = True
        for match in file_inventory:
            if all(entry[x] == match[x] for x in xrange(3)):
                should_add = False
                break
        if should_add:
            results.append(entry)
    
    return results


def inventory_filesystem():
    """
    Composes an inventory of all attaches filesystems and finds files with the
    SUID or GUID bits set.
    
    :return: a list of tuples containing (file, modification time, hash)
    """
    results = []
    
    ####
    # Get a list of mounted disks.
    mount_info = subprocess.check_output(['/bin/df', '-Pl']).split('\n')
    mount_info = [x for x in mount_info if x][1:]
    mounted_disks = []
    for disk in mount_info:
        disk = "/{}".format(disk.split(' /')[1])
        mounted_disks.append(disk)

    ####
    # Find all the offending files on each disk.
    bad_files_on_disks = {}
    for disk in mounted_disks:
        bad_files_on_disks[disk] = find_bad_files_on_disk(disk)

    ####
    # Iterate over the files found.
    for files in bad_files_on_disks.values():
        for file in files:
            # Ensure we have an absolute path.
            file = os.path.abspath(file)
            # Check file exists.
            if os.path.isfile(file):
                try:
                    mtime = str(int(os.path.getmtime(file)))
                except:
                    mtime = ""
                try:
                    file_hash = get_hash(file)
                except:
                    file_hash = ""
                results.append((file, mtime, file_hash))
    
    return results


def get_hash(file):
    """
    Uses the given hash function to generate a hash of the given file.

    :return: the hash of the file
    """
    if not os.path.isfile(file):
        # It's not a file, which means... it doesn't have a hash. So don't hash
        # it. I thought about checking for specific errors but I don't want to
        # interrupt output.
        return None
    path = os.path.abspath(file)

    # Split the hash function into command form, then add the path.
    hasher = shlex.split(HASHER)
    hasher.append(path)
    if not os.path.isfile(hasher[0]):
        # The given hash function doesn't work.
        raise ValueError("Invalid hash function given: {}").format(haseher[0])

    try:
        # Get the hash of the file.
        info = subprocess.check_output(hasher).split('\n')[0]
        # I assume that the hash will be the last thing outputted on the line.
        file_hash = info.split()[-1]
        return file_hash
    except subprocess.CalledProcessError:
        # Something went wrong.
        return None


def find_bad_files_on_disk(disk=None):
    if not disk:
        disk = '/'
    tm_vol = get_tm_volume()
    if tm_vol:
        if disk == tm_vol:
            return []
    # print("Checking disk '{}'").format(disk)
    find_command = [
        '/usr/bin/find',
        disk,
        '-xdev',
        '-uid', '0',
        '-type', 'f',
        '-perm', '+6000',
    ]
    output = subprocess.check_output(find_command).split('\n')
    output = [x for x in output if x]

    return output


def get_tm_volume():
    """
    Finds the currently-set Time Machine volume, if there is one.

    :return: TM volume mountpoint, or else None if there isn't one
    """
    try:
        tm_dir = subprocess.check_output(['/usr/bin/tmutil', 'machinedirectory'], stderr=subprocess.STDOUT).split('\n')[0]
    except subprocess.CalledProcessError:
        tm_dir = None
    if not tm_dir:
        return None
    
    # Get the filesystem the Time Machine volume is mounted on.
    # (e.g. /dev/disk1s1)
    df = ['/bin/df', '-P', '-k', str(tm_dir)]
    df_info = subprocess.check_output(df).split('\n')
    df_info = [x for x in df_info if x]
    if len(df_info) != 2:
        # Unable to find responsible Time Machine filesystem.
        return None
    info = df_info[1].split()
    index = 0
    for index in range(len(info)):
        try:
            int(info[index])
            break
        except ValueError:
            pass
    fs_id = ' '.join(info[:index])
    
    # Take the filesystem identifier and get the volume name.
    fs_info = subprocess.check_output(['/sbin/mount']).split('\n')
    result = [x for x in fs_info if fs_id == x.split(' on')[0]]
    if len(result) != 1:
        # Something went wrong, but we don't want to halt execution.
        return None

    result = result[0].split('on ')[1]
    return result.split(' (')[0]


def version():
    """
    :return: the version information for this program
    """
    return ("suid_scan, version {version}\n").format(version=__version__)


def usage():
    """
    Prints the usage information for this program.
    """
    print(version())
    
    print('''\
usage: suid_scan [--input IN_FILE] [--output OUT_FILE] [--mailto ADDRESS]
                 [--hash HASHFUNCTION]

Track down any files on the system with the SUID or SGID bits set.

    --help
        Print this help message and quit.
    --version
        Print the version information and quit.

    --input IN_FILE
        Compare results against those in the file IN_FILE.
    --output OUT_FILE
        Write the findings of this scan to OUT_FILE.
    --mailto ADDRESS
        Send an email to ADDRESS (an email address) containing the findings.
        This email will be sent as root, and will contain a list of all the
        files found during the scan (exactly as they would be written to the
        OUT_FILE if you were to use the `--output` option).
            ADDRESS can actually have multiple addresses specified, if they are
            all comma-separated, e.g. "me@test.com,you@test.com".
    --hash HASHFUNCTION
        You can specify your own hashing function for the hashes generated by
        the search. For example, we use Radmind at the University of Utah and so
        we like to use `--hash '/usr/local/bin/fsdiff -1 -c sha1'`. Note that
        the hash function you specify must take the path of the file to be
        hashed as its last argument.
''').format()


if __name__ == '__main__':
    # Gotta have access to the special places.
    if os.geteuid() != 0:
        print("Must be root to run this script!")
        sys.exit(1)
    
    # Parse for command line arguments.
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument('--help', action='store_true')
    parser.add_argument('--version', action='store_true')
    parser.add_argument('--input')
    parser.add_argument('--output')
    parser.add_argument('--mailto')
    parser.add_argument('--hash')
    
    args = parser.parse_args()
    
    if args.help:
        usage()
        sys.exit(0)
    
    if args.version:
        print(version())
        sys.exit(0)
    
    if args.hash:
        HASHER = args.hash
    
    if not os.path.isfile(shlex.split(HASHER)[0]):
        sys.stderr.write("ERROR: Invalid hash function given: {}\n".format(shlex.split(HASHER)[0]))
        sys.stderr.write("    Please specify the full path to the hash function.\n")
        sys.stderr.flush()
        sys.exit(1)
    
    main(args.input, args.output, args.mailto)