✏️ Editing: ossec.py Read Only

"""Module for managing OSSEC configuration.

We care about following aspects of OSSEC configuration:

* its relation to Imunify PAM module;
* rules configuration.

To cooperate with PAM module, OSSEC ships two configuration files:
ossec-no-pam.conf and ossec-pam.conf.  A symlink ossec.conf is created during
OSSEC packages installation that points to ossec-no-pam.conf.  On switching
PAM module on this symlink is changed to point to ossec-pam.conf.  Then OSSEC
services are restarted.

When PAM is disabled, similar operations are performed: symlink switched to
ossec-no-pam.conf and services are restarted.

Rules are downloaded from Imunify360 files server.  After downloading is
complete a hook is called which should update rules in OSSEC configuration
directory and restart OSSEC service.  Rules are copied into
/var/ossec/etc/VERSIONS/<version> directory.  Then symlinks are created
from /var/ossec/etc/dirname.d -> VERSIONS/<version>/dirname.
"""
import asyncio
import contextlib
import logging
import os
import pathlib
import shutil
from functools import lru_cache
from typing import List, Optional

from defence360agent.subsys import svcctl
from defence360agent.utils import CheckRunError, check_run
from defence360agent.utils.common import LooseVersion
from im360 import files
from .pam import PamService, PamServiceStatusValue

logger = logging.getLogger(__name__)

ETC_DIR = pathlib.Path("/var/ossec/etc")
_VERSIONS_DIR = ETC_DIR / "VERSIONS"
CONF_NAME = "ossec.conf"
DOVECOT = "dovecot"
PAM_RULES_NAMES = {
    PamService.SSHD: "rules_pam.d/320_pam_switch.xml",
    DOVECOT: "rules_pam.d/320_pam_switch_dovecot.xml",
    PamService.FTP: "rules_pam.d/320_pam_ftp.xml",
}
_PAM_CONFIG_DISABLED = {
    PamService.SSHD: PamServiceStatusValue.disabled,
    DOVECOT: PamServiceStatusValue.disabled,
    PamService.FTP: PamServiceStatusValue.disabled,
}

_RULES_DIRS = ["decoders", "rules", "rules_pam"]
_VERSION_FILE = "VERSION"
_VERSIONS_TO_KEEP = 2

rules_update_lock = asyncio.Lock()

class OssecRulesError(Exception):
    pass

@lru_cache(maxsize=1)
def _ossec_service():
    return svcctl.adaptor("ossec-hids")

def _get_pam_config_state():
    result = dict()
    for service, rule in PAM_RULES_NAMES.items():
        if (ETC_DIR / rule).exists():
            result[service] = PamServiceStatusValue.enabled
            with contextlib.suppress(FileNotFoundError):
                (ETC_DIR / (rule + ".disabled")).unlink()
        elif (ETC_DIR / (rule + ".disabled")).exists():
            result[service] = PamServiceStatusValue.disabled
        else:
            raise OssecRulesError("Absent rule {}".format(rule))
    return result

def _change_state_of_rules(target: str, enabled: bool) -> bool:
    """Check consistency of enabled/disabled rule file"""
    conf = ETC_DIR / target
    conf_disabled = ETC_DIR / (target + ".disabled")

config_changed = False
    if enabled and conf_disabled.exists():
        conf_disabled.rename(conf)
        config_changed = True
    elif not enabled and conf.exists():
        conf.rename(conf_disabled)
        config_changed = True
    return config_changed

async def configure_for_pam(pam_status: dict, force_restart=False) -> None:
    """Configure OSSEC to work with PAM enabled (if needed)."""
    config_changed = False
    current_state = _get_pam_config_state()
    if pam_status != current_state:
        for pam_service, state in pam_status.items():
            config_changed |= (
                False
                if (pam_service not in PAM_RULES_NAMES)
                else _change_state_of_rules(
                    PAM_RULES_NAMES[pam_service],
                    state == PamServiceStatusValue.enabled,
                )
            )
    if config_changed or force_restart:
        try:
            await _ossec_service().restart()
        except CheckRunError as exc:
            raise OssecRulesError("Failed to apply ossec rules {}".format(exc))

