1#@!#!123s

��JFIF�� 1#@!#!123s
: / lib / fixperms /
Filename : fixperms_base.py
back
"""Common fixperms classes"""
import os
from shlex import quote
from pathlib import Path
from typing import Union
from stat import S_IMODE, S_ISREG, S_ISDIR, S_ISLNK
import re
from fixperms_cli import Args
from fixperms_ids import IDCache


class PermMap:
    """Base class for fixperms"""

    def __init__(
        self,
        ids: IDCache,
        args: Args,
        user: str,
        all_docroots: list[str],
        docroot_chmod: int,
        docroot_chown: tuple[str, str],
    ):
        self.args = args
        self.skip = self.args.skip.copy()
        self.all_docroots = all_docroots
        self.log = args.logger
        self.hard_links = HardLinkTracker(self)
        self.ids = ids
        self.user = user
        pwuser = self.ids.getpwnam(user)
        self.uid = pwuser.pw_uid
        self.gid = pwuser.pw_gid
        doc_uid = ids.getpwnam(docroot_chown[0]).pw_uid
        doc_gid = ids.getgrnam(docroot_chown[1]).gr_gid
        self.docroot_perms = Rule('', (None, docroot_chmod), (doc_uid, doc_gid))
        self.homedir = os.path.realpath(pwuser.pw_dir)
        if not re.match(r'\/home\d*\/', self.homedir):
            raise ValueError(f"{user}: unexpected homedir: {self.homedir!r}")
        self.home_re = re.escape(pwuser.pw_dir)
        self.perm_map: list['Rule'] = []

    def add_rule(
        self,
        regex: str,
        modes: tuple[Union[int, None], Union[int, None]],
        chown: tuple[int, int],
    ) -> None:
        """Add a fixperms path rule. ^HOMEDIR is automatically added"""
        # no actual ^ becasue we use .match, not .search
        self.perm_map.append(Rule(f"{self.home_re}{regex}", modes, chown))

    def lchown(self, path: str, stat: os.stat_result, uid: int, gid: int):
        """Runs os.lchown"""
        if uid == gid == -1:
            return
        tgt_uid = stat.st_uid if uid == -1 else uid
        tgt_gid = stat.st_gid if gid == -1 else gid
        if (stat.st_uid, stat.st_gid) == (tgt_uid, tgt_gid):
            return
        if not self.args.noop:
            try:
                os.lchown(path, uid, gid)
            except OSError as exc:
                self.log.error(exc)
                return
        old_user = self.ids.uid_label(stat.st_uid)
        old_group = self.ids.gid_label(stat.st_gid)
        new_user = self.ids.uid_label(tgt_uid)
        new_group = self.ids.gid_label(tgt_gid)
        self.log.debug(
            'Changed ownership of %s from %s:%s to %s:%s',
            quote(path),
            old_user,
            old_group,
            new_user,
            new_group,
        )

    def lchmod(
        self,
        path: str,
        stat: os.stat_result,
        mode: Union[int, None],
    ):
        """Runs os.chmod if the path is not a symlink"""
        if mode is None:
            return
        orig = S_IMODE(stat.st_mode)
        if orig == mode:
            return
        if S_ISLNK(stat.st_mode):
            return  # Linux does not support follow_symlinks=False
        if not self.args.noop:
            try:
                os.chmod(path, mode)
            except OSError as exc:
                self.log.error(exc)
                return
        self.log.debug(
            'Changed mode of %s from %s to %s',
            quote(path),
            oct(orig)[2:],
            oct(mode)[2:],
        )

    def walk(self, path: str, ignore_skips: bool = False):
        """os.walk/os.lstat to yield a path and all of its contents"""
        for entry in self._walk(path, ignore_skips):
            try:
                stat = os.lstat(entry)
            except OSError as exc:
                self.log.error(exc)
                continue
            yield stat, entry

    def _walk(self, top_dir: str, ignore_skips: bool = False):
        if not ignore_skips and self.should_skip(top_dir):
            return
        yield top_dir
        if not os.path.isdir(top_dir):
            return
        for dirpath, dirnames, filenames in os.walk(top_dir):
            for filename in filenames:
                path = os.path.join(dirpath, filename)
                if ignore_skips or not self.should_skip(path):
                    yield path
            skip_dirs = []
            for dirname in dirnames:
                path = os.path.join(dirpath, dirname)
                if not ignore_skips and self.should_skip(path):
                    skip_dirs.append(path)
                else:
                    yield path
            if skip_dirs:
                # editing dirnames[:] in-place causes os.walk to not traverse it
                dirnames[:] = [x for x in dirnames if x not in skip_dirs]

    def run(self) -> None:
        """To be called from fixperms_main.py - processes this user"""
        self.fixperms()
        self.hard_links.handle()

    def fixperms(self) -> None:
        """Iterate over a user's files and chown/chmod as needed"""
        for stat, path in self.walk(self.homedir):
            try:
                self.check_path(stat, path)
            except OSError as exc:
                self.log.error(exc)

    def with_exec_bits(self, stat: os.stat_result, new_mode: Union[None, int]):
        """Get a new file mode including old mode's exec bits"""
        if new_mode is None:
            return None
        if self.args.preserve_exec:
            exec_bits = stat.st_mode & 0o111
            return new_mode | exec_bits
        return new_mode

    def check_path(self, stat: os.stat_result, path: str):
        """Chown and chmod files as necessary"""
        rule = self.find_rule(str(path))
        file_mode, dir_mode = rule.modes
        if S_ISREG(stat.st_mode):  # path is a regular file
            new_mode = self.with_exec_bits(stat, file_mode)
            if stat.st_nlink > 1:
                self.hard_links.add(path, stat, rule.chown, new_mode)
                return
        elif S_ISDIR(stat.st_mode):  # path is a directory
            new_mode = dir_mode
        elif S_ISLNK(stat.st_mode):  # path is a symlink
            new_mode = None
        else:  # path is socket/device/fifo/etc
            self.log.warning("Skipping unexpected path type at %s", path)
            return
        if new_mode is not None:
            self.lchmod(path, stat, new_mode)
        self.lchown(path, stat, *rule.chown)

    def find_rule(self, path: str) -> 'Rule':
        """Find the matching ``Rule`` for a given path"""
        assert isinstance(path, str)
        if path in self.all_docroots:
            return self.docroot_perms
        for rule in self.perm_map:
            if rule.regex.match(path):
                return rule
        raise ValueError(f"No matching rule for {path}")

    def should_skip(self, path: str):
        """Determine if a path should be skipped based on --skip args"""
        for skip in self.skip:
            if path == skip:
                return True
            if Path(path).is_relative_to(skip):
                return True
        return False


