Viewing File: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils/tenant_path.py

"""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}/")