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/tenant_path.py (4456B)

"""Tenant identity carried by the path, for cluster mode.

A resource names its tenant in its own path — ``/app1/var/www/html/x`` —
so nothing else has to carry an application id. This module is the codec
for that form; both products read and write it.
"""

import os
from typing import Union

from defence360agent.utils import is_cluster

def to_prefixed(path: Union[str, os.PathLike], user: "str | None") -> str:
    """Serialize a (user, path) pair into the canonical tenant-prefixed
    string used in the DB and on the MRS/server wire on k8s:
    ``/var/www/x`` + ``app1`` -> ``/app1/var/www/x``.

No-op (returns ``str(path)``) when the user is falsy, outside k8s,
    or the path already carries this user's prefix — so it is safe to
    call at every serialization boundary without double-prefixing.
    """
    s = os.fspath(path)
    if not user or not is_cluster():
        return s
    prefix = f"/{user}"
    if s == prefix or s.startswith(prefix + "/"):
        return s
    return prefix + s

def split_prefixed(path: Union[str, os.PathLike]) -> "tuple[str | None, str]":
    """Purely syntactic split of a tenant-prefixed path:
    ``/app1/var/www/x`` -> ``("app1", "/var/www/x")``.

Returns ``(None, path)`` for relative paths or paths with fewer than
    two components (``/app1`` alone is not a prefixed path). No registry
    validation happens here — callers at input boundaries must validate
    the candidate user against registered applications.
    """
    s = os.fspath(path)
    if not s.startswith("/"):
        return None, s
    first, sep, rest = s[1:].partition("/")
    if not sep or not first or not rest:
        return None, s
    return first, "/" + rest

def strip_prefix(path: Union[str, os.PathLike], user: "str | None") -> str:
    """Strip ``/<user>`` iff the path starts with it; else return unchanged.

Used on the read/display side when the tenant is already known
    (e.g. from the ``MalwareHit.user`` column). Gated on k8s like
    ``to_prefixed`` — on a standalone host a user named e.g. ``home``
    must never get ``/home/home/x`` mangled to ``/home/x``.
    """
    s = os.fspath(path)
    if not user or not is_cluster():
        return s
    prefix = f"/{user}"
    if s.startswith(prefix + "/"):
        return s[len(prefix) :]
    return s

#: Tenant name of an ignore-list entry that applies to every application.
#: Not a legal application id, so it cannot collide with a real tenant.
GLOBAL_TENANT = "*"

def under_prefix(field, user: str):
    """peewee predicate: ``field`` is ``/<user>`` or lives beneath it.

A half-open range instead of ``LIKE '/user/%'``: exact under BINARY
    collation (LIKE is case-insensitive in SQLite and would let ``/App-1/``
    read ``/app-1/``'s rows), free of ``%``/``_`` escaping, and index-usable.
    """
    prefix = f"/{user}/"
    upper = prefix[:-1] + chr(ord(prefix[-1]) + 1)
    return (field == f"/{user}") | ((field >= prefix) & (field < upper))

def strip_tenant(path, user: "str | None" = None) -> str:
    """A stored path as a consumer outside the agent's tables wants it.

The wildcard marker always comes off, and ``user``'s own prefix when a
    tenant is given. Anything else is returned untouched: the split is
    syntactic, so a bare host path (/usr/share/..., which no cluster write
    ever prefixes) must not lose its first directory, and another tenant's
    entry is left as it is — as a scanner exclusion pattern it simply never
    matches inside this container.
    """
    s = os.fspath(path)
    if not is_cluster():
        return s
    tenant, bare = split_prefixed(s)
    if tenant == GLOBAL_TENANT or (user is not None and tenant == user):
        return bare
    return s

def split_marked(path, applications) -> "tuple[str | None, str]":
    """Split a stored path into (tenant, in-container path), but only when
    the tenant is one we can prove: the wildcard, or a registered
    application. A host path such as /usr/share/... must not lose its first
    directory to a syntactic guess.
    """
    s = os.fspath(path)
    tenant, bare = split_prefixed(s)
    if tenant == GLOBAL_TENANT or (
        tenant is not None and tenant in applications
    ):
        return tenant, bare
    return None, s

def is_under_prefix(path, user: str) -> bool:
    """Python counterpart of :func:`under_prefix`."""
    s = os.fspath(path)
    return s == f"/{user}" or s.startswith(f"/{user}/")

cmd: