CWE-114: Process Control - Python

Overview

Process control vulnerabilities in Python applications occur when untrusted user input decides which process to start, signal, or stop. The request then reaches the operating system with the privileges of the account the application server runs as.

Key Security Issues:

• Unauthorized Process Termination: Killing critical system or application processes
• Resource Exhaustion: Spawning unlimited processes to cause denial of service
• Privilege Escalation: Manipulating process priorities or spawning privileged processes
• Command Injection: Shell metacharacters in a PID or priority parameter that reaches a shell=True call
• Information Disclosure: Accessing process information to map system architecture

Primary Defence: Use subprocess.run() with explicit argument lists (not shell=True), allowlist the executables and signals a request may name, authorize each operation against the application's own record of the processes it started, and bound the child with resource.setrlimit() or container limits.

Common Python Scenarios:

• Web applications killing background worker processes based on user input
• Admin panels allowing process restart with user-supplied PIDs
• Job schedulers terminating jobs using unvalidated job IDs
• Container orchestration accepting process control commands
• Service managers with insufficient authorization checks
• Monitoring tools displaying process details without access control

Why this matters in Python:

• os.kill(), subprocess, and signal modules provide powerful but dangerous process control
• Microservices and containerized apps often need process management
• Django/Flask admin panels frequently implement process control features
• Celery workers and task queues require process lifecycle management

Common Vulnerable Patterns

Unvalidated Process Termination

import os
import signal
from flask import Flask, request

app = Flask(__name__)

@app.route('/admin/kill-process', methods=['POST'])
def kill_process():
    # DANGEROUS: User controls which process to kill
    pid = int(request.form.get('pid'))

    try:
        os.kill(pid, signal.SIGKILL)  # No validation or authorization
        return f"Process {pid} terminated"
    except ProcessLookupError:
        return "Process not found", 404

Why this is vulnerable:

• No authorization check - any user can kill any process
• No validation of PID - can target system processes
• No ownership verification
• No audit logging of who killed what

Command Injection via Process Control

import subprocess

def set_process_priority(pid, priority):
    # DANGEROUS: Command injection vulnerability
    command = f"renice {priority} -p {pid}"
    subprocess.run(command, shell=True)  # shell=True is dangerous

Why this is vulnerable:

• Using shell=True enables command injection
• String interpolation allows injection of shell metacharacters
• No validation of priority or PID parameters
• User input directly in shell command

Unrestricted Process Spawning

import subprocess
from flask import request

@app.route('/run-job', methods=['POST'])
def run_job():
    # DANGEROUS: Unrestricted process spawning
    job_type = request.form.get('job_type')
    job_args = request.form.get('args', '').split()

    # No validation or resource limits
    subprocess.Popen([f'/opt/jobs/{job_type}.py'] + job_args)
    return "Job started"

Why this is vulnerable:

• No rate limiting - can spawn unlimited processes
• Path traversal possible in job_type
• No resource limits on spawned processes
• Arguments not validated

Signal Handling Without Authorization

import os
import signal

def pause_process(pid):
    # DANGEROUS: No authorization or validation
    os.kill(pid, signal.SIGSTOP)  # Can pause any process

def resume_process(pid):
    os.kill(pid, signal.SIGCONT)  # Can resume any process

Why this is vulnerable:

• No check that the caller owns the process, or is allowed process control at all
• Can suspend system-critical processes, which stay holding their locks and file descriptors
• No logging or audit trail

Process Information Disclosure

import psutil

@app.route('/process-info')
def process_info():
    pid = int(request.args.get('pid'))

    # DANGEROUS: Exposes all process information
    proc = psutil.Process(pid)
    return {
        'name': proc.name(),
        'cmdline': proc.cmdline(),  # May contain secrets
        'environ': proc.environ(),  # Environment variables
        'cwd': proc.cwd(),
        'connections': [str(c) for c in proc.connections()]
    }

Why this is vulnerable:

• No authorization - anyone can view any process
• Command line may contain passwords or API keys
• Environment variables often contain secrets
• Network connections reveal internal architecture

Race Condition in Process Management

import logging
import os
import psutil

logger = logging.getLogger(__name__)

def manage_worker(action, worker_id):
    # DANGEROUS: Time-of-check-time-of-use vulnerability
    pid = get_worker_pid(worker_id)

    if pid and psutil.pid_exists(pid):
        # Race condition: PID could change or be reused here
        if action == 'kill':
            os.kill(pid, signal.SIGKILL)