def get_rules_version() -> Optional[str]:
    """Return version of OSSEC rules downloaded from server."""
    path = os.path.join(files.Index.files_path(files.OSSEC), _VERSION_FILE)
    try:
        with open(path) as f:
            return f.read().strip()
    except OSError as exc:
        logger.warning("Error '%s' occurs when reading version file.", exc)
    return None

def get_rules_installed_version():
    versions = _sorted_versions()
    if versions:
        return versions[0]

async def _is_conf_valid() -> bool:
    try:
        await check_run(["/var/ossec/bin/ossec-logtest", "-t"])
    except CheckRunError as exc:
        logger.error("Ossec configuration is not valid: %s", exc)
        return False
    return True

def _do_prepare_new_version() -> None:
    """Copy new files to appropriate subdirectory in OSSEC config tree."""
    files_prefix = pathlib.Path(files.Index.files_path(files.OSSEC))
    version = (files_prefix / _VERSION_FILE).read_text().strip()

with contextlib.suppress(FileNotFoundError):
        shutil.rmtree(str(_VERSIONS_DIR / version))
    tmp_dir = _VERSIONS_DIR / (version + ".tmp")
    with contextlib.suppress(FileNotFoundError):
        shutil.rmtree(str(tmp_dir))

tmp_dir.mkdir()
    for dir_name in _RULES_DIRS:
        shutil.copytree(
            str(files_prefix / dir_name), str(tmp_dir / (dir_name + ".d"))
        )
    tmp_dir.rename(_VERSIONS_DIR / version)

async def _prepare_new_version() -> None:
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _do_prepare_new_version)

def _switch_version_to(version: LooseVersion) -> None:
    """Activate configuration for `version`."""
    logger.info("Selecting %s version of OSSEC configuration", version)
    for name in _RULES_DIRS:
        ossec_dir = ETC_DIR / (name + ".d")
        if ossec_dir.exists():
            if ossec_dir.is_symlink():
                ossec_dir.unlink()
            else:
                shutil.rmtree(str(ossec_dir))
        # creating relative symlink because OSSEC chroots to /var/ossec
        # during startup and absolute links will not work.
        ossec_dir.symlink_to(
            (_VERSIONS_DIR / str(version) / ossec_dir.name).relative_to(
                ETC_DIR
            )
        )

def _sorted_versions(skip_invalid: bool = True) -> List[LooseVersion]:
    """Return a list of prepared OSSEC configuration versions.

If `skip_invalid` is True (default) then only versions (directories) not
    ending in ".tmp" are returned.

Versions are sorted in descending order (latest first)."""
    return sorted(
        (
            LooseVersion(d.name)
            for d in _VERSIONS_DIR.glob("*")
            if not skip_invalid or d.suffix != ".tmp"
        ),
        reverse=True,
    )

async def _select_version() -> None:
    """Select latest version if it is valid, or second to latest otherwise."""
    try:
        pam_config_state = _get_pam_config_state()
    except OssecRulesError:
        pam_config_state = _PAM_CONFIG_DISABLED
    versions = _sorted_versions()
    _switch_version_to(versions[0])
    if len(versions) >= 2 and not await _is_conf_valid():
        _switch_version_to(versions[1])
    await configure_for_pam(pam_config_state, force_restart=True)

def _cleanup_old_versions() -> None:
    kept = 0
    for version in _sorted_versions(skip_invalid=False):
        path = pathlib.Path(_VERSIONS_DIR / str(version))
        if path.suffix == ".tmp" or kept >= _VERSIONS_TO_KEEP:
            shutil.rmtree(str(path))
        else:
            kept += 1

async def on_files_update(_, is_updated: bool) -> None:
    if not is_updated:
        return
    async with rules_update_lock:
        _VERSIONS_DIR.mkdir(0o755, exist_ok=True)
        try:
            await _prepare_new_version()
            await _select_version()
        except (OSError, OssecRulesError) as exc:
            logger.error("Failed to update OSSEC configuration: %s", exc)
        finally:
            _cleanup_old_versions()

🔐 Sid Gifari File Manager Pro