#!/opt/imunify360/venv/bin/python
"""Fix file ownership/permissions for /var/ossec.

Usage:

  $ imunify360-ossec-fix-var-ossec [--log-level <level>] <var-ossec-dir>

It supports cases when [orphaned/alian] files in /var/ossec are owned
by non-existing/unrelated to root/ossec* users (DEF-12616).

It replicates ownership/permissions specified in the %files sections
of the ossec spec file. Remaining [orphaned/alian] files are renamed
(+.rpmsave suffix is appended)

"""
import argparse
import functools
import grp
import logging
import os
import pwd
import shutil
import stat
import sys
from contextlib import suppress
from pathlib import Path


__all__ = ["main"]

#: File attributes are generated by the `inv get-permissions-ownership` command
#   based on the ossec spec's %attr data
# !!! DO NOT EDIT BY HAND !!!
FILE_ATTRS = [
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/active-response",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": True,
        "glob": "%{_localstatedir}/ossec/active-response/bin",
    },
    {
        "mode": "0o700",
        "user": "root",
        "group": "ossec",
        "recursive": True,
        "glob": "%{_localstatedir}/ossec/active-response/quarantine",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": True,
        "glob": "%{_localstatedir}/ossec/agentless",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/bin",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/etc",
    },
    {
        "mode": "0o770",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/etc/shared",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/etc/templates",
    },
    {
        "mode": "0o640",
        "user": "ossec",
        "group": "ossec",
        "recursive": True,
        "glob": "%{_localstatedir}/ossec/etc/templates/*",
    },
    {
        "mode": "0o770",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/logs",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue",
    },
    {
        "mode": "0o770",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/ossec",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/diff",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/var",
    },
    {
        "mode": "0o770",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/var/run",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "root",
        "recursive": True,
        "glob": "%{_localstatedir}/ossec/bin/ossec-lua*",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/tmp",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "root",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/lua/*",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/logs",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/queue/alerts",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/queue/diff",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/queue/ossec",
    },
    {
        "mode": "0o775",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/queue/rids",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/queue/syscheck",
    },
    {
        "mode": "0o1750",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/tmp",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/var",
    },
    {
        "mode": "0o770",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/ossec-agent/var/run",
    },
    {
        "mode": "0o600",
        "user": "root",
        "group": "root",
        "recursive": True,
        "glob": "%{_sysconfdir}/ossec-init.conf",
    },
    {
        "mode": "0o640",
        "user": "ossec",
        "group": "ossec",
        "recursive": True,
        "glob": "%{_localstatedir}/ossec/etc/shared/agent.conf",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/logs/archives",
    },
    {
        "mode": "0o770",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/logs/alerts",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/logs/firewall",
    },
    {
        "mode": "0o755",
        "user": "ossecr",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/agent-info",
    },
    {
        "mode": "0o755",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/agentless",
    },
    {
        "mode": "0o770",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/alerts",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/fts",
    },
    {
        "mode": "0o755",
        "user": "ossecr",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/rids",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/rootcheck",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/queue/syscheck",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/rules",
    },
    {
        "mode": "0o750",
        "user": "ossec",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/stats",
    },
    {
        "mode": "0o550",
        "user": "root",
        "group": "ossec",
        "recursive": False,
        "glob": "%{_localstatedir}/ossec/agentless",
    },
]
OSSEC_DIR_TEMPLATE = "%{_localstatedir}/ossec"
OSSEC_USER = "ossec"
OSSEC_GROUP = OSSEC_USER


logger = logging.getLogger("imunify360-ossec-fix-var-ossec")


def main(argv=None):
    "Entry point for the script."
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--log-level",
        default="INFO",
        choices=[
            "CRITICAL",
            "FATAL",
            "ERROR",
            "WARN",
            "WARNING",
            "INFO",
            "DEBUG",
            "NOTSET",
        ],
    )
    parser.add_argument("ossec_dir")
    args = parser.parse_args(args=argv[1:] if argv is not None else None)
    logging.basicConfig(
        level=getattr(logging, args.log_level),
        format=parser.prog + ": %(message)s",
    )
    ossec_path = Path(args.ossec_dir)
    logger.info("Processing %s", ossec_path)
    if not ossec_path.is_dir():
        return  # nothing to fix

    success = True
    native_users = set()  # users known from %files
    for d in FILE_ATTRS:
        if not d["glob"].startswith(OSSEC_DIR_TEMPLATE):
            continue  # ignore non /var/ossec files
        glob_path = ossec_path / Path(d["glob"]).relative_to(
            OSSEC_DIR_TEMPLATE
        )
        success &= set_ownership_permissions(
            glob_path,
            mode=int(d["mode"], 0),
            user=d["user"],
            group=d["group"],
            recursive=d["recursive"],
        )
        if d["user"]:
            native_users.add(d["user"])
    # workaround from DEF-12616
    success &= set_ownership_permissions(
        ossec_path / "queue/fts/*", user=OSSEC_USER, group=OSSEC_GROUP
    )

    success &= move_alians(ossec_path, native_users)
    return not success