Why this is vulnerable:

• PID can be reused between check and use
• No atomic operation
• Could kill wrong process
• No verification process is still the expected one

Insufficient Process Isolation

import subprocess

def start_user_job(username, script_path):
    # DANGEROUS: Insufficient isolation
    subprocess.Popen([
        'python3',
        script_path
    ], env={'USER': username})  # Only sets USER, inherits everything else

Why this is vulnerable:

• Inherits parent environment (PATH, secrets, etc.)
• No resource limits (CPU, memory, file descriptors)
• Runs with same user as parent process
• No sandboxing or containerization

Celery Task Control Without Authorization

from celery import current_app

@app.route('/admin/revoke-task', methods=['POST'])
def revoke_task():
    # DANGEROUS: No authorization
    task_id = request.form.get('task_id')

    current_app.control.revoke(task_id, terminate=True)
    return f"Task {task_id} revoked"

Why this is vulnerable:

• Any user can revoke any task
• No validation of task_id format
• No verification user owns the task
• Can disrupt critical background jobs

Secure Patterns

Process Termination with Authorization

import logging
import subprocess
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, Set

import psutil

@dataclass
class ManagedProcess:
    pid: int
    name: str
    owner: str
    started_at: datetime
    # psutil's create_time() for this PID, recorded at registration. PIDs are
    # reused; the (pid, create_time) pair is what actually identifies a process.
    create_time: float

logger = logging.getLogger(__name__)

class SecureProcessManager:
    def __init__(self, authorized_users: Set[str]):
        self.managed_processes: Dict[str, ManagedProcess] = {}
        # Passed in, not defaulted to an empty set. A manager constructed with
        # no authorized users rejects every request, including the legitimate
        # ones, and looks identical to a working one until someone tries it.
        self.authorized_users: Set[str] = set(authorized_users)

    def register_process(self, process_id: str, popen: subprocess.Popen, owner: str):
        """Register a process for management"""
        proc = psutil.Process(popen.pid)
        self.managed_processes[process_id] = ManagedProcess(
            pid=popen.pid,
            name=proc.name(),
            owner=owner,
            started_at=datetime.now(timezone.utc),
            create_time=proc.create_time(),
        )
        logger.info("Process registered: %s (PID: %d) by %s", process_id, popen.pid, owner)

    def _resolve(self, managed: ManagedProcess) -> psutil.Process:
        """Return the live process, or raise if this PID is no longer the same one."""
        proc = psutil.Process(managed.pid)   # raises NoSuchProcess if gone
        # Comparing the name is not enough: a reused PID belonging to another
        # 'python3' passes that check. create_time is unique per process.
        if proc.create_time() != managed.create_time:
            raise ProcessLookupError("PID has been reused - refusing to signal it")
        return proc

    def kill_process(self, process_id: str, current_user: str) -> bool:
        """Safely terminate a process with authorization"""
        # Authorization check
        if current_user not in self.authorized_users:
            logger.warning(f"Unauthorized kill attempt by {current_user}")
            raise PermissionError(f"User {current_user} not authorized for process control")

        # Validate process ID
        process = self.managed_processes.get(process_id)
        if not process:
            logger.warning(f"Attempted to kill unmanaged process: {process_id}")
            raise ValueError(f"Process {process_id} not under management")

        # Ownership verification
        if process.owner != current_user and current_user != 'admin':
            logger.warning(
                f"User {current_user} attempted to kill process owned by {process.owner}"
            )
            raise PermissionError("Cannot kill process owned by another user")

        try:
            # Resolve through psutil and signal the object, not the bare PID.
            # os.kill(pid, 0) followed by os.kill(pid, SIGTERM) is two syscalls
            # with a window between them - the same race this page's
            # "Race Condition in Process Management" pattern warns about.
            # proc.terminate() checks create_time internally before signalling.
            proc = self._resolve(process)
            proc.terminate()  # SIGTERM on Unix, TerminateProcess on Windows

            logger.info("SIGTERM sent to %s (PID: %d) by %s",
                        process_id, process.pid, current_user)

            try:
                proc.wait(timeout=10)
            except psutil.TimeoutExpired:
                logger.warning("Process %s ignored SIGTERM; sending SIGKILL", process_id)
                proc.kill()
                proc.wait(timeout=5)

            del self.managed_processes[process_id]
            return True

        except (psutil.NoSuchProcess, ProcessLookupError):
            logger.warning("Process %s (PID: %d) no longer exists", process_id, process.pid)
            del self.managed_processes[process_id]
            return False
        except psutil.AccessDenied as e:
            logger.error("Permission denied killing process %s: %s", process_id, e)
            raise

