✏️ Editing: snapshot.py Read Only

# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
import json
import logging
import os
import pwd
import shutil
import sys
import zlib
from collections.abc import Iterator
from contextlib import contextmanager
from typing import BinaryIO

SNAPSHOT_PATH = "/var/lve/snapshots"
SNAPSHOT_EXT = ".snapshot"
SNAPSHOT_EXT_LEN = len(SNAPSHOT_EXT)

class Snapshot:
    """
    Class responsible for loading and saving snapshot files for the interval.

The files will be saved in a format of: `/var/lve/snapshots/[uid]/dump_time.snapshot`

dump_time is the timestamp/integer.

The directories `/var/lve/snapshots/[uid]` and the dump files themselves will be owned by user.
    They will not be readable by other users.
    """

def __init__(self, incident: dict, compresslevel: int = 1) -> None:
        self.compresslevel = compresslevel
        self.incident = incident
        self.log = logging.getLogger("lib-snapshot")

def save(self, data: dict) -> str:
        dump_date = data["dump_time"]
        assert dump_date is not None

# convert possible non-ascii data to unicode
        self._replace_unicode_data(data)
        json_compressed = zlib.compress(json.dumps(data).encode(), self.compresslevel)
        with self.create_file(dump_date) as f:
            f.write(json_compressed)
            self.log.debug("Snapshot dumped to file %s", f.name)
        return f.name

@staticmethod
    def _to_unicode(obj):
        if isinstance(obj, bytes):
            return obj.decode("utf-8", "replace")
        return obj

def _replace_unicode_data(self, data: dict) -> None:
        u_queries = []
        for query in data.get("snap_sql", []):
            u_queries.append(list(map(self._to_unicode, query)))
        data["snap_sql"] = u_queries

u_urls = []
        for http in data.get("snap_http", []):
            u_urls.append(list(map(self._to_unicode, http)))
        data["snap_http"] = u_urls

def get_file_list(self) -> list[str]:
        dir_ = self.get_dir()
        if os.path.isdir(dir_):
            return os.listdir(dir_)
        return []

def get_ts_list(
        self,
        from_ts: int | None,
        to_ts: int | None,
    ) -> list[int]:
        """
        Return ordered list of timestamps when snapshots for this user were created.

:param from_ts: Start timestamp for filtering snapshots (inclusive).
        If None, starts from the earliest snapshot.
        :param to_ts: End timestamp for filtering snapshots (inclusive).
        If None, includes all snapshots up to the latest.
        :return: List of timestamps ordered for the given period.
        """
        return self.snapshot_filter(self.get_file_list(), from_ts, to_ts)

def get_snapshots(
        self,
        from_ts: int | None,
        to_ts: int | None,
    ) -> Iterator[dict]:
        """
        Get snapshots for a given period.

Yields one snapshot at a time for memory efficiency.

:param from_ts: Start timestamp for filtering snapshots (inclusive).
        If None, starts from the earliest snapshot.
        :param to_ts: End timestamp for filtering snapshots (inclusive).
        If None, includes all snapshots up to the latest.
        :return: Iterator of snapshot dictionaries.
        """
        for ts in self.get_ts_list(from_ts, to_ts):
            try:
                filename = self.get_file_name(self.ts_to_name(ts))
                # SECURITY: read_file_content has a write-back path (recompresses
                # uncompressed files). drop_privileges ensures this happens as the
                # user, not root, preventing symlink-based file corruption attacks
                # via user-owned snapshot directories. The directory is intentionally
                # user-owned so a user can only affect their own diagnostic data.
                if not os.geteuid():
                    with drop_privileges(self.incident["uid"]):
                        content = self.read_file_content(filename)
                else:
                    content = self.read_file_content(filename)

snapshot_data = json.loads(content)
                yield snapshot_data

except (ValueError, OSError) as ve:
                self.log.warning(
                    "Corrupted file: %s (%s)",
                    self.get_file_name(self.ts_to_name(ts)),
                    str(ve),
                )

