MAINHACK

Viewing / Editing File: pam.py

File is not writable. Editing disabled.
import asyncio
import functools
import logging
import signal
from enum import Enum
from typing import List, Dict

import yaml

from defence360agent.utils import CheckRunError, atomic_rewrite, check_run, run
from defence360agent.utils.kwconfig import KWConfig


_PAM_EXECUTABLE = "imunify360-pam"
logger = logging.getLogger(__name__)


class PamService:
    DOVECOT_NATIVE = "dovecot-native"
    DOVECOT_PAM = "dovecot-pam"
    FTP = "ftp"
    SSHD = "sshd"


class PamServiceStatusValue:
    enabled = "enabled"
    disabled = "disabled"


class DovecotStatus(str, Enum):
    DISABLED = "disabled"
    PAM = "pam"
    NATIVE = "native"

    def __str__(self):
        return self.value


_DEFAULT_STATUS = {
    PamService.DOVECOT_NATIVE: PamServiceStatusValue.disabled,
    PamService.DOVECOT_PAM: PamServiceStatusValue.disabled,
    PamService.FTP: PamServiceStatusValue.disabled,
    PamService.SSHD: PamServiceStatusValue.disabled,
}


class PAMError(Exception):
    pass


class _Config(KWConfig):
    SEARCH_PATTERN = r"^\s*{}\s*=\s*(.*?)\s*$"
    WRITE_PATTERN = "{}={}"
    DEFAULT_FILENAME = "/etc/pam_imunify/i360.ini"

    _IP_WHITELIST_OPTION = "whitelisted_ips_path"
    _IP_WHITELIST_DEFAULT = "/var/i360_pam_imunify/wl/ips.txt"

    def get_default(self, default: str) -> str:
        v = self.get()
        return v if v is not None else default

    @classmethod
    def ip_whitelist_path(cls) -> str:
        try:
            return (
                cls(cls._IP_WHITELIST_OPTION).get_default(
                    cls._IP_WHITELIST_DEFAULT
                )
                # whitelisted_ips_path is comma separated list
                # where user ip list path goes the last
                .split(",")[-1]
            )
        except FileNotFoundError:
            return cls._IP_WHITELIST_DEFAULT


async def _export_list(path: str, values: List[str]) -> None:
    loop = asyncio.get_event_loop()
    content = "".join([v + "\n" for v in values])
    writer = functools.partial(atomic_rewrite, path, content, backup=False)
    await loop.run_in_executor(None, writer)


async def export_ip_whitelist(networks: List[str]) -> None:
    """Save a list of `networks` into IP address whitelist."""
    await _export_list(_Config.ip_whitelist_path(), networks)


async def get_status() -> Dict:
    cmd = [_PAM_EXECUTABLE, "status", "--yaml"]
    try:
        returncode, output, err = await run(cmd)
    except FileNotFoundError:
        return _DEFAULT_STATUS
    except OSError as exc:
        raise PAMError("PAM status failed") from exc
    else:
        if returncode == -signal.SIGTERM:
            # Don't send SIGTERM to Sentry
            # (presumably on shutdown on systemctl stop imunify360)
            logger.warning(
                "SIGTERM while getting pam status, rc: %s, out: %s, err: %s",
                returncode,
                output,
                err,
            )
            return _DEFAULT_STATUS
        elif returncode != 0 or output.strip() == b"":
            raise PAMError(
                "PAM status failed: run(%r) = %r"
                % (cmd, (returncode, output, err))
            )
    log_response_warnings(output)
    try:
        return yaml.safe_load(output.decode())["status"]
    except (UnicodeDecodeError, yaml.YAMLError, TypeError, KeyError) as e:
        raise PAMError(f"Can't get pam status from {output!r}") from e


async def enable(module, dry_run=False) -> None:
    """Enable PAM module. Raises PAMError."""
    pam_command = {
        PamService.SSHD: ["enable"],
        PamService.DOVECOT_PAM: ["set-dovecot", "pam"],
        PamService.DOVECOT_NATIVE: ["set-dovecot", "native"],
        PamService.FTP: ["enable-ftp"],
    }[module]
    try:
        output = await check_run(
            [
                _PAM_EXECUTABLE,
                *pam_command,
                *(["--dry-run"] if dry_run else []),
                "--yaml",
            ]
        )
        log_response_warnings(output)
    except CheckRunError as exc:
        raise PAMError("failed to enable PAM for %s" % module) from exc


async def disable(module) -> None:
    """Disable PAM module. Raises PAMError."""
    pam_command = {
        PamService.SSHD: ["disable"],
        PamService.DOVECOT_PAM: ["set-dovecot", "disabled"],
        PamService.DOVECOT_NATIVE: ["set-dovecot", "disabled"],
        PamService.FTP: ["disable-ftp"],
    }[module]
    try:
        output = await check_run([_PAM_EXECUTABLE, *pam_command, "--yaml"])
        log_response_warnings(output)
    except CheckRunError as exc:
        raise PAMError("failed to disable PAM for %s" % module) from exc


def log_response_warnings(output):
    try:
        response = yaml.safe_load(output.decode())
    except (yaml.YAMLError, UnicodeDecodeError):
        logger.warning("Not yaml response for %s: %s", _PAM_EXECUTABLE, output)
        response = None
    warnings = response and response.get("warnings")
    if warnings:
        logger.warning("imunify360-pam warnings: %s", warnings)
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: pam.py

Server Info

System Features

User Info