Why this works:

• The request names a process_id from the application's own records, never a system PID, so a process the application did not start cannot be reached at all.
• Authorization and ownership are both checked before anything is signalled.
• The authorized-user set is a constructor argument. Defaulting it to an empty set produces a manager that denies every request - the tests for "an unauthorized user is rejected" still pass, and nothing legitimate works.
• (pid, create_time) identifies the process rather than the PID alone, so a recycled PID belonging to something else is refused instead of terminated. Comparing proc.name() instead, as the priority and signal examples below do, is weaker: another python3 passes.
• proc.terminate() re-checks identity inside psutil, closing the window that os.kill(pid, 0) followed by os.kill(pid, SIGTERM) leaves open.
• SIGTERM first with a bounded wait, then SIGKILL, so a process that handles shutdown gets the chance and one that ignores it still goes away.

Safe Process Priority Management

import sys

import psutil


class ProcessPriorityManager:
    # Unix nice values. 0 is normal; higher means lower priority. Lowering the
    # value requires CAP_SYS_NICE, so the allowed range only ever de-prioritises.
    MIN_NICE = 0
    MAX_NICE = 19

    def set_process_priority(
        self,
        process_id: str,
        nice_value: int,
        current_user: str,
        process_manager: SecureProcessManager,
    ):
        """Lower a managed process's scheduling priority (Unix)."""
        if sys.platform == "win32":
            # psutil.Process.nice() takes a Windows priority CLASS here, not a
            # nice value - psutil.BELOW_NORMAL_PRIORITY_CLASS and friends.
            # Passing 0..19 raises OSError [WinError 87] on every value in the
            # range, so the two platforms need genuinely different code.
            raise NotImplementedError("Use set_process_priority_windows on Windows")

        if not (self.MIN_NICE <= nice_value <= self.MAX_NICE):
            raise ValueError(
                f"Nice value must be between {self.MIN_NICE} and {self.MAX_NICE}"
            )

        process = process_manager.managed_processes.get(process_id)
        if not process:
            raise ValueError(f"Process {process_id} not under management")

        if process.owner != current_user and current_user != "admin":
            raise PermissionError("Cannot modify process owned by another user")

        # Identity check, not just existence - see _resolve above
        proc = process_manager._resolve(process)
        proc.nice(nice_value)

        logger.info(
            "Process %s priority set to %d by %s", process_id, nice_value, current_user
        )

    def set_process_priority_windows(
        self,
        process_id: str,
        priority_class: int,
        current_user: str,
        process_manager: SecureProcessManager,
    ):
        """Lower a managed process's priority class (Windows)."""
        allowed = {
            psutil.IDLE_PRIORITY_CLASS,
            psutil.BELOW_NORMAL_PRIORITY_CLASS,
            psutil.NORMAL_PRIORITY_CLASS,
        }
        if priority_class not in allowed:
            raise ValueError("Priority class not permitted")

        process = process_manager.managed_processes.get(process_id)
        if not process:
            raise ValueError(f"Process {process_id} not under management")

        if process.owner != current_user and current_user != "admin":
            raise PermissionError("Cannot modify process owned by another user")

        process_manager._resolve(process).nice(priority_class)

Why this works:

• The allowed range only ever de-prioritises. On Unix, lowering a nice value below its current one needs CAP_SYS_NICE, so 0..19 cannot be used to starve other work; on Windows the allowlist stops at NORMAL_PRIORITY_CLASS for the same reason.
• The process is resolved through the manager's (pid, create_time) check, so a recycled PID is not re-prioritised by mistake.
• Ownership is verified against the application's own record, not against the OS user - see the note on check_process_ownership below for why those are not the same thing.

nice() is not the cross-platform call it looks like. psutil presents one method name on both platforms and the argument means different things: a Unix nice value in -20..19, or a Windows priority class constant (NORMAL_PRIORITY_CLASS is 32, IDLE_PRIORITY_CLASS is 64). Verified on psutil 7.2 / Windows: proc.nice(0) and proc.nice(19) both raise OSError: [WinError 87] The parameter is incorrect, so a single "cross-platform" method over the 0..19 range fails on every value it accepts. Where a library offers one signature over two operating-system concepts, run it on both before describing it as portable.

