scripts/subprocess_utils.py - main - ump-team/ump-infra

scripts/subprocess_utils.py
158 строк8 KB
Dmitry Kochenov
v0.0.5: рефакторинг валидации фаз и улучшение PR-логики
19 июл 2026, 23:40
19 июл 2026, 23:40a686387
"""subprocess_utils — безопасные обёртки над subprocess.run.

P0-fix (бывший P0-7 аудита): заменяет `subprocess.run(cmd, shell=True, ...)`
на list-form, чтобы исключить shell-инъекции через интерполируемые переменные
(branch name из YAML, file path из plans, и т.д.).

До фикса: `subprocess.run(f'git push origin {step_branch}', shell=True, ...)`
  → если step_branch = 'main; rm -rf /', выполнится `git push origin main`
     и затем `rm -rf /`.

После фикса: `run_list(['git', 'push', 'origin', step_branch], ...)`
  → step_branch передаётся как один argv-элемент, shell-метасимволы
     игнорируются git'ом как часть имени ветки.

API:
  - run_list(args, check=True, cwd=None, timeout=120) — основной безопасный
    способ запуска. args — list[str], без shell.
  - run_str(cmd, check=True, cwd=None, timeout=120) — парсит cmd через
    shlex.split() и вызывает run_list. Используется для механической замены
    старой функции run(). ВАЖНО: shlex.split НЕ выполняет shell-подстановки
    ($VAR, `cmd`, *), НЕ поддерживает конвейеры (|), перенаправления (> <),
    логические операторы (&& || ;). Если в cmd есть такие конструкции —
    нужно явно переписать на Python-логику.
  - shell_forbidden(cmd: str) → bool — True, если cmd содержит shell-метасимволы,
    которые shlex.split не обработает корректно.
"""
from __future__ import annotations

import shlex
import subprocess
import sys
from pathlib import Path

# Список shell-конструкций, которые shlex.split НЕ обработает корректно.
# Если в cmd есть любая из них — caller должен переписать на list-form
# или на явную Python-логику.
_SHELL_METACHARS = ('&&', '||', ';', '|', '>', '<', '$(', '`', '*?', '?*', )

# Glob-паттерны в shell (например, '*.yaml'). Если встречаются — нужно
# использовать Python glob.glob() вместо shell.
_GLOB_CHARS = ('*', '?', '[', ']', '{', '}',)


def shell_forbidden(cmd: str) -> list[str]:
    """Вернуть список shell-конструкций, найденных в cmd.

    Пустой список = cmd можно безопасно парсить через shlex.split().
    Непустой список = caller должен переписать на list-form или Python-логику.

    >>> shell_forbidden('git push origin main')
    []
    >>> shell_forbidden('git stash 2>/dev/null || true')
    ['>', '||']
    """
    found: list[str] = []
    for mc in _SHELL_METACHARS:
        if mc in cmd:
            found.append(mc)
    # Глоббинг — отдельный случай: только если wildcard-символ встречается
    # в аргументе, а не в начале команды (например, '*.yaml' — glob, но
    # 'a*b' как строковый аргумент — нет).
    # Эвристика: если '*' или '?' идут после пробела в середине аргумента.
    # Для простоты считаем любой '*' или '?' признаком glob.
    if '*' in cmd or '?' in cmd:
        found.append('*|?')
    return found


def run_list(
    args: list[str],
    check: bool = True,
    cwd: Path | None = None,
    timeout: int = 120,
    capture: bool = True,
) -> subprocess.CompletedProcess[str]:
    """Безопасный запуск subprocess в list-form.

    Args:
        args: список аргументов, args[0] — исполняемый файл, остальное — argv.
            Никакого shell=True, никаких интерполяций.
        check: если True — печатает diagnose в stderr при ненулевом exit code,
            но НЕ поднимает исключение (для совместимости со старой семантикой).
        cwd: рабочая директория (None = наследуется от родителя).
        timeout: таймаут в секундах.
        capture: если True — захватывает stdout/stderr в CompletedProcess.

    Returns:
        subprocess.CompletedProcess с stdout, stderr, returncode.

    Raises:
        subprocess.TimeoutExpired — если команда превысила timeout.
    """
    # P2-4 fix (mypy): передаём явные kwargs вместо dict unpacking —
    # это позволяет mypy проверить типы. dict unpacking теряет информацию.
    result = subprocess.run(
        args,
        text=True,
        cwd=cwd,
        timeout=timeout,
        capture_output=capture,
    )
    if check and result.returncode != 0:
        print(f'ERROR: command failed: {" ".join(args)}', file=sys.stderr)
        if capture:
            print(f'stdout: {(result.stdout or "")[:500]}', file=sys.stderr)
            print(f'stderr: {(result.stderr or "")[:500]}', file=sys.stderr)
    return result


class ShellInjectionError(ValueError):
    """P0-security: caller передал команду с shell-метасимволами.

    Ранее (до v0.0.2-audit) run_str fallback'ал на subprocess.run(shell=True)
    в таком случае — это позволяло shell-инъекции через интерполируемые
    переменные. Теперь — fail-loud: caller обязан переписать вызов на
    run_list([...]) или на явную Python-логику (glob.glob, subprocess.run
    с list-form, и т.д.).
    """


def run_str(
    cmd: str,
    check: bool = True,
    cwd: Path | None = None,
    timeout: int = 120,
    capture: bool = True,
) -> subprocess.CompletedProcess[str]:
    """Запустить cmd, распарсенную через shlex.split().

    БЕЗ shell=True. Это безопасно от инъекций: интерполируемые переменные
    воспринимаются как один argv-элемент.

    ⚠️ ВАЖНО: shlex.split НЕ поддерживает shell-конструкции:
      - перенаправления: 2>/dev/null, >file, <input
      - конвейеры: cmd1 | cmd2
      - логика: &&, ||, ;
      - подстановки: $(cmd), `cmd`, $VAR
      - glob: *.yaml, ?.txt

    Если в cmd есть такие конструкции — поднимается ShellInjectionError.
    Caller должен переписать вызов:
      - `git stash 2>/dev/null || true` → `run_list(['git', 'stash'], check=False)`
      - `cat *.yaml` → `for f in glob.glob('*.yaml'): ...`
      - `cmd1 | cmd2` → два run_list() + Python-обработка stdout первого

    P0-security-fix (audit v0.0.2): ранее fallback на shell=True печатал WARNING
    и выполнял команду — это High-severity finding (B602 bandit). Теперь —
    fail-loud через ShellInjectionError, никаких shell=True.
    """
    forbidden = shell_forbidden(cmd)
    if forbidden:
        raise ShellInjectionError(
            f'Command contains shell metachars {forbidden!r}: {cmd!r}. '
            f'Rewrite to run_list([...]) or Python logic (glob, multiple run_list calls). '
            f'shell=True fallback removed in v0.0.2-audit for security (B602).'
        )
    args = shlex.split(cmd)
    return run_list(args, check=check, cwd=cwd, timeout=timeout, capture=capture)