1#@!#!123s

: / lib / fixperms /
Filename : fixperms_cpanel.py
back
"""Fixperms class for cPanel"""

import os
import re
from shlex import quote, join as cmd_join
from subprocess import CalledProcessError, check_call
from stat import S_ISLNK, S_ISREG, S_ISDIR
import rads
from fixperms_base import PermMap
from fixperms_cli import Args
from fixperms_ids import IDCache


class CpanelPermMap(PermMap):
    """Fixperms class for cPanel"""

    def __init__(self, ids: IDCache, args: Args, user: str):
        super().__init__(
            ids=ids,
            args=args,
            user=user,
            all_docroots=rads.UserData(user).all_roots,
            docroot_chmod=0o750,
            docroot_chown=(user, 'nobody'),
        )
        self.is_shared = rads.IMH_ROLE == 'shared'
        # always skip ~/etc and ~/mail in the main os.walk - that's what
        # self.mailperms is for
        self.skip.add(os.path.join(self.homedir, 'mail'))
        self.skip.add(os.path.join(self.homedir, 'etc'))
        self.bad_links = []
        # pylint: disable=duplicate-code
        # Order these rules more specific to less specific regex.
        uid, gid = self.uid, self.gid
        # sensitive passwords: ~/.accesshash, ~/.pgpass, ~/.my.cnf
        self.add_rule(
            r"\/\.(?:accesshash|pgpass|my\.cnf)$", (0o600, None), (uid, gid)
        )
        # ~/.imh/nginx - ngxconf & cache manager files
        self.add_rule(r"\/\.imh\/nginx(?:$|\/)", (0o664, 0o775), (uid, gid))
        # ~/.imh directory and contents
        self.add_rule(r"\/\.imh(?:$|\/)", (0o644, 0o755), (0, 0))
        # ~/.ssh directory and contents
        self.add_rule(r"\/\.ssh(?:$|\/)", (0o600, 0o700), (uid, gid))
        # ~/.pki dir and subdirs
        self.add_rule(r"\/\.pki(?:$|\/)", (None, 0o740), (uid, gid))
        # .cgi and .pl files
        self.add_rule(r"\/.*\.(?:pl|cgi)$", (0o755, None), (uid, gid))
        # homedir folder itself
        self.add_rule("$", (None, 0o711), (uid, gid))
        # restrict access to sensitive CMS config files
        self.add_rule(
            r"\/.+\/(?:(?:wp-config|conf|[cC]onfig|[cC]onfiguration|"
            r"LocalSettings|settings)(?:\.inc)?\.php|"
            r"local\.xml|mt-config\.cgi)$",
            (0o640, None),
            (uid, gid),
        )
        # contents of homedir which do not match a previous regex
        self.add_rule(r"\/", (0o644, 0o755), (uid, gid))
        # full path to symlink sources which are safe
        self.safe_link_src = {
            os.path.join(self.homedir, '.cphorde/meta/latest'),
            os.path.join(self.homedir, 'www'),
        }
        # regex for symlink sources which are safe
        safe_link_src_re = [
            fr'(?:{self.home_re}\/(?:etc|mail|logs)\/)',
            r'(?:.*\/\.ea-php-cli\.cache$)',
        ]
        self.safe_link_src_re = re.compile('|'.join(safe_link_src_re))
        # full path to symlink destinations which are safe
        self.safe_link_dest = {
            self.homedir,
            os.path.join('/usr/local/apache/domlogs', self.user),
            os.path.join('/etc/apache2/logs/domlogs', self.user),
            os.path.join('/var/log/apache2/domlogs', self.user),
            '/home/shrusr/SharedHtDocsDir',
            '/var/lib/mysql/mysql.sock',
            '/var/run/postgres/.s.PGSQL.5432',
            '/run/postgres/.s.PGSQL.5432',
            '/usr/local/cpanel/base/frontend/paper_lantern/styled/retro',
        }

    def link_unsafe(self, path: str) -> bool:
        """Determine if a symlink is "unsafe" for a shared server"""
        if not self.is_shared:
            return False
        if path in self.safe_link_src:
            return False
        if os.path.realpath(path) in self.safe_link_dest:
            return False
        if self.safe_link_src_re.match(path):
            return False
        bad_link = f'{quote(path)} -> {quote(os.readlink(path))}'
        self.bad_links.append(bad_link)
        self.log.warning('Potentially malicious symlink detected: %s', bad_link)
        os.unlink(path)
        return True

    def mailperms(self):
        """Run /scripts/mailperm"""
        if self.args.skip_mail:
            return
        self.mailperm_fix('mail', self.gid)
        self.mailperm_fix('etc', self.ids.getgrnam('mail').gr_gid)
        cmd_args = [
            '/usr/local/cpanel/scripts/mailperm',
            '--skiplocaldomains',
            '--skipmxcheck',
            self.user,
        ]
        self.log.debug('Running: %s', cmd_join(cmd_args))
        if self.args.noop:
            return
        try:
            check_call(cmd_args)
        except (CalledProcessError, OSError):
            self.log.error('Error running: %s', cmd_join(cmd_args))
            raise

    def fix_cpanel_db_files(self):
        """Fix ownership/permissions on /var/cpanel/databases/ files"""
        db_dir = '/var/cpanel/databases'
        # (filename, uid, gid, mode)
        targets = [
            (f'{self.user}.json', 0, self.gid, 0o640),
            (f'grants_{self.user}.yaml', 0, 0, 0o600),
            (f'grants_{self.user}.cache', 0, 0, 0o600),
        ]
        for filename, uid, gid, mode in targets:
            path = os.path.join(db_dir, filename)
            try:
                stat = os.lstat(path)
            except FileNotFoundError:
                self.log.debug(
                    '%s does not exist, skipping', path
                )
                continue
            except OSError as exc:
                self.log.error(exc)
                continue
            self.lchown(path, stat, uid, gid)
            self.lchmod(path, stat, mode)

    def fixperms(self) -> None:
        super().fixperms()
        self.send_str()
        self.mailperms()
        self.fix_cpanel_db_files()

    def check_path(self, stat: os.stat_result, path: str):
        if S_ISLNK(stat.st_mode) and self.link_unsafe(path):
            return
        super().check_path(stat, path)

    def mailperm_fix(self, subdir: str, dir_gid: int):
        """Fix permissions not caught by cPanel's mailperm script"""
        top_dir = os.path.join(self.homedir, subdir)
        mail_gid = self.ids.getgrnam('mail').gr_gid
        dir_gids = {self.gid, dir_gid}
        for stat, path in self.walk(top_dir, ignore_skips=True):
            if S_ISREG(stat.st_mode):  # path is a regular file
                if stat.st_gid in (self.gid, mail_gid):
                    gid = -1
                else:
                    gid = self.gid
                if self.uid != stat.st_uid and stat.st_nlink > 1:
                    self.hard_links.add(path, stat, (self.uid, gid), None)
                    continue
                self.lchown(path, stat, self.uid, gid)
            elif S_ISDIR(stat.st_mode):  # path is a directory
                # for each directory with a group not set to the user or mail
                # chgrp to user:mail if ~/etc, user:user if ~/mail
                if stat.st_gid in dir_gids:
                    self.lchown(path, stat, self.uid, -1)
                else:
                    self.lchown(path, stat, self.uid, dir_gid)
            elif S_ISLNK(stat.st_mode):  # path is a symlink
                if self.link_unsafe(path):
                    continue
                self.lchown(path, stat, self.uid, self.gid)
            else:  # path is socket/device/fifo/etc
                self.log.warning("Skipping unexpected path type at %s", path)
                continue

    def send_str(self):
        """Send an email to str@imhadmin.net if malicious symlinks are found"""
        if not self.bad_links:
            return
        bad_links = "\n".join(self.bad_links)
        top = (
            "Fixperms detected and removed the following symlinks. While these "
            "symlinks have been removed from the account in question the "
            "account requires further investigation"
        )
        self.log.info("An STR will now be sent for review by T2S staff")
        if self.args.noop:
            return
        try:
            rads.send_email(
                to_addr='str@imhadmin.net',
                subject=f'AUTO STR: bad symlinks on {self.user}',
                body=f'{top}\n\n{bad_links}',
                errs=True,
            )
        except OSError as exc:
            self.log.error(str(exc))
            self.log.info(
                "Failed to send STR. An escalation must be sent to an",
                "available T2S. Include the following information\n\n",
                bad_links,
            )