Restricted Process Spawning with Resource Limits

import logging
import os
import resource
import subprocess
import sys
from typing import Dict, List

logger = logging.getLogger(__name__)

class SecureProcessSpawner:
    # Allowlist of permitted executables
    ALLOWED_EXECUTABLES = {
        'worker': '/opt/app/worker.py',
        'processor': '/opt/app/processor.py',
        'analyzer': '/opt/app/analyzer.py'
    }

    # Maximum concurrent processes per user
    MAX_PROCESSES_PER_USER = 5

    # Not /tmp: it is world-writable, so anything the job leaves there is
    # readable and replaceable by any local user (CWE-377).
    WORK_DIR = '/var/lib/app/work'

    def __init__(self):
        self.user_process_count: Dict[str, int] = {}

    def spawn_process(
        self,
        job_type: str,
        args: List[str],
        current_user: str
    ) -> subprocess.Popen:
        """Spawn process with security controls"""
        # Validate job type against allowlist
        if job_type not in self.ALLOWED_EXECUTABLES:
            logger.warning(f"Unauthorized job type requested: {job_type}")
            raise ValueError(f"Job type '{job_type}' not permitted")

        executable = self.ALLOWED_EXECUTABLES[job_type]

        # Rate limiting per user
        user_count = self.user_process_count.get(current_user, 0)
        if user_count >= self.MAX_PROCESSES_PER_USER:
            raise RuntimeError(
                f"User {current_user} has reached process limit of {self.MAX_PROCESSES_PER_USER}"
            )

        # Validate arguments (no path traversal, no shell metacharacters)
        validated_args = self._validate_arguments(args)

        # Prepare secure environment
        secure_env = {
            'PATH': '/usr/bin:/bin',
            'USER': current_user,
            'HOME': f'/home/{current_user}',
            'PYTHONDONTWRITEBYTECODE': '1',
            'PYTHONUNBUFFERED': '1'
        }

        # Create process with resource limits
        def set_limits():
            # Limit CPU time to 1 hour
            resource.setrlimit(resource.RLIMIT_CPU, (3600, 3600))
            # Limit memory to 1GB
            resource.setrlimit(resource.RLIMIT_AS, (1024*1024*1024, 1024*1024*1024))
            # Limit number of file descriptors
            resource.setrlimit(resource.RLIMIT_NOFILE, (1024, 1024))
            # Prevent core dumps
            resource.setrlimit(resource.RLIMIT_CORE, (0, 0))

        try:
            process = subprocess.Popen(
                [sys.executable, executable, '--'] + validated_args,
                env=secure_env,
                # preexec_fn runs between fork() and exec() in the child. The
                # subprocess docs call it unsafe in the presence of threads, and
                # a Flask or Django worker is threaded - the child can deadlock
                # on a lock another thread held at fork time. Only the rlimits
                # need to be set there, and only on POSIX.
                preexec_fn=set_limits if os.name == 'posix' else None,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,  # One pipe, so one reader suffices
                stdin=subprocess.DEVNULL,  # No input
                cwd=self.WORK_DIR,
                start_new_session=True  # Child is not in our process group
            )

            self.user_process_count[current_user] = user_count + 1

            logger.info(
                "Process spawned: %s by %s (PID: %d)", job_type, current_user, process.pid
            )

            return process

        except Exception:
            logger.exception("Failed to spawn %s for %s", job_type, current_user)
            raise

    def wait_for(self, process: subprocess.Popen, current_user: str,
                 timeout: int = 300) -> str:
        """Collect a spawned process's output and release its rate-limit slot."""
        try:
            # communicate() reads the pipe while waiting. Calling wait() on a
            # process whose stdout is a PIPE deadlocks as soon as the child
            # fills the OS buffer: it blocks writing, the parent blocks waiting.
            output, _ = process.communicate(timeout=timeout)
        except subprocess.TimeoutExpired:
            process.kill()
            output, _ = process.communicate()
            raise
        finally:
            # Release the slot here, in finally. A counter that is only
            # decremented on the success path leaks a slot per failure, and
            # after MAX_PROCESSES_PER_USER failures the user is locked out for
            # the lifetime of the worker.
            self._release(current_user)

        return output.decode('utf-8', errors='replace')

    def _release(self, current_user: str):
        count = self.user_process_count.get(current_user, 0)
        # max(0, ...) - an unmatched release must not push the count negative,
        # which would hand the user extra slots rather than fewer.
        self.user_process_count[current_user] = max(0, count - 1)

    def _validate_arguments(self, args: List[str]) -> List[str]:
        """Validate process arguments"""
        validated = []

        for arg in args:
            # Reject path traversal attempts
            if '..' in arg or arg.startswith('/'):
                raise ValueError(f"Invalid argument: {arg}")

            # Reject shell metacharacters
            dangerous_chars = set(';&|`$(){}[]<>*?~')
            if any(c in arg for c in dangerous_chars):
                raise ValueError(f"Argument contains dangerous characters: {arg}")

            # Limit argument length
            if len(arg) > 255:
                raise ValueError(f"Argument too long: {len(arg)} chars")

            validated.append(arg)

        return validated

Why this works:

• The job_type key maps to an absolute script path, so the request never supplies one.
• Arguments are validated, and -- stops the script's own argument parser reading an attacker-chosen value as an option.
• sys.executable is the interpreter already running, so the child does not depend on which python3 the (replaced) PATH resolves to.
• Rate limiting per user prevents resource exhaustion, and the slot is released in a finally so a failed or timed-out job does not consume one permanently.
• resource.setrlimit bounds CPU, address space and file descriptors in the child, which is enforcement the parent cannot be talked out of.
• The environment is replaced rather than inherited, so nothing from the request path or the parent's secrets reaches the child.
• start_new_session=True puts the child in its own process group, so a signal sent to the web process's group does not take the job with it - and killing the job's group does not reach back.
• communicate() reads the pipe while waiting, which is what stops a chatty child blocking on a full buffer forever.

Safe Signal Handling

import logging
import os
import signal
import psutil

logger = logging.getLogger(__name__)

class SecureSignalManager:
    # Only allow safe signals
    ALLOWED_SIGNALS = {
        'TERM': signal.SIGTERM,  # Graceful termination
        'HUP': signal.SIGHUP,    # Reload configuration
        'USR1': signal.SIGUSR1,  # User-defined
        'USR2': signal.SIGUSR2   # User-defined
    }

    def send_signal(
        self,
        process_id: str,
        signal_name: str,
        current_user: str,
        process_manager: SecureProcessManager
    ):
        """Send signal to process with authorization"""
        # Validate signal
        if signal_name not in self.ALLOWED_SIGNALS:
            raise ValueError(f"Signal '{signal_name}' not permitted")

        sig = self.ALLOWED_SIGNALS[signal_name]

        # Get managed process
        process = process_manager.managed_processes.get(process_id)
        if not process:
            raise ValueError(f"Process {process_id} not under management")

        # Authorization check
        if process.owner != current_user and current_user != 'admin':
            logger.warning(
                f"User {current_user} attempted to signal process owned by {process.owner}"
            )
            raise PermissionError("Cannot signal process owned by another user")

        try:
            # Identity check via (pid, create_time). Comparing proc.name() alone
            # accepts any other process with the same executable name that
            # happens to have inherited the PID.
            proc = process_manager._resolve(process)
            proc.send_signal(sig)

            logger.info(
                f"Signal {signal_name} sent to process {process_id} "
                f"(PID: {process.pid}) by {current_user}"
            )

        except psutil.NoSuchProcess:
            raise ProcessLookupError(f"Process {process_id} no longer exists")

Why this works:

• The signal allowlist excludes SIGKILL and SIGSTOP, which cannot be handled or blocked - a process that is stopped rather than terminated stays holding its locks and file descriptors, and nothing in it can notice.
• Authorization and ownership are checked against the application's record before anything is sent.
• _resolve compares (pid, create_time), so a recycled PID is refused rather than signalled. proc.name() on its own does not distinguish two processes running the same executable.
• Both the allowed and the refused paths are logged with the requesting user.

Secure Process Information Disclosure

import psutil
from typing import Dict, Any

class SecureProcessInfo:
    def get_process_info(
        self,
        process_id: str,
        current_user: str,
        process_manager: SecureProcessManager
    ) -> Dict[str, Any]:
        """Get process information with authorization"""
        # Get managed process
        process = process_manager.managed_processes.get(process_id)
        if not process:
            raise ValueError(f"Process {process_id} not under management")

        # Authorization - only owner or admin
        if process.owner != current_user and current_user != 'admin':
            raise PermissionError("Cannot view process owned by another user")

        try:
            proc = psutil.Process(process.pid)

            # Return only safe, sanitized information
            return {
                'id': process_id,
                'pid': process.pid,
                'name': process.name,
                'status': proc.status(),
                'cpu_percent': proc.cpu_percent(interval=0.1),
                'memory_mb': proc.memory_info().rss / 1024 / 1024,
                'started_at': process.started_at.isoformat(),
                'owner': process.owner
                # Do NOT include: cmdline, environ, connections, open files
            }

        except psutil.NoSuchProcess:
            raise ProcessLookupError(f"Process {process_id} no longer exists")

