FM / utils

/opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils/

/opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
antivirus_mode.py	497	0644	edit dl rm
async_utils.py	718	0644	edit dl rm
benchmark.py	538	0644	edit dl rm
buffer.py	1945	0644	edit dl rm
check_db.py	7988	0644	edit dl rm
check_lock.py	856	0644	edit dl rm
cli.py	9273	0644	edit dl rm
common.py	14757	0644	edit dl rm
completions.py	9988	0644	edit dl rm
config.py	1695	0644	edit dl rm
cronjob.py	902	0644	edit dl rm
doctor.py	5503	0644	edit dl rm
fd_ops.py	7573	0644	edit dl rm
hyperscan.py	149	0644	edit dl rm
importer.py	2722	0644	edit dl rm
ipecho.py	3247	0644	edit dl rm
json.py	953	0644	edit dl rm
kwconfig.py	1756	0644	edit dl rm
net.py	595	0644	edit dl rm
net_transport.py	16289	0644	edit dl rm
parsers.py	11879	0644	edit dl rm
resource_limits.py	5158	0644	edit dl rm
safe_fileops.py	13672	0644	edit dl rm
safe_sequence.py	363	0644	edit dl rm
serialization.py	2410	0644	edit dl rm
sshutil.py	15174	0644	edit dl rm
subprocess.py	1570	0644	edit dl rm
support.py	5329	0644	edit dl rm
tenant_path.py	4456	0644	edit dl rm
threads.py	1005	0644	edit dl rm
validate.py	4375	0644	edit dl rm
whmcs.py	7784	0644	edit dl rm
wordpress_mu_plugin.py	1440	0644	edit dl rm
zipsafe.py	720	0644	edit dl rm
_shutil.py	4019	0644	edit dl rm
__init__.py	69706	0644	edit dl rm

Edit: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils/serialization.py (2410B)

"""JSON persistence helpers for small agent state files (no pickle at runtime)."""

import collections
import functools
import json
import logging
import os
from asyncio import iscoroutinefunction
from typing import Any, Callable, Union

logger = logging.getLogger(__name__)

def _to_jsonable(obj: Any) -> Any:
    if isinstance(obj, collections.deque):
        return [_to_jsonable(item) for item in obj]
    if isinstance(obj, dict):
        return {k: _to_jsonable(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_to_jsonable(item) for item in obj]
    return obj

def _dump(path, obj):
    """Atomically write ``obj`` to ``path`` as JSON."""
    payload = json.dumps(_to_jsonable(obj))
    tmp = "{}.tmp".format(path)
    with open(tmp, "w", encoding="utf-8") as w:
        w.write(payload)
    os.replace(tmp, path)

def serialize_attr(*, path: str, attr: str):
    """Decorator: after the wrapped method runs, persist ``self.<attr>``
    to ``path`` as JSON."""

def decorator(f):
        @functools.wraps(f)
        def wrapper(self, *args, **kwargs):
            result = f(self, *args, **kwargs)
            obj = getattr(self, attr)
            logger.debug("Write %r to %r", obj, path)
            _dump(path, obj)
            return result

@functools.wraps(f)
        async def async_wrapper(self, *args, **kwargs):
            result = await f(self, *args, **kwargs)
            obj = getattr(self, attr)
            logger.debug("Write %r to %r", obj, path)
            _dump(path, obj)
            return result

if iscoroutinefunction(f):
            return async_wrapper
        return wrapper

return decorator

def unserialize(*, path: str, fallback: Union[Callable, object] = None):
    """Restore an object from ``path`` (JSON); a top-level list becomes a
    deque to match the legacy queue API, and missing/unparseable input
    returns ``fallback`` (called if callable)."""
    try:
        with open(path, "r", encoding="utf-8") as r:
            obj = json.load(r)
    except FileNotFoundError:
        logger.warning("Can't find %s to unserialize", path)
    except Exception as e:
        logger.error("Unserialize failed with %r. Returning fallback", e)
    else:
        if isinstance(obj, list):
            return collections.deque(obj)
        return obj
    return fallback() if callable(fallback) else fallback

cmd: