ObjSee ProcessMonitor wrapper

procmon.py

#!/usr/bin/env python
from __future__ import print_function
import os
import subprocess
import threading
import json

# ProcessMonitor need to download from https://objective-see.com/products/utilities.html 
# and put to /Applications/ folder
# also give Terminal.app Full Disk Access
# Launch with sudo
# sudo python ./procmon.py

PROCMON_BIN = "/Applications/ProcessMonitor.app/Contents/MacOS/ProcessMonitor"
# using script command to emulate tty
SCRIPT_BIN = "/usr/bin/script"

"""
Event example:
{
  "event": "ES_EVENT_TYPE_NOTIFY_EXEC",
  "timestamp": "2021-09-25 07:32:50 +0000",
  "process": {
    "pid": 674,
    "name": "Activity Monitor",
    "path": "/System/Applications/Utilities/Activity Monitor.app/Contents/MacOS/Activity Monitor",
    "uid": 501,
    "architecture": "Intel",
    "arguments": [
      "/System/Applications/Utilities/Activity Monitor.app/Contents/MacOS/Activity Monitor"
    ],
    "ppid": 1,
    "rpid": 0,
    "ancestors": [
      1
    ],
    "signing info (reported)": {
      "csFlags": 570456849,
      "platformBinary": 1,
      "signingID": "com.apple.ActivityMonitor",
      "teamID": "",
      "cdHash": "897D9A325FFE3E93560320393D8ABFD031763241"
    },
    "signing info (computed)": {
      "signatureID": "com.apple.ActivityMonitor",
      "signatureStatus": 0,
      "signatureSigner": "Apple",
      "signatureAuthorities": [
        "Software Signing",
        "Apple Code Signing Certification Authority",
        "Apple Root CA"
      ]
    }
  }
}
"""


def event_handler(event):
    """
    Outputs process only exec events:
        0     688   <1>    xpcproxy com.apple.ActivityMonitor.20448
        501   688   <1>    /System/Applications/Utilities/Activity Monitor.app/Contents/MacOS/Activity Monitor
    """

    if event["event"] == "ES_EVENT_TYPE_NOTIFY_EXEC":
        proc = event["process"]
        print("{:<5} {:<5} {:<7}{}".format(
            proc["uid"],
            proc["pid"],
            "<{}>".format(proc["ppid"]),
            " ".join(proc["arguments"])
        ), end="\r\n")  # using \r\n as script command modifies output


class ProcessMonitorStreamReader:

    def __init__(self, cb):
        self.cb = cb

    def run(self):
        if not os.path.exists(PROCMON_BIN):
            print("ProcessMonitor.app not found in /Applications/ folder")
            return

        procmon_args = [PROCMON_BIN]
        script_args = [SCRIPT_BIN, "-q", "/dev/null"]
        p = subprocess.Popen(script_args + procmon_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        event_json = ""
        for line in iter(p.stdout.readline, b''):
            if line.startswith("{"):
                event_json = "{\n"
                if line.rstrip().endswith("}"):
                    event_json = ""
                    self._process_event(line)

            elif line.startswith("}"):
                event_json += "}"
                self._process_event(event_json)
                event_json = ""
            else:
                event_json += line

        print(p.stderr.read(), end="")

    def _process_event(self, event_json):
        try:
            event_raw = json.loads(event_json)
            self.cb(event_raw)

        except Exception as e:
            pass
            print(e)


if __name__ == "__main__":
    ProcessMonitorStreamReader(event_handler).run()