Why this works:

• Only the owner or an admin can read a process, and only one the application is managing
• Returns status, CPU and memory use, start time and owner
• Excludes command line arguments, which may contain passwords or API keys
• Excludes environment variables, which often contain secrets
• Excludes network connections and open file handles, which reveal internal architecture

Celery Task Management with Authorization

import logging
from celery import current_app
from flask import request, g
import re

logger = logging.getLogger(__name__)

class SecureCeleryManager:
    # Valid task ID format (UUID)
    TASK_ID_PATTERN = re.compile(
        r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
    )

    def __init__(self):
        self.task_owners = {}  # task_id -> username mapping

    def register_task(self, task_id: str, owner: str):
        """Register task ownership"""
        self.task_owners[task_id] = owner

    def revoke_task(self, task_id: str, current_user: str, terminate: bool = False):
        """Revoke Celery task with authorization"""
        # Validate task ID format
        if not self.TASK_ID_PATTERN.match(task_id):
            raise ValueError(f"Invalid task ID format: {task_id}")

        # Check ownership
        owner = self.task_owners.get(task_id)
        if not owner:
            raise ValueError(f"Task {task_id} not found or not owned by you")

        if owner != current_user and current_user != 'admin':
            logger.warning(
                f"User {current_user} attempted to revoke task owned by {owner}"
            )
            raise PermissionError("Cannot revoke task owned by another user")

        # Revoke with appropriate signal
        current_app.control.revoke(
            task_id,
            terminate=terminate,  # Only if explicitly requested
            signal='SIGTERM' if terminate else None  # Graceful termination
        )

        logger.info(
            f"Task {task_id} revoked by {current_user} "
            f"(terminate={terminate})"
        )

        del self.task_owners[task_id]

Why this works:

• A task id that is not a UUID is rejected before it reaches the broker
• register_task records who owns a task, and the owner is checked before the revoke, so one user cannot cancel another's job
• terminate defaults to False, and when it is requested the signal is SIGTERM, so a running task gets the chance to shut down
• Both the revoke and a refused attempt are logged with the requesting user

Docker Container Process Control

import docker
import logging
import os
from typing import Dict

logger = logging.getLogger(__name__)

class SecureContainerManager:
    def __init__(self):
        self.client = docker.from_env()
        self.managed_containers: Dict[str, str] = {}  # container_id -> owner

    def start_container(
        self,
        image: str,
        command: str,
        current_user: str
    ) -> str:
        """Start container with security controls"""
        # Image allowlist
        ALLOWED_IMAGES = {
            'worker': 'myapp/worker:latest',
            'analyzer': 'myapp/analyzer:latest'
        }

        if image not in ALLOWED_IMAGES:
            raise ValueError(f"Image '{image}' not permitted")

        # Start container with security options
        container = self.client.containers.run(
            ALLOWED_IMAGES[image],
            command,
            detach=True,
            # Security options
            read_only=True,  # Read-only root filesystem
            tmpfs={'/tmp': ''},  # ...so give it somewhere to write
            mem_limit='512m',  # Memory limit
            cpu_quota=50000,  # CPU limit (50% of one core against the 100000 default period)
            pids_limit=100,  # Limit number of processes
            network_mode='none',  # No network. 'bridge' is Docker's DEFAULT, not isolation
            cap_drop=['ALL'],  # Drop all capabilities
            security_opt=['no-new-privileges'],  # Prevent privilege escalation
            # userns_mode is deliberately NOT set to 'host': that value OPTS OUT of
            # user-namespace remapping, which is the opposite of what the name suggests.
            # Leave it unset so the daemon's userns-remap setting applies.
            user='65534:65534'  # nobody:nogroup inside the container
        )

        self.managed_containers[container.id] = current_user

        logger.info(
            f"Container {container.id} started by {current_user} "
            f"(image: {image})"
        )

        return container.id

    def stop_container(self, container_id: str, current_user: str):
        """Stop container with authorization"""
        # Check ownership
        owner = self.managed_containers.get(container_id)
        if not owner:
            raise ValueError(f"Container {container_id} not found")

        if owner != current_user and current_user != 'admin':
            raise PermissionError("Cannot stop container owned by another user")

        container = self.client.containers.get(container_id)
        container.stop(timeout=10)  # Graceful stop with timeout

        logger.info(f"Container {container_id} stopped by {current_user}")

        del self.managed_containers[container_id]

