monitor.py

import argparse
import os
import platform
import time
from datetime import datetime

import psutil

from Process_OS.windows_process import WHITELIST as WINDOWS_WHITELIST
from Process_OS.windows_process import WHITELIST_PID as WINDOWS_WHITELIST_PID
from Process_OS.linux_process import WHITELIST as LINUX_WHITELIST
from Process_OS.unix_process import WHITELIST as UNIX_WHITELIST


def get_os_policy():
    os_name = platform.system().lower()

    if os_name == "windows":
        whitelist = set(WINDOWS_WHITELIST)
        whitelist_pid = set(WINDOWS_WHITELIST_PID)
        normalizer = lambda name: name
        return "windows", whitelist, whitelist_pid, normalizer

    if os_name == "linux":
        whitelist = set(name.lower().strip() for name in LINUX_WHITELIST)
        whitelist_pid = {1}
        normalizer = lambda name: (name or "").lower().strip()
        return "linux", whitelist, whitelist_pid, normalizer

    if os_name == "darwin":
        whitelist = set(os.path.basename(name).lower().strip() for name in UNIX_WHITELIST)
        whitelist_pid = {1}
        normalizer = lambda name: os.path.basename(name or "").lower().strip()
        return "darwin", whitelist, whitelist_pid, normalizer

    raise RuntimeError(f"Unsupported operating system: {os_name}")


def detect_blocked_processes(whitelist, whitelist_pid, normalizer):
    blocked = []

    for proc in psutil.process_iter(["pid", "name", "username"]):
        try:
            pid = proc.info["pid"]
            name = proc.info["name"] or ""
            user = proc.info.get("username") or "unknown"

            if pid in whitelist_pid:
                continue

            normalized_name = normalizer(name)
            if normalized_name in whitelist:
                continue

            blocked.append(
                {
                    "pid": pid,
                    "name": name,
                    "normalized_name": normalized_name,
                    "user": user,
                    "time": datetime.now().strftime("%H:%M:%S"),
                    "reason": "Process not in whitelist",
                }
            )
        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            continue

    return blocked


def monitor(interval_seconds=1.0, once=False):
    os_name, whitelist, whitelist_pid, normalizer = get_os_policy()
    print(f"Monitoring OS: {os_name}")
    print(f"Scan interval: {interval_seconds:.2f}s")
    print("Alert mode only: processes are NOT terminated.")

    alerted_names = set()

    while True:
        blocked = detect_blocked_processes(whitelist, whitelist_pid, normalizer)
        current_blocked_names = set()

        for proc in blocked:
            blocked_name_key = proc["normalized_name"]
            current_blocked_names.add(blocked_name_key)
            if blocked_name_key in alerted_names:
                continue
            print(
                f"{proc['time']} PID={proc['pid']:<6} User={proc['user']:<12}\tName={proc['name']}"
                #f"User={proc['user']} | Reason={proc['reason']}"
            )
            alerted_names.add(blocked_name_key)

        alerted_names.intersection_update(current_blocked_names)

        if once:
            break

        time.sleep(interval_seconds)


def parse_args():
    parser = argparse.ArgumentParser(
        description="Monitor running processes against OS whitelist and alert to stdout."
    )
    parser.add_argument(
        "--interval",
        type=float,
        default=1.0,
        help="Seconds between scans (default: 1.0)",
    )
    parser.add_argument(
        "--once",
        action="store_true",
        help="Run a single scan and exit.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.interval <= 0:
        raise ValueError("--interval must be greater than 0")

    try:
        monitor(interval_seconds=args.interval, once=args.once)
    except KeyboardInterrupt:
        print("\nMonitoring stopped by user.")


if __name__ == "__main__":
    main()