def set_ownership_permissions(
    glob_path, *, mode=None, user=None, group=None, recursive=False
) -> bool:
    """
    Set ownership to *user*/*group* and permissions to *mode* for *glob_path.*

    Apply recursively to all files/dirs if *recursive*.

    Return whether setting ownership/permissions was successful.
    """
    success = True
    if not glob_path.parent.exists():
        return success  # skip non-existent paths

    for path in glob_path.parent.glob(glob_path.name):
        success &= _fix_path(
            str(path),
            mode=mode,
            user=user,
            group=group,
            recursive=recursive,
        )
    return success


@functools.lru_cache(maxsize=None)
def _resolve_owner(user, group):
    """Resolve user/group names to uid/gid.  Returns (uid, gid)."""
    uid = pwd.getpwnam(user).pw_uid if user else -1
    gid = grp.getgrnam(group).gr_gid if group else -1
    return uid, gid


def _fix_path(path_str, *, mode=None, user=None, group=None, recursive=False):
    """Apply ownership/permissions via fd-anchored ops (CWE-59 safe)."""
    try:
        st = os.lstat(path_str)
    except OSError as e:
        logger.warning("lstat failed: %s", e)
        return False

    if stat.S_ISLNK(st.st_mode):
        logger.warning("refusing to fix symlink: %s", path_str)
        return True

    open_flags = os.O_RDONLY | os.O_NOFOLLOW
    if stat.S_ISDIR(st.st_mode):
        open_flags |= os.O_DIRECTORY

    try:
        fd = os.open(path_str, open_flags)
    except OSError as e:
        logger.warning("open(%s) failed: %s", path_str, e)
        return False

    try:
        fd_st = os.fstat(fd)
        if (fd_st.st_dev, fd_st.st_ino) != (st.st_dev, st.st_ino):
            logger.warning(
                "inode changed between lstat and open: %s", path_str
            )
            return False
        return _apply_to_fd(
            fd,
            fd_st,
            path_str,
            mode=mode,
            user=user,
            group=group,
            recursive=recursive,
        )
    finally:
        os.close(fd)


def _apply_to_fd(
    fd, fd_st, display_path, *, mode, user, group, recursive
):
    """Apply mode/owner to an already-pinned fd; optionally recurse."""
    success = True

    if mode is not None:
        logger.info(
            "Setting %s permissions for %s",
            stat.filemode(mode),
            display_path,
        )
        try:
            os.fchmod(fd, mode)
        except OSError as e:
            success = False
            logger.warning("chmod failed: %s", e)

    if user or group:
        logger.info(
            "Setting ownership to %s:%s for %s",
            user or "",
            group or "",
            display_path,
        )
        try:
            uid, gid = _resolve_owner(user, group)
            os.fchown(fd, uid, gid)
        except (LookupError, OSError) as e:
            success = False
            logger.warning("chown failed: %s", e)

    if recursive and stat.S_ISDIR(fd_st.st_mode):
        try:
            children = os.listdir(fd)
        except OSError as e:
            success = False
            logger.warning("listdir failed: %s", e)
            return success
        for name in children:
            child_display = os.path.join(display_path, name)
            try:
                child_st = os.lstat(name, dir_fd=fd)
            except OSError as e:
                success = False
                logger.warning("lstat failed: %s", e)
                continue
            if stat.S_ISLNK(child_st.st_mode):
                logger.warning(
                    "refusing to fix symlink: %s", child_display
                )
                continue
            child_flags = os.O_RDONLY | os.O_NOFOLLOW
            if stat.S_ISDIR(child_st.st_mode):
                child_flags |= os.O_DIRECTORY
            try:
                child_fd = os.open(name, child_flags, dir_fd=fd)
            except OSError as e:
                success = False
                logger.warning(
                    "open(%s) failed: %s", child_display, e
                )
                continue
            try:
                child_fd_st = os.fstat(child_fd)
                if (child_fd_st.st_dev, child_fd_st.st_ino) != (
                    child_st.st_dev,
                    child_st.st_ino,
                ):
                    success = False
                    logger.warning(
                        "inode changed: %s", child_display
                    )
                    continue
                success &= _apply_to_fd(
                    child_fd,
                    child_fd_st,
                    child_display,
                    mode=mode,
                    user=user,
                    group=group,
                    recursive=recursive,
                )
            finally:
                os.close(child_fd)

    return success


def move_alians(ossec_path, native_users, suffix=".rpmsave") -> bool:
    """
    Move files&dirs from *ossec_path* if they are not owned by *native_users*
    """
    success = True
    for path in ossec_path.iterdir():
        if path.name.endswith(suffix):
            continue  # do nothing if the path already has the suffix
        try:
            alian = path.owner() not in native_users
        except KeyError:
            alian = True
        if not alian:
            with suppress(OSError):
                if path.is_dir():
                    success &= move_alians(path, native_users, suffix)
            continue  # keep it

        # move path
        new_path = path.with_name(path.name + suffix)
        logger.info("Renaming %s -> %s", path, new_path)
        for last in range(2):
            try:
                path.replace(new_path)
            except OSError as err:
                if not last:  # remove target & retry
                    logger.info("Removing %s", new_path)
                    if isinstance(err, IsADirectoryError):
                        shutil.rmtree(str(new_path), ignore_errors=True)
                    else:  # assume file
                        with suppress(OSError):
                            new_path.unlink()
                else:  # last
                    success = False
                    logger.warning(
                        "Can't rename %s, reason: %s", path, str(err)
                    )
            else:
                break

    return success


if __name__ == "__main__":
    sys.exit(main())