Why this works:

• Image allowlist prevents arbitrary container execution: the request supplies a key, and the tag it maps to is fixed in the application.
• Resource limits (memory, CPU quota, PID count) bound what one container can consume.
• Read-only root filesystem with an explicit tmpfs for the paths that must be writable.
• All capabilities dropped and no-new-privileges set, so a compromised workload cannot regain privileges through a setuid binary.
• Runs as UID 65534 rather than root.
• network_mode='none' means no network at all.

Two of those settings are easy to get backwards.

network_mode='bridge' is Docker's default, not isolation: the container reaches the internet and anything routable from the host. 'none' is the isolating value. Where the workload does need to talk to one service, put both on a user-defined network rather than leaving it on bridge.

userns_mode='host' reads as "enable user namespaces" and does the opposite - it opts this container out of the daemon's userns-remap, so root inside the container is root on the host. Leaving the parameter unset is what lets the daemon's remapping apply. This is a recurring shape: check what a value means rather than what the parameter name implies, and be most suspicious where the name is the argument for the line.

Key Security Functions

Process Validator

import re
from typing import Optional

class ProcessValidator:
    """Validate process control parameters"""

    # Valid process ID pattern (internal identifier)
    PROCESS_ID_PATTERN = re.compile(r'^[a-z0-9_-]{1,64}$')

    @staticmethod
    def validate_process_id(process_id: str) -> bool:
        """Validate process identifier format"""
        if not isinstance(process_id, str):
            raise TypeError("Process ID must be a string")

        if not ProcessValidator.PROCESS_ID_PATTERN.match(process_id):
            raise ValueError(
                f"Invalid process ID format: {process_id}. "
                "Must be alphanumeric, dash, or underscore (1-64 chars)"
            )

        return True

    @staticmethod
    def validate_pid(pid: int) -> bool:
        """Validate system PID"""
        if not isinstance(pid, int):
            raise TypeError("PID must be an integer")

        if pid <= 0:
            raise ValueError(f"Invalid PID: {pid}")

        # Deliberately no "pid < 100 is a system process" rule. That is a blocklist
        # over a numbering scheme the target does not have to sit in: inside a
        # container the workload is PID 1, and on a busy host the database is
        # somewhere in the tens of thousands. The control that works is looking the
        # PID up in the application's own record of processes it started.
        return True

    @staticmethod
    def validate_signal_name(signal_name: str) -> bool:
        """Validate signal name"""
        ALLOWED_SIGNALS = {'TERM', 'HUP', 'USR1', 'USR2'}

        if signal_name not in ALLOWED_SIGNALS:
            raise ValueError(
                f"Signal '{signal_name}' not allowed. "
                f"Permitted: {', '.join(ALLOWED_SIGNALS)}"
            )

        return True

Process Ownership Checker

import logging
import os
import psutil

logger = logging.getLogger(__name__)

def check_process_ownership(pid: int, expected_os_user: str) -> bool:
    """Verify a process runs as the expected OPERATING SYSTEM account.

    This is not an application-level authorization check and must not be used as
    one. Every process a web application spawns runs as the same OS account - the
    one the application server runs as - so passing the logged-in username here
    returns False for everything, and passing the service account returns True for
    every process the application started, whichever user requested it.

    Its actual use is a sanity check before signalling: confirm the target runs as
    the service account rather than as root or another service. Application-level
    ownership comes from SecureProcessManager.managed_processes, which records who
    asked for each process.
    """
    try:
        proc = psutil.Process(pid)
        process_user = proc.username()

        if process_user != expected_os_user:
            logger.warning(
                "Process %d runs as %s, expected %s", pid, process_user, expected_os_user
            )
            return False

        return True

    except psutil.NoSuchProcess:
        return False
    except psutil.AccessDenied:
        logger.warning(f"Access denied checking ownership of PID {pid}")
        return False

Considerations

