#!/usr/bin/python3 import uuid import random import os import pathlib import posix1e import shutil import pwd import grp import stat import yaml import json import tabulate import itertools import sys try: from samba import param WITH_SAMBA = True lp = param.LoadParm() lp.load('/etc/samba/smb.conf') except: WITH_SAMBA = False temp_umask = 0o022 UMASK = os.umask(temp_umask) os.umask(UMASK) SUBJECTS = ['owner', 'group owner', 'other', 'mask', 'users', 'groups'] ACL_TAG_TYPE_DICT = {1: 'owner:{nid}:{perm}', 2: 'u:{nid}:{perm}', 4: 'group:{nid}:{perm}', 8: 'g:{nid}:{perm}', 16: 'mask:{perm}', 32: 'other:{perm}'} def permset_to_rwx(permset): """Return permission set as well known r, w, x representation permset: Permission set permset type: posix1e.Permset return: permissions represented by r, w, x, - characters return type: str """ permset_string = '' if permset.read: permset_string += 'r' else: permset_string += '-' if permset.write: permset_string += 'w' else: permset_string += '-' if permset.execute: permset_string += 'x' else: permset_string += '-' return permset_string def format_comparison(comparison): """Return list of strings """ if not comparison: return comparison return [f"{c[1]}" for c in comparison] def format_acl(path): """Return formatted acl entry path: path of file path type: pathlib.Path return: formatted string listing permissons return type: str """ acl = posix1e.ACL(file=path.as_posix()) human_readable_entries = {} for entry in acl: if entry.tag_type == posix1e.ACL_USER_OBJ: human_readable_entries.setdefault('owner', []) human_readable_entries['owner'].append(ACL_TAG_TYPE_DICT[entry.tag_type].format(nid=path.owner(), perm=permset_to_rwx(entry.permset))) elif entry.tag_type == posix1e.ACL_GROUP_OBJ: human_readable_entries.setdefault('group owner', []) human_readable_entries['group owner'].append(ACL_TAG_TYPE_DICT[entry.tag_type].format(nid=path.group(), perm=permset_to_rwx(entry.permset))) elif entry.tag_type == posix1e.ACL_OTHER: human_readable_entries.setdefault('other', []) human_readable_entries['other'].append(ACL_TAG_TYPE_DICT[entry.tag_type].format(perm=permset_to_rwx(entry.permset))) elif entry.tag_type == posix1e.ACL_MASK: human_readable_entries.setdefault('mask', []) human_readable_entries['mask'].append(ACL_TAG_TYPE_DICT[entry.tag_type].format(perm=permset_to_rwx(entry.permset))) elif entry.tag_type == posix1e.ACL_USER: human_readable_entries.setdefault('users', []) human_readable_entries['users'].append(ACL_TAG_TYPE_DICT[entry.tag_type].format(nid=pwd.getpwuid(entry.qualifier).pw_name, perm=permset_to_rwx(entry.permset))) elif entry.tag_type == posix1e.ACL_GROUP: human_readable_entries.setdefault('groups', []) human_readable_entries['groups'].append(ACL_TAG_TYPE_DICT[entry.tag_type].format(nid=grp.getgrgid(entry.qualifier).gr_name, perm=permset_to_rwx(entry.permset))) human_readable_list = sorted(human_readable_entries.items(), key=lambda x: (SUBJECTS.index(x[0]), x[1])) human_readable_digest = '\n'.join(itertools.chain(*[hre[1] for hre in human_readable_list])) return human_readable_digest def diff_perm_entries(ref, subject): """ Return string encoding differences between ref and subject permessions. 0 stands for no differences, - for missing permission, + for added permission. Differences are evaluated for r, w, x in that order and concatenated in one string. ref: permissions ref type: int subject: permissions set subject type: posix1e.Entry """ diff = [] for perm_code in [posix1e.ACL_READ, posix1e.ACL_WRITE, posix1e.ACL_EXECUTE]: if subject.permset.test(perm_code): if ref & perm_code == perm_code: diff.append('0') else: diff.append('+') else: if ref & perm_code == perm_code: diff.append('-') else: diff.append('0') return ''.join(diff) def compare_acl(default, parent, effective, subject, exhaustive=False, report_root=True, umask=0o0000): """ default: parent folder acl default type: posix1e.ACL effective: current folder acl effective type: posix1e.ACL exhaustive: if False, inspection stops at first difference, else lists all differences exhaustive type: boolean umask: mask used globally for filtering permissions of new folders and files umask type: int """ comparisons = [] other_mask = umask%0o10 group_mask = (umask>>3)%0o10 user_mask = (umask>>6)%0o10 acl_mask_mode = 7 user_entries = [e for e in effective if e.tag_type == posix1e.ACL_USER] group_entries = [e for e in effective if e.tag_type == posix1e.ACL_GROUP] for ref in default: ref_mode = 0 if ref.permset.read: ref_mode = ref_mode | 4 if ref.permset.write: ref_mode = ref_mode | 2 if ref.permset.execute: ref_mode = ref_mode | 1 ref_mode = ref_mode & acl_mask_mode if ref.tag_type == posix1e.ACL_USER_OBJ: filtered_mode = ref_mode & ~user_mask entry = [e for e in effective if e.tag_type == ref.tag_type][0] diff = diff_perm_entries(filtered_mode, entry) if report_root or not (parent.owner() != subject.owner() and parent.owner() == 'root'): key = ACL_TAG_TYPE_DICT[ref.tag_type].format(nid=parent.owner(), perm=diff) comparisons.append((2, key, diff, '!' if diff != '000' else '=')) if parent.owner() != subject.owner(): key = ACL_TAG_TYPE_DICT[ref.tag_type].format(nid=subject.owner(), perm='+++') comparisons.append((1, key, '+++', '!')) elif ref.tag_type == posix1e.ACL_USER: entries = [e for e in user_entries if e.qualifier == ref.qualifier] if entries: filtered_mode = ref_mode & ~user_mask entry = entries[0] diff = diff_perm_entries(filtered_mode, entries[0]) user_entries.pop(user_entries.index(entry)) else: diff = '---' key = ACL_TAG_TYPE_DICT[ref.tag_type].format(nid=pwd.getpwuid(ref.qualifier).pw_name, perm=diff) comparisons.append((6, key, diff)) elif ref.tag_type == posix1e.ACL_GROUP_OBJ: filtered_mode = ref_mode & ~group_mask entry = [e for e in effective if e.tag_type == ref.tag_type][0] diff = diff_perm_entries(filtered_mode, entry) if report_root or not (parent.group() != subject.group() and parent.group() == 'root'): key = ACL_TAG_TYPE_DICT[ref.tag_type].format(nid=parent.group(), perm=diff) comparisons.append((4, key, diff)) if parent.group() != subject.group(): key = ACL_TAG_TYPE_DICT[ref.tag_type].format(nid=subject.group(), perm='+++') comparisons.append((3, key, '+++')) elif ref.tag_type == posix1e.ACL_GROUP: entries = [e for e in group_entries if e.qualifier == ref.qualifier] if entries: filtered_mode = ref_mode & ~group_mask entry = entries[0] diff = diff_perm_entries(filtered_mode, entries[0]) group_entries.pop(group_entries.index(entry)) else: diff = '---' key = ACL_TAG_TYPE_DICT[ref.tag_type].format(nid=grp.getgrgid(ref.qualifier).gr_name, perm=diff) comparisons.append((7, key, diff)) elif ref.tag_type == posix1e.ACL_OTHER: filtered_mode = ref_mode & ~other_mask entry = [e for e in effective if e.tag_type == ref.tag_type][0] diff = diff_perm_entries(filtered_mode, entry) key = ACL_TAG_TYPE_DICT[ref.tag_type].format(perm=diff) comparisons.append((5, key, diff)) if diff != '000' and not exhaustive: break for user in user_entries: key = ACL_TAG_TYPE_DICT[2].format(nid=pwd.getpwuid(user.qualifier).pw_name, perm='+++') comparisons.append((6, key, '+++')) for group in group_entries: key = ACL_TAG_TYPE_DICT[8].format(nid=grp.getgrgid(group.qualifier).gr_name, perm='+++') comparisons.append((7, key, '+++')) return sorted(comparisons) def compare_entities_acl(parent, subject, detail=False, inherited_only=True, report_root=True): """ Return list of tuples with permissions diverging folders and their differences as dictionnary. For each entity (user, group, etc.) differences are expressed as a three characters chain, one character for r, w, x each, with 0 standing for no difference, + for permission not present in parent folder and - for permission not present in current folder. For additional entities (users and groups from extended acl), --- denotes an entity absent from current folder and +++ an entity absent from parent folder. parent: parent folder from which getting default acl and permissions parent type: pathlib.Path subject: folder to inspect subject type: pathlib.Path detail: if False, inspection stops at first difference, else lists all differences detail type: boolean inherited_only: if True, compare only default acl with applied permissions, else also compare with current parent folder permissions inherited_only type: boolean """ if subject.is_file(): umask = UMASK else: umask = 0o0000 parent_default_acl = posix1e.ACL(filedef=parent.as_posix()) if [default_acl for default_acl in parent_default_acl]: comparison = compare_acl(parent_default_acl, parent, posix1e.ACL(file=subject.as_posix()), subject, exhaustive=detail, report_root=report_root, umask=umask) elif not inherited_only: parent_acl = posix1e.ACL(file=parent.as_posix()) comparison = compare_acl(parent_acl, parent, posix1e.ACL(file=subject.as_posix()), subject, exhaustive=detail, report_root=report_root, umask=umask) else: comparison = None return comparison def permissions_modification_detected(entity, detail=False): """ Return True if entity's permissions do not follow parent folder restrictions entity: root folder where recursive inspection begins entity type: pathlib.Path detail: if False, inspection stops at first difference, else lists all differences detail type: boolean """ perms = {'setuid': False, 'setgid': False, 'owner': False, 'group': False, 'acl': False} parent = entity.parent parent_setuid = parent.stat().st_mode | stat.S_ISUID == parent.stat().st_mode parent_setgid = parent.stat().st_mode | stat.S_ISGID == parent.stat().st_mode perms['owner'] = parent.owner() != entity.owner() perms['group'] = parent.group() != entity.group() if parent_setuid and parent.owner() != entity.owner(): perms['setuid'] = True if parent_setgid and parent.group() != entity.group(): perms['setgid'] = True acl_comparison = compare_entities_acl(parent, entity, detail=detail, inherited_only=False) if acl_comparison: if any([c[2] != '000' for c in acl_comparison]): perms['acl'] = True return any(perms.values()) if __name__ == '__main__': import argparse def human_readable(report): tabulated_report = [] for report_element in report: acl = report[report_element]['acl'] parent_acl = report[report_element]['parent_acl'] diff = report[report_element]['diff'] if tabulate.__version__ < '0.8.1': tabulated_report.extend(itertools.zip_longest([report_element], acl.split('\n'), parent_acl.split('\n'),diff, fillvalue='')) tabulated_report.append(['', '', '']) else: tabulated_report.append([report_element, acl, '\n'.join(diff)]) tabulated_report.append(tabulate.SEPARATING_LINE) return tabulate.tabulate(tabulated_report, headers=['dossier', 'ACL observée', 'ACL du parent', 'diff'], tablefmt="orgtbl") def parse_tree(root): modified_paths = [] modified_paths.append(root) tested_paths = 0 for step in os.walk(root): folders = [pathlib.Path(step[0]).joinpath(p) for p in step[1]] tested_paths += len(folders) modified_paths.extend([folder for folder in folders if permissions_modification_detected(folder)]) return modified_paths def exec_cmd(args): modified_paths = [] if args.root: for root_path in args.root: modified_paths.extend(parse_tree(pathlib.Path(root_path))) elif WITH_SAMBA: for share in [lp.get('path', s) for s in lp.services() if s not in ['homes', 'profiles']]: modified_paths.extend(parse_tree(pathlib.Path(share))) perm_digest = {mp.as_posix(): {'acl': format_acl(mp), 'parent_acl': format_acl(mp.parent), 'diff': format_comparison(compare_entities_acl(mp.parent, mp, inherited_only=False, report_root=False, detail=True))} for mp in sorted(modified_paths)} if not perm_digest: raise Exception('Aucun dossier à analyser : déclarez des dossiers partagés dans la configuration du serveur ou indiquez des dossiers dans les paramètres de l’action') if args.format == 'json': perm_digest = json.dumps(perm_digest) elif args.format == 'yaml': perm_digest = yaml.dump(perm_digest, Dumper=yaml.Dumper, default_flow_style=False, indent=4) elif args.format == 'tabular': perm_digest = human_readable(perm_digest) if args.output: dest = pathlib.Path(args.output) if not dest.exists(): raise Exception(f'{dest} does not exist') with open(dest.joinpath(f'modified_acl.{args.format}'), 'w') as perm_digest_fh: perm_digest_fh.write(perm_digest) else: print(perm_digest) parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(help='Aide des sous-commandes') parser_exec = subparsers.add_parser('exec', help='Execute command') parser_exec.add_argument('-r', '--root', nargs='*', required=WITH_SAMBA is False) parser_exec.add_argument('-f', '--format', help='Format de sortie de la commande', default='json') parser_exec.add_argument('-o', '--output', help='Dossier de destination') parser_exec.set_defaults(func=exec_cmd) args = parser.parse_args() if hasattr(args, 'func'): args.func(args) else: parser.print_usage()