"""
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 functools
import logging
import os
import pwd
import warnings
from collections import namedtuple
from functools import partial
from typing import List, Sequence
from defence360agent.contracts.config import Malware, UserType
from defence360agent.contracts.license import LicenseError
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.permissions import (
MS_IGNORE_LIST_EDIT,
check_permission,
)
from defence360agent.feature_management.constants import AV, AV_REPORT, FULL
from defence360agent.feature_management.lookup import feature
from defence360agent.model.instance import db
from defence360agent.model.simplification import run_in_executor
from defence360agent.rpc_tools.lookup import (
CommonEndpoints,
RootEndpoints,
bind,
)
from defence360agent.rpc_tools.utils import run_in_executor_decorator
from defence360agent.rpc_tools.validate import (
ValidationError,
validate_av_plus_license,
)
from defence360agent.utils import (
Scope,
does_path_belong_to_user,
get_path_owner,
is_cloudways,
is_cluster,
safe_fileops,
)
from imav.malwarelib.cleanup.storage import CleanupStorage
from imav.malwarelib.config import (
MalwareHitStatus,
MalwareScanResourceType,
MalwareScanType,
)
from imav.malwarelib.difflib import DiffError, MalwareHitDiff
from imav.malwarelib.ignore_paths import scope_to_user
from imav.malwarelib.model import (
MalwareHistory,
MalwareHit,
MalwareIgnorePath,
)
from imav.malwarelib.rpc.endpoints.ondemand import split_args
from imav.malwarelib.scan.crontab import get_crontab
from imav.malwarelib.scan.queue_supervisor_sync import (
QueueSupervisorSync as ScanQueue,
)
from imav.malwarelib.subsys.malware import MalwareAction
from imav.malwarelib.tenant_path import (
GLOBAL_TENANT,
is_under_prefix,
strip_prefix,
to_prefixed,
)
from imav.malwarelib.utils.user_list import split_stored_paths
from imav.malwarelib.utils import malware_response, user_list
from imav.malwarelib.utils.user_list import resolve_tenant_path
from imav.malwarelib.utils.endpoints import MaliciousEndpointStatus
from imav.malwarelib.utils.submit import (
FALSE_NEGATIVE,
FALSE_POSITIVE,
submit_malware,
)
from defence360agent.wordpress.site_repository import get_sites_for_user
logger = logging.getLogger(__name__)
IgnoreParameters = namedtuple(
"IgnoreParameters",
["path", "app_name", "db_host", "db_port", "db_name", "signature_id"],
defaults=(None, None, None, None, None, None),
)
#: Cap on a single cluster-mode malware-read chunk. The syncer reads this
#: many bytes into memory, base64-encodes them (~1.33x) and the agent
#: decodes another copy, so an unbounded client-supplied limit is a
#: memory-exhaustion vector against both — but the UI reads whole flagged
#: files, so the ceiling is generous.
MAX_K8S_READ_LIMIT = 100 * 1024 * 1024 # 100 MiB
class SubmitEndpoints(RootEndpoints):
SCOPE = Scope.AV
_SEND_FILES_DISABLED_BANNER = """\
Warning: This server’s security can be enhanced \
by enabling the MALWARE_SCANNING.sends_file_for_analysis option. \
This may minimize the number of undetected malware, \
making your system more resistant to new threats.
The command below can be used to enable the option:
imunify-antivirus config update \
'{"MALWARE_SCANNING": {"sends_file_for_analysis": true}}'
- or -
imunify360-agent config update \
'{"MALWARE_SCANNING": {"sends_file_for_analysis": true}}'
"""
@bind("submit", "false-positive")
async def submit_fp(self, filename, reason, scanner=None):
# WARNING: scanner parameter is deprecated
try:
result = await submit_malware(
filename, FALSE_POSITIVE, reason=reason
)
return {"items": result}
except LicenseError as e:
raise ValidationError(e)
except FileNotFoundError:
raise ValidationError("File {} doesn't exist.".format(filename))
@bind("submit", "false-negative")
async def submit_fn(self, filename):
try:
result = await submit_malware(filename, FALSE_NEGATIVE)
if not Malware.SEND_FILES:
warnings.warn(self._SEND_FILES_DISABLED_BANNER)
return {"items": result}
except LicenseError as e:
raise ValidationError(e)
except FileNotFoundError:
raise ValidationError("File {} doesn't exist.".format(filename))
class MaliciousEndpoints(CommonEndpoints):
SCOPE = Scope.AV
def __init__(self, sink):
super().__init__(sink)
self.queue = ScanQueue(sink=sink)
@feature(AV, [FULL, AV_REPORT])
@bind("malware", "malicious", "list")
@run_in_executor_decorator
def malicious_list(self, user=None, **kwargs):
user_sites = []
if user and kwargs.get("site_search"):
if is_cluster():
# K8s app users are not in /etc/passwd, so we cannot
# enumerate their sites via get_sites_for_user. Pass the
# requested site_search as a single-entry user_sites so
# the model still applies the primary
# `orig_file LIKE site_search%` filter; there are no
# longer subsite paths to exclude in a single-app pod.
# Rows are stored tenant-prefixed, so filter in that form.
kwargs["site_search"] = to_prefixed(
kwargs["site_search"], user
)
user_sites = [kwargs["site_search"]]
else:
user_info = pwd.getpwnam(user)
user_sites = get_sites_for_user(user_info)
return MalwareHit.malicious_list(
user=user,
user_sites=user_sites,
**kwargs,
)
@feature(AV, [FULL, AV_REPORT])
@bind("malware", "malicious", "summary")
@run_in_executor_decorator
def malicious_summary(self, user=None, **kwargs):
return MalwareHit.malicious_summary(user=user, **kwargs)
@classmethod
@feature(AV, [FULL, AV_REPORT])
@bind("malware", "read")
async def read_file(cls, path, offset, limit, user=None, **_):
if is_cluster() and user:
return await cls._read_file_k8s(path, offset, limit, user)
return await cls._read_file_local(path, offset, limit, user)
@classmethod
async def _read_file_k8s(cls, path, offset, limit, user):
from imav.malwarelib.cleanup.storage import CleanupStorageK8s
offset = max(int(offset), 0)
limit = min(max(int(limit), 0), MAX_K8S_READ_LIMIT)
# The syncer reads with its own (root-in-pod) privileges, so kernel
# DAC does not scope the read the way it does on the local branch.
# Constrain it to a path this tenant already has a hit for — the
# endpoint previews flagged files, so an arbitrary in-pod path (a
# service-account token, another user's file) must not be readable.
prefixed = to_prefixed(path, user)
hits = await run_in_executor(
asyncio.get_event_loop(),
partial(lambda p: list(MalwareHit.get_hits([p])), prefixed),
)
if not hits:
raise FileNotFoundError("notifications.fileNotFound")
# The syncer runs inside the tenant container and addresses files by
# their bare in-container path; strip the prefix the UI round-trips.
bare = strip_prefix(prefixed, user)
try:
result = await CleanupStorageK8s.read_file(
app_id=user, path=bare, offset=offset, limit=limit
)
except FileNotFoundError:
raise FileNotFoundError("notifications.fileNotFound")
except OSError as e:
raise PermissionError("notifications.permissionError") from e
import base64
raw = base64.b64decode(result["data"])
text = raw.decode("utf-8", errors="ignore")
return {
"data": {
"chunk": text,
"eof": result["eof"],
"limit": limit,
"offset": offset,
"size": result["size"],
},
}
@classmethod
async def _read_file_local(cls, path, offset, limit, user):
mode = "rb"
if not os.path.exists(path):
raise FileNotFoundError("notifications.fileNotFound")
if user:
open_fun = functools.partial(
safe_fileops.safe_open_file,
path,
mode,
user,
respect_homedir=False,
)
else:
open_fun = functools.partial(open, path, mode)
try:
with open_fun() as f:
f.seek(offset)
chunk = f.read(limit)
eof = False if chunk else True
text = chunk.decode("utf-8", errors="ignore")
return {
"data": {
"chunk": text,
"eof": eof,
"limit": limit,
"offset": offset,
"size": os.fstat(f.fileno()).st_size,
},
}
except asyncio.CancelledError:
raise
except Exception as e:
raise PermissionError("notifications.permissionError") from e
@staticmethod
@feature(AV, [FULL, AV_REPORT])
@bind("malware", "malicious", "remove-from-list")
async def malicious_remove_from_list(ids, user=None):
hits_to_remove = MalwareHit.malicious_select(ids, user=user)
MalwareHit.delete_instances(hits_to_remove)
return MaliciousEndpointStatus(hits_to_remove, [])
@feature(AV, [FULL, AV_REPORT])
@bind("malware", "malicious", "move-to-ignore")
async def malicious_move_to_ignore(self, ids, user=None):
ignored = await self._malicious_move_to_ignore(ids, user)
return len(ignored)
async def _malicious_move_to_ignore(self, ids, user=None):
await check_permission(MS_IGNORE_LIST_EDIT, user)
hits = await run_in_executor(
asyncio.get_event_loop(),
partial(MalwareHit.malicious_select, ids, user=user),
)
# flush found
malicious_found = (
h for h in hits if h.status == MalwareHitStatus.FOUND
)
await run_in_executor(
asyncio.get_event_loop(),
partial(MalwareHit.delete_instances, malicious_found),
)
file_hits = [
hit
for hit in hits
if hit.resource_type == MalwareScanResourceType.FILE.value
]
file_items = [
IgnoreParameters(path=hit.orig_file) for hit in file_hits
]
db_items = [
IgnoreParameters(
hit.orig_file,
hit.app_name,
hit.db_host,
hit.db_port,
hit.db_name,
)
for hit in hits
if hit.resource_type == MalwareScanResourceType.DB.value
]
ignored = await IgnoreEndpoints(self._sink).try_add_to_ignore(
file_items,
resource_type=MalwareScanResourceType.FILE.value,
user=user,
) + await IgnoreEndpoints(self._sink).try_add_to_ignore(
db_items,
resource_type=MalwareScanResourceType.DB.value,
user=user,
)
if Malware.SEND_FILES:
for hit in file_hits:
# DEF-31406 send the original file (not cleaned)
content_path = str(
CleanupStorage.get_hit_store_path(hit)
if hit.status in MalwareHitStatus.CLEANED
else hit.orig_file
)
file = malware_response.MalwareHitPath(
content_path=content_path,
# the stored (tenant-prefixed on k8s) path is exactly
# what MRS should record
real_path=hit.orig_file,
)
try:
await MalwareAction.submit_for_analysis(
type=FALSE_POSITIVE,
reason=hit.type,
path=file,
file_owner=hit.owner,
file_user=hit.user,
initiator=user,
signature_id=hit.signature_id,
)
except ValidationError as e:
# e.g. the owning app was deregistered — keep moving
# the remaining hits to ignore instead of aborting
logger.warning(
"MRS submit skipped for %s: %s", hit.orig_file, e
)
return ignored
@feature(AV, [FULL, AV_REPORT])
async def _malicious_diff_for_cleaned_file(
self, differ: MalwareHitDiff, user=None
) -> bytes:
if is_cluster():
return await self._malicious_diff_k8s_cleaned(differ)
return await differ.get_unified_diff_for_cleaned_file()
@feature(AV, [FULL])
@validate_av_plus_license
async def _malicious_diff_for_infected_file(
self, differ: MalwareHitDiff, user=None
) -> bytes:
if is_cluster():
return await self._malicious_diff_k8s_found(differ)
return await differ.clean_and_get_unified_diff()
async def _malicious_diff_k8s_cleaned(
self, differ: MalwareHitDiff
) -> bytes:
"""K8s CLEANED diff: fetch current file via read, diff with backup."""
from imav.malwarelib.cleanup.storage import CleanupStorageK8s
from imav.malwarelib.difflib.differ import get_files_diff
import io
# Infected version is in local cleanup_storage (backup)
infected_path = CleanupStorage.get_hit_store_path(differ.hit)
if not infected_path.exists():
raise DiffError(f"Backup not found for hit(id={differ.hit.id}).")
# Current (cleaned) version is on the app pod — fetch via syncer
try:
result = await CleanupStorageK8s.read_file(
app_id=differ.hit.user,
path=str(differ.hit.orig_file_path),
offset=0,
limit=10 * 1024 * 1024, # 10 MB max for diff
)
if not result["eof"]:
# a partial body would silently diff as if the tail of the
# backup had been deleted — refuse, like the local path does
# for oversized files
raise DiffError(
"File is too large to compute diff"
f" (over {10 * 1024 * 1024} bytes,"
f" hit_id={differ.hit.id})."
)
cleaned_content = base64.b64decode(result["data"])
cleaned_file = io.BytesIO(cleaned_content)
except FileNotFoundError:
# File was removed during cleanup — diff against empty
cleaned_file = io.BytesIO(b"")
except OSError as e:
raise DiffError(f"Failed to read file for diff: {e}")
with open(infected_path, "rb") as infected_file:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, get_files_diff, infected_file, cleaned_file
)
async def _malicious_diff_k8s_found(self, differ: MalwareHitDiff) -> bytes:
"""Diff for FOUND (infected) files — syncer runs procu2.php."""
from imav.malwarelib.cleanup.storage import CleanupStorageK8s
from imav.malwarelib.difflib.differ import get_files_diff
import io
try:
result = await CleanupStorageK8s.clean_preview(
app_id=differ.hit.user,
path=str(differ.hit.orig_file_path),
)
except FileNotFoundError:
raise DiffError(f"File not found for hit(id={differ.hit.id}).")
except OSError as e:
raise DiffError(f"Clean preview failed: {e}")
infected_file = io.BytesIO(base64.b64decode(result["original"]))
cleaned_file = io.BytesIO(base64.b64decode(result["cleaned"]))
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, get_files_diff, infected_file, cleaned_file
)
@bind("malware", "malicious", "diff")
async def malicious_diff(self, id, user=None, **kwargs):
"""
Return the base64 encoded difference between infected and cleaned file
in unified diff format
"""
differ = MalwareHitDiff(id=id, user=user)
try:
if differ.hit.status in MalwareHitStatus.CLEANED:
diff = await self._malicious_diff_for_cleaned_file(
differ, user=user
)
elif differ.hit.status == MalwareHitStatus.FOUND:
diff = await self._malicious_diff_for_infected_file(
differ, user=user
)
else:
logger.warning(
"Malware hit has unexpected status=%s. Use the empty"
" diff.",
differ.hit.status,
)
return {"diff": ""}
except DiffError as exc:
raise ValidationError(str(exc)) from exc
except safe_fileops.UnsafeFileOperation as exc:
raise PermissionError("notifications.permissionError") from exc
return {"diff": base64.b64encode(diff).decode("utf-8")}
@feature(AV, [FULL, AV_REPORT])
@bind("malware", "history", "list")
@run_in_executor_decorator
def get_history(self, user=None, **kwargs):
return MalwareHistory.get_history(user=user, **kwargs)
@bind("malware", "user", "list")
async def user_list(
self, offset, limit, search=None, order_by=None, user=None, ids=None
):
if user:
# user endpoint
_, users = await user_list.fetch_user_list(
self.queue.get_scans_from_paths, match={user}
)
max_count = len(users)
elif search:
# search
_, users = await user_list.fetch_user_list(
self.queue.get_scans_from_paths, match=search
)
max_count = len(users)
elif ids:
# filter by ids
max_count, users = await user_list.fetch_user_list(
self.queue.get_scans_from_paths, match=ids
)
else:
# all users
max_count, users = await user_list.fetch_user_list(
self.queue.get_scans_from_paths
)
# sort
users = user_list.sort(users)
for order in reversed(order_by or []):
users = user_list.sort(users, order.column_name, desc=order.desc)
# limit and offset
start = offset
end = offset + limit
return max_count, users[start:end]
@bind("malware", "user", "scan")
async def user_scan(
self, scan_file, scan_db, background=False, **scan_args
):
if not scan_file and not scan_db:
raise ValidationError(
"Either --scan-file or --scan-db should be specified"
)
if background and self.queue.status().get("background"):
raise ValidationError("Background scan pending")
if background:
scan_type = MalwareScanType.BACKGROUND
else:
scan_type = MalwareScanType.ON_DEMAND
users = await user_list.panel_users()
if scan_db:
if is_cluster():
# DB scans keep bare local paths, so the tenant travels
# beside them; one put per application.
for u in users:
await self.queue.put(
paths=[u["home"]],
resource_type=MalwareScanResourceType.DB,
scan_type=scan_type,
user=u["user"],
**split_args(scan_args),
)
else:
await self.queue.put(
paths=[user["home"] for user in users],
resource_type=MalwareScanResourceType.DB,
scan_type=scan_type,
**split_args(scan_args),
)
if scan_file:
# The prefix is the identity; off-cluster to_prefixed is a
# no-op, so this is the same flat put on every deployment.
await self.queue.put(
paths=[
to_prefixed(user["home"], user["user"]) for user in users
],
resource_type=MalwareScanResourceType.FILE,
scan_type=scan_type,
**split_args(scan_args),
)
if background and Malware.CRONTABS_SCAN_ENABLED:
crontab_paths = [
p
for user in users
if os.path.exists(p := get_crontab(user["user"]))
]
if crontab_paths:
await self.queue.put(
paths=crontab_paths,
resource_type=MalwareScanResourceType.FILE,
scan_type=scan_type,
**split_args(scan_args),
)
def get_file_ownership(path, user) -> tuple[str, str]:
if is_cloudways():
# return a file owner and a user who keeps a file for Cloudways
owner = get_path_owner(path)
user = user or UserType.ROOT
else:
# use the same value for owner and user in other cases
owner = user = user or get_path_owner(path)
return owner, user
@feature(AV, [FULL, AV_REPORT])
class IgnoreEndpoints(CommonEndpoints):
@bind("malware", "ignore", "list")
async def ignore_list(self, user=None, **kwargs):
await check_permission(MS_IGNORE_LIST_EDIT, user)
if user is not None:
kwargs["user"] = user
count, items = MalwareIgnorePath.paths_count_and_list(**kwargs)
if is_cluster():
items = await split_stored_paths(items)
return count, items
@bind("malware", "ignore", "delete-ui")
async def ignore_delete_ui(self, ids, user=None):
return await self.ignore_delete(
ids=ids,
user=user,
skip_rescan=False,
)
@bind("malware", "ignore", "delete")
async def ignore_delete(self, ids, user=None, skip_rescan=False):
await check_permission(MS_IGNORE_LIST_EDIT, user)
ignore_paths: List[MalwareIgnorePath] = list(
MalwareIgnorePath.select().where(MalwareIgnorePath.id.in_(ids))
)
if user is not None:
if is_cluster():
ignore_paths = [
p for p in ignore_paths if is_under_prefix(p.path, user)
]
else:
user_crontab_path = get_crontab(user)
ignore_paths = [
ignore_path
for ignore_path in ignore_paths
if does_path_belong_to_user(ignore_path.path, user)
or ignore_path.path == user_crontab_path
]
file_paths = [
ignore_path.path
for ignore_path in ignore_paths
if ignore_path.resource_type == MalwareScanResourceType.FILE.value
]
with db.atomic():
for ignore_path in ignore_paths:
file_owner, file_user = get_file_ownership(
ignore_path.path, user
)
MalwareAction.delete_from_ignore_sync(
path=ignore_path.path,
file_owner=file_owner,
file_user=file_user,
initiator=user or UserType.ROOT,
resource_type=ignore_path.resource_type,
signature_id=None,
)
if file_paths:
if not skip_rescan:
await self._sink.process_message(
# In cluster the paths are tenant-prefixed and live in
# other containers; only the rescan types skip the
# agent-local existence check and route by prefix, so the
# re-scan must be a rescan there.
MessageType.MalwareScanTask(
filelist=file_paths,
scan_type=(
MalwareScanType.RESCAN
if is_cluster()
else MalwareScanType.REALTIME
),
)
)
await self._sink.process_message(
MessageType.MalwareIgnorePathUpdated()
)
return len(ignore_paths)
@bind("malware", "ignore", "add")
async def ignore_add(self, resource_type, paths, user=None):
if (
is_cluster()
and resource_type == MalwareScanResourceType.FILE.value
):
# Both input forms are accepted (bare path + --user, or an
# already-prefixed path); either way the entry is stored in the
# canonical prefixed form, which is what scopes it to its tenant.
registered = await user_list.registered_apps()
resolved = [
await resolve_tenant_path(
str(path), user, registered=registered
)
for path in paths
]
# no tenant resolved means an admin-wide entry: scope it to the
# wildcard tenant so it applies inside every application
items = [
IgnoreParameters(
tp.to_prefixed()
if tp.user
else f"/{GLOBAL_TENANT}{tp.display_path}"
)
for tp in resolved
]
elif (
is_cluster()
and user is not None
and resource_type == MalwareScanResourceType.DB.value
):
# A db entry carries no tenant prefix and the resident matches
# it syntactically across every application, so a tenant request
# cannot be honoured without suppressing other tenants'
# detections. Only an admin (user is None) may add one.
raise ValidationError(
"Database ignore-list entries are not available to a"
" tenant in cluster mode"
)
else:
items = [IgnoreParameters(path) for path in paths]
added = await self.try_add_to_ignore(items, resource_type, user)
return len(added)
async def try_add_to_ignore(
self,
items: Sequence[IgnoreParameters],
resource_type: str,
user: str | None = None,
) -> List[str]:
await check_permission(MS_IGNORE_LIST_EDIT, user)
if (
is_cluster()
and user is not None
and resource_type == MalwareScanResourceType.DB.value
):
# A tenant cannot own a db ignore in cluster mode: the resident
# matches db entries across every application. ignore_add raises
# for a direct request; a mixed move-to-ignore batch funnels
# here instead, so drop the db items yet keep ignoring the file
# hits rather than aborting the whole batch.
return []
added = []
items = [item for item in items if os.path.isabs(item.path)]
if (
resource_type == MalwareScanResourceType.FILE.value
and not is_cluster()
):
# the resident is handed this path as-is, so resolve away what the
# user controls instead of checking the spelling they supplied
scoped = []
for item in items:
path = scope_to_user(item.path, user)
if path is not None:
scoped.append(item._replace(path=path))
items = scoped
elif user is not None and not is_cluster():
# in cluster mode the tenant is the path prefix, and ownership
# inside the container is not the agent's to judge
user_crontab_path = get_crontab(user)
items = (
i
for i in items
if does_path_belong_to_user(i.path, user)
or i.path == user_crontab_path
)
already_ignored = MalwareIgnorePath.path_list(
resource_type=resource_type
)
items = (items for items in items if items.path not in already_ignored)
for item in items:
file_owner, file_user = get_file_ownership(item.path, user)
result = await MalwareAction.ignore(
path=item.path,
resource_type=resource_type,
file_owner=file_owner,
file_user=file_user,
initiator=user or UserType.ROOT,
app_name=item.app_name,
db_host=item.db_host,
db_port=item.db_port,
db_name=item.db_name,
signature_id=item.signature_id,
)
if result.successful:
added.append(item.path)
if added:
await self._sink.process_message(
MessageType.MalwareIgnorePathUpdated()
)
return added
async def is_path_ignored(self, check_path, user=None):
assert os.path.isabs(check_path)
if user is not None:
belongs = (
is_under_prefix(check_path, user)
if is_cluster()
else does_path_belong_to_user(check_path, user)
)
if not belongs:
return False
return await MalwareIgnorePath.is_path_ignored(check_path)
class MalwareRebuildPatterns(RootEndpoints):
@bind("malware", "rebuild", "patterns")
async def rebuild_patterns(self):
await self._sink.process_message(
MessageType.MalwareIgnorePathUpdated()
)
return {}
class MalwareRescanEdnpoints(RootEndpoints):
@bind("malware", "rescan")
async def malware_rescan_files(self, files: List[str]):
await self._sink.process_message(
MessageType.MalwareRescanFiles(files=files)
)
return {}
class MalwareSendFiles(RootEndpoints):
@bind("malware", "send", "files")
async def malware_send_files(self, reason: str, files: List[str]):
await self._sink.process_message(
MessageType.MalwareSendFiles(reason=reason, files=files)
)
return {}