A PID is a number, not a capability. os.kill() will signal anything the interpreter's user owns, and PIDs are reused as the table wraps - on Linux, the kernel's usual compiled-in default for /proc/sys/kernel/pid_max is 32768, though CONFIG_BASE_SMALL lowers it and boot initialization can raise it with CPU count. On 64-bit systems using systemd's upstream defaults, systemd 243 introduced a pid_max setting of 4194304 (50-pid-max.conf), which distributions may override. Either way, a host can recycle PIDs; the interval depends on workload. Every pattern here routes the request through the application's own record of processes it started, and that indirection is the control. Validating the number is shape checking, and a "system PIDs are below 100" rule is a blocklist over a namespace the target need not sit in. The finding applies when a caller-supplied PID can target a process without verifying the caller's authorization for that specific process.

Application ownership and OS ownership are different questions. proc.username() tells you which account a process runs as, which for anything a web application spawned is the service account - the same answer for every user's job. Whether this requester may control that job is a fact the application recorded when it started it, and the only place it exists. Reaching for psutil to answer an authorization question is a sign the record was never kept.

preexec_fn is not safe under a threaded server. It runs in the child between fork() and exec(), where only async-signal-safe calls are legal, and a Flask, Django or Celery process is threaded - if another thread held a lock at fork time, the child can hang holding a copy of it, with no error anywhere. resource.setrlimit is the case that genuinely needs it. Where the limits can be applied another way - a systemd slice, a cgroup, container --memory and --pids-limit, or ulimit in a wrapper script - prefer that, because the limit then survives someone editing the Python.

Where the rate-limit counter lives. self.user_process_count is per-worker. Under Gunicorn with four workers, MAX_PROCESSES_PER_USER = 5 is really twenty, and after a worker restart it is whatever the new worker thinks. That is usually acceptable for a courtesy limit and never acceptable as a control. If the limit is load-bearing, keep it in Redis with the release in a finally, or let a job queue own concurrency and drop the counter entirely.

Is a spawn the right shape at all? Timeouts, output caps and per-user counters are what you write when a request handler owns a child process, and each of them is bounding something that should not be on the request path. Celery, RQ or Dramatiq give you the concurrency limit, the retry, the timeout and the audit trail, and the request returns a job id. Keep the controls on this page for the cases where the work genuinely has to complete before the response does.

When resource limits are worth setting individually. RLIMIT_AS bounds address space, not resident memory, so a process that maps a large file or uses a garbage-collected runtime can fail on an allocation while using very little RAM - cgroup memory limits measure what you actually care about. RLIMIT_CPU counts CPU seconds, not wall-clock, so a job that mostly waits on IO is not bounded by it and needs the communicate(timeout=...) as well. Set both, and know which one will fire.

Testing

A Bandit or Semgrep rule for shell=True or os.kill reports on the shape of the call, not on whether the authorization around it functions. Every defect of this kind that has actually shipped passed a re-scan: an authorized-user set that was empty, a rate-limit counter that was never decremented, a nice() call that raises on every value it accepts, and a container flag that means the opposite of its name.

• kill_process("job-1", "alice") succeeds for a process alice registered. A manager built with an empty authorized_users set passes every "unauthorized user is rejected" test and fails this one, which is why this assertion has to exist.
• kill_process("job-1", "bob") raises PermissionError, and kill_process("unknown", "admin") raises ValueError. Assert both, so a method that raises unconditionally is distinguishable from one that works.
• Register a process, mutate the stored create_time, and assert _resolve raises ProcessLookupError. This is the PID-reuse case; comparing proc.name() instead passes it, because two python3 processes share a name.
• Spawn a job that prints 200 KB and assert wait_for returns the output. Popen.wait() with stdout=PIPE and no reader blocks at the OS pipe buffer - verified: the same call returns immediately for a one-line child and does not return at all for this one, so a test with a quiet child never reaches the bug.
• Exhaust MAX_PROCESSES_PER_USER with jobs that fail, then assert the same user can start another. A slot released only on the success path leaves the user locked out for the life of the worker.
• On Windows, assert set_process_priority raises NotImplementedError rather than OSError [WinError 87]. psutil's nice() takes a priority class there, so every value in the documented 0..19 range fails.
• docker inspect a started container and assert .HostConfig.NetworkMode == "none" and that UsernsMode is empty. userns_mode='host' opts the container out of user-namespace remapping - the setting reads as enabling it and does the reverse.

Additional Resources

• Celery Security
• CWE-114 Details
• Docker Security
• Linux Capabilities
• OWASP Access Control
• psutil Documentation
• Python resource Module
• Python subprocess Security