Da

✏️ Editing: cagefs.py

"""
CageFS is a CloudLinux feature which allows to restrict access to files of other users.
This module contains utility function to work with it.
"""

from parallels.core import messages
from parallels.core.logging import get_logger
from parallels.core.runners.base import BaseRunner
from parallels.core.runners.entities import ExecutionOptions
from parallels.core.utils.common import cached

logger = get_logger(__name__)

CAGEFS_BINARY = '/usr/sbin/cagefsctl'

@cached
def is_cagefs_enabled_for_server(server):
    """Check whether CageFS is enabled on the server

:type server: parallels.core.connections.server.Server
    :rtype: bool
    """
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        if runner.file_exists(CAGEFS_BINARY):
            try:
                output = runner.execute_command(
                    '{cagefsctl} --cagefs-status',
                    dict(cagefsctl=CAGEFS_BINARY),
                    ExecutionOptions(ignore_exit_code=True)
                ).stdout
                if 'enabled' in output.lower():
                    logger.fdebug(messages.CAGEFS_IS_ENABLED_FOR_SERVER, server=server.description())
                    return True
                else:
                    return False
            except Exception as e:
                logger.exception()
                logger.fwarn(
                    messages.FAILED_TO_DETECT_CAGEFS_STATUS,
                    reason=e, server=server.description()
                )
                # Failed to detect status - consider that CageFS is disabled
                return False
        else:
            # No CageFS binary, most probably it is not CloudLinux or CageFS is not installed
            return False

def is_cagefs_enabled_for_user(server, username):
    """Check whether CageFS is enabled on the server for the specified user

:type server: parallels.core.connections.server.Server
    :type username: str | unicode
    :rtype: bool
    """
    if not is_cagefs_enabled_for_server(server):
        return False

with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        try:
            output = runner.execute_command(
                '{cagefsctl} --user-status {username}',
                dict(cagefsctl=CAGEFS_BINARY, username=username),
                ExecutionOptions(ignore_exit_code=True)
            ).stdout
            if 'enabled' in output.lower():
                logger.fdebug(messages.CAGEFS_IS_ENABLED_FOR_USER, server=server.description(), username=username)
                return True
            else:
                return False
        except Exception as e:
            logger.exception()
            logger.fwarn(
                messages.FAILED_TO_DETECT_CAGEFS_STATUS_FOR_USER,
                reason=e, server=server.description(), username=username
            )
            # Failed to detect status - consider that CageFS is disabled
            return False

def disable_cagefs(server, username):
    """Disable CageFS for specified user

:type server: parallels.core.connections.server.Server
    :type username: str | unicode
    :rtype: None
    """
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        runner.execute_command(
            u'{cagefsctl} --disable {username}',
            dict(cagefsctl=CAGEFS_BINARY, username=username)
        )

def enable_cagefs(server, username):
    """Enable CageFS for specified user

:type server: parallels.core.connections.server.Server
    :type username: str | unicode
    :rtype: None
    """
    with server.runner() as runner:
        assert isinstance(runner, BaseRunner)
        runner.execute_command(
            u'{cagefsctl} --enable {username}',
            dict(cagefsctl=CAGEFS_BINARY, username=username)
        )

📁 Sid Gifari File Manager