def read_file_content(self, filename: str) -> str:
        with open(filename, "rb") as f:
            content = f.read()
        try:
            content = zlib.decompress(content)
        except zlib.error:
            compressed_content = zlib.compress(content, self.compresslevel)
            with open(filename, "wb") as f:
                f.write(compressed_content)
        return content.decode()

def get_incident_snapshots(self) -> list:
        """
        Load all snapshots for given incident
        :return: list of snapshots
        """
        return list(self.get_snapshots(self.incident["incident_start_time"], self.incident["incident_end_time"]))

def get_dir(self) -> str:
        return os.path.join(SNAPSHOT_PATH, str(self.incident["uid"]))

def get_file_name(self, name: str) -> str:
        return os.path.join(self.get_dir(), name)

def create_file(self, dump_date: int) -> BinaryIO:
        """
        create file, change its ownership & permissions if needed. Create directories if needed as well
        :param dump_date: int timestamp used as file name
        :return: open File object
        """
        dir_ = self.get_dir()
        if not os.path.exists(dir_):
            try:  # sacrifice security if we cannot setup ownership properly
                os.makedirs(dir_)
                # SECURITY: directory is user-owned so each user can only access
                # their own snapshots. This is safe because snapshots are purely
                # diagnostic data (process lists, HTTP requests, SQL queries at
                # LVE fault time). Forging own snapshots has no security impact —
                # LVE enforcement is in the kernel, incident records are in the DB.
                os.chmod(dir_, 0o751)
                os.chown(dir_, self.incident["uid"], 0)
            except OSError as e:
                self.log.error("Unable to create dir %s (%s)", dir_, str(e))
        file_name = self.get_file_name(self.ts_to_name(dump_date))
        # SECURITY: drop_privileges ensures file creation and permission
        # setting happen as the user, preventing symlink attacks from
        # escalating to root-owned file corruption.
        with drop_privileges(self.incident["uid"]):
            file_ = open(file_name, "wb")  # pylint: disable=consider-using-with
            try:
                os.fchmod(file_.fileno(), 0o400)
            except OSError as e:
                self.log.error("Unable to set file permissions %s (%s)", file_name, str(e))
        return file_

def delete_old(self, to_ts: int) -> None:
        """
        Delete old snapshots. If there are no more
        :param to_ts: up to which timestamp to remove snapshots
        :return: None
        """
        _dir = self.get_dir()
        files = os.listdir(_dir)
        all_snapshots = self.snapshot_filter(files)
        ts_to_remove = self.snapshot_filter(files, to_ts=to_ts)
        if all_snapshots == ts_to_remove:
            shutil.rmtree(_dir, ignore_errors=True)
        else:
            for ts in ts_to_remove:
                os.remove(self.get_file_name(self.ts_to_name(ts)))

@staticmethod
    def get_ts(file_: str) -> int | None:
        if file_.endswith(SNAPSHOT_EXT):
            ts = file_[0:-SNAPSHOT_EXT_LEN]
            if ts.isdigit():
                try:
                    return int(ts)
                except ValueError:
                    pass
        return None

@staticmethod
    def snapshot_filter(
        files: list[str],
        from_ts: int | None = None,
        to_ts: int | None = None,
    ) -> list[int]:
        if from_ts is None:
            from_ts = 0
        if to_ts is None:
            to_ts = sys.maxsize
        result = []
        for filename in files:
            ts = Snapshot.get_ts(filename)
            if ts is not None and from_ts <= ts <= to_ts:
                result.append(ts)
        return sorted(result)

@staticmethod
    def ts_to_name(ts: int) -> str:
        return str(ts) + SNAPSHOT_EXT

@contextmanager
def drop_privileges(uid: int):
    old_uid, old_gid, old_groups = os.getuid(), os.getgid(), os.getgroups()
    gid = pwd.getpwnam("nobody")[3]
    os.setgroups([])
    os.setegid(gid)
    os.seteuid(uid)
    try:
        yield
    finally:
        os.seteuid(old_uid)
        os.setegid(old_gid)
        os.setgroups(old_groups)

🔐 Sid Gifari File Manager Pro