/opt/imunify360/venv/lib/python3.11/site-packages/imav/subsys
Edit: /opt/imunify360/venv/lib/python3.11/site-packages/imav/subsys/realtime_av.py (15118B)
"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see
.
Copyright © 2019 Cloud Linux Software Inc.
This software is also available under ImunifyAV commercial license,
see
"""
import asyncio
import base64
import logging
import os
import re
import shutil
from pathlib import Path
from typing import Callable, Iterable, List, Optional, Set, Tuple
from defence360agent.contracts.config import ANTIVIRUS_MODE, Malware
from defence360agent.subsys import svcctl
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.utils import check_run
from defence360agent.utils.common import LooseVersion
from imav.malwarelib.model import MalwareIgnorePath
from imav.malwarelib.scan.crontab import crontab_path
logger = logging.getLogger(__name__)
# location of admin provided watched and ignored paths
_ADMIN_PATH = Path("/etc/sysconfig/imunify360/malware-filters-admin-conf")
# location of internal configs, shipped with imunify360-firewall
_INTERNAL_PATH = Path("/var/imunify360/files/realtime-av-conf/v1")
# location of processed configs
_PROCESSED_PATH = _ADMIN_PATH / "processed"
_PD_NAME = "pd-combined.txt"
_INTERNAL_NAME = "av-internal.txt"
_INTERNAL_OPTIMIZED_NAME = "av-internal-optimized.txt"
_ADMIN_NAME = "av-admin.txt"
_ADMIN_PATHS_NAME = "av-admin-paths.txt"
_IGNORED_SUB_DIR = "ignored"
_MAX_PATTERN_LENGTH = 64000
REALTIME_PACKAGE = "imunify-realtime-av"
REALTIME_SERVICE_NAME = "imunify-realtime-av"
_PD_PREPARE = "/usr/bin/i360-exclcomp"
_BIN_PATH = Path("/usr/sbin/imunify-realtime-av")
class PatternLengthError(Exception):
"""Raised when pattern's length is too big."""
pass
def _save_basedirs(dir: Path, basedirs: Set[str]) -> None:
"""Save list of basedirs in a file inside dir."""
with (dir / "basedirs-list.txt").open("w") as f:
for basedir in sorted(basedirs):
f.write(os.path.realpath(basedir) + "\n")
def _split_paths(paths: List[str]) -> Tuple[List[str], List[str]]:
"""Split paths into two lists: absolute and relative.
Relative paths start with +. This + sign is removed from resulting path."""
absolute, relative = [], []
for path in paths:
if path.startswith("+"):
relative.append(path[1:])
else:
absolute.append(path)
return absolute, relative
def _read_list(path: Path) -> List[str]:
"""Read file at path and return its lines as a list.
Empty lines or lines starting with '#' symbol are skipped. Lines are
stripped of leading and trailing whitespace. If the file does not exist,
empty list is returned."""
try:
with path.open() as f:
lines = [line.strip() for line in f]
return [x for x in lines if len(x) > 0 and not x.startswith("#")]
except FileNotFoundError:
return []
class _Watched(list):
"""Holds a list of watched glob patterns ready to be saved."""
def __init__(self, w: List[str], basedirs: Set[str]) -> None:
super().__init__()
absolute, relative = _split_paths(w)
self.extend(
os.path.realpath(p)
for p in absolute + self._extend_relative(relative, basedirs)
if self._is_valid(p)
)
@staticmethod
def _is_valid(pattern: str) -> bool:
"""Return True if watched pattern is valid."""
if not pattern.startswith("/"):
logger.warning(
"skipping watched path %s: not starts with /", pattern
)
return False
return True
@staticmethod
def _extend_relative(paths: List[str], basedirs: Set[str]) -> List[str]:
"""Join basedirs with all paths and return resulting list."""
extended = []
for path in paths:
for basedir in basedirs:
extended.append(os.path.join(basedir, path))
return extended
def save(self, path: Path) -> None:
"""Save watched list at specified path."""
with path.open("w") as f:
f.write("\n".join(self))
class _Ignored(str):
"""Holds a list of ignored regexp patterns ready to be saved."""
@staticmethod
def _is_valid_relative(pattern: str) -> bool:
"""Return True if relative ignored pattern is valid."""
if pattern.startswith("^"):
logger.warning(
"skipping relative ignored path %s: starts with ^", pattern
)
return False
return True
@staticmethod
def _remove_leading_slash(pattern: str) -> str:
"""Remove leading slash from pattern, if present."""
if pattern.startswith("/"):
return pattern[1:]
return pattern
@staticmethod
def _compiles(pattern: str) -> bool:
"""Return True if pattern successfully compiles as regexp."""
try:
re.compile(pattern)
return True
except Exception:
logger.warning(
"skipping ignored pattern %s: invalid regex", pattern
)
return False
@classmethod
def from_patterns(
cls, patterns: List[str], basedirs: Set[str]
) -> "_Ignored":
"""Build single ignored regexp from given patterns and basedirs."""
absolute, relative = _split_paths(patterns)
absolute = [p for p in absolute if cls._compiles(p)]
relative = [
cls._remove_leading_slash(p)
for p in relative
if cls._is_valid_relative(p) and cls._compiles(p)
]
if len(basedirs) > 0 and len(relative) > 0:
relative_pattern = "^(?:{})/(?:{})".format(
"|".join(basedirs), "|".join(relative)
)
absolute.append(relative_pattern)
pat = "|".join(absolute)
if pat == "":
pat = "^$"
return _Ignored(pat)
def save(self, path: Path):
"""Save ignored list at specified path."""
if len(self) > _MAX_PATTERN_LENGTH:
raise PatternLengthError(
"{} pattern is too long ({})".format(path, len(self))
)
with path.open("w") as f:
f.write(self)
def _read_configs(panel: str, name: str) -> Tuple[List[str], List[str]]:
"""Read internal and admin lists from files with given name."""
common_dir = _INTERNAL_PATH / "common"
internal = _read_list(common_dir / name)
panel_path = _INTERNAL_PATH / panel.lower()
if panel_path.exists():
internal.extend(_read_list(panel_path / name))
return internal, _read_list(_ADMIN_PATH / name)
class _WatchedCtx:
def __init__(self, internal: _Watched, admin: _Watched) -> None:
self.internal = internal
self.admin = admin
def save(self, dir: Path) -> None:
w = dir / "watched"
w.mkdir(exist_ok=True)
self.internal.save(w / _INTERNAL_NAME)
self.admin.save(w / _ADMIN_NAME)
def _watched_context(
panel_name: str, basedirs: Set[str], *, extra: Iterable[str]
) -> _WatchedCtx:
internal_watched, admin_watched = _read_configs(panel_name, "watched.txt")
internal_watched.extend(extra)
return _WatchedCtx(
_Watched(internal_watched, basedirs), _Watched(admin_watched, basedirs)
)
class _IgnoredCtx:
def __init__(
self,
internal: _Ignored,
admin: _Ignored,
pd: _Ignored,
internal_optimized: Optional[_Ignored] = None,
) -> None:
self.internal = internal
self.admin = admin
self.pd = pd
self.internal_optimized = internal_optimized
def save(self, dir: Path) -> None:
w = dir / _IGNORED_SUB_DIR
w.mkdir(exist_ok=True)
self.internal.save(w / _INTERNAL_NAME)
self.admin.save(w / _ADMIN_NAME)
self.pd.save(w / _PD_NAME)
if self.internal_optimized is not None:
self.internal_optimized.save(w / _INTERNAL_OPTIMIZED_NAME)
def _ignored_context(panel_name: str, basedirs: Set[str]) -> _IgnoredCtx:
internal_ignored, admin_ignored = _read_configs(panel_name, "ignored.txt")
# Read optimized list if available (DEF-35124).
# Built from optimized_ignored only (no admin), same as av-internal.txt
# is built from internal_ignored only. Scanners read av-admin.txt
# separately, so admin patterns must not be baked in here.
optimized_ignored, _ = _read_configs(panel_name, "ignored-optimized.txt")
internal_optimized = None
if optimized_ignored:
internal_optimized = _Ignored.from_patterns(
optimized_ignored, basedirs
)
return _IgnoredCtx(
_Ignored.from_patterns(internal_ignored, basedirs),
_Ignored.from_patterns(admin_ignored, basedirs),
_Ignored.from_patterns(internal_ignored + admin_ignored, basedirs),
internal_optimized,
)
def _admin_ignored_paths(dir: Path) -> None:
ignored_paths = MalwareIgnorePath.path_list()
ignored_paths_base64 = b"".join(
base64.b64encode(os.fsencode(path)) + b"\n" for path in ignored_paths
)
target = dir / _IGNORED_SUB_DIR / _ADMIN_PATHS_NAME
target.write_bytes(ignored_paths_base64)
def _contain_changes(dir1: Path, dir2: Path) -> bool:
"""Compare content of two folders if files in this directory are the
same return False."""
for file in dir1.iterdir():
if file.is_dir():
if _contain_changes(file, dir2 / file.name):
return True
if not file.is_file():
continue
other = dir2 / file.name
if not other.exists():
return True
if file.read_bytes() != other.read_bytes():
return True
return False
def _save_configs(dir: Path, savers: List[Callable[[Path], None]]) -> bool:
"""Save configs in directory dir using saves callable.
Each function in savers will be called with single dir argument."""
temp = dir.with_suffix(".tmp")
if temp.exists():
shutil.rmtree(str(temp))
temp.mkdir()
for save in savers:
save(temp)
if dir.exists():
backup = dir.with_name(".backup")
if backup.exists():
shutil.rmtree(str(backup))
dir.rename(backup)
try:
temp.rename(dir)
except Exception:
backup.rename(dir)
raise
return _contain_changes(dir, backup)
else:
temp.rename(dir)
return True
def _update_pd_symlink() -> None:
target = _PROCESSED_PATH / _IGNORED_SUB_DIR / _PD_NAME
source = _ADMIN_PATH / _PD_NAME
try:
# source.exists() returns False for broken symlink.
# so call lstat() and if it throws exception, source does not exist.
_ = source.lstat()
except FileNotFoundError:
source.symlink_to(target)
else:
if not (
source.is_symlink() and os.readlink(str(source)) == str(target)
):
source.unlink()
source.symlink_to(target)
def generate_configs() -> bool:
"""Generate new malware paths filters config."""
panel = HostingPanel()
basedirs = panel.basedirs()
extra_watched = set()
if Malware.CRONTABS_SCAN_ENABLED:
extra_watched.add(str(crontab_path()))
changed = _save_configs(
_PROCESSED_PATH,
[
lambda dir: _save_basedirs(dir, {*basedirs, *extra_watched}),
_watched_context(panel.NAME, basedirs, extra=extra_watched).save,
_ignored_context(panel.NAME, basedirs).save,
_admin_ignored_paths,
],
)
_update_pd_symlink()
return changed
def is_installed() -> bool:
return _BIN_PATH.exists()
# the alternatives symlink resolves to .imrt2 or .legacy
_IMRT2_ALTERNATIVES_SUFFIX = ".imrt2"
# older imrt2 could absorb a config change landing during its in-process
# reload cooldown; only the agent-side restart masked that
_MIN_SELF_RELOAD_VERSION = "8.0.20"
_VERSION_RE = re.compile(rb"version\s+(\d+(?:\.\d+)+)")
_self_reload_cache: Tuple[Optional[Tuple[str, float]], bool] = (None, False)
async def _self_reload_capable() -> bool:
"""Whether the active realtime binary picks up config changes itself."""
global _self_reload_cache
try:
real = _BIN_PATH.resolve(strict=True)
except OSError:
return False
if not real.name.endswith(_IMRT2_ALTERNATIVES_SUFFIX):
return False
try:
key = (str(real), real.stat().st_mtime)
except OSError:
return False
cached_key, cached_value = _self_reload_cache
if cached_key == key:
return cached_value
capable = False
try:
out = await check_run([str(_BIN_PATH), "-version"])
match = _VERSION_RE.search(out)
if match:
capable = LooseVersion(match.group(1).decode()) >= LooseVersion(
_MIN_SELF_RELOAD_VERSION
)
except asyncio.CancelledError:
raise
except Exception as e:
logger.warning("realtime_av version probe failed: %s", e)
_self_reload_cache = (key, capable)
return capable
async def reload_services() -> None:
"""Apply freshly regenerated malware-filter configs."""
# a self-reloading imrt2 picks the configs up in-process; the historical
# restart also resurrected a stopped unit, so keep that via an idempotent
# start-when-inactive; ANY probe failure falls back to the restart
action: Optional[str] = None
if not await _self_reload_capable():
action = "restart"
else:
try:
unit = svcctl.adaptor(REALTIME_SERVICE_NAME)
if not await unit.is_active():
# a failed unit is also inactive; clear its state (and any
# start-rate-limit counter) so the plain start below cannot
# be refused — no-op on a cleanly stopped unit
await unit.reset_failed()
action = "start"
except Exception as e:
logger.warning(
"realtime_av.reload_services unit-state check failed: %s", e
)
action = "restart"
tasks = []
if action:
tasks.append(check_run(["service", REALTIME_SERVICE_NAME, action]))
tasks.append(check_run([_PD_PREPARE]))
for t in tasks:
try:
await t
except asyncio.CancelledError:
raise
except Exception as e:
logger.warning("realtime_av.reload_services exception: %s", e)
def should_be_running() -> bool:
return not ANTIVIRUS_MODE and Malware.INOTIFY_ENABLED