#!/usr/bin/env python3
# Copyright (c) 2018 Egor Tensin <Egor.Tensin@gmail.com>
# This file is part of the "audit-scripts" project.
# For details, see https://github.com/egor-tensin/audit-scripts.
# Distributed under the MIT License.
import argparse
import contextlib
import grp
import logging
import logging.config
import logging.handlers
import multiprocessing as mp
import os
import pwd
import sys
def console_log_formatter():
fmt = '%(asctime)s | %(processName)s | %(levelname)s | %(message)s'
datefmt = '%Y-%m-%d %H:%M:%S'
return logging.Formatter(fmt=fmt, datefmt=datefmt)
def console_log_level():
return logging.DEBUG
def console_log_handler():
handler = logging.StreamHandler()
handler.setFormatter(console_log_formatter())
handler.setLevel(console_log_level())
return handler
@contextlib.contextmanager
def launch_logging_thread(queue):
process = mp.current_process()
process.name = 'scandir'
listener = logging.handlers.QueueListener(queue, console_log_handler())
listener.start()
try:
yield
finally:
listener.stop()
@contextlib.contextmanager
def setup_logging(queue):
config = {
'version': 1,
'handlers': {
'sink': {
'class': 'logging.handlers.QueueHandler',
'queue': queue,
},
},
'root': {
'handlers': ['sink'],
'level': 'DEBUG',
},
}
logging.config.dictConfig(config)
try:
yield
except Exception as e:
logging.exception(e)
def validate_uid(uid):
try:
return pwd.getpwuid(uid).pw_uid
except KeyError:
return None
def validate_gid(gid):
try:
return grp.getgrgid(gid).gr_gid
except KeyError:
return None
def map_user_name(user_name):
try:
return pwd.getpwnam(user_name).pw_uid
except KeyError:
return None
def map_group_name(group_name):
try:
return grp.getgrnam(group_name).gr_gid
except KeyError:
return None
def parse_user_name(src):
uid = map_user_name(src)
if uid is None:
raise argparse.ArgumentTypeError('unknown user name: {}'.format(src))
return uid
def parse_group_name(src):
gid = map_group_name(src)
if gid is None:
raise argparse.ArgumentTypeError('unknown group name: {}'.format(src))
return gid
def parse_uid(src):
try:
uid = int(src)
except ValueError:
return parse_user_name(src)
uid = validate_uid(uid)
if uid is None:
raise argparse.ArgumentTypeError('unknown UID: {}'.format(src))
return uid
def parse_gid(src):
try:
gid = int(src)
except ValueError:
return parse_group_name(src)
gid = validate_gid(gid)
if gid is None:
raise argparse.ArgumentTypeError('unknown GID: {}'.format(src))
return gid
def get_primary_gid(uid):
return pwd.getpwuid(uid).pw_gid
def parse_args(argv=None):
if argv is None:
argv = sys.argv[1:]
parser = argparse.ArgumentParser()
parser.add_argument('root_dir', default='/', nargs='?', metavar='DIR',
help='set root directory')
parser.add_argument('--user', '-u', dest='uid', required=True,
metavar='USER', type=parse_uid,
help='set new process\' UID')
parser.add_argument('--group', '-g', dest='gid',
metavar='GROUP', type=parse_gid,
help='set new process\' GID')
args = parser.parse_args(argv)
if args.gid is None:
args.gid = get_primary_gid(args.uid)
return args
def dump_process_info():
ruid, euid, suid = os.getresuid()
logging.info('User IDs:')
logging.info('\tReal: %d', ruid)
logging.info('\tEffective: %d', euid)
logging.info('\tSaved: %d', suid)
rgid, egid, sgid = os.getresgid()
logging.info('Group IDs:')
logging.info('\tReal: %d', rgid)
logging.info('\tEffective: %d', egid)
logging.info('\tSaved: %d', sgid)
def check_root():
if os.getuid() == 0:
return True
logging.error('Must be run as root')
return False
def scandir(dir_path):
try:
entry_it = os.scandir(dir_path)
except (PermissionError, FileNotFoundError) as e:
logging.warning('%s', e)
return
with entry_it:
yield from entry_it
def enum_dirs(dir_path):
for entry in scandir(dir_path):
if entry.is_dir(follow_symlinks=False):
yield entry.path
def is_writable_via_access(dir_path):
return os.access(dir_path, os.W_OK | os.X_OK)
def is_writable(dir_path):
return is_writable_via_access(dir_path)
def drop_privileges(uid, gid):
if not check_root():
return
os.setgroups([])
os.setgid(gid)
os.setuid(uid)
os.umask(0o77)
@contextlib.contextmanager
def close_queue(queue):
try:
yield
finally:
queue.put(None)
def access_loop(access_queue, scandir_queue):
for dir_list in iter(access_queue.get, None):
denied_dir_list = []
for dir_path in dir_list:
if is_writable(dir_path):
logging.info('Writable: %s', dir_path)
else:
denied_dir_list.append(dir_path)
if not denied_dir_list:
break
scandir_queue.put(denied_dir_list)
def access_main(args, log_queue, access_queue, scandir_queue):
with close_queue(scandir_queue), setup_logging(log_queue):
drop_privileges(args.uid, args.gid)
dump_process_info()
access_loop(access_queue, scandir_queue)
def scandir_loop(access_queue, scandir_queue):
for parent_dir_list in iter(scandir_queue.get, None):
child_dir_list = [child_dir
for parent_dir in parent_dir_list
for child_dir in enum_dirs(parent_dir)]
if not child_dir_list:
break
access_queue.put(child_dir_list)
def scandir_main(access_queue, scandir_queue):
with close_queue(access_queue):
if not check_root():
return
dump_process_info()
scandir_loop(access_queue, scandir_queue)
def main(argv=None):
log_queue = mp.Queue()
with launch_logging_thread(log_queue), setup_logging(log_queue):
prog_args = parse_args(argv)
access_queue = mp.SimpleQueue()
scandir_queue = mp.SimpleQueue()
access_process_args = prog_args, log_queue, access_queue, scandir_queue
access_process = mp.Process(target=access_main,
args=access_process_args,
name='access')
access_queue.put([prog_args.root_dir])
access_process.start()
scandir_main(access_queue, scandir_queue)
access_process.join()
if __name__ == '__main__':
mp.set_start_method('spawn')
main()
