"""
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 pwd
import os
import re
import shutil
import time
import uuid
from contextlib import suppress
from functools import partial
from logging import getLogger
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
from defence360agent import utils
from defence360agent.api import inactivity
from defence360agent.contracts.config import (
Malware as Config,
)
from defence360agent.contracts.config import (
MalwareTune,
MyImunifyConfig,
)
from defence360agent.contracts.hook_events import HookEvent
from defence360agent.contracts.license import LicenseCLN
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.permissions import myimunify_protection_enabled
from defence360agent.contracts.plugins import (
MessageSink,
MessageSource,
expect,
)
from defence360agent.internals.global_scope import g
from defence360agent.model import tls_check
from defence360agent.subsys.persistent_state import register_lock_file
from defence360agent.utils import (
Scope,
is_system_user,
nice_iterator,
recurring_check,
safe_cancel_task,
split_for_chunk,
)
from defence360agent.utils.check_lock import check_lock
from defence360agent.utils.common import DAY, MINUTE, rate_limit
from imav.contracts.messages import MalwareDatabaseRestoreTask
from imav.malwarelib.cleanup.cleaner import (
CleanupResult,
MalwareCleaner,
MalwareCleanupProxy,
)
from imav.malwarelib.cleanup.reasons import local_details, log_diagnostic
from imav.malwarelib.cleanup.storage import CleanupStorage
from imav.malwarelib.config import (
MalwareHitStatus,
MalwareScanResourceType,
MalwareScanType,
)
from imav.malwarelib.hit_removed_hook import (
EXPIRED,
delete_and_report,
hooked,
)
from imav.malwarelib.model import MalwareHistory, MalwareHit
from imav.malwarelib.scan import ScanAlreadyCompleteError
from imav.malwarelib.scan.crontab import is_crontab
from imav.malwarelib.scan.mds.cleaner import MalwareDatabaseCleaner
from imav.malwarelib.scan.mds.detached import (
MDSDetachedCleanup,
MDSDetachedRestore,
)
from imav.malwarelib.scan.mds.restore import MalwareDatabaseRestore
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from imav.malwarelib.stop_process import cron as stop_cron
from imav.malwarelib.stop_process import mode as stop_mode
from imav.malwarelib.stop_process import registry as stop_registry
from imav.malwarelib.stop_process import proc as stop_proc
from imav.malwarelib.stop_process import report as stop_report
from imav.malwarelib.stop_process.controller import StopProcessController
from imav.malwarelib.stop_process.registry import release_orphans
from imav.malwarelib.stop_process.stopper import (
COULD_NOT_FREEZE,
FILE_NOT_CLEANED,
FREEZE_TIMEOUT,
READ_FAILED,
RESPAWNING,
STOP_CAP,
cleaner_outcome,
)
from imav.malwarelib.subsys.malware import HackerTrapHitsSaver, MalwareAction
from imav.malwarelib.tenant_path import gather_exists, safe_exists
from imav.malwarelib.utils import malware_response
from imav.malwarelib.utils.user_list import (
get_username_by_uid,
is_uid,
)
from imav.malwarelib.model import MalwareIgnorePath
logger = getLogger(__name__)
#: seconds a small cleaner pass may still finish once the agent stops;
#: then its curator is terminated, as released cleanup does on cancel
SMALL_PASS_GRACE = 2.0
#: the owner's other detected files looked at before one of its processes
#: is stopped (newest first)
MAX_VETO_HITS = 10_000
#: seconds between two looks at the cron store's pending re-enables
CRON_RETRY_INTERVAL = 60.0
_NOT_CLEANED_YET = (
MalwareHitStatus.FOUND,
MalwareHitStatus.CLEANUP_STARTED,
MalwareHitStatus.CLEANUP_REQUIRES_MYIMUNIFY_PROTECTION,
MalwareHitStatus.RESTORE_FROM_BACKUP_STARTED,
)
def _other_detected_hits(users, batch_ids) -> List[MalwareHit]:
"""The file hits of these accounts (the indexed home owner) detected
and not cleaned yet, outside the batch: a process linked to one of
them is never stopped. Runs in the executor."""
with suppress(tls_check.OverridingReset):
tls_check.reset()
rows = [
hit
for hit in MalwareHit.select(
MalwareHit.id,
MalwareHit.orig_file,
MalwareHit.user,
MalwareHit.owner,
MalwareHit.file_dev,
MalwareHit.file_ino,
MalwareHit.file_uid,
)
.where(
MalwareHit.user.in_(sorted(users)),
MalwareHit.resource_type == MalwareScanResourceType.FILE.value,
MalwareHit.status.in_(_NOT_CLEANED_YET),
)
.order_by(MalwareHit.id.desc())
.limit(MAX_VETO_HITS + len(batch_ids) + 1)
if hit.id not in batch_ids
]
if len(rows) > MAX_VETO_HITS:
logger.info(
"More than %d detected files of %s are not cleaned yet: the"
" oldest are not checked before a process is stopped",
MAX_VETO_HITS,
", ".join(sorted(users)),
)
return rows[:MAX_VETO_HITS]
# Deferred missing-file evidence must not prevent the agent going idle.
MAX_GONE_STOP_ATTEMPTS = 5
COUNT_OF_ATTEMPTS_TO_CLEANUP_PER_DAY = 4
# A path we keep cleaning successfully still burns a full backup copy on
# every failed round, so the reset above needs a ceiling of its own.
COUNT_OF_FAILURES_TO_CLEANUP_PER_DAY = 16
RESCAN_ATTEMPTS = 2
LOCK_FILE = register_lock_file("cleanup", Scope.AV_IM360)
def _record_stop_history(out, hits, cause, initiator):
"""Write one account's rows in a worker-owned DB connection."""
with suppress(tls_check.OverridingReset):
tls_check.reset()
database = MalwareHistory._meta.database
with database.connection_context():
stop_report.record_locally(out, hits, cause=cause, initiator=initiator)
# accounts web servers execute client code as: www-data, apache, EL7 nobody.
# They stay cleanable on our side; the released cleaner still refuses 48/99
# itself until its guard adopts the same set, so until then they behave as
# before this guard existed.
WEB_EXEC_UIDS = frozenset((33, 48, 99))
def _system_owned(uid: int) -> bool:
return uid not in WEB_EXEC_UIDS and is_system_user(uid)
def _refused_by_owner_guard(hit) -> bool:
return (
is_uid(hit.owner)
and _system_owned(int(hit.owner))
and not is_crontab(hit.orig_file_path)
)
_group_by_status = partial(MalwareHit.group_by_attribute, attribute="status")
_group_by_user = partial(MalwareHit.group_by_attribute, attribute="owner")
def _homes(users):
homes = {}
for user in users:
try:
homes[user] = pwd.getpwnam(user).pw_dir
except Exception as error:
logger.debug("Resolving cleanup account home failed: %r", error)
homes[user] = ""
return homes
def _kernel() -> str:
return os.uname().release
throttled_log_error = rate_limit(period=DAY, on_drop=logger.warning)(
logger.error
)
def filter_cleanable(hits: Iterable[MalwareHit]) -> Iterable:
return (hit for hit in hits if hit.status == MalwareHitStatus.FOUND)
def drop_stored_originals(hits: list) -> list:
"""Forget hits that point inside the cleanup storage.
They are copies of files detected elsewhere, and they can never be
cleaned: only root may enter the storage, while the cleanup drops
privileges to the owner of the file the copy was made from.
"""
stored = [hit for hit in hits if CleanupStorage.contains(hit.orig_file)]
if not stored:
return hits
logger.warning(
"Removing %s hit(s) inside %s: a stored original is not a detection"
" of its own",
len(stored),
CleanupStorage.path,
)
MalwareHit.delete_instances(stored)
stored_ids = {hit.id for hit in stored}
return [hit for hit in hits if hit.id not in stored_ids]
def release_frozen_orphans():
"""SIGCONT every process the previous agent left frozen (its registry
entry, pid and start time still matching); the entries released."""
try:
released = release_orphans()
except Exception:
logger.exception("Could not release processes frozen earlier")
return []
for fp in released:
logger.info(
"Released pid %s, left frozen for %s by an earlier stop step",
fp.pid,
fp.file,
)
return released
class _StopCleanupCancelled(asyncio.CancelledError):
def __init__(self, state):
super().__init__()
self.state = state
class Cleanup(MessageSink, MessageSource):
def __init__(self):
self._cleanup_task = None
self._store_original_task = None
self._running = False
self._loop = None
self._sink = None
self._proxy = None
self._cleaner = None
self._stop_controller = None
self._retry_task = None
self._gone_pending = [] # (hit, cause, initiator) of gone files
self._gone_attempts = {} # id(entry) -> (owned entry, attempts)
self._cron_checked_at = float("-inf")
self._run_reports = [] # (out, hits, cause, initiator) of one run
self._rollup_loaded = False
async def create_source(self, loop, sink):
self._loop = loop
self._sink = sink
# Load before constructors as well as before the first await. If
# startup fails, shutdown may save only state it actually owns.
stop_report.load_rollup()
self._rollup_loaded = True
self._proxy = MalwareCleanupProxy()
self._cleaner = MalwareCleaner(loop=loop, sink=sink)
if self._stop_controller is None:
self._stop_controller = StopProcessController()
# a process frozen when the agent died must not stay frozen
released, release_cancelled = await stop_report._complete_owned(
loop.run_in_executor(None, release_frozen_orphans)
)
report_cancelled = False
try:
await stop_report.report_released(
self._sink,
released,
kernel=_kernel(),
panel=HostingPanel().NAME,
)
except asyncio.CancelledError:
report_cancelled = True
except Exception:
logger.exception("Could not report the processes released")
if release_cancelled or report_cancelled:
raise asyncio.CancelledError
self._cleanup_task = loop.create_task(self.cleanup())
async def create_sink(self, loop):
pass
async def shutdown(self):
if self._cleanup_task:
await safe_cancel_task(self._cleanup_task)
try:
await self._stop_retry_task()
finally:
if stop_registry.REGISTRY_PATH.exists():
# no process may stay frozen once the agent is gone
await self._loop.run_in_executor(None, release_frozen_orphans)
async def _stop_retry_task(self):
if self._retry_task is not None and not self._retry_task.done():
task = self._retry_task
task.cancel()
caller_cancelled = False
while not task.done():
try:
await asyncio.shield(task)
except asyncio.CancelledError:
current = asyncio.current_task()
caller_cancelled = caller_cancelled or bool(
current and current.cancelling()
)
try:
task.result()
except asyncio.CancelledError:
pass
if caller_cancelled:
raise asyncio.CancelledError
# the roll-up is not sent at exit (the agent exits when idle):
# the next agent sends it when its hour is over
if self._rollup_loaded:
stop_report.save_rollup()
@expect(MessageType.MalwareCleanupTask)
async def process_cleanup_task(self, message: Dict):
cause = message.get("cause")
initiator = message.get("initiator")
post_action = message.get("post_action")
scan_id = message.get("scan_id")
standard_only = message.get("standard_only")
manual_cleanup = cause is None
# In case another scan already found some of the hits
# and the cleanup for them has started.
origin_hits_num = len(message["hits"])
hits = MalwareHit.refresh_hits(
message["hits"], include_scan_info=not manual_cleanup
)
hits = filter_cleanable(hits)
hits = [
hit
for hit in hits
if hit.resource_type == MalwareScanResourceType.FILE.value
]
hits = drop_stored_originals(hits)
if (
not manual_cleanup
): # don't use any limits when run cleanup manually
rescan_hits, hits = self._split_hits_by_scan_type(
hits, [MalwareScanType.RESCAN, MalwareScanType.RESCAN_OUTDATED]
)
rescan_hits = await self._filter_rescan_hits(rescan_hits)
hits = rescan_hits + await self._filter_failed_to_cleanup_hits(
hits
)
if filtered := origin_hits_num - len(hits):
logger.info(
"%s/%s hits filtered before cleanup",
filtered,
origin_hits_num,
)
self._store_original_task = self._loop.create_task(
self._store_original(
hits, cause, initiator, post_action, scan_id, standard_only
)
)
@staticmethod
def _split_hits_by_scan_type(
hits: list, scan_types: List[MalwareScanType]
) -> Tuple[list, list]:
target_hits, other_hits = [], []
for hit in hits:
if hit.scanid.type in scan_types:
target_hits.append(hit)
else:
other_hits.append(hit)
return target_hits, other_hits
async def _filter_failed_to_cleanup(
self,
hits: list,
*,
time_range: float,
allowed_total: int,
allowed_after_success: Optional[int] = None,
report_give_up: bool = False,
) -> list:
hits_to_clean = []
suppressed = []
if hits:
since = time.time() - time_range
attempts = {}
for hits_chunk in split_for_chunk(hits, chunk_size=200):
counted = MalwareHistory.get_failed_cleanup_attempts(
[hit.orig_file for hit in hits_chunk], since=since
)
for path, total, after_success in counted:
attempts[path] = (total, after_success)
await asyncio.sleep(0)
for hit in hits:
total, after_success = attempts.get(hit.orig_file, (0, 0))
exhausted = total >= allowed_total or (
allowed_after_success is not None
and after_success >= allowed_after_success
)
if exhausted:
throttled_log_error(
"Skip cleanup file '%s', since there are too many "
"attempts to cleanup it in %s sec [%s, %s since the "
"last successful cleanup]",
hit.orig_file,
time_range,
total,
after_success,
)
if hit.orig_file not in suppressed:
suppressed.append(hit.orig_file)
continue
hits_to_clean.append(hit)
if suppressed and report_give_up:
await self._report_suppressed_cleanup(suppressed, time_range)
return hits_to_clean
async def _report_suppressed_cleanup(self, paths: list, time_range: float):
# The only signal that the agent decided not to try: without it a
# path that is being skipped looks exactly like one with nothing
# left to do.
await self._sink.process_message(
MessageType.CleanupFailed(
message=(
"Automatic cleanup suppressed for {} path(s) that"
" exhausted the attempt budget in {} sec: {}".format(
len(paths), time_range, ", ".join(paths[:3])
)
),
exception="CleanupAttemptsExhausted",
err="",
timestamp=int(time.time()),
)
)
async def _filter_rescan_hits(self, hits: list) -> list:
# A burst guard, not a give-up decision: a success minutes ago is no
# reason to keep hammering the same path, so no reset applies here.
return await self._filter_failed_to_cleanup(
hits, time_range=5 * MINUTE, allowed_total=RESCAN_ATTEMPTS
)
async def _filter_failed_to_cleanup_hits(self, hits: list) -> list:
return await self._filter_failed_to_cleanup(
hits,
time_range=DAY,
report_give_up=True,
allowed_total=(
MalwareTune.CLEANUP_FAILURES_PER_DAY
or COUNT_OF_FAILURES_TO_CLEANUP_PER_DAY
),
allowed_after_success=(
MalwareTune.CLEANUP_ATTEMPTS_AFTER_SUCCESS
or COUNT_OF_ATTEMPTS_TO_CLEANUP_PER_DAY
),
)
async def _store_original(
self, hits, cause, initiator, post_action, scan_id, standard_only
):
MalwareHit.set_status(hits, MalwareHitStatus.CLEANUP_STARTED)
original_status = _group_by_status(hits)
# Re-check ignore list after marking as started, before any destructive action
hits_to_keep = []
ignored_hits = []
for hit in hits:
try:
if await MalwareIgnorePath.is_path_ignored(hit.orig_file):
ignored_hits.append(hit)
else:
hits_to_keep.append(hit)
except Exception as exc:
# Be conservative: if ignore check fails, keep the hit
logger.exception(
"Ignore re-check failed for file %s: %s; keeping hit",
hit.orig_file,
exc,
)
hits_to_keep.append(hit)
if ignored_hits:
# Transition ignored hits to a terminal non-in-progress state
MalwareHit.set_status(ignored_hits, MalwareHitStatus.FOUND)
# Do not proceed with storage/cleanup for ignored hits
hits = hits_to_keep
# _clean_files refuses these before the cleaner ever runs, so a
# backup copy of them is work nobody will read.
refused = [hit for hit in hits if _refused_by_owner_guard(hit)]
if refused:
refused_set = set(refused)
hits = [hit for hit in hits if hit not in refused_set]
with inactivity.track.task("cleanup_storage"):
succeeded, failed, not_exist = await CleanupStorage.store_all(hits)
for hit, exc in failed.items():
await self._sink.process_message(
MessageType.CleanupFailed(
message=(
"Failed to store the original from {} to {}".format(
hit.orig_file, CleanupStorage.path
)
),
exception=type(exc).__name__,
err=str(exc),
timestamp=int(time.time()),
)
)
self._add_to_proxy(
# succeeded may be a query, and + would union it, not extend it
[*succeeded, *refused] if refused else succeeded,
cause,
initiator,
post_action,
scan_id,
standard_only,
)
failed_ids = {hit.id for hit in failed}
for status, hit_list in original_status.items():
MalwareHit.set_status(
[hit for hit in hit_list if hit.id in failed_ids], status
)
# Own missing-file evidence before deleting its database row or
# awaiting history I/O. Either operation can fail or be cancelled,
# and the recorded inode is then the only input left for a retry.
self._gone_pending.extend(
(hit, cause, initiator)
for hit in not_exist
if hit.resource_type == MalwareScanResourceType.FILE.value
)
MalwareHit.delete_instances(not_exist)
await MalwareAction.not_exist(
not_exist, cause=cause, initiator=initiator
)
def _add_to_proxy(
self, hits, cause, initiator, post_action, scan_id, standard_only
):
standard_only_hits = []
advanced_hits = []
for hit in hits:
standard_only_user = decide_if_standard_signatures_only(
initiator, standard_only
)
if standard_only_user:
standard_only_hits.append(hit)
else:
advanced_hits.append(hit)
self._proxy.add(
cause,
initiator,
post_action,
scan_id,
True,
standard_only_hits,
)
self._proxy.add(
cause,
initiator,
post_action,
scan_id,
standard_only, # None if default action otherwise False
advanced_hits,
)
@staticmethod
def _user_hits(hits):
user_hits = _group_by_user(hits)
return user_hits
def _cloud_assisted_hits(self):
action_hits = self._proxy.flush()
for (
cause,
initiator,
post_action,
scan_id,
standard_only,
all_hits,
) in action_hits:
blacklist = [
hit
for hit in all_hits
if re.match(r"\w+-BLKH-|cloudhash\.|cld-", hit.type)
]
regular_hits = [hit for hit in all_hits if hit not in blacklist]
yield (
regular_hits,
blacklist,
cause,
initiator,
post_action,
scan_id,
standard_only,
)
async def _start_hook(self, cleanup_id, started, hits):
dump = [hit.as_dict() for hit in hits]
cleanup_started = HookEvent.MalwareCleanupStarted(
cleanup_id=cleanup_id,
started=started,
total_files=len(hits),
DUMP=dump,
)
await self._sink.process_message(cleanup_started)
async def _clean_files(
self,
hits,
blacklist=None,
cause=None,
initiator=None,
post_action=None,
scan_id=None,
standard_only=None,
gone_entries=None,
):
user_hits = self._user_hits(hits)
user_hits_black = self._user_hits(blacklist or [])
gone_entries = gone_entries if gone_entries is not None else []
users = sorted({*user_hits, *user_hits_black})
def unvisited_hits(after_index):
remaining = []
for unvisited_user in users[after_index + 1 :]:
remaining.extend(user_hits.get(unvisited_user, []))
remaining.extend(user_hits_black.get(unvisited_user, []))
return remaining
def requeue(pending):
if pending:
self._add_to_proxy(
pending,
cause,
initiator,
post_action,
scan_id,
standard_only,
)
for user_index, user in enumerate(users):
hits_regular = user_hits.get(user, [])
hits_black = user_hits_black.get(user, [])
# Filter out ignored paths before invoking cleaner (DEF-36692)
filtered_hits_regular = []
for hit in hits_regular:
try:
if await MalwareIgnorePath.is_path_ignored(hit.orig_file):
# Transition to a terminal state instead of silently skipping
MalwareHit.set_status([hit], MalwareHitStatus.FOUND)
else:
filtered_hits_regular.append(hit)
except Exception as exc:
# be conservative: if ignore check fails, do not drop the hit
logger.exception(
"Ignore check failed for file %s: %s; keeping hit",
hit.orig_file,
exc,
)
filtered_hits_regular.append(hit)
filtered_hits_black = []
for hit in hits_black:
try:
if await MalwareIgnorePath.is_path_ignored(hit.orig_file):
MalwareHit.set_status([hit], MalwareHitStatus.FOUND)
else:
filtered_hits_black.append(hit)
except Exception as exc:
logger.exception(
"Ignore check failed for file %s: %s; keeping hit",
hit.orig_file,
exc,
)
filtered_hits_black.append(hit)
hits_regular = filtered_hits_regular
hits_black = filtered_hits_black
user_hits_all = hits_regular + hits_black
# DEF-40165: skip files that no longer exist on disk so they
# are not dispatched to procu2 (guaranteed to fail and loop).
# exists_async() checks concurrently in k8s without blocking
# the event loop.
all_hits = hits_regular + hits_black
existence = await gather_exists(
[h.orig_file_path for h in all_hits]
)
preflight_not_exist = [
h for h, exists in zip(all_hits, existence) if not exists
]
# a deleted file may still be running or held: the stop step
# looks for its processes by the recorded inode (the cleaner
# is never asked about it); the owner's files already gone at
# the storage step join here and leave the pending list
gone = [
h
for h in preflight_not_exist
if h.resource_type == MalwareScanResourceType.FILE.value
]
mine = [e for e in gone_entries if e[0].owner == user]
for e in mine:
gone_entries[:] = [
pending for pending in gone_entries if pending is not e
]
gone.append(e[0])
new_gone_entries = [
(hit, cause, initiator)
for hit in gone
if all(entry[0] is not hit for entry in mine)
]
# Claim newly vanished hits before deleting their DB rows or
# awaiting history writes. Cancellation/error can then retry
# the exact in-memory evidence on a later cleanup tick.
self._gone_pending.extend(new_gone_entries)
owned_gone_entries = mine + new_gone_entries
if preflight_not_exist:
preflight_set = set(preflight_not_exist)
hits_regular = [
h for h in hits_regular if h not in preflight_set
]
hits_black = [h for h in hits_black if h not in preflight_set]
MalwareHit.delete_instances(preflight_not_exist)
await MalwareAction.not_exist(
preflight_not_exist,
cause=cause,
initiator=initiator,
)
user_hits_all = [
h for h in user_hits_all if h not in preflight_set
]
# Skip if nothing to clean after filtering
if not hits_regular and not hits_black:
if gone:
try:
await self._stop_running_processes(
user,
[],
[],
standard_only,
cause,
initiator,
gone,
gone_entries=owned_gone_entries,
)
except _StopCleanupCancelled:
requeue(unvisited_hits(user_index))
raise asyncio.CancelledError from None
continue
# the cleaner wrapper needs bare in-container paths; the
# tenant rides in --username (k8s prefix stripped)
files = [str(hit.orig_file_path) for hit in hits_regular]
black = [str(hit.orig_file_path) for hit in hits_black]
logger.debug("Cleaning files: %s", files + black)
cleanup_id = uuid.uuid4().hex
started = time.time()
if is_uid(user): # non panel user
uid = int(user)
async def _refuse(error, refused_hits):
await self._sink.process_message(
MessageType.MalwareCleanup(
hits=refused_hits,
result={},
cleanup_id=cleanup_id,
started=started,
error=error,
cause=cause,
initiator=initiator,
post_action=post_action,
scan_id=scan_id,
args=[],
)
)
if _system_owned(uid):
# crontabs stay cleanable: the cleaner exempts
# /var/spool/cron from its own guard by path
crontab_ids = {
hit.id
for hit in [*user_hits_all, *gone]
if is_crontab(hit.orig_file_path)
}
refused = [
hit
for hit in user_hits_all
if hit.id not in crontab_ids
]
if refused:
logger.warning(
"Refusing to clean files owned by system"
" account %s, manual cleanup required",
uid,
)
await _refuse(
"Cleanup skipped. File is owned by a system"
" account, manual cleanup required",
refused,
)
if not crontab_ids:
self._ack_gone_entries(owned_gone_entries)
continue
hits_regular = [
h for h in hits_regular if h.id in crontab_ids
]
hits_black = [h for h in hits_black if h.id in crontab_ids]
user_hits_all = hits_regular + hits_black
files = [str(hit.orig_file_path) for hit in hits_regular]
black = [str(hit.orig_file_path) for hit in hits_black]
if not files and not black:
if gone:
try:
await self._stop_running_processes(
user,
[],
[],
standard_only,
cause,
initiator,
gone,
gone_entries=owned_gone_entries,
)
except _StopCleanupCancelled:
requeue(unvisited_hits(user_index))
raise asyncio.CancelledError from None
continue
logger.warning(
"Cleaning system-owned crontabs as a deliberate"
" guard exception: %s",
files + black,
)
if not LicenseCLN.is_unlimited():
logger.error(
f"Can't clean files for non panel user {uid=}, "
"since license is limited"
)
await _refuse(
"Cleanup failed. Automatic cleanup of files not"
" owned by a panel user requires an unlimited"
" license",
user_hits_all,
)
self._ack_gone_entries(owned_gone_entries)
continue
if not (username := await get_username_by_uid(uid)):
logger.warning(f"Can't find username for {uid=}")
await _refuse(
"Cleanup failed. File owner cannot be resolved",
user_hits_all,
)
self._ack_gone_entries(owned_gone_entries)
continue
user = username
await self._start_hook(cleanup_id, started, user_hits_all)
# files whose running process is stopped are cleaned first, in
# their own small pass, and never a second time
stop_cancelled = False
try:
(
result,
cleaned_first,
error,
cmd,
) = await self._stop_running_processes(
user,
hits_regular,
hits_black,
standard_only,
cause,
initiator,
gone,
gone_entries=owned_gone_entries,
)
except _StopCleanupCancelled as cancelled:
state = cancelled.state
result, cleaned_first, error, cmd = (
state["result"],
state["cleaned"],
state["error"],
state["cmd"],
)
stop_cancelled = True
files = [f for f in files if f not in cleaned_first]
black = [f for f in black if f not in cleaned_first]
if (files or black) and not stop_cancelled:
rest, rest_error, cmd = await self._cleaner.start(
user,
files,
soft=Config.CLEANUP_TRIM,
blacklist=black,
standard_only=standard_only,
)
if cleaned_first:
rest.update(result)
rest.attempts.update(getattr(result, "attempts", {}))
error = rest_error or error
else:
error = rest_error
result = rest
reported_hits = user_hits_all
if stop_cancelled:
reported_hits = [
hit
for hit in user_hits_all
if str(hit.orig_file_path) in cleaned_first
]
unattempted = [
hit for hit in user_hits_all if hit not in reported_hits
]
unattempted.extend(unvisited_hits(user_index))
requeue(unattempted)
if not reported_hits:
raise asyncio.CancelledError
message = MessageType.MalwareCleanup(
hits=reported_hits,
result=result,
cleanup_id=cleanup_id,
started=started,
error=error,
cause=cause,
initiator=initiator,
post_action=post_action,
scan_id=scan_id,
args=cmd,
cleanup_user=user,
)
if stop_cancelled:
task = asyncio.ensure_future(
self._sink.process_message(message)
)
while not task.done():
try:
await asyncio.shield(task)
except asyncio.CancelledError:
pass
task.result()
raise asyncio.CancelledError
await self._sink.process_message(message)
async def _stop_running_processes(
self,
user,
hits_regular,
hits_black,
standard_only,
cause,
initiator,
gone=(),
gone_entries=(),
):
"""Stop the processes that run or hold the files about to be
cleaned. The KILL candidates' files are cleaned here, in their own
small pass, so a process is only killed once its file is clean.
*gone* are hits whose file no longer exists: their processes are
still looked for by the recorded inode; the cleaner never sees
those files (the stop step kills only when the held inode has no
link left on disk).
Returns (result of that pass, the files it cleaned, error, cmd)."""
empty = (CleanupResult(), set(), None, [])
if self._stop_controller is None or not stop_mode.evaluation_wanted():
self._ack_gone_entries(gone_entries)
return empty
# The cleaner may replace a numeric account with its NSS username
# before this boundary. Authorize each immutable hit owner instead
# of the mutable dispatch name, including retry-only gone entries.
refused_entries = [
entry
for entry in gone_entries
if _refused_by_owner_guard(entry[0])
]
if refused_entries:
self._ack_gone_entries(refused_entries)
hits_regular = [
hit for hit in hits_regular if not _refused_by_owner_guard(hit)
]
hits_black = [
hit for hit in hits_black if not _refused_by_owner_guard(hit)
]
gone = [hit for hit in gone if not _refused_by_owner_guard(hit)]
file_hits = [
h
for h in hits_regular + hits_black + list(gone)
if h.resource_type == MalwareScanResourceType.FILE.value
]
if not file_hits:
self._ack_gone_entries(gone_entries)
return empty
self._note_gone_attempts(gone_entries)
state = {
"result": CleanupResult(),
"cleaned": set(),
"error": None,
"cmd": [],
}
if is_uid(user):
# The normal cleaner path applies this gate before replacing a
# numeric owner with its NSS name. Gone-only batches reach this
# common boundary directly and must use the same eligibility.
if not LicenseCLN.is_unlimited():
self._ack_gone_entries(gone_entries)
return empty
try:
user = await get_username_by_uid(user) or user
except asyncio.CancelledError:
raise _StopCleanupCancelled(state) from None
cleaner_paths = {
hit.orig_file: str(hit.orig_file_path)
for hit in hits_regular + hits_black
}
black_files = {cleaner_paths[h.orig_file] for h in hits_black}
gone_ids = frozenset(h.id for h in gone)
async def clean(paths):
by_path = {path: cleaner_paths.get(path, path) for path in paths}
regular = sorted(
{f for f in by_path.values() if f not in black_files}
)
black = sorted({f for f in by_path.values() if f in black_files})
# claimed before the pass runs: a pass that fails or is
# cancelled must not be repeated by the normal pass while the
# curator may still be at work on these files
state["cleaned"].update(by_path.values())
cleaner = asyncio.create_task(
self._cleaner.start(
user,
regular,
soft=Config.CLEANUP_TRIM,
blacklist=black or None,
standard_only=standard_only,
)
)
cancelled = False
grace_end = None
loop = asyncio.get_event_loop()
while not cleaner.done():
try:
if grace_end is None:
await asyncio.shield(cleaner)
elif loop.time() < grace_end:
await asyncio.wait_for(
asyncio.shield(cleaner), grace_end - loop.time()
)
else:
# the curator is terminated, so no pass of it can
# overlap a later cleanup of these files
cleaner.cancel()
await asyncio.wait({cleaner})
except asyncio.CancelledError:
cancelled = True
if grace_end is None:
grace_end = loop.time() + SMALL_PASS_GRACE
except asyncio.TimeoutError:
continue
except Exception:
break
if cleaner.cancelled():
# nothing was cleaned: the files are cleaned after restart
state["cleaned"].difference_update(by_path.values())
raise asyncio.CancelledError
try:
result, error, cmd = cleaner.result()
except Exception:
if cancelled:
logger.exception(
"The owned cleaner failed while stop cleanup was"
" being cancelled"
)
raise asyncio.CancelledError from None
raise
reported = dict(result)
attempts = dict(getattr(result, "attempts", {}))
# CleanupResult lookup uses the DB path, not the cleaner operand.
for original, file in cleaner_paths.items():
if file in result:
reported[original] = result[file]
if file in attempts:
attempts[original] = attempts[file]
for file in cleaner_paths.values():
if file not in cleaner_paths:
reported.pop(file, None)
attempts.pop(file, None)
state["result"].update(reported)
state["result"].attempts.update(attempts)
state["error"] = error
state["cmd"] = cmd
outcomes = {
path: cleaner_outcome(result[file] if file in result else None)
for path, file in by_path.items()
}
if cancelled:
raise asyncio.CancelledError
return outcomes
try:
# only a run that may kill needs them; without them (None)
# nothing is stopped in the run, also when a rule turns KILL
# during the search
veto_hits = None
if stop_mode.base_mode()[0] is stop_mode.Mode.KILL:
veto_hits = await self._loop.run_in_executor(
None,
_other_detected_hits,
{h.owner for h in file_hits} | {h.user for h in file_hits},
{h.id for h in file_hits},
)
out = await self._stop_controller.evaluate(
file_hits,
clean,
gone_ids=gone_ids,
user=user,
veto_hits=veto_hits,
)
# reported once per cleanup run, with the other accounts' steps
self._run_reports.append((out, file_hits, cause, initiator))
if out.cancelled:
# One hit may be linked to more than one process. A result
# for one pid does not prove that cancellation did not skip
# another, and deleted database rows cannot recreate this
# work. Retain the exact owned batch for the next tick.
raise _StopCleanupCancelled(state)
if getattr(out, "skipped", False):
# not an attempt: an earlier search is still stuck
self._forget_gone_attempt(gone_entries)
if getattr(out, "timed_out", False):
# nothing was looked at: a gone file's recorded inode is the
# only way to find its holder again, so it is kept for the
# next tick (bounded by its attempts)
return (
state["result"],
state["cleaned"],
state["error"],
state["cmd"],
)
# Evaluation owns these exact in-memory entries. Once it has
# completed, later reporting failure must not repeat a
# physical stop or its history on the next cleanup tick.
by_pid = {}
for evaluation in out.evaluations:
by_pid.setdefault(evaluation.result.pid, []).append(evaluation)
deferred_hit_ids = set()
retryable = {
RESPAWNING,
STOP_CAP,
COULD_NOT_FREEZE,
FREEZE_TIMEOUT,
READ_FAILED,
}
for evaluations in by_pid.values():
intended = [
evaluation
for evaluation in evaluations
if evaluation.mode is stop_mode.Mode.KILL
]
deferred = any(
evaluation.result.reason in retryable
for evaluation in intended
) or any(
not evaluation.result.file_gone
and evaluation.result.reason == FILE_NOT_CLEANED
for evaluation in evaluations
)
if deferred:
deferred_hit_ids.update(
evaluation.finding.hit_id
for evaluation in intended
if evaluation.result.file_gone
)
self._ack_gone_entries(
[
entry
for entry in gone_entries
if entry[0].id not in deferred_hit_ids
]
)
except asyncio.CancelledError:
raise _StopCleanupCancelled(state) from None
except Exception:
logger.exception("Stopping processes during cleanup failed")
return state["result"], state["cleaned"], state["error"], state["cmd"]
def _note_gone_attempts(self, entries):
owned = {id(entry) for entry in self._gone_pending}
seen = set()
for entry in entries:
key = id(entry)
if key not in owned or key in seen:
continue
seen.add(key)
prior = self._gone_attempts.get(key)
attempts = (
prior[1] if prior is not None and prior[0] is entry else 0
)
self._gone_attempts[key] = (entry, attempts + 1)
def _forget_gone_attempt(self, entries):
for entry in entries:
state = self._gone_attempts.get(id(entry))
if state is not None and state[0] is entry:
attempts = state[1] - 1
if attempts > 0:
self._gone_attempts[id(entry)] = (entry, attempts)
else:
del self._gone_attempts[id(entry)]
def _expire_gone_entries(self):
owned = {id(entry): entry for entry in self._gone_pending}
self._gone_attempts = {
key: state
for key, state in self._gone_attempts.items()
if key in owned and state[0] is owned[key]
}
expired = {
key
for key, (_, attempts) in self._gone_attempts.items()
if attempts >= MAX_GONE_STOP_ATTEMPTS
}
if expired:
self._gone_pending[:] = [
entry
for entry in self._gone_pending
if id(entry) not in expired
]
for key in expired:
self._gone_attempts.pop(key, None)
logger.info(
"Stopped retrying %d missing-file process checks after %d"
" attempts",
len(expired),
MAX_GONE_STOP_ATTEMPTS,
)
def take_gone_hits(self):
"""The hits whose file was already gone at the storage step, handed
over once to the run's stop step."""
pending, self._gone_pending = self._gone_pending, []
self._gone_attempts.clear()
return [hit for hit, _, _ in pending]
def _take_gone_entries(self):
self._expire_gone_entries()
return list(self._gone_pending)
def _ack_gone_entries(self, entries):
"""Forget only entries deliberately evaluated by this run."""
for entry in entries:
self._gone_attempts.pop(id(entry), None)
for index, pending in enumerate(self._gone_pending):
if pending is entry:
self._gone_pending.pop(index)
break
async def _stop_for_gone_hits(self, entries):
"""The stop step for files already gone at the storage step, per
owner and cleanup cause, without any cleaner pass."""
groups = {}
for hit, cause, initiator in entries:
groups.setdefault((hit.owner, cause, initiator), []).append(hit)
for (user, cause, initiator), hits in groups.items():
owned = [
entry
for entry in entries
if (entry[0].owner, entry[1], entry[2])
== (user, cause, initiator)
]
await self._stop_running_processes(
user,
[],
[],
None,
cause,
initiator,
hits,
gone_entries=owned,
)
async def _report_run(self):
"""One report per cleanup run: the stop steps of all its accounts
in one MALWARE_PROCESS_STOP_LIST (process items plus one counters
item with runs 1) or, when nothing was linked, one roll-up entry;
the admin-visible record per step in LOG and KILL. The record is
written first: a failed or cancelled send never loses it."""
parts, self._run_reports = self._run_reports, []
# a search that gave up or was skipped examined nothing, but the
# run is counted
parts = [
p
for p in parts
if p[0].evaluations or p[0].examined or p[0].timed_out
]
if not parts:
return
merged = stop_report.merge_run(
[(out, hits) for out, hits, _, _ in parts]
)
loop = asyncio.get_event_loop()
for out, hits, cause, initiator in parts:
try:
await loop.run_in_executor(
None,
_record_stop_history,
out,
hits,
cause,
initiator,
)
except asyncio.CancelledError:
raise
except Exception:
logger.exception(
"Writing one account's process-stop history failed"
)
items = []
kernel, panel, btime = (
_kernel(),
HostingPanel().NAME,
stop_proc.boot_time(),
)
users = {h.owner for _, hits, _, _ in parts for h in hits}
try:
homes = await asyncio.get_event_loop().run_in_executor(
None, _homes, users
)
except Exception as e:
logger.debug("Resolving cleanup account homes failed: %r", e)
homes = {user: "" for user in users}
for out, hits, _, _ in parts:
built = stop_report.build_items(
out,
hits,
home_of=homes,
kernel=kernel,
panel=panel,
btime=btime,
)
items.extend(i for i in built if i["record_type"] == "process")
items.append(stop_report.counters_item(merged, kernel, panel))
await stop_report.report_run(self._sink, items)
active_now = stop_mode.base_mode().mode in (
stop_mode.Mode.LOG,
stop_mode.Mode.KILL,
)
if active_now or any(
ev.mode in (stop_mode.Mode.LOG, stop_mode.Mode.KILL)
for ev in merged.evaluations
):
stop_report.log_timings(merged)
async def _cleanup(self):
if self._running:
return
# Expiry precedes the tracked work path so a failed backlog can idle.
self._expire_gone_entries()
if not self._proxy.hits and not self._gone_pending:
await self._retry_left_overs_coalesced()
if not self._proxy.hits and not self._gone_pending:
self._proxy.reset()
return
self._running = True
cleanup_cancelled = False
cleanup_error = None
with inactivity.track.task("cleanup"):
try:
if self._stop_controller is not None:
self._stop_controller.begin_run() # the run's stop cap
await self._retry_left_overs_coalesced()
# files already gone at the storage step join the stop
# step of the owner's batch of this run; the rest (owners
# without a batch this run) get their own stop step
gone_entries = self._take_gone_entries()
data = self._cloud_assisted_hits()
for (
all_hits,
blacklist,
cause,
initiator,
post_action,
scan_id,
standard_only,
) in data:
await self._clean_files(
all_hits,
blacklist=blacklist,
cause=cause,
initiator=initiator,
post_action=post_action,
scan_id=scan_id,
standard_only=standard_only,
gone_entries=gone_entries,
)
if gone_entries:
await self._stop_for_gone_hits(gone_entries)
except asyncio.CancelledError:
cleanup_cancelled = True
raise
except Exception as error:
cleanup_error = error
raise
finally:
# review 6 L7: a later account's failure never loses the
# history and telemetry of kills already made
try:
report_task = asyncio.ensure_future(self._report_run())
report_cancelled = cleanup_cancelled
report_error = None
while not report_task.done():
try:
await asyncio.shield(report_task)
except asyncio.CancelledError:
report_cancelled = True
except Exception as error:
report_error = error
break
current = asyncio.current_task()
if current is not None and current.cancelling():
report_cancelled = True
if report_error is None:
try:
report_task.result()
except Exception as error:
report_error = error
if report_error is not None:
if not report_cancelled and cleanup_error is None:
raise report_error
logger.error(
"Could not report process-stop results %s",
"during cancellation"
if report_cancelled
else "after cleanup failed",
exc_info=(
type(report_error),
report_error,
report_error.__traceback__,
),
)
if report_cancelled:
raise asyncio.CancelledError
finally:
if self._stop_controller is not None:
self._stop_controller.end_run() # review 6 L9
self._running = False
async def _retry_left_overs(self):
"""Every cleanup tick: the cron re-enables a slow crontab left
pending and the frozen processes whose SIGCONT failed at start are
tried again, each only while its state file exists (it is removed
once empty), so an idle tick costs a stat."""
now = self._loop.time()
if (
now - self._cron_checked_at >= CRON_RETRY_INTERVAL
and stop_cron.DISABLED_PATH.exists()
):
# pending re-enables are retried at most once a minute: the
# store can hold hundreds of disabled lines
self._cron_checked_at = now
try:
await stop_cron.retry_pending_reenables()
except Exception:
logger.exception("Could not retry the pending cron re-enables")
if not stop_registry.REGISTRY_PATH.exists():
return
future = self._loop.run_in_executor(None, release_frozen_orphans)
cancelled = False
while not future.done():
try:
await asyncio.shield(future)
except asyncio.CancelledError:
cancelled = True
released = future.result()
if released:
report_task = asyncio.ensure_future(
stop_report.report_released(
self._sink,
released,
kernel=_kernel(),
panel=HostingPanel().NAME,
)
)
while not report_task.done():
try:
await asyncio.shield(report_task)
except asyncio.CancelledError:
cancelled = True
try:
report_task.result()
except Exception:
logger.exception("Could not report the processes released")
if cancelled:
raise asyncio.CancelledError
async def _retry_left_overs_coalesced(self):
task = self._retry_task
if task is None or task.done():
task = asyncio.ensure_future(self._retry_left_overs())
self._retry_task = task
try:
await asyncio.shield(task)
finally:
if task.done() and self._retry_task is task:
self._retry_task = None
@recurring_check(1)
async def cleanup(self):
try:
await self._cleanup()
except asyncio.CancelledError:
# Counters are already durable. Do not delay or replace caller
# cancellation with a new transport operation.
raise
except Exception:
# Earlier completed runs have already persisted their counters;
# a later account failure must not indefinitely postpone the
# hourly send.
try:
await stop_report.flush_rollup(self._sink)
except asyncio.CancelledError:
raise
except Exception:
logger.exception(
"Could not flush process-stop counters after cleanup"
)
raise
else:
# Preserve the pre-existing successful-tick behavior: a send
# failure remains visible to the recurring-check wrapper.
await stop_report.flush_rollup(self._sink)
class ResultProcessor(MessageSink, MessageSource):
SCOPE = Scope.AV
async def create_sink(self, loop):
pass
async def create_source(self, loop, sink):
self._sink = sink
@staticmethod
def _set_hit_status(hits: List[MalwareHit], status: str, cleaned_at=None):
MalwareHit.set_status(hits, status, cleaned_at)
for hit in hits:
hit.status = status
hit.cleaned_at = cleaned_at
@expect(MessageType.MalwareCleanup)
async def store_result(self, message):
hits: List[MalwareHit] = message["hits"]
result: CleanupResult = message["result"]
cause = message.get("cause")
initiator = message.get("initiator")
now = time.time()
processed = [hit for hit in hits if hit in result]
unprocessed = [hit for hit in hits if hit not in result]
if processed:
await result.mark_unchanged(
{
hit.orig_file: hit.hash
for hit in processed
if not hit.orig_file_path.is_remote
}
)
error = message.get("error")
if unprocessed:
sample = [h.orig_file for h in unprocessed[:3]]
logger.warning(
"Cleanup: %d/%d hits unmatched in result"
" (error=%r, sample=%s)",
len(unprocessed),
len(hits),
error,
sample,
)
if error and result:
logger.warning(
"Cleanup: partial failure — error=%r"
" but %d result entries present",
error,
len(result),
)
not_exist = []
async for hit in nice_iterator(processed, chunk_size=100):
# in case if procu2.php tries to clean user file in root dirs,
# it will be marked as non-existent due to 'Permission denied'
# error which confuses users, so consider it as unable to cleanup.
if result[hit].not_exist(): # pragma: no cover
if await safe_exists(hit.orig_file_path):
unprocessed.append(hit)
else:
not_exist.append(hit)
# DEF-40165: files missing from procu2 result that no longer exist
# on disk must be removed instead of being restored to FOUND,
# otherwise they get re-dispatched for cleanup indefinitely.
still_unprocessed = []
unprocessed_existence = await gather_exists(
[h.orig_file_path for h in unprocessed]
)
for hit, exists in zip(unprocessed, unprocessed_existence):
if not exists:
not_exist.append(hit)
else:
still_unprocessed.append(hit)
unprocessed = still_unprocessed
details = {
hit.orig_file: local_details(
{
**getattr(result, "attempts", {}).get(
hit.orig_file,
getattr(result, "attempts", {}).get(
str(hit.orig_file_path), {}
),
),
**result.get(hit.orig_file, {}),
},
message.get("cleanup_user", hit.user),
bool(error),
)
for hit in hits
}
message["cleanup_details"] = {}
def record_diagnostics(outcome_hits, event):
for hit in outcome_hits:
detail = details[hit.orig_file]
message["cleanup_details"][hit.orig_file] = detail
log_diagnostic(logger, hit.orig_file, detail, event)
record_diagnostics(unprocessed, "unable_to_cleanup")
await MalwareAction.cleanup_unable(
unprocessed, cause=cause, initiator=initiator, details=details
)
requires_myimunify_protection = [
hit
for hit in processed
if result[hit].requires_myimunify_protection()
]
await MalwareAction.cleanup_requires_myimunify_protection(
requires_myimunify_protection, cause=cause, initiator=initiator
)
self._set_hit_status(
requires_myimunify_protection,
MalwareHitStatus.CLEANUP_REQUIRES_MYIMUNIFY_PROTECTION,
now,
)
failed = [hit for hit in processed if result[hit].is_failed()]
record_diagnostics(failed, "failed_to_cleanup")
await MalwareAction.cleanup_failed(
failed, cause=cause, initiator=initiator, details=details
)
cleaned = [hit for hit in processed if result[hit].is_cleaned()]
await MalwareAction.cleanup_done(
cleaned, cause=cause, initiator=initiator
)
self._set_hit_status(cleaned, MalwareHitStatus.CLEANUP_DONE, now)
removed = [hit for hit in processed if result[hit].is_removed()]
await MalwareAction.cleanup_removed(
removed, cause=cause, initiator=initiator
)
self._set_hit_status(removed, MalwareHitStatus.CLEANUP_REMOVED, now)
# a result row matching no outcome predicate (e.g. an error code
# this agent does not know yet) must still reach a terminal state,
# otherwise the hit stays in cleanup_started until agent restart
unclassified = [
hit
for hit in processed
if not (
result[hit].not_exist()
or result[hit].requires_myimunify_protection()
or result[hit].is_failed()
or result[hit].is_cleaned()
or result[hit].is_removed()
)
]
if unclassified:
logger.warning(
"Cleanup: %d/%d hits with unclassified result entries"
" (sample=%s)",
len(unclassified),
len(hits),
[(h.orig_file, result[h]["e"]) for h in unclassified[:3]],
)
record_diagnostics(unclassified, "unable_to_cleanup")
await MalwareAction.cleanup_unable(
unclassified, cause=cause, initiator=initiator, details=details
)
MalwareHit.delete_instances(not_exist)
await MalwareAction.not_exist(
not_exist, cause=cause, initiator=initiator
)
for status, hit_list in _group_by_status(
unprocessed, failed, unclassified
).items():
self._set_hit_status(hit_list, status)
await self.send_failed_to_cleanup_hits_to_mrs(failed)
return message
async def send_failed_to_cleanup_hits_to_mrs(self, failed_to_cleanup_hits):
if failed_to_cleanup_hits:
await self._sink.process_message(
MessageType.MalwareMRSUpload(
hits=[
malware_response.HitInfo(hit.orig_file, hit.hash)
for hit in failed_to_cleanup_hits
],
upload_reason="cleanup_failure_current",
)
)
await self._sink.process_message(
MessageType.MalwareMRSUpload(
hits=[
malware_response.HitInfo(
str(CleanupStorage.get_hit_store_path(hit)),
hit.hash,
)
for hit in failed_to_cleanup_hits
],
upload_reason="cleanup_failure_original",
)
)
class StorageController(MessageSink):
"""Remove old backed up files from storage"""
def __init__(self):
self._clear_task = None
self._keep = Config.CLEANUP_KEEP
async def create_sink(self, loop):
self._clear_task = loop.create_task(self.daily_clear())
async def shutdown(self):
if self._clear_task:
await safe_cancel_task(self._clear_task)
async def _clear(self):
now = time.time()
keep_hits = now - self._keep * DAY
keep_orig = now - (self._keep + 1) * DAY # keep files one more day
if hooked():
await delete_and_report(EXPIRED, MalwareHit.cleaned_at < keep_hits)
MalwareHit.delete().where(MalwareHit.cleaned_at < keep_hits).execute()
cleared = await CleanupStorage.clear(keep_orig)
if cleared:
logger.info(
"Cleanup storage have cleaned. Files removed: %s", cleared
)
@expect(MessageType.ConfigUpdate)
@utils.log_error_and_ignore()
async def config_updated(self, _):
if self._keep != Config.CLEANUP_KEEP:
self._keep = Config.CLEANUP_KEEP
await self._clear()
@recurring_check(
check_lock,
check_period_first=True,
check_lock_period=DAY,
lock_file=LOCK_FILE,
)
async def daily_clear(self):
await self._clear()
def decide_if_standard_signatures_only(user, standard_only):
"""Root user or user with MyImunify can use advanced signatures"""
if not MyImunifyConfig.ENABLED:
return False
if user is None or user == "root" or myimunify_protection_enabled(user):
return standard_only
return True
class ResultProcessorIm360(ResultProcessor):
"""Imunify360 specialization of ResultProcessor, which removes all
cleaned and removed files from HackerTrap
"""
SCOPE = Scope.IM360
@expect(MessageType.MalwareCleanup)
async def store_result(self, message):
message = await super().store_result(message)
to_remove = [
Path(hit)
for hit, state in message["result"].items()
if (state.is_cleaned() or state.is_removed())
]
await HackerTrapHitsSaver.update_sa_hits([], to_remove)
class _TakenCompletions:
"""Ids of MDS operations whose completion has already been taken.
Both the completion notification and the reconciler deliver a
completion. The first delivery removes the run directory whatever the
outcome, so a later one can only repeat the outcome or, finding no
report, fail the operation queued next.
"""
MAX_IDS = 256
def __init__(self):
self._ids: Dict[str, None] = {}
def take(self, operation_id: str) -> bool:
if operation_id in self._ids:
return False
self._ids[operation_id] = None
if len(self._ids) > self.MAX_IDS:
del self._ids[next(iter(self._ids))]
return True
class CleanupDb(MessageSink):
SCOPE = Scope.IM360
def __init__(self):
self._loop = None
self._completions = _TakenCompletions()
@staticmethod
async def _start_cleaner(path, app_name):
cleanup_id = uuid.uuid4().hex
await MalwareDatabaseCleaner(cleanup_id, path, app_name).start()
async def _cleanup_next(self):
if (
MalwareHit.db_hits_under_cleanup().exists()
or (
next_hit := MalwareHit.db_hits_pending_cleanup()
.order_by(MalwareHit.timestamp.asc())
.first()
)
is None
):
return
# One --clean pass cleans the whole DB, so batch all its pending hits.
same_db_hits = list(
MalwareHit.db_hits_pending_cleanup().where(
(MalwareHit.orig_file == next_hit.orig_file)
& (MalwareHit.app_name == next_hit.app_name)
)
)
logger.info(
"Cleaning hit: (%s::%s::%s)",
next_hit.orig_file,
next_hit.app_name,
next_hit.type,
)
MalwareHit.set_status(same_db_hits, MalwareHitStatus.CLEANUP_STARTED)
await self._start_cleaner(next_hit.orig_file, next_hit.app_name)
async def create_sink(self, loop):
self._loop = loop
await self._cleanup_next()
@expect(MessageType.MalwareCleanupTask)
async def process_cleanup_task(self, message):
hits = MalwareHit.refresh_hits(message["hits"])
hits_to_clean = filter_cleanable(hits)
db_hits = [
hit
for hit in hits_to_clean
if hit.resource_type == MalwareScanResourceType.DB.value
]
if not db_hits:
return
MalwareHit.set_status(db_hits, MalwareHitStatus.CLEANUP_PENDING)
await self._cleanup_next()
@expect(MessageType.MalwareCleanComplete)
async def parse_cleanup_results(self, message):
clean_id = message["scan_id"]
if not self._completions.take(clean_id):
logger.info(
"Ignoring a duplicate completion of cleanup %s", clean_id
)
return
detached_cleanup = MDSDetachedCleanup(clean_id)
try:
cleanup_outcome = await detached_cleanup.complete()
except ScanAlreadyCompleteError:
# This happens when AV is woken up by AiBolit. See DEF-11078.
logger.warning(
"Cannot complete cleanup %s, assuming it is already complete",
clean_id,
)
return
finally:
shutil.rmtree(
str(detached_cleanup.detached_dir), ignore_errors=True
)
await g.sink.process_message(cleanup_outcome)
@expect(MessageType.MalwareDatabaseCleanup)
async def update_cleaned_hits_status(
self, message: MessageType.MalwareDatabaseCleanup
):
cleaned_hits = MalwareHit.db_hits_under_cleanup_in(message.succeeded)
failed_hits = MalwareHit.db_hits_under_cleanup_in(message.failed)
MalwareHit.set_status(
cleaned_hits, MalwareHitStatus.CLEANUP_DONE, time.time()
)
MalwareHit.set_status(failed_hits, MalwareHitStatus.FOUND)
# The queue is serial, so any hit still under cleanup belongs to
# this operation; without a terminal status it would block the
# queue forever. The sweep is deliberately not scoped to this
# operation's (path, app): a run that fails before reaching any
# database reports no hits at all, and only an unscoped sweep
# finalizes its hits then. The trade-off is that hits stranded in
# cleanup_started by an unrelated earlier crash are finalized (and
# get a failed_to_cleanup event) on this run's behalf.
if leftover_hits := list(MalwareHit.db_hits_under_cleanup()):
MalwareHit.set_status(leftover_hits, MalwareHitStatus.FOUND)
await MalwareAction.cleanup_failed(
leftover_hits, cause=None, initiator=None
)
await self._cleanup_next()
@expect(MessageType.MalwareDatabaseCleanupFailed)
async def update_failed_hits_status(self, message):
"""
Clear the queue when the cleanup fails,
set hits' status back to infected
"""
# We assume here that all CLEANUP_STARTED hits are part of the
# same cleanup operation
hits = list(MalwareHit.db_hits_under_cleanup())
MalwareHit.set_status(hits, MalwareHitStatus.FOUND)
await MalwareAction.cleanup_failed(hits, cause=None, initiator=None)
await self._cleanup_next()
@expect(MessageType.MalwareDatabaseCleanup)
async def save_cleanup_events_in_history(
self, message: MessageType.MalwareDatabaseCleanup
):
cause = None
initiator = None
cleaned_hits = MalwareHit.get_db_hits(message.succeeded)
await MalwareAction.cleanup_done(
cleaned_hits, cause=cause, initiator=initiator
)
failed_hits = MalwareHit.get_db_hits(message.failed)
await MalwareAction.cleanup_failed(
failed_hits, cause=cause, initiator=initiator
)
class RestoreOriginalDb(MessageSink):
SCOPE = Scope.IM360
def __init__(self):
self.loop = None
self._completions = _TakenCompletions()
@staticmethod
async def _restore_next():
if (
MalwareHit.db_hits_under_restoration().exists()
or (
hit_to_restore := (
MalwareHit.db_hits_pending_cleanup_restore()
.order_by(MalwareHit.timestamp.asc())
.first()
)
)
is None
):
return
signature_id = (
hit_to_restore.signature_id
if hit_to_restore.status
== MalwareHitStatus.CLEANUP_REMOTE_RESTORE_PENDING
else None
)
logger.info(
"Restoring from cleanup hit: (%s::%s::%s)",
hit_to_restore.orig_file,
hit_to_restore.app_name,
signature_id,
)
await MalwareDatabaseRestore(
path=hit_to_restore.orig_file,
app_name=hit_to_restore.app_name,
signature_id=signature_id,
).restore()
MalwareHit.set_status(
[hit_to_restore], MalwareHitStatus.CLEANUP_RESTORE_STARTED
)
async def create_sink(self, loop):
self.loop = loop
await self._restore_next()
@staticmethod
def _filter_under_restore(
hits: Iterable[MalwareHit],
) -> Iterable[MalwareHit]:
return (
hit
for hit in hits
if hit.status == MalwareHitStatus.CLEANUP_RESTORE_STARTED
)
@expect(MessageType.MalwareDatabaseRestoreTask)
async def queue_db_restore(self, message: MalwareDatabaseRestoreTask):
query = (
MalwareHit.db_hits()
.where(MalwareHit.orig_file == message.path)
.where(MalwareHit.app_name == message.app_name)
)
restore_status = MalwareHitStatus.CLEANUP_RESTORE_PENDING
if message.signature_id:
query = query.where(MalwareHit.type == message.signature_id)
restore_status = MalwareHitStatus.CLEANUP_REMOTE_RESTORE_PENDING
MalwareHit.set_status(
query,
restore_status,
)
await self._restore_next()
@expect(MessageType.MalwareRestoreComplete)
async def parse_restore_results(self, message):
restore_id = message["scan_id"]
if not self._completions.take(restore_id):
logger.info(
"Ignoring a duplicate completion of restore %s", restore_id
)
return
detached_restore = MDSDetachedRestore(restore_id)
try:
restore_message = await detached_restore.complete()
except ScanAlreadyCompleteError:
# This happens when AV is woken up by AiBolit. See DEF-11078.
logger.warning(
"Cannot complete restore %s, assuming it is already complete",
restore_id,
)
return
finally:
shutil.rmtree(
str(detached_restore.detached_dir), ignore_errors=True
)
await g.sink.process_message(restore_message)
@expect(MessageType.MalwareDatabaseRestore)
async def update_restored_hits_status(self, message):
restored_hits = MalwareHit.get_db_hits(message.succeeded)
MalwareHit.set_status(
self._filter_under_restore(restored_hits), MalwareHitStatus.FOUND
)
# Failed hits leave the under-restoration status here so the
# leftover sweep does not report them a second time; their
# history event comes from save_restore_events_in_history.
failed_hits = MalwareHit.get_db_hits(message.failed)
MalwareHit.set_status(
self._filter_under_restore(failed_hits),
MalwareHitStatus.CLEANUP_DONE,
)
# The queue is serial, so any hit still under restoration belongs
# to this operation; without a terminal status it would block the
# queue forever. Unscoped for the same reason as the cleanup
# sweep: a run that fails before reaching any database reports no
# hits at all.
if leftover_hits := list(MalwareHit.db_hits_under_restoration()):
MalwareHit.set_status(leftover_hits, MalwareHitStatus.CLEANUP_DONE)
for hit in leftover_hits:
await MalwareAction.cleanup_failed_restore(
path=hit.orig_file,
app_name=hit.app_name,
resource_type=MalwareScanResourceType.DB.value,
file_owner=hit.owner,
file_user=hit.user,
initiator=message.get("initiator"),
cause=message.get("cause"),
db_host=hit.db_host,
db_port=hit.db_port,
db_name=hit.db_name,
signature_id=hit.signature_id,
)
await self._restore_next()
@expect(MessageType.MalwareDatabaseRestore)
async def save_restore_events_in_history(self, message):
cause = message.get("cause")
initiator = message.get("initiator")
restored_hits = MalwareHit.get_db_hits(message.succeeded)
for hit in restored_hits:
# FIXME: change cleanup_restored_original to accept multiple
# values
await MalwareAction.cleanup_restored_original(
path=hit.orig_file,
app_name=hit.app_name,
resource_type=MalwareScanResourceType.DB.value,
file_owner=hit.owner,
file_user=hit.user,
initiator=initiator,
cause=cause,
db_host=hit.db_host,
db_port=hit.db_port,
db_name=hit.db_name,
signature_id=hit.signature_id,
)
failed_hits = MalwareHit.get_db_hits(message.failed)
for hit in failed_hits:
await MalwareAction.cleanup_failed_restore(
path=hit.orig_file,
app_name=hit.app_name,
resource_type=MalwareScanResourceType.DB.value,
file_owner=hit.owner,
file_user=hit.user,
initiator=initiator,
cause=cause,
db_host=hit.db_host,
db_port=hit.db_port,
db_name=hit.db_name,
signature_id=hit.signature_id,
)
@expect(MessageType.MalwareDatabaseRestoreFailed)
async def update_failed_hits_status(self, message):
"""
Clear the queue when the restore fails,
set hits' status back to cleanup_done
"""
hits = list(MalwareHit.db_hits_under_restoration())
MalwareHit.set_status(hits, MalwareHitStatus.CLEANUP_DONE)
for hit in hits:
await MalwareAction.cleanup_failed_restore(
path=hit.orig_file,
app_name=hit.app_name,
resource_type=MalwareScanResourceType.DB.value,
file_owner=hit.owner,
file_user=hit.user,
initiator=message.get("initiator"),
cause=message.get("cause"),
db_host=hit.db_host,
db_port=hit.db_port,
db_name=hit.db_name,
signature_id=hit.signature_id,
)
await self._restore_next()