class HardLinkTracker:
    """Tracks and handles hard links discovered while walking through a
    user's files"""

    def __init__(self, perm_map: PermMap):
        self.perm_map = perm_map
        self.chowns: dict[int, tuple[int, int]] = {}
        self.stats: dict[int, os.stat_result] = {}
        self.modes: dict[int, int] = {}
        self.paths: dict[int, list[str]] = {}

    def add(
        self,
        path: str,
        stat: os.stat_result,
        chown: tuple[int, int],
        mode: Union[int, None],
    ):
        """Used to add a hard link found during the fixperms run which might
        be unsafe to operate on"""
        inum = stat.st_ino
        self.stats[inum] = stat  # will be the same for all ends of the link
        if inum in self.paths:
            self.paths[inum].append(path)
        else:
            self.paths[inum] = [path]
        if inum in self.chowns:
            uid, gid = chown
            prev_uid, prev_gid = self.chowns[inum]
            if uid == -1:
                uid = prev_uid
            if gid == -1:
                gid = prev_gid
            self.chowns[inum] = [uid, gid]
        else:
            self.chowns[inum] = chown
        if mode is not None:
            self.modes[inum] = mode

    def handle(self):
        """If self.hard_links was populated with any items, handle any that are
        safe, or log any that are not"""
        for inum, stat in self.stats.items():
            # for each distinct inode found with hard links...
            if stat.st_nlink == len(self.paths[inum]):
                # If we came across every end of the link in this run, then it's
                # safe to operate on. Chmod the first instance of it; the rest
                # will change with it.
                path = self.paths[inum][0]
                self.perm_map.lchown(path, stat, *self.chowns[inum])
                self.perm_map.lchmod(path, stat, self.modes.get(inum, None))
                continue
            # Otherwise these hard links can't be trusted.
            for path in self.paths[inum]:
                self.perm_map.log.error(
                    '%s is hardlinked and not owned by the user',
                    quote(path),
                )


class Rule:
    """Fixperms path rule"""

    def __init__(
        self,
        regex: str,
        modes: tuple[Union[int, None], Union[int, None]],
        chown: tuple[int, int],
    ):
        """Fixperms path rule

        Args:
            regex (str): regular expression
            file tuple[(int | None), (int | None)]: (file, dir) modes if matched
            chown tuple[int, int]: if a matching file/dir is found, chown to
                this UID/GID. Use -1 to make no change.
        """
        self.regex = re.compile(regex)
        assert isinstance(modes, tuple)
        assert isinstance(chown, tuple)
        self.modes = modes
        self.chown = chown