tsale-EDR-Telemetry/Tools/Telemetry-Generator/macOS/native.py

#!/usr/bin/env python3
"""Native macOS helpers (no external binaries).

This module intentionally avoids spawning system utilities. It uses ctypes to
call macOS frameworks/libraries directly where practical.
"""

from __future__ import annotations

import ctypes
import ctypes.util
from typing import Iterable, List, Optional, Tuple


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


def xattr_set(path: str, name: str, value: bytes, flags: int = 0) -> int:
    """Set an extended attribute via libc setxattr."""

    _libc.setxattr.argtypes = [
        ctypes.c_char_p,
        ctypes.c_char_p,
        ctypes.c_void_p,
        ctypes.c_size_t,
        ctypes.c_uint32,
        ctypes.c_int,
    ]
    _libc.setxattr.restype = ctypes.c_int

    bpath = path.encode("utf-8")
    bname = name.encode("utf-8")
    buf = ctypes.create_string_buffer(value)
    return int(_libc.setxattr(bpath, bname, buf, len(value), 0, int(flags)))


def xattr_get(path: str, name: str) -> Optional[bytes]:
    """Get an extended attribute via libc getxattr."""

    _libc.getxattr.argtypes = [
        ctypes.c_char_p,
        ctypes.c_char_p,
        ctypes.c_void_p,
        ctypes.c_size_t,
        ctypes.c_uint32,
        ctypes.c_int,
    ]
    _libc.getxattr.restype = ctypes.c_ssize_t

    bpath = path.encode("utf-8")
    bname = name.encode("utf-8")

    size = int(_libc.getxattr(bpath, bname, None, 0, 0, 0))
    if size < 0:
        return None
    if size == 0:
        return b""

    buf = ctypes.create_string_buffer(size)
    got = int(_libc.getxattr(bpath, bname, buf, size, 0, 0))
    if got < 0:
        return None
    return buf.raw[:got]


def xattr_remove(path: str, name: str) -> int:
    """Remove an extended attribute via libc removexattr."""

    _libc.removexattr.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int]
    _libc.removexattr.restype = ctypes.c_int

    bpath = path.encode("utf-8")
    bname = name.encode("utf-8")
    return int(_libc.removexattr(bpath, bname, 0))


class _CoreFoundation:
    def __init__(self) -> None:
        path = ctypes.util.find_library("CoreFoundation")
        if not path:
            raise RuntimeError("CoreFoundation not found")
        self.cf = ctypes.CDLL(path)

        self.CFRelease = self.cf.CFRelease
        self.CFRelease.argtypes = [ctypes.c_void_p]
        self.CFRelease.restype = None

        self.CFStringGetCString = self.cf.CFStringGetCString
        self.CFStringGetCString.argtypes = [
            ctypes.c_void_p,
            ctypes.c_char_p,
            ctypes.c_long,
            ctypes.c_uint32,
        ]
        self.CFStringGetCString.restype = ctypes.c_bool


def _cfstring_to_str(cfstring: int) -> Optional[str]:
    if not cfstring:
        return None

    cf = _CoreFoundation()
    # kCFStringEncodingUTF8
    kCFStringEncodingUTF8 = 0x08000100
    buf = ctypes.create_string_buffer(1024)
    ok = cf.CFStringGetCString(cfstring, buf, ctypes.sizeof(buf), kCFStringEncodingUTF8)
    try:
        return buf.value.decode("utf-8") if ok else None
    finally:
        cf.CFRelease(cfstring)


def get_console_user() -> Tuple[Optional[str], Optional[int], Optional[int]]:
    """Return (username, uid, gid) for the active console user.

    Uses SystemConfiguration's SCDynamicStoreCopyConsoleUser.
    """

    sc_path = "/System/Library/Frameworks/SystemConfiguration.framework/SystemConfiguration"
    sc = ctypes.CDLL(sc_path)

    sc.SCDynamicStoreCopyConsoleUser.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(ctypes.c_uint32),
        ctypes.POINTER(ctypes.c_uint32),
    ]
    sc.SCDynamicStoreCopyConsoleUser.restype = ctypes.c_void_p

    uid = ctypes.c_uint32(0)
    gid = ctypes.c_uint32(0)
    cf_user = sc.SCDynamicStoreCopyConsoleUser(None, ctypes.byref(uid), ctypes.byref(gid))
    username = _cfstring_to_str(cf_user) if cf_user else None
    return username, int(uid.value) if cf_user else None, int(gid.value) if cf_user else None


class _LibProc:
    def __init__(self) -> None:
        # libproc is usually available via find_library('proc')
        path = ctypes.util.find_library("proc") or "/usr/lib/libproc.dylib"
        self.lp = ctypes.CDLL(path)

        self.proc_listallpids = self.lp.proc_listallpids
        self.proc_listallpids.argtypes = [ctypes.c_void_p, ctypes.c_int]
        self.proc_listallpids.restype = ctypes.c_int

        self.proc_name = self.lp.proc_name
        self.proc_name.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_uint32]
        self.proc_name.restype = ctypes.c_int


def list_processes() -> List[Tuple[int, str]]:
    """Return a list of (pid, process_name) for running processes."""

    lp = _LibProc()
    count = lp.proc_listallpids(None, 0)
    if count <= 0:
        return []

    # proc_listallpids returns number of PIDs when buffer is NULL
    pids = (ctypes.c_int * count)()
    written = lp.proc_listallpids(pids, ctypes.sizeof(pids))
    if written <= 0:
        return []

    results: List[Tuple[int, str]] = []
    name_buf = ctypes.create_string_buffer(1024)

    for i in range(min(written, count)):
        pid = int(pids[i])
        if pid <= 0:
            continue
        name_buf.raw = b"\x00" * ctypes.sizeof(name_buf)
        n = lp.proc_name(pid, name_buf, ctypes.sizeof(name_buf))
        if n > 0:
            try:
                results.append((pid, name_buf.value.decode("utf-8", errors="replace")))
            except Exception:
                continue

    return results


def find_processes_by_name(names: Iterable[str]) -> List[Tuple[int, str]]:
    wanted = {n.lower() for n in names}
    return [(pid, name) for pid, name in list_processes() if name.lower() in wanted]