Управление процессами и файловой системой в Python для DevOps

Глубокое погружение в автоматизацию системного администрирования Linux. Курс научит эффективно управлять файлами через pathlib, запускать внешние утилиты с помощью subprocess и создавать надежные скрипты для бэкапов и обслуживания системы.

Современная работа с путями: переход от os.path к объектно-ориентированному pathlib

config_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'conf', 'nginx.conf') Эта конструкция до боли знакома любому, кто писал скрипты автоматизации. Попытка подняться на две директории вверх и зайти в папку с конфигурацией превращается в нечитаемую матрешку из вызовов функций. Проблема кроется в самом подходе: модуль os.path работает с путями как с обычными строками. Для него /var/log/syslog — это просто набор символов, который можно резать по слэшам. Но в реальности файловый путь — это сущность со своими правилами, иерархией, расширениями и привязкой к жесткому диску.

Переход к модулю pathlib, добавленному в стандартную библиотеку Python, меняет парадигму. Вместо манипуляций со строками мы начинаем работать с объектами.

Объектно-ориентированный подход к путям

Базовым элементом pathlib является класс Path. Когда мы передаем ему строку, он конструирует объект, который «понимает» структуру файловой системы.

from pathlib import Path

# Создание объекта пути
log_dir = Path("/var/log/nginx")

В Linux Path автоматически создает экземпляр класса PosixPath, который учитывает особенности UNIX-систем (например, прямой слэш / как разделитель). Главное визуальное отличие pathlib от старого подхода — использование оператора деления / для конструирования путей.

В Python существует механизм перегрузки операторов, позволяющий менять поведение стандартных математических символов для пользовательских классов. Разработчики pathlib перегрузили оператор /, чтобы он выполнял безопасное склеивание путей, заменяя громоздкий os.path.join.

# Старый стиль (os.path)
# access_log = os.path.join("/var/log/nginx", "access.log")

# Новый стиль (pathlib)
base_dir = Path("/var/log/nginx")
access_log = base_dir / "access.log"
error_log = base_dir / "error" / "error.log"

Оператор / работает, если хотя бы один из операндов (обычно левый) является объектом Path. Это делает код декларативным: мы визуально видим структуру директорий прямо в коде.

Анатомия пути: извлечение компонентов

В скриптах резервного копирования или ротации логов постоянно требуется отделять имя файла от расширения или получать путь к родительской директории. os.path предлагал для этого функции basename(), dirname() и splitext(). У объекта Path вся эта информация уже вычислена и доступна через свойства (атрибуты).

Рассмотрим путь к архиву логов: /var/log/nginx/access.log.gz.

  • access_log.name вернет 'access.log.gz' (полное имя файла с расширениями).
  • access_log.parent вернет объект Path('/var/log/nginx').
  • access_log.suffix вернет '.gz' (последнее расширение).
  • access_log.suffixes вернет список ['.log', '.gz'] (удобно для архивов).
  • access_log.stem вернет 'access.log' (имя без последнего расширения).

Свойство .parent возвращает не строку, а новый объект Path. Это означает, что к нему можно применять те же методы или оператор /. Если нужно подняться на несколько уровней вверх (как в примере из начала статьи), используется кортеж .parents.

script_path = Path("/opt/myapp/bin/start.py")

# Поднимаемся на один уровень (в /opt/myapp/bin)
bin_dir = script_path.parent

# Поднимаемся на два уровня (в /opt/myapp)
app_dir = script_path.parents[1]

# Формируем путь к конфигу
config_path = script_path.parents[1] / "conf" / "app.ini"

Индекс 0 в .parents соответствует прямому родителю (эквивалент .parent), индекс 1 — «дедушке» и так далее. Это полностью устраняет необходимость во вложенных вызовах os.path.dirname.

Модификация путей на лету

Частая задача DevOps-инженера — создать резервную копию файла перед его изменением, добавив суффикс .bak или изменив расширение. Строковые манипуляции здесь чреваты ошибками (например, replace('.conf', '.bak') может случайно заменить часть имени директории, если она тоже называется .conf).

Объект Path предоставляет безопасные методы для замены частей пути, которые возвращают новый объект Path (сами объекты путей неизменяемы):

  • .with_name(name) — заменяет имя файла, оставляя директорию прежней.
  • .with_suffix(suffix) — заменяет или добавляет расширение.
original_config = Path("/etc/ssh/sshd_config")

# Создаем путь для бэкапа: /etc/ssh/sshd_config.bak
backup_config = original_config.with_suffix(".bak")

# Создаем путь для временного файла в той же директории: /etc/ssh/tmp_config
temp_config = original_config.with_name("tmp_config")

Разрешение путей и символические ссылки

В Linux-системах символические ссылки (symlinks) используются повсеместно. Например, /var/run часто является ссылкой на /run, а /etc/alternatives/java может вести через цепочку ссылок к конкретной версии JRE в /usr/lib/jvm/.

Когда автоматика работает с путями, важно понимать, где физически находится файл. Метод .resolve() выполняет полную нормализацию пути. Он делает три вещи одновременно:

  1. Преобразует относительный путь в абсолютный (от корня /).
  2. Устраняет переходы . (текущая директория) и .. (родительская директория).
  3. Проходит по всем символическим ссылкам до конечного целевого файла.
# Допустим, мы находимся в /home/user
# Путь содержит переходы и симлинки
complex_path = Path("../../var/run/docker.sock")

# Получаем реальный абсолютный путь
real_path = complex_path.resolve()
# Результат: PosixPath('/run/docker.sock')

По умолчанию в Python 3.6+ метод .resolve() не вызывает ошибку, если конечного файла не существует (он просто нормализует строку и разрешает те симлинки, которые может найти). Если требуется жесткая проверка, используется аргумент strict=True — в этом случае, если путь ведет в никуда, будет выброшено исключение FileNotFoundError.

Быстрое чтение и запись файлов

В предыдущем материале разбирался классический паттерн работы с файлами через контекстный менеджер with open(...). Он незаменим для ленивого чтения больших логов, когда данные подгружаются в память построчно. Однако в DevOps-задачах часто нужно просто прочитать небольшой конфигурационный файл целиком, изменить пару значений и записать обратно.

Для таких сценариев pathlib предлагает методы-шорткаты, которые инкапсулируют открытие, чтение/запись и безопасное закрытие файлового дескриптора:

  • .read_text(encoding="utf-8") — возвращает содержимое файла в виде строки.
  • .write_text(data, encoding="utf-8") — записывает строку в файл (перезаписывая его).
  • .read_bytes() и .write_bytes(data) — для работы с бинарными данными.
ssh_config = Path("/etc/ssh/sshd_config")

# Читаем весь файл в строку (файл автоматически закроется)
content = ssh_config.read_text(encoding="utf-8")

# Проверяем наличие настройки
if "PermitRootLogin yes" in content:
    # Заменяем строку
    safe_content = content.replace("PermitRootLogin yes", "PermitRootLogin no")

    # Записываем обратно
    ssh_config.write_text(safe_content, encoding="utf-8")

Этот подход сокращает код на несколько строк и делает его более линейным. Важно помнить об ограничении: .read_text() загружает весь файл в оперативную память. Использовать его для разбора файла /var/log/syslog размером в 5 гигабайт — фатальная ошибка, которая приведет к исчерпанию RAM (OOM Killer убьет процесс). Для больших файлов объект Path поддерживает метод .open(), который работает точно так же, как встроенная функция open(), возвращая итератор для ленивого чтения.

Проверки файловой системы

Методы проверок из os.path (exists, isfile, isdir) в pathlib реализованы как методы самого объекта:

target = Path("/etc/nginx/sites-available/default")

if not target.exists():
    print("Файл или директория не существует")

if target.is_symlink():
    print(f"Это ссылка, ведущая на {target.resolve()}")
elif target.is_file():
    print("Это обычный файл")
elif target.is_dir():
    print("Это директория")

Здесь проявляется еще одно архитектурное преимущество: автодополнение в современных IDE. Когда вы пишете os.path., редактор предлагает десятки функций, из которых нужно выбрать подходящую. Когда вы ставите точку после объекта Path (target.), IDE показывает только те методы, которые применимы к конкретному пути.

Разделение логики: PurePath против Path

В pathlib заложен важный архитектурный принцип: разделение вычислительных операций (манипуляции со строками) и операций ввода-вывода (обращение к диску).

Класс Path наследуется от класса PurePath. Все свойства (.name, .parent, .suffix) и методы конструирования (/, .with_name()) реализованы на уровне PurePath. Они работают исключительно в оперативной памяти и никогда не обращаются к ядру ОС. Вы можете создать объект PurePath('/root/secret') от имени обычного пользователя, извлекать из него суффиксы и менять имена — ошибок прав доступа не возникнет, так как реальный диск не опрашивается.

Методы, требующие системных вызовов (.exists(), .resolve(), .read_text(), .stat()), реализованы только в классе Path. Как только вызывается такой метод, Python обращается к виртуальной файловой системе Linux.

Это разделение полезно при написании тестов или кроссплатформенных скриптов. Например, если на Linux-машине нужно обработать пути, пришедшие из Windows-системы (с обратными слэшами \), нельзя использовать стандартный Path — он сломается о чужой синтаксис. Для этого используется PureWindowsPath, который позволит безопасно разобрать Windows-путь на компоненты, находясь внутри Linux.

Совместимость со старым кодом

Хотя pathlib стал стандартом де-факто, в экосистеме Python всё ещё встречаются старые библиотеки или модули, которые ожидают на вход строго строковый тип str, а не объект Path. Если передать объект Path в такую функцию, она может завершиться с ошибкой TypeError.

Для решения этой проблемы в Python был введен протокол __fspath__. Большинство современных встроенных функций (включая open(), модули subprocess, shutil) автоматически распознают объекты путей и извлекают из них строку.

Если же вы работаете со сторонней библиотекой, которая не поддерживает этот протокол, объект Path легко конвертируется в обычную строку явным приведением типов:

import old_legacy_module

config_path = Path("/etc/app/config.yml")

# Явное преобразование объекта в строку
string_path = str(config_path)

# Теперь можно передавать в старую функцию
old_legacy_module.load_config(string_path)

Переход от os.path к pathlib — это не просто смена синтаксиса, а изменение способа мышления. Скрипты автоматизации становятся более устойчивыми к ошибкам: исчезают проблемы с потерянными слэшами при склеивании, упрощается навигация по дереву каталогов, а операции чтения и записи конфигураций сводятся к одному вызову метода. Использование объектов Path закладывает надежный фундамент для более сложных операций с файловой системой, таких как рекурсивный обход директорий и массовое управление метаданными.

Манипуляция файлами и директориями: создание, перемещение и рекурсивный обход

Манипуляция файлами и директориями: создание, перемещение и рекурсивный обход

Скрипт резервного копирования падает в три часа ночи с ошибкой FileExistsError, потому что директория для бэкапов, которую он пытается создать, уже существует со вчерашнего дня. Или ломается с загадочным [Errno 18] Invalid cross-device link при попытке перенести архив из временной папки /tmp на примонтированный NFS-диск. Автоматизация работы с файловой системой в Linux требует предсказуемости: скрипт должен корректно отрабатывать независимо от того, запускается он впервые на чистом сервере или в сотый раз на системе с уже существующей структурой папок.

Идемпотентное создание инфраструктуры

В DevOps существует фундаментальный принцип идемпотентности: многократное выполнение операции должно приводить к тому же результату, что и однократное, без вызова ошибок. При работе с путями через pathlib этот принцип реализуется встроенными аргументами методов создания.

Метод .mkdir() создает новую директорию. По умолчанию он ведет себя как базовая команда mkdir в bash: если родительских папок нет — выдаст ошибку FileNotFoundError, если целевая папка уже существует — выдаст FileExistsError.

Для написания надежных скриптов метод .mkdir() всегда следует использовать с двумя флагами:

  1. parents=True — аналог ключа -p в bash. Указывает Python автоматически создать все недостающие промежуточные директории в пути.
  2. exist_ok=True — подавляет ошибку, если конечная директория уже существует.
from pathlib import Path

backup_dir = Path("/var/backups/nginx/2023/10")

# Безопасное создание: создаст /var/backups, затем nginx, и так далее.
# Если запустить скрипт дважды, на второй раз он просто проигнорирует команду.
backup_dir.mkdir(parents=True, exist_ok=True)

Для создания пустых файлов (или обновления времени их модификации) используется метод .touch(). Он работает идентично утилите touch в Linux. Если файл не существует, он будет создан нулевого размера. Если существует — обновится его метаданные (timestamp), но содержимое останется нетронутым. Аргумент exist_ok=True здесь включен по умолчанию.

Перемещение и ловушка границ файловых систем

Логика перемещения файлов в Linux скрывает в себе важный архитектурный нюанс. Когда вы переименовываете файл или перемещаете его в пределах одного логического диска (одной файловой системы), операционная система не копирует сами данные на диске. Она лишь меняет запись в индексной таблице (inode), указывая новое имя или новый путь к тем же физическим блокам памяти. Это происходит мгновенно, независимо от того, весит файл 1 килобайт или 100 гигабайт.

В pathlib за эту легковесную операцию отвечает метод .rename().

log_file = Path("/var/log/app/current.log")
archive_file = Path("/var/log/app/archive/app_old.log")

# Мгновенное переименование/перемещение внутри одного раздела
log_file.rename(archive_file)

Но скрипты автоматизации часто работают с несколькими томами. Например, скрипт собирает архив в /tmp (который в Linux часто смонтирован в оперативную память как tmpfs), а затем пытается переместить его в /mnt/nfs_share (внешнее сетевое хранилище).

Вызов Path("/tmp/backup.tar").rename("/mnt/nfs_share/backup.tar") завершится фатальной ошибкой OSError: [Errno 18] Invalid cross-device link. Метод .rename() вызывает системную функцию ядра, которая физически не умеет переносить данные между разными файловыми системами — она умеет только переписывать указатели.

Чтобы решить эту проблему, необходимо использовать модуль стандартной библиотеки shutil (shell utilities). Функция shutil.move() реализует умный алгоритм:

  1. Сначала она пытается выполнить быстрое перемещение на уровне указателей (как .rename()).
  2. Если ядро возвращает ошибку cross-device link, shutil.move() переключается в режим копирования: побайтово читает исходный файл, записывает его в целевую файловую систему, а затем удаляет оригинал.
import shutil
from pathlib import Path

source = Path("/tmp/db_dump.sql")
destination = Path("/mnt/s3_bucket/db_dump.sql")

# Гарантированно сработает, даже если пути находятся на разных дисках
shutil.move(source, destination)

shutil.move — это высокоуровневая обертка, которая делает перемещение файлов независимым от топологии дисковой подсистемы сервера. В скриптах автоматизации, где вы не контролируете точки монтирования, всегда используйте shutil.move вместо .rename().

Удаление: хирургическая точность против полного уничтожения

Удаление объектов файловой системы разделено на три разных механизма в зависимости от уровня опасности операции.

Удаление одного файла выполняется методом .unlink(). Начиная с Python 3.8, в него добавили параметр missing_ok=True. Если его не указать, попытка удалить несуществующий файл вызовет FileNotFoundError. С параметром missing_ok=True метод работает как rm -f — удаляет файл, если он есть, и молча идет дальше, если его нет.

lock_file = Path("/var/run/my_script.lock")
lock_file.unlink(missing_ok=True)

Удаление пустой директории выполняется методом .rmdir(). Это исключительно безопасный метод: операционная система откажется удалять папку, если внутри нее есть хотя бы один скрытый файл. Это защищает от случайного удаления важных данных из-за ошибки в формировании пути.

Рекурсивное удаление директории со всем содержимым (аналог rm -rf) в pathlib отсутствует намеренно, чтобы усложнить случайный прострел ноги. Для этой задачи снова привлекается модуль shutil и его функция shutil.rmtree().

import shutil
from pathlib import Path

cache_dir = Path("/var/cache/app_data")

if cache_dir.exists() and cache_dir.is_dir():
    # Удалит саму папку app_data и абсолютно всё внутри неё
    shutil.rmtree(cache_dir)

При использовании shutil.rmtree() критически важно убедиться, что путь указывает именно туда, куда вы ожидаете. Если переменная пути формируется динамически и из-за бага окажется равной Path("/") или Path("/etc"), скрипт, запущенный от имени root, уничтожит операционную систему за несколько секунд.

Итерация и поиск: от плоских списков к паттернам

Для анализа содержимого директорий pathlib предоставляет инструменты, которые возвращают генераторы. Это означает, что при запросе списка файлов в папке с миллионом логов, Python не будет пытаться загрузить весь миллион путей в оперативную память. Он будет выдавать их по одному в момент итерации цикла.

Метод .iterdir() выполняет плоский обход — возвращает объекты Path для всех файлов и папок, находящихся строго на первом уровне указанной директории (аналог обычного ls -a).

log_dir = Path("/var/log/nginx")

for item in log_dir.iterdir():
    if item.is_file():
        print(f"Найден файл: {item.name}")

Для фильтрации на лету используется метод .glob(), который принимает строковый паттерн (wildcard). Паттерны поддерживают символ * (любое количество любых символов) и ? (строго один любой символ).

# Найдет access.log, error.log, но проигнорирует access.log.gz
for log in log_dir.glob("*.log"):
    print(log.stem)

Рекурсивный обход дерева: заглядываем в каждую папку

Когда нужно найти файлы не только в целевой папке, но и во всех её вложенных подпапках на любую глубину, применяется метод .rglob() (recursive glob). Это прямой аналог утилиты find в Linux.

Метод .rglob() незаменим для задач инвентаризации или массовой очистки. Например, если конфигурационные файлы раскиданы по сложной структуре микросервисов:

config_root = Path("/opt/microservices")

# Найдет все файлы .yaml на любой глубине вложенности
for config_file in config_root.rglob("*.yaml"):
    # Читаем содержимое, парсим, проверяем синтаксис...
    pass

Проблема прав доступа при глубоком обходе

При использовании .rglob() в системных директориях (например, /var или /etc) скрипт почти гарантированно столкнется с ловушкой прав доступа. Если обход дерева натыкается на подпапку, для чтения которой у текущего пользователя нет прав (отсутствует атрибут r для директории), генератор выбросит исключение PermissionError и полностью прервет цикл. Остальные, доступные файлы, найдены не будут.

К сожалению, стандартный .rglob() не умеет молча пропускать запрещенные директории. Если скрипт запускается не от root и должен сканировать широкие участки файловой системы, использование .rglob() становится рискованным. В таких специфических случаях DevOps-инженеры вынуждены спускаться на уровень ниже и использовать os.walk() из модуля os, перехватывая ошибки вручную, либо запускать скрипт с повышенными привилегиями. Однако для пользовательских директорий, папок приложений и логов конкретных сервисов .rglob() остается самым лаконичным и читаемым инструментом.

Практический сценарий: ротация логов по расширению

Соберем изученные концепции в единый скрипт. Задача: обойти директорию приложения, найти все старые логи (файлы с расширением .old), перенести их в отдельную структуру папок для архивации, сохранив исходную иерархию, а затем удалить пустые папки, оставшиеся после переноса.

import shutil
from pathlib import Path

app_dir = Path("/opt/myapp")
archive_dir = Path("/mnt/archive/myapp_logs")

# 1. Рекурсивно ищем все файлы .old
for old_log in app_dir.rglob("*.old"):

    # Вычисляем относительный путь от корня приложения
    # Например: logs/module1/debug.old
    relative_path = old_log.relative_to(app_dir)

    # Формируем целевой путь в архиве: /mnt/archive/myapp_logs/logs/module1/debug.old
    target_path = archive_dir / relative_path

    # 2. Идемпотентно создаем родительские папки в архиве
    target_path.parent.mkdir(parents=True, exist_ok=True)

    # 3. Безопасно перемещаем файл (учитывая возможные разные ФС)
    shutil.move(old_log, target_path)
    print(f"Архивирован: {target_path}")

# 4. Очистка: удаляем пустые директории в исходной папке
# Сортируем пути по длине в обратном порядке, чтобы сначала удалять самые глубокие папки
all_dirs = sorted(app_dir.rglob("*"), key=lambda p: len(p.parts), reverse=True)

for d in all_dirs:
    if d.is_dir():
        try:
            d.rmdir() # Упадет с OSError, если папка не пуста
            print(f"Удалена пустая директория: {d}")
        except OSError:
            # Игнорируем ошибку: папка не пуста, значит удалять нельзя
            pass

В этом сценарии метод .relative_to() позволяет элегантно отсечь базовый путь /opt/myapp, оставив только внутреннюю структуру папок. Затем оператор / присоединяет этот хвост к новому корню архива. Конструкция try/except OSError при вызове .rmdir() используется как штатный механизм проверки: мы просто пытаемся удалить папку, и операционная система сама блокирует действие, если внутри остались активные логи.

Автоматизация на уровне файловой системы требует аккуратного баланса. Использование parents=True и exist_ok=True делает скрипты устойчивыми к повторным запускам. Применение shutil.move страхует от невидимых границ между дисками. А понимание разницы между безопасным .rmdir() и радикальным shutil.rmtree() гарантирует, что рутинная задача по очистке кэша не превратится в инцидент с потерей данных.

Управление метаданными и правами доступа в контексте автоматизации

Скрипт автоматизации успешно копирует конфигурационные файлы на удаленный сервер, перезапускает службу, но сервис отказывается стартовать. В логах появляется лаконичная запись: Bad owner or permissions on /etc/ssh/sshd_config. Файл на месте, его содержимое корректно, но операционная система блокирует к нему доступ. В Linux наличие файла — это лишь половина дела; вторая половина заключается в его метаданных. Автоматизация инфраструктуры требует точного управления тем, кто владеет файлом, какие у него права и когда он был изменён.

Чтение метаданных: заглядываем под капот

Любой файл или директория в Linux описывается индексным дескриптором (inode), который хранит метаданные: размер, владельца, права доступа и временные метки. В Python для получения этой информации используется метод .stat() объекта Path. Он делает системный вызов stat и возвращает объект os.stat_result.

from pathlib import Path

config_path = Path("/etc/resolv.conf")
metadata = config_path.stat()

print(f"Размер: {metadata.st_size} байт")
print(f"Владелец (UID): {metadata.st_uid}")

Объект os.stat_result содержит множество атрибутов, начинающихся с префикса st_. Наиболее востребованные в задачах DevOps — это временные метки. Linux отслеживает три основных времени для каждого файла, и все они хранятся в формате Unix timestamp (количество секунд, прошедших с 1 января 1970 года).

  1. st_mtime (Modification Time) — время последнего изменения содержимого файла. Обновляется при записи данных.
  2. st_atime (Access Time) — время последнего чтения файла.
  3. st_ctime (Change Time) — время последнего изменения метаданных (прав доступа, владельца, количества ссылок) или содержимого.

Частая ошибка начинающих инженеров — расшифровывать ctime как «creation time» (время создания). В классических файловых системах Linux (ext4, xfs) исторически не было стандартизированного атрибута времени создания файла. ctime — это именно время изменения состояния (change time). Если вы сделаете chmod для файла, его mtime останется прежним, а ctime обновится.

Поскольку время возвращается в виде секунд (тип float), для удобного вывода или сложной логики его конвертируют в объекты datetime.

from datetime import datetime
import time

# Получаем mtime
mtime_seconds = config_path.stat().st_mtime

# Конвертация в человекочитаемый формат
readable_time = datetime.fromtimestamp(mtime_seconds)
print(f"Файл изменен: {readable_time.strftime('%Y-%m-%d %H:%M:%S')}")

В автоматизации временные метки чаще всего используются для очистки старых данных. Например, скрипт ротации бэкапов должен удалить архивы старше 30 дней. Вместо конвертации в даты, проще и быстрее работать с сырыми секундами, вычитая их из текущего системного времени, которое возвращает функция time.time().

import time

backup_dir = Path("/var/backups/db")
# 30 дней * 24 часа * 60 минут * 60 секунд
age_limit_seconds = 30 * 24 * 3600
current_time = time.time()

for archive in backup_dir.glob("*.tar.gz"):
    file_age = current_time - archive.stat().st_mtime

    if file_age > age_limit_seconds:
        archive.unlink()
        print(f"Удален старый бэкап: {archive.name}")

Анатомия прав доступа и восьмеричная система

В Linux права доступа делятся на три категории пользователей: владелец (User), группа (Group) и все остальные (Others). Для каждой категории задаются три вида прав: чтение (Read), запись (Write) и выполнение (eXecute).

Каждое право — это один бит информации: включено (1) или выключено (0). Три бита (r, w, x) образуют число от 0 до 7. Именно поэтому права доступа в Linux исторически записываются в восьмеричной системе счисления.

В Python восьмеричные числа (литералы) обозначаются префиксом 0o (ноль и буква «o»).

  • 0o755 — владелец может всё (7), группа и остальные могут читать и выполнять (5).
  • 0o600 — только владелец может читать и писать (6), остальным доступ закрыт (0).
  • 0o644 — владелец читает и пишет (6), остальные только читают (4).

Если попытаться прочитать права файла через атрибут st_mode, результат может сбить с толку:

ssh_key = Path("/home/user/.ssh/id_rsa")
mode = ssh_key.stat().st_mode
print(mode)  # Выведет что-то вроде 33152

Число 33152 (в десятичной системе) не похоже на привычные 0600. Дело в том, что st_mode — это 16-битное целое число, которое хранит не только права доступа, но и тип файла (обычный файл, директория, символическая ссылка, сокет), а также специальные биты (SUID, SGID, Sticky bit).

Чтобы извлечь из этого «комбайна» только младшие 9 бит, отвечающие за права доступа (rwxrwxrwx), необходимо применить побитовое «И» (оператор &) с маской. В модуле stat для этого предусмотрена константа stat.S_IMODE. Она содержит маску 0o7777 (включая специальные биты).

import stat

# Извлекаем только права доступа
permissions = stat.S_IMODE(ssh_key.stat().st_mode)

# Выводим в восьмеричном формате с префиксом 0o
print(oct(permissions))  # Выведет 0o600

Использование функции oct() здесь необходимо только для визуального отображения (чтобы человек увидел 0o600, а не десятичное 384). Внутри скрипта, при сравнении прав, конвертация в строку не нужна, числа сравниваются напрямую: if permissions == 0o600:.

Модификация прав: ловушка десятичных чисел

Для изменения прав доступа используется метод .chmod() объекта Path. Он принимает целое число, представляющее новые права. И здесь кроется одна из самых коварных ошибок при написании Python-скриптов для Linux.

Рассмотрим два варианта изменения прав:

script_path = Path("/usr/local/bin/deploy.sh")

# Вариант А (ОШИБКА)
script_path.chmod(755)

# Вариант Б (ПРАВИЛЬНО)
script_path.chmod(0o755)

В первом варианте программист передал десятичное число 755. Python честно передаст его ядру Linux. Но ядро ожидает битовую маску. Десятичное число 755 в восьмеричной системе равно 0o1363. В результате скрипт получит права --wxrw--wx (плюс установится Sticky bit), что сделает систему уязвимой и сломает выполнение скрипта.

Всегда используйте префикс 0o при работе с правами.

При автоматизации развертывания часто требуется рекурсивно задать права на директорию: например, директории должны иметь права 0o755 (чтобы в них можно было переходить — бит x), а файлы — 0o644. Метод .chmod() применяется только к одному объекту, поэтому для рекурсивного изменения используется обход дерева.

web_root = Path("/var/www/html")

# Устанавливаем права на саму корневую директорию
web_root.chmod(0o755)

# Рекурсивный обход
for item in web_root.rglob("*"):
    if item.is_dir():
        item.chmod(0o755)
    elif item.is_file():
        item.chmod(0o644)

Управление владельцами: UID, GID и модуль shutil

Права доступа неразрывно связаны с тем, кто является владельцем файла. В Linux пользователи и группы идентифицируются числами — UID (User ID) и GID (Group ID). Имена пользователей (например, nginx или root) — это лишь удобная текстовая абстракция, хранящаяся в файлах /etc/passwd и /etc/group.

Объект os.stat_result возвращает именно числовые идентификаторы (st_uid, st_gid). Объект Path предоставляет удобные методы .owner() и .group(), которые автоматически читают системные базы данных и возвращают строковые имена владельца и группы.

nginx_conf = Path("/etc/nginx/nginx.conf")

print(f"UID: {nginx_conf.stat().st_uid}")
print(f"Владелец (имя): {nginx_conf.owner()}")

Однако, когда дело доходит до изменения владельца, pathlib не предоставляет метода .chown(). Для этой задачи в стандартной библиотеке есть два инструмента: os.chown() и shutil.chown().

Функция os.chown(path, uid, gid) работает на низком уровне и требует передачи числовых идентификаторов. Если вы хотите передать имя пользователя, вам придется самостоятельно разрешать его в UID через модуль pwd. Это усложняет код.

Функция shutil.chown(path, user=None, group=None) является высокоуровневой оберткой. Она принимает строковые имена, сама выполняет поиск соответствующих UID/GID и применяет их. Это делает её идеальным выбором для скриптов автоматизации.

import shutil

app_dir = Path("/opt/myapp")

# Смена владельца и группы по текстовым именам
shutil.chown(app_dir, user="appuser", group="appgroup")

Важный нюанс безопасности: в Linux изменять владельца файла (выполнять системный вызов chown) имеет право только суперпользователь (root). Даже если вы являетесь владельцем файла, вы не можете «подарить» его другому пользователю. Если скрипт автоматизации, запущенный от обычного пользователя, попытается выполнить shutil.chown(), будет выброшено исключение PermissionError. Поэтому скрипты инициализации инфраструктуры всегда запускаются с повышенными привилегиями (через sudo или системами управления конфигурациями).

Работа с символическими ссылками: lstat и lchown

Символическая ссылка (symlink) — это особый файл, который содержит путь к другому файлу. При чтении метаданных симлинка возникает развилка: чьи метаданные мы хотим получить — самой ссылки или того файла, на который она указывает?

По умолчанию метод Path.stat() прозрачно переходит по ссылке и возвращает метаданные целевого файла. Если целевой файл удален (битая ссылка), вызов завершится ошибкой FileNotFoundError.

Чтобы получить метаданные самой ссылки (например, узнать время её создания или права), используется метод Path.lstat() (от слова link-stat).

link = Path("/etc/localtime") # Обычно это симлинк на /usr/share/zoneinfo/...

# Метаданные целевого файла (самой зоны)
target_meta = link.stat()
print(f"Размер цели: {target_meta.st_size}")

# Метаданные самой ссылки
link_meta = link.lstat()
print(f"Размер ссылки: {link_meta.st_size}") # Обычно равен длине строки пути

Аналогичное правило действует для изменения владельца. Если применить shutil.chown() к симлинку, изменится владелец целевого файла. Если нужно изменить владельца самой ссылки (что требуется редко, но бывает необходимо при специфичных настройках безопасности), используется низкоуровневая функция os.lchown().

Аудит и приведение конфигураций к эталону

Понимание работы с метаданными позволяет создавать скрипты-аудиторы, которые проверяют состояние инфраструктуры и автоматически исправляют отклонения. Рассмотрим практический сценарий: на сервере есть директория с конфигурациями базы данных /etc/db_configs/. Требования безопасности гласят, что все файлы внутри должны принадлежать пользователю postgres, а права должны быть строго 0o600.

Скрипт должен просканировать директорию, найти нарушения, исправить их и залогировать свои действия.

import stat
import shutil
from pathlib import Path

def enforce_security_policy(target_dir: str):
    config_dir = Path(target_dir)

    if not config_dir.exists() or not config_dir.is_dir():
        print(f"Директория {target_dir} не найдена.")
        return

    expected_user = "postgres"
    expected_mode = 0o600

    for file_path in config_dir.glob("*.conf"):
        # Читаем текущее состояние
        current_stat = file_path.stat()
        current_mode = stat.S_IMODE(current_stat.st_mode)
        current_owner = file_path.owner()

        # Флаг для отслеживания изменений
        fixed = False

        # Проверка и исправление прав
        if current_mode != expected_mode:
            file_path.chmod(expected_mode)
            print(f"[FIX] Права {file_path.name}: {oct(current_mode)} -> {oct(expected_mode)}")
            fixed = True

        # Проверка и исправление владельца
        if current_owner != expected_user:
            try:
                shutil.chown(file_path, user=expected_user)
                print(f"[FIX] Владелец {file_path.name}: {current_owner} -> {expected_user}")
                fixed = True
            except PermissionError:
                print(f"[ERROR] Нет прав для смены владельца {file_path.name}")

        if not fixed:
            print(f"[OK] {file_path.name} соответствует политике.")

# Запуск аудита
enforce_security_policy("/etc/db_configs")

Этот подход реализует принцип идемпотентности, рассмотренный в предыдущих главах. Скрипт не применяет chmod и chown слепо ко всем файлам при каждом запуске. Он сначала читает состояние системы, сравнивает его с эталоном и выполняет модифицирующие системные вызовы только там, где есть отклонение. Это экономит ресурсы дисковой подсистемы и позволяет использовать скрипт в системах непрерывного мониторинга состояния (compliance as code).

Контроль над метаданными файлов — это фундамент, на котором строится безопасность операционной системы. Умение корректно читать битовые маски, избегать ловушек с восьмеричными числами и различать временные метки позволяет писать надежные инструменты администрирования.

Основы запуска процессов: использование subprocess.run для простых bash-команд

Любой скрипт автоматизации рано или поздно упирается в границы стандартной библиотеки языка. Python обладает мощными инструментами для работы с сетью, файловой системой и форматами данных, но в операционной системе Linux существуют сотни специализированных утилит, переписывать которые на Python нецелесообразно. Управление системными службами через systemctl, синхронизация директорий через rsync, управление пакетами через apt или dnf — всё это требует выхода за пределы интерпретатора и прямого взаимодействия с бинарными файлами ОС.

Для решения этой задачи в Python предусмотрен модуль subprocess. Он предоставляет интерфейс для порождения новых процессов, подключения к их каналам ввода-вывода и получения кодов возврата. Исторически в Python существовало множество способов запуска внешних команд (например, os.system или os.popen), но в современных версиях языка стандартом де-факто стала функция subprocess.run(). Она объединяет в себе безопасность, гибкость и предсказуемость поведения.

Анатомия вызова: почему аргументы передаются списком

Самая частая ошибка при переходе от написания bash-скриптов к Python — попытка передать команду в виде единой строки. В терминале администратор пишет ls -l /var/log, и оболочка (bash или zsh) берет на себя задачу разбора этой строки на составные части.

При использовании subprocess.run() по умолчанию применяется другой механизм. Функция ожидает не строку, а список строк (массив аргументов).

import subprocess

# Правильный подход: каждый элемент команды — отдельная строка в списке
subprocess.run(["ls", "-l", "/var/log"])

Такое требование продиктовано архитектурой операционной системы. Как было изучено ранее, запуск новой программы в Linux опирается на системный вызов ядра execve. Этот вызов не умеет парсить пробелы, кавычки или экранированные символы. Он принимает строго структурированные данные: путь к исполняемому файлу и массив аргументов (включая нулевой аргумент — само имя программы).

Передавая список в subprocess.run(), интерпретатор Python транслирует его напрямую в ядро Linux, полностью минуя командную оболочку. Первый элемент списка становится исполняемым файлом, а остальные — его аргументами.

Этот подход решает одну из главных проблем автоматизации — обработку путей, содержащих пробелы или специальные символы.

В bash пробел является разделителем аргументов. Если необходимо удалить директорию с именем old backups, в терминале приходится использовать экранирование или кавычки: rm -rf "old backups". Если забыть кавычки, утилита rm попытается удалить две разные директории: old и backups.

В Python при передаче аргументов списком эта проблема исчезает на фундаментальном уровне.

import subprocess

target_dir = "/mnt/data/old backups"

# Ядро Linux получит ровно два аргумента: флаг "-rf" и строку "/mnt/data/old backups"
# Никакого дополнительного разбиения по пробелам не произойдет.
subprocess.run(["rm", "-rf", target_dir])

Поскольку строка target_dir передается как единый элемент списка, execve воспринимает её как один монолитный аргумент. Никакие пробелы внутри этой строки не заставят систему разделить её на части. Это делает скрипты невосприимчивыми к некорректным именам файлов, обеспечивая высокую надежность при массовой обработке данных.

Интеграция с объектно-ориентированными путями

В современных скриптах пути редко хранятся в виде обычных строк. Для работы с файловой системой используется модуль pathlib и его объекты Path. Разработчики модуля subprocess учли эту тенденцию: начиная с Python 3.8, функция run() нативно поддерживает объекты, реализующие протокол os.PathLike.

Это означает, что при формировании списка аргументов нет необходимости принудительно конвертировать объекты Path в строки с помощью str().

import subprocess
from pathlib import Path

backup_dir = Path("/var/backups/nginx")
config_file = Path("/etc/nginx/nginx.conf")

# Объекты Path передаются напрямую в список аргументов
subprocess.run(["cp", config_file, backup_dir])

Интерпретатор автоматически извлечет строковое представление путей перед передачей их системному вызову. Это позволяет элегантно комбинировать манипуляции с файловой системой и вызов внешних утилит. Например, можно найти нужный файл с помощью методов pathlib, а затем передать его в системный архиватор:

import subprocess
from pathlib import Path

log_dir = Path("/var/log/app")
# Находим самый старый лог-файл
oldest_log = min(log_dir.glob("*.log"), key=lambda p: p.stat().st_mtime)

archive_name = oldest_log.with_suffix(".tar.gz")

# Создаем архив с помощью системной утилиты tar
subprocess.run(["tar", "-czf", archive_name, oldest_log])

Механика shell=True: удобство против безопасности

Иногда передача команды списком кажется избыточной, особенно если команда содержит специфичные для bash конструкции: конвейеры (|), перенаправления потоков (>), раскрытие шаблонов (*) или вызов встроенных команд оболочки (например, source или export).

Для таких случаев в subprocess.run() предусмотрен именованный аргумент shell=True. При его активации функция принимает не список, а единую строку.

import subprocess

# Использование shell=True позволяет применять bash-шаблоны (*)
subprocess.run("rm -rf /tmp/cache/*", shell=True)

Чтобы понять, почему этот подход требует осторожности, необходимо рассмотреть, что происходит под капотом. При вызове с shell=True Python не запускает целевую утилиту (в данном случае rm) напрямую. Вместо этого он запускает системную оболочку /bin/sh и передает ей переданную строку с помощью флага -c.

Фактически, вызов subprocess.run("rm -rf /tmp/cache/*", shell=True) эквивалентен следующему вызову без оболочки: subprocess.run(["/bin/sh", "-c", "rm -rf /tmp/cache/*"])

Появление промежуточного процесса /bin/sh кардинально меняет правила игры:

  1. Разбор строки: Оболочка берет на себя парсинг переданной строки. Она самостоятельно ищет пробелы, применяет кавычки, раскрывает переменные окружения (вида $USER) и шаблоны (вида *.txt).
  2. Накладные расходы: Запуск дополнительного процесса оболочки требует времени и ресурсов ОС. Для единичного вызова это незаметно, но в цикле из тысяч итераций разница в производительности станет критической.
  3. Безопасность: Если часть строки формируется из внешних данных (имени файла, ввода пользователя, значения из базы данных), оболочка может интерпретировать спецсимволы внутри этих данных как команды. Это открывает вектор для уязвимости Shell Injection (подробный разбор которой будет дан в последующих главах).

Использование shell=True оправдано только в тех случаях, когда скрипту действительно требуются уникальные возможности bash, которые сложно или невозможно воспроизвести средствами самого Python. Если задача сводится к простому запуску бинарного файла с фиксированными флагами — стандартный режим (список строк) всегда предпочтительнее.

Контроль успешности: fail-fast подход и check=True

При запуске внешней команды через subprocess.run() её вывод (stdout и stderr) по умолчанию направляется в терминал, где запущен сам Python-скрипт. Процесс Python приостанавливает свое выполнение и ждет, пока дочерний процесс завершит работу.

После завершения внешней программы subprocess.run() возвращает специальный объект — экземпляр класса CompletedProcess. Этот объект содержит метаданные о выполненном процессе, главными из которых на базовом этапе являются переданные аргументы и код возврата (exit status).

Важнейшая особенность subprocess.run() заключается в том, что по умолчанию функция не генерирует исключение, если внешняя команда завершилась с ошибкой (вернула ненулевой код).

import subprocess

# Попытка остановить несуществующую службу
result = subprocess.run(["systemctl", "stop", "nonexistent.service"])

print("Скрипт продолжает работу!")

В примере выше утилита systemctl выведет в терминал сообщение об ошибке и завершится с кодом 5. Однако Python-скрипт воспримет это как штатное завершение системного вызова. Переменная result получит объект CompletedProcess(args=['systemctl', 'stop', 'nonexistent.service'], returncode=5), и интерпретатор перейдет к следующей строке кода.

В контексте DevOps и системного администрирования такое поведение (тихий провал) часто является источником катастрофических сбоев. Если команда создания бэкапа завершилась ошибкой, скрипт не должен переходить к следующему шагу — удалению старых архивов.

Чтобы изменить это поведение и заставить скрипт падать при любой ошибке внешнего процесса, используется аргумент check=True.

import subprocess

try:
    # Аргумент check=True включает строгую валидацию кода возврата
    subprocess.run(["systemctl", "restart", "nginx"], check=True)
    print("Nginx успешно перезапущен.")
except subprocess.CalledProcessError as e:
    print(f"Критическая ошибка! Команда завершилась с кодом {e.returncode}")
    # Здесь можно добавить логику отката (rollback) или отправки алерта

При наличии check=True функция проверяет атрибут returncode после завершения процесса. Если он не равен нулю (что в семантике Linux означает ошибку), функция немедленно выбрасывает исключение subprocess.CalledProcessError.

Использование check=True реализует паттерн fail-fast (быстрый отказ). Скрипт прерывает выполнение ровно в той точке, где произошел сбой, предотвращая эффект домино, при котором некорректное состояние системы передается последующим функциям.

Жизненный цикл вызова и изоляция среды

Когда Python вызывает subprocess.run(), дочерний процесс наследует окружение родительского процесса. Это означает, что все переменные окружения, текущая рабочая директория (CWD) и маска прав доступа (umask) передаются запускаемой утилите без изменений.

Если скрипт на Python был запущен пользователем root из директории /opt/scripts, то и команда subprocess.run(["pwd"]) отработает с правами root, а её рабочей директорией будет /opt/scripts.

Иногда требуется запустить команду в изолированном контексте или в другой директории, не меняя при этом состояние самого Python-скрипта. Для этого subprocess.run() предоставляет аргумент cwd (Current Working Directory).

import subprocess
from pathlib import Path

repo_path = Path("/var/www/html/webapp")

# Команда git pull будет выполнена внутри директории /var/www/html/webapp,
# хотя сам Python-скрипт может быть запущен откуда угодно.
subprocess.run(["git", "pull"], cwd=repo_path, check=True)

Изменение рабочей директории через аргумент cwd влияет только на конкретный дочерний процесс. Это гораздо безопаснее, чем использовать os.chdir() перед вызовом команды, так как os.chdir() глобально меняет директорию для всего процесса Python, что может сломать логику работы других модулей или параллельных потоков.

Запуск процессов из Python — это мост между высокоуровневой логикой скрипта и низкоуровневыми бинарными инструментами операционной системы. Понимание того, как ядро обрабатывает список аргументов, почему shell=True запускает дополнительную оболочку и как контролировать коды возврата через check=True, позволяет писать надежные обертки над любыми системными утилитами. На этом фундаменте строится вся дальнейшая работа с процессами: от захвата их вывода для парсинга до построения сложных конвейеров обработки данных.

Захват и обработка вывода: работа с stdout, stderr и кодами возврата

Захват и обработка вывода: работа с stdout, stderr и кодами возврата

Скрипт резервного копирования запускается по расписанию cron каждую ночь. Утром системный администратор видит, что код возврата равен нулю, скрипт отчитался об успехе, но архив базы данных пуст. Проблема кроется в том, что утилита дампа столкнулась с некритичной ошибкой блокировки таблиц, вывела предупреждение в поток ошибок, но штатно завершила работу. Поскольку Python-скрипт проверял только статус завершения и игнорировал текстовый вывод утилиты, критическая инфраструктурная аномалия осталась незамеченной.

Автоматизация на стыке Python и Linux требует не просто запускать команды, а уметь «слушать» то, что они возвращают. Операционная система предоставляет каждому процессу три стандартных файловых дескриптора: stdin (ввод), stdout (вывод) и stderr (ошибки). По умолчанию, когда мы вызываем внешнюю команду через модуль subprocess, её потоки вывода остаются привязанными к терминалу, в котором запущен сам Python. Текст печатается на экран, но внутри скрипта переменные остаются пустыми. Чтобы программно реагировать на результаты работы утилит, эти потоки необходимо перехватить и направить в память интерпретатора.

Механика перехвата: от байтов к строкам

Для захвата вывода в современных версиях Python используется аргумент capture_output=True. При его передаче функция subprocess.run() создает каналы связи между родительским и дочерним процессами, аккумулируя всё, что утилита пишет в консоль.

Однако операционная система ничего не знает о кодировках текста — она оперирует сырыми байтами. Если мы просто захватим вывод, мы получим объекты типа bytes.

import subprocess

# Захват без указания текстового режима
result = subprocess.run(
    ["uname", "-r"],
    capture_output=True
)

print(repr(result.stdout))
# Выведет: b'6.1.0-13-amd64\n'

Работать с байтовыми строками (префикс b) при парсинге логов или конфигураций неудобно: к ним нельзя применить большинство привычных строковых методов без явного декодирования. Чтобы Python автоматически переводил байты в строки, необходимо использовать аргумент text=True (в старых версиях языка он назывался universal_newlines=True).

При включении текстового режима крайне важно явно указывать кодировку через аргумент encoding. В средах Linux стандартом де-факто является UTF-8, однако при обработке вывода специфичных утилит или при работе в гетерогенных сетях скрипт может столкнуться с неожиданными символами. Если инструмент выведет байт, не соответствующий UTF-8, скрипт упадет с исключением UnicodeDecodeError. Для отказоустойчивых систем автоматизации применяется аргумент errors="replace", который заменяет нечитаемые символы на безопасный знак вопроса ?, позволяя скрипту продолжить работу.

# Отказоустойчивый захват вывода
result = subprocess.run(
    ["cat", "/var/log/syslog"],
    capture_output=True,
    text=True,
    encoding="utf-8",
    errors="replace"
)

Раздельный анализ: stdout для данных, stderr для диагностики

Философия Unix гласит, что программы должны выводить полезные данные в stdout, а диагностические сообщения и ошибки — в stderr. Раздельный захват позволяет нам строить четкую логику: мы парсим stdout как структурированные данные, а stderr логируем для аудита.

Рассмотрим задачу инвентаризации дисковой подсистемы. Утилита lsblk умеет отдавать данные в формате JSON, что идеально подходит для интеграции с Python.

import subprocess
import json
import sys

result = subprocess.run(
    ["lsblk", "-J", "-o", "NAME,SIZE,TYPE,MOUNTPOINT"],
    capture_output=True,
    text=True,
    encoding="utf-8"
)

if result.returncode != 0:
    print(f"Критическая ошибка lsblk: {result.stderr.strip()}", file=sys.stderr)
    sys.exit(1)

try:
    # Парсинг захваченного stdout
    disk_data = json.loads(result.stdout)

    for device in disk_data.get("blockdevices", []):
        if device.get("type") == "disk":
            print(f"Обнаружен диск: {device.get('name')} объемом {device.get('size')}")

except json.JSONDecodeError:
    print("Ошибка: lsblk вернул некорректный JSON", file=sys.stderr)

В этом сценарии разделение потоков критически важно. Если бы утилита lsblk вывела предупреждение (например, о недоступности какого-то специфичного устройства) в тот же поток, где находится JSON, метод json.loads() не смог бы распарсить строку из-за наличия постороннего текста. Благодаря тому, что capture_output=True разделяет потоки, result.stdout содержит исключительно валидный JSON, а любые системные жалобы безопасно оседают в result.stderr.

Слияние потоков: когда хронология важнее структуры

Бывают ситуации, когда разделение потоков вредит делу. Представьте, что вы автоматизируете обновление пакетов через apt-get upgrade или запускаете длительную миграцию базы данных. Такие инструменты часто пишут информацию о прогрессе в stdout, а предупреждения — в stderr.

Если захватить их раздельно, а затем попытаться сохранить в единый лог-файл, вы потеряете хронологию. Сначала запишется весь stdout, а затем весь stderr. Вы не сможете понять, после какого именно шага возникло предупреждение.

Для решения этой задачи потоки необходимо объединить на уровне операционной системы, направив stderr в тот же канал, что и stdout. В subprocess это делается путем отказа от макроса capture_output=True в пользу прямой настройки дескрипторов.

# Объединение потоков для сохранения хронологии
result = subprocess.run(
    ["apt-get", "update"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    encoding="utf-8"
)

# Теперь result.stderr равен None,
# а result.stdout содержит смешанный вывод в правильном порядке
with open("/var/log/automation/apt_update.log", "a", encoding="utf-8") as log_file:
    log_file.write(result.stdout)

Здесь константа subprocess.PIPE указывает Python создать канал в оперативной памяти для захвата стандартного вывода. Константа subprocess.STDOUT, переданная в аргумент stderr, дает ядру Linux команду: «всё, что этот процесс пытается написать в поток ошибок, перенаправь в его поток стандартного вывода». В результате мы получаем единую, хронологически верную ленту событий.

Тонкая работа с кодами возврата (Exit Codes)

Ранее мы использовали аргумент check=True, который реализует паттерн fail-fast: если команда завершается с кодом, отличным от нуля, Python немедленно выбрасывает исключение CalledProcessError. Это отлично работает для команд вроде mkdir или tar, где любой ненулевой код — это безусловная авария.

Однако в мире Linux код возврата 0\neq 0 далеко не всегда означает ошибку. Часто он используется для передачи логического состояния. Классический пример — утилита grep.

Согласно документации grep:

  • Код 00 возвращается, если искомая строка найдена.
  • Код 11 возвращается, если строка не найдена (это не ошибка выполнения, а результат поиска).
  • Код 22 возвращается при реальных ошибках (например, файл не существует или нет прав на чтение).

Если мы применим check=True к grep, наш скрипт упадет с исключением при первой же попытке поиска отсутствующей строки. Для таких утилит необходимо обрабатывать атрибут returncode вручную.

def check_user_in_group(username, group_file="/etc/group"):
    result = subprocess.run(
        ["grep", f"^{username}:", group_file],
        capture_output=True,
        text=True
    )

    if result.returncode == 0:
        return True
    elif result.returncode == 1:
        return False
    else:
        # Реальная ошибка (код 2 или другой)
        raise RuntimeError(f"Ошибка чтения {group_file}: {result.stderr.strip()}")

is_admin = check_user_in_group("devops_user", "/etc/sudoers")

Другой частый пример в практике SRE — проверка статуса служб через systemctl. Команда systemctl is-active nginx вернет строку "active" и код 00, если служба работает. Но если служба остановлена, она выведет "inactive" и завершится с кодом 33. Использование check=True здесь приведет к краху скрипта мониторинга ровно в тот момент, когда он должен зафиксировать падение службы и отправить алерт.

Обработка плохо спроектированных CLI-инструментов

В идеальном мире POSIX-совместимых утилит коды возврата и потоки используются строго по назначению. В реальной инфраструктуре DevOps-инженеру регулярно приходится автоматизировать работу проприетарных бинарников, legacy-скриптов или утилит от разработчиков, которые игнорируют стандарты.

Самый опасный паттерн — утилита сталкивается с критической ошибкой, выводит текст "ERROR: Connection timeout" в stdout (вместо stderr) и завершается с кодом 00 (успех).

Если мы будем опираться только на returncode, наша система автоматизации проигнорирует сбой. В таких случаях контракт успешности должен определяться не операционной системой, а логикой Python-скрипта на основе анализа захваченного текста.

def run_legacy_backup():
    result = subprocess.run(
        ["/opt/legacy_app/bin/backup_tool", "--destination", "/mnt/nfs"],
        capture_output=True,
        text=True,
        encoding="utf-8"
    )

    # Утилита всегда возвращает 0, поэтому проверяем текст
    output = result.stdout.upper()

    if "ERROR" in output or "FAILED" in output:
        # Искусственно эскалируем проблему в Python
        raise ValueError(f"Legacy backup failed. Output: {result.stdout.strip()}")

    if "SUCCESS" not in output:
        # Защита от молчаливого изменения формата вывода в новых версиях утилиты
        raise ValueError("Unknown output format from backup tool.")

    return True

Этот подход называется защитным программированием. Мы не доверяем коду возврата, а ищем явные маркеры успеха или неудачи в тексте. Обратите внимание на проверку if "SUCCESS" not in output. Если разработчики legacy-утилиты выпустят обновление, где изменят формат вывода, и скрипт перестанет находить слова "ERROR", он не сочтет операцию успешной ошибочно. Отсутствие явного подтверждения успеха трактуется как сбой.

Понимание того, как операционная система маршрутизирует байты между процессами, и умение гибко перехватывать эти потоки в Python — это граница, отделяющая хрупкие скрипты от надежных инструментов автоматизации. Явное декодирование, обработка нетипичных кодов возврата и анализ текстового вывода позволяют строить системы, которые корректно реагируют на любые сбои в управляемой инфраструктуре.

Продвинутое управление процессами: конвейеры (pipes) и передача данных на stdin

Продвинутое управление процессами: конвейеры (pipes) и передача данных на stdin

В практике системного администрирования редко удается решить задачу вызовом одной изолированной утилиты. Логи Linux-сервера фильтруются через grep, передаются в awk для извлечения колонок и сортируются через sort. Базы данных выгружаются через pg_dump, на лету сжимаются gzip и отправляются на удаленный сервер. В командной оболочке bash эта элегантная маршрутизация данных реализуется одним символом — вертикальной чертой | (pipe). При переносе таких скриптов на Python возникает соблазн использовать аргумент shell=True, чтобы сохранить привычный синтаксис bash. Однако этот подход делегирует контроль над процессами промежуточной оболочке, усложняя перехват ошибок и открывая векторы для инъекций. Нативная реализация конвейеров средствами Python требует понимания того, как операционная система работает со стандартными потоками ввода-вывода.

Управление стандартным вводом (stdin) через аргумент input

Любой процесс в Linux имеет три стандартных файловых дескриптора: 0 (stdin), 1 (stdout) и 2 (stderr). В предыдущих материалах разбирался захват вывода (stdout/stderr). Теперь фокус смещается на передачу данных внутрь процесса.

Многие системные утилиты спроектированы так, чтобы читать данные из стандартного ввода, если они не переданы через аргументы командной строки. Это особенно критично для передачи чувствительной информации. Передача паролей или токенов через аргументы (например, mysql -u root -pMySecretPassword) оставляет их видимыми в таблице процессов ОС, и любой пользователь может прочитать их с помощью команды ps aux.

Функция subprocess.run() предоставляет аргумент input для безопасной отправки данных в stdin дочернего процесса.

import subprocess

# Безопасное изменение пароля пользователя через утилиту chpasswd
# Формат ввода для chpasswd: "username:new_password"
credentials = "devops_user:SuperSecure!123"

# Передаем строку в stdin процесса.
# text=True автоматически кодирует строку в байты перед отправкой.
result = subprocess.run(
    ["sudo", "chpasswd"],
    input=credentials,
    text=True,
    check=True
)

В этом сценарии Python открывает процесс chpasswd, записывает строку credentials в его файловый дескриптор stdin, отправляет сигнал EOF (End of File), чтобы утилита поняла, что передача данных завершена, и дожидается окончания работы процесса. Пароль ни на секунду не появляется в аргументах командной строки.

Аргумент input принимает байты (bytes), но при наличии флага text=True (или encoding="utf-8") модуль subprocess автоматически выполняет кодирование строковых данных.

Наивный конвейер: Python в роли посредника

Самый очевидный способ объединить две команды в Python — запустить первую, сохранить ее вывод в переменную, а затем передать эту переменную в input второй команды.

Рассмотрим задачу: получить список всех установленных пакетов в Debian/Ubuntu и отфильтровать только те, что связаны с python3.

import subprocess

# Шаг 1: Получаем полный список пакетов
dpkg_process = subprocess.run(
    ["dpkg", "-l"],
    capture_output=True,
    text=True,
    check=True
)

# Шаг 2: Передаем результат первой команды на вход второй
grep_process = subprocess.run(
    ["grep", "python3"],
    input=dpkg_process.stdout,
    capture_output=True,
    text=True,
    check=False # grep возвращает 1, если ничего не найдено
)

print(grep_process.stdout)

Этот подход абсолютно корректен для небольших объемов данных. Он легко читается и отлаживается: можно в любой момент вывести содержимое dpkg_process.stdout и проверить, что именно передается на следующий этап.

Проблема наивного конвейера кроется в потреблении оперативной памяти. В данном примере Python дожидается полного завершения dpkg, загружает весь его вывод в оперативную память (создавая гигантскую строку), и только потом запускает grep. Если первая команда генерирует дамп базы данных размером 50 GB50 \text{ GB}, скрипт неизбежно завершится с ошибкой MemoryError или спровоцирует срабатывание системного OOM Killer (Out Of Memory), так как попытается выделить 50 GB50 \text{ GB} RAM для хранения переменной dpkg_process.stdout.

В настоящем bash-конвейере dpkg -l | grep python3 процессы работают параллельно. Операционная система создает между ними канал связи (pipe) в оперативной памяти фиксированного размера (обычно 64 KB в современных ядрах Linux). Как только dpkg заполняет этот буфер, ядро приостанавливает его работу до тех пор, пока grep не прочитает часть данных. Память не расходуется бесконтрольно.

Истинные конвейеры и класс Popen

Для реализации параллельного выполнения с прямой передачей данных между процессами функции subprocess.run() недостаточно. Она по определению является блокирующей (синхронной) — интерпретатор останавливается на строке вызова и ждет завершения процесса.

Здесь на сцену выходит класс subprocess.Popen. Это низкоуровневый интерфейс, на котором базируется run(). Создание объекта Popen запускает процесс в фоновом режиме (асинхронно) и немедленно возвращает управление Python-скрипту.

Чтобы связать два процесса напрямую, необходимо перенаправить stdout первого процесса в stdin второго.

import subprocess

# Запускаем первую команду.
# subprocess.PIPE указывает ядру создать канал связи, а не выводить данные на экран.
p1 = subprocess.Popen(
    ["cat", "/var/log/syslog"],
    stdout=subprocess.PIPE
)

# Запускаем вторую команду.
# В качестве stdin передаем файловый дескриптор stdout первого процесса.
p2 = subprocess.Popen(
    ["grep", "sshd"],
    stdin=p1.stdout,
    stdout=subprocess.PIPE,
    text=True
)

В этот момент оба процесса запущены и работают параллельно. Данные перетекают из cat в grep на уровне ядра Linux. Python не участвует в копировании строк, его потребление памяти остается минимальным и константным, независимо от размера файла /var/log/syslog.

Золотое правило закрытия дескрипторов

Приведенный выше код содержит критическую логическую уязвимость, которая является самой частой причиной зависания DevOps-скриптов.

Когда создается p1, Python получает файловый дескриптор для чтения из его stdout. Когда создается p2, этот дескриптор передается ему. Теперь у потока данных p1.stdout есть два потенциальных читателя: процесс p2 (на уровне ОС) и сам интерпретатор Python (через объект p1).

Процесс p2 (grep) будет читать данные до тех пор, пока не получит сигнал EOF. Ядро Linux отправит EOF только тогда, когда все файловые дескрипторы, указывающие на пишущий конец канала, будут закрыты. Если p1 завершит работу, его дескриптор закроется. Но Python все еще держит свою копию дескриптора открытой! В результате grep будет бесконечно ждать новых данных, а скрипт зависнет навсегда.

Чтобы этого избежать, Python должен явно закрыть свою ссылку на p1.stdout сразу после того, как передаст ее второму процессу.

import subprocess

p1 = subprocess.Popen(["cat", "/var/log/syslog"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", "sshd"], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)

# КРИТИЧЕСКИ ВАЖНО: Закрываем дескриптор в Python.
# Теперь только p2 может читать из p1.stdout.
# Когда p1 завершится, p2 корректно получит EOF.
p1.stdout.close()

# Дожидаемся завершения p2 и получаем его вывод
output, errors = p2.communicate()

Метод .communicate() выполняет сразу три задачи:

  1. Читает данные из stdout и stderr процесса до самого конца.
  2. Дожидается завершения процесса.
  3. Возвращает кортеж (stdout_data, stderr_data).

Использование .communicate() вместо прямого вызова p2.wait() и последующего чтения p2.stdout.read() обязательно. Если процесс сгенерирует много данных, а Python попытается сначала дождаться его завершения (wait()), 64-килобайтный буфер ОС переполнится. Процесс заблокируется в попытке записать данные, а Python заблокируется в ожидании завершения процесса. Возникнет классический deadlock (взаимная блокировка). Метод .communicate() использует внутренние механизмы ОС (select/poll) для параллельного чтения потоков, предотвращая переполнение буферов.

Сборка сложных многоступенчатых конвейеров

Логика связывания через Popen масштабируется на любое количество процессов. Главное — строго соблюдать последовательность: создать процесс, передать его stdout следующему, закрыть stdout в Python.

Рассмотрим реалистичную задачу SRE-инженера: необходимо сделать дамп базы данных PostgreSQL, сжать его и зашифровать симметричным ключом перед отправкой в хранилище. Использование временных файлов на диске недопустимо из-за требований безопасности и ограничений I/O.

Bash-эквивалент: pg_dump dbname | gzip -9 | gpg -c --passphrase "secret" > backup.gz.gpg

Реализация на Python:

import subprocess

# 1. Запуск выгрузки базы данных
p_dump = subprocess.Popen(
    ["pg_dump", "production_db"],
    stdout=subprocess.PIPE
)

# 2. Запуск сжатия. Читает из p_dump.stdout
p_gzip = subprocess.Popen(
    ["gzip", "-9"],
    stdin=p_dump.stdout,
    stdout=subprocess.PIPE
)

# Закрываем ссылку на stdout первого процесса
p_dump.stdout.close()

# 3. Запуск шифрования. Читает из p_gzip.stdout
# gpg требует пароль. Мы передадим его через --passphrase-fd 0 (читать пароль из stdin)
# Но stdin уже занят потоком от gzip!
# В таких случаях используют аргументы командной строки или переменные окружения,
# либо gpg позволяет передать пароль через отдельный файловый дескриптор.
# Для упрощения примера используем аргумент (в реальности лучше использовать GPG-агенты).
p_gpg = subprocess.Popen(
    ["gpg", "--symmetric", "--batch", "--passphrase", "SecureKey123"],
    stdin=p_gzip.stdout,
    stdout=subprocess.PIPE
)

# Закрываем ссылку на stdout второго процесса
p_gzip.stdout.close()

# 4. Получаем финальный зашифрованный бинарный поток
encrypted_backup, _ = p_gpg.communicate()

# Записываем результат на диск средствами Python
with open("backup.gz.gpg", "wb") as f:
    f.write(encrypted_backup)

В этой архитектуре данные непрерывно текут от PostgreSQL через архиватор в шифратор. Оперативная память расходуется только на буферы ОС (несколько мегабайт в сумме), даже если размер исходной базы составляет терабайты.

Обработка ошибок в конвейерах

В bash, если первая команда в конвейере завершается с ошибкой (например, pg_dump не смог подключиться к базе), конвейер продолжает работу. Код возврата всего конвейера по умолчанию равен коду возврата последней команды (в нашем примере gpg). Это приводит к созданию пустых или битых бэкапов, так как gpg успешно зашифрует пустой поток от упавшего pg_dump. В bash эта проблема решается директивой set -o pipefail.

В Python при использовании Popen необходимо проверять коды возврата каждого узла конвейера вручную. Свойство .returncode становится доступным только после того, как процесс завершится.

Поскольку метод .communicate() дожидается завершения только последнего процесса (p_gpg), статус предыдущих процессов нужно проверять методом .wait().

# ... (код создания конвейера из предыдущего примера) ...

encrypted_backup, _ = p_gpg.communicate()

# Теперь безопасно проверить статусы всех процессов
dump_status = p_dump.wait()
gzip_status = p_gzip.wait()
gpg_status = p_gpg.returncode # Уже известен после communicate()

if dump_status != 0:
    print(f"Ошибка: pg_dump завершился с кодом {dump_status}")
    # Логика алертинга
elif gzip_status != 0:
    print(f"Ошибка: gzip завершился с кодом {gzip_status}")
elif gpg_status != 0:
    print(f"Ошибка: gpg завершился с кодом {gpg_status}")
else:
    print("Бэкап успешно создан и зашифрован.")

Важно понимать порядок вызовов: .wait() для промежуточных процессов вызывается после .communicate() финального процесса. Если вызвать p_dump.wait() до того, как данные будут вычитаны из конца трубы, конвейер заблокируется. pg_dump не сможет завершиться, пока gzip не прочитает его данные, gzip не прочитает данные, пока gpg не освободит место в своем буфере, а gpg ждет, пока Python вызовет .communicate().

Построение конвейеров через Popen требует дисциплины в управлении ресурсами. Забытый вызов .close() на промежуточном потоке или неправильный порядок ожидания завершения процессов гарантированно приводят к зависаниям (deadlocks), которые крайне сложно отлаживать в production-среде. Однако именно этот подход обеспечивает надежную, масштабируемую и безопасную автоматизацию тяжелых системных задач без утечек памяти и рисков инъекций, характерных для shell=True.

Безопасность и тайм-ауты: предотвращение зависаний и инъекций в системных вызовах

Безопасность и тайм-ауты: предотвращение зависаний и инъекций в системных вызовах

Ночной скрипт резервного копирования запускает утилиту rsync для синхронизации файлов с удалённым сервером. В середине процесса сетевой коммутатор на стороне дата-центра перезагружается. Соединение обрывается без отправки TCP-пакета RST. Утилита rsync не получает уведомления об ошибке и остаётся висеть в оперативной памяти, бесконечно ожидая ответа. Следующей ночью cron запускает новый экземпляр скрипта. Через месяц на сервере скапливаются десятки зависших процессов, которые исчерпывают лимит файловых дескрипторов, и операционная система начинает отказывать в обслуживании легитимным сервисам.

Автоматизация инфраструктуры — это не только запуск команд, но и гарантия их детерминированного завершения. Внешние утилиты могут зависать из-за сетевых проблем, ожидания пользовательского ввода (например, запроса пароля) или состояния взаимной блокировки. Если Python-скрипт делегирует задачу системному вызову, он обязан контролировать время её выполнения и потребляемые ресурсы, а также гарантировать, что передаваемые данные не приведут к выполнению произвольного кода.

Управление временем: аргумент timeout и исключение TimeoutExpired

Самый прямолинейный способ защитить скрипт от бесконечного ожидания — установить жесткий лимит времени на выполнение внешней команды. Функция subprocess.run() принимает аргумент timeout, который задаёт максимально допустимое время работы дочернего процесса в секундах.

Если процесс не завершается за отведённое время, Python прерывает ожидание и генерирует исключение subprocess.TimeoutExpired.

import subprocess
import sys

try:
    # Запуск сетевой утилиты с жестким лимитом в 5 секунд
    result = subprocess.run(
        ["ping", "-c", "10", "8.8.8.8"],
        capture_output=True,
        text=True,
        timeout=5
    )
    print("Успешно выполнено")
except subprocess.TimeoutExpired as e:
    print(f"Процесс {e.cmd} превысил лимит времени ({e.timeout} сек).")
    # В объекте исключения могут остаться частичные данные вывода
    if e.stdout:
        print(f"Успели получить: {e.stdout.decode('utf-8', errors='replace')}")

Механика работы аргумента timeout внутри subprocess.run() скрывает важную деталь. Когда время истекает, Python не просто выбрасывает исключение и идёт дальше. Под капотом интерпретатор отправляет дочернему процессу сигнал SIGKILL (безусловное завершение), затем вызывает метод .wait() для очистки таблицы процессов ОС, и только после этого поднимает исключение TimeoutExpired. Это гарантирует, что непосредственно вызванный процесс будет уничтожен и не превратится в «зомби».

Однако эта встроенная защита имеет критическую уязвимость, когда речь заходит о древовидных структурах процессов.

Проблема осиротевших процессов (Orphans) и группы процессов

Представим, что мы запускаем не бинарный файл напрямую, а shell-скрипт, который внутри себя вызывает другие утилиты.

# Уязвимый паттерн
try:
    subprocess.run(["bash", "backup_wrapper.sh"], timeout=10)
except subprocess.TimeoutExpired:
    print("Тайм-аут резервного копирования")

Допустим, внутри backup_wrapper.sh выполняется архивация tar, которая занимает 30 секунд. На 10-й секунде срабатывает тайм-аут в Python. Интерпретатор отправляет сигнал SIGKILL процессу bash. Оболочка bash мгновенно умирает.

Но процесс tar, порождённый этой оболочкой, ничего не знает о тайм-ауте. Он теряет своего родителя (становится сиротой — orphan process). Ядро Linux автоматически переназначает родителем такого процесса процесс инициализации системы (обычно systemd с PID 1). В результате Python-скрипт рапортует об ошибке тайм-аута и продолжает работу, а tar продолжает нагружать диск и процессор в фоновом режиме.

Чтобы убить не только сам вызванный процесс, но и всех его потомков, необходимо использовать концепцию сессий и групп процессов (Process Groups). В Linux каждый процесс принадлежит к определённой группе. Сигналы можно отправлять не отдельному PID, а целому PGID (Process Group ID).

Для реализации этого паттерна мы отступаем от высокоуровневого subprocess.run() и используем subprocess.Popen с аргументом start_new_session=True. Это заставляет ядро ОС создать новую группу процессов, где лидером становится наш дочерний процесс.

import subprocess
import os
import signal
import time

# Запускаем процесс как лидера новой группы
process = subprocess.Popen(
    ["bash", "backup_wrapper.sh"],
    start_new_session=True
)

try:
    # Ожидаем завершения с тайм-аутом
    process.wait(timeout=10)
except subprocess.TimeoutExpired:
    print("Тайм-аут! Уничтожаем всё дерево процессов...")
    # Получаем ID группы процессов (он совпадает с PID лидера)
    pgid = os.getpgid(process.pid)

    # Отправляем SIGKILL всей группе (обратите внимание на минус перед pgid)
    # В системном вызове kill отрицательное значение означает "отправить группе"
    os.killpg(pgid, signal.SIGKILL)

    # Ожидаем завершения лидера, чтобы избежать появления зомби
    process.wait()

В этом сценарии вызов os.killpg() гарантирует, что сигнал SIGKILL будет доставлен и оболочке bash, и утилите tar, и любым другим подпроцессам, которые они успели породить. Это единственный надёжный способ очистки ресурсов при тайм-аутах сложных инфраструктурных скриптов.

Инъекции команд: за пределами shell=True

Безопасность системных вызовов традиционно сводится к правилу: никогда не использовать shell=True при работе с пользовательским вводом. При передаче аргументов в виде списка строк (например, ["ls", "-l", user_input]), данные передаются напрямую в системный вызов execve, минуя парсер оболочки. Символы вроде ;, | или && теряют свою магическую силу и воспринимаются утилитой просто как часть имени файла.

Однако отказ от shell=True не является абсолютной панацеей. Уязвимости могут возникать на уровне самих утилит, если они обладают собственным синтаксисом исполнения команд.

Рассмотрим утилиту find. Она имеет флаг -exec, который позволяет выполнить команду для каждого найденного файла.

user_input = "{} ; rm -rf /" # Злонамеренный ввод

# shell=False, аргументы переданы списком. Кажется, что это безопасно.
subprocess.run([
    "find", "/var/log", "-name", "*.log", "-exec", "echo", user_input, ";"
])

Несмотря на отсутствие shell=True, утилита find сама парсит аргументы после -exec. Она увидит маркер ;, решит, что команда echo завершена, и попытается интерпретировать следующие аргументы. Это архитектурная инъекция аргументов (Argument Injection), от которой списочная передача спасает не всегда. Защита здесь строится на строгой валидации ввода (например, проверке по белому списку допустимых символов) и использовании безопасных альтернатив (например, обработке путей внутри самого Python через pathlib.Path.rglob()).

Безопасное экранирование через shlex

Бывают ситуации, когда использование оболочки неизбежно. Например, если требуется использовать сложный конвейер из нескольких утилит с перенаправлением потоков ввода-вывода, который нецелесообразно переписывать на чистом Python, или если команда передаётся по SSH на удалённый сервер (где она в любом случае попадёт в shell).

В таких случаях на помощь приходит стандартный модуль shlex (Shell Lexical Analyzer). Его функция shlex.quote() принимает строку и возвращает её безопасную версию, экранированную таким образом, что оболочка POSIX гарантированно воспримет её как единый строковый литерал.

import shlex
import subprocess

# Ввод от пользователя (например, имя файла из веб-формы)
untrusted_filename = "report.pdf; cat /etc/passwd"

# Экранируем строку
safe_filename = shlex.quote(untrusted_filename)
print(f"Экранированная строка: {safe_filename}")
# Вывод: 'report.pdf; cat /etc/passwd'

# Теперь можно безопасно интерполировать её в shell-команду
command = f"scp {safe_filename} user@backup-server:/data/"

subprocess.run(command, shell=True)

Механика shlex.quote() элегантна в своей простоте. В большинстве случаев она просто оборачивает переданную строку в одинарные кавычки. В POSIX-совместимых оболочках (bash, sh, zsh) всё, что находится внутри одинарных кавычек, интерпретируется буквально — никакие подстановки переменных ($VAR), выполнения команд ($()) или разделители (;) не работают. Если же в самом вводе уже есть одинарная кавычка, shlex разрывает строку, вставляет экранированную кавычку \' и снова открывает строку. Строка O'Reilly превратится в 'O'\''Reilly'.

Ограничение ресурсов: предотвращение DoS-атак

Даже если команда безопасна с точки зрения синтаксиса, она может быть опасна с точки зрения потребления ресурсов. Представьте скрипт, который автоматически распаковывает архивы, загружаемые пользователями. Злоумышленник загружает «zip-бомбу» — архив размером в несколько килобайт, который при распаковке разворачивается в петабайты данных. Дочерний процесс unzip начнёт потреблять всю доступную оперативную память и процессорное время, что приведёт к отказу в обслуживании (Denial of Service, DoS).

Для ограничения ресурсов в Linux используется подсистема rlimit. В Python доступ к ней предоставляет модуль resource.

Сложность заключается в том, что лимиты нужно применить не к самому Python-скрипту, а исключительно к дочернему процессу. Для этого в subprocess.Popen существует аргумент preexec_fn. Это функция, которая будет выполнена в дочернем процессе после системного вызова fork(), но до вызова exec().

import subprocess
import resource

def set_limits():
    # Ограничение виртуальной памяти: 500 Мегабайт
    # Формула: мегабайты * 1024 * 1024
    mem_limit = 500 * 1024 * 1024

    # Устанавливаем мягкий и жесткий лимиты (soft, hard)
    resource.setrlimit(resource.RLIMIT_AS, (mem_limit, mem_limit))

    # Ограничение процессорного времени: 10 секунд
    resource.setrlimit(resource.RLIMIT_CPU, (10, 10))

try:
    # Запускаем потенциально опасную утилиту
    process = subprocess.Popen(
        ["unzip", "untrusted_archive.zip", "-d", "/tmp/extract"],
        preexec_fn=set_limits
    )
    process.wait()
except Exception as e:
    print(f"Ошибка при выполнении: {e}")

Если процесс unzip попытается выделить больше 500 МБ памяти, системный вызов malloc (или mmap) на уровне ядра Linux вернёт ошибку, и утилита аварийно завершится с ошибкой нехватки памяти (обычно код возврата 1 или специфичный для утилиты), не затронув остальные сервисы ОС.

Опасность preexec_fn в многопоточной среде

Несмотря на мощь preexec_fn, её использование сопряжено с критическим архитектурным риском при работе в многопоточных (multi-threaded) Python-приложениях.

В POSIX-системах, когда многопоточный процесс вызывает fork(), ядро копирует память родительского процесса, но в дочернем процессе продолжает выполняться только тот поток, который вызвал fork(). Все остальные потоки внезапно исчезают. Если один из исчезнувших потоков в момент форка удерживал блокировку (lock) — например, внутреннюю блокировку модуля logging, блокировку сборщика мусора или GIL — эта блокировка в дочернем процессе останется в состоянии «захвачено» навсегда. Никто её не освободит, потому что поток-владелец не был скопирован.

Если функция set_limits, переданная в preexec_fn, попытается выполнить операцию, требующую этой блокировки (например, выведет лог через logging.info()), дочерний процесс навсегда зависнет в состоянии взаимной блокировки (Deadlock), так и не дойдя до вызова exec().

Поэтому правило использования preexec_fn звучит строго: внутри этой функции можно использовать только безопасные, примитивные системные вызовы. Никакого логирования, никаких сложных вычислений, никакого создания объектов. Вызов resource.setrlimit() безопасен, так как это прямая обёртка над системным вызовом C.

Если инфраструктурный скрипт активно использует потоки (например, ThreadPoolExecutor), более безопасным и современным подходом является использование утилиты prlimit (доступна в современных дистрибутивах Linux) в качестве обёртки прямо в списке аргументов, что позволяет вообще отказаться от preexec_fn:

# Альтернативный безопасный подход без preexec_fn
# Устанавливаем лимит виртуальной памяти (AS) в 500 МБ (байты)
mem_limit = 500 * 1024 * 1024
command = [
    "prlimit",
    f"--as={mem_limit}",
    "unzip",
    "untrusted_archive.zip",
    "-d",
    "/tmp/extract"
]

subprocess.run(command)

Этот подход делегирует настройку лимитов внешней утилите, полностью исключая риски, связанные с поведением fork() в многопоточной среде Python.

Надёжная автоматизация требует параноидального отношения к внешним вызовам. Предполагайте, что сеть оборвётся, утилита зависнет, а пользователь передаст в качестве имени файла вредоносный скрипт. Использование жестких тайм-аутов, изоляция в группы процессов для предотвращения утечек, строгое экранирование ввода и ограничение потребляемых ресурсов превращают хрупкие скрипты в отказоустойчивые SRE-инструменты, способные безопасно функционировать в непредсказуемой среде production-серверов.

Архивация и сжатие данных: использование модулей tarfile, zipfile и shutil

Архивация и сжатие данных: использование модулей tarfile, zipfile и shutil

Переполнение дискового пространства из-за накопившихся логов, дампов баз данных или старых релизов — классическая причина отказа серверов. Текстовые данные, из которых состоит большая часть системной информации, обладают высокой избыточностью и отлично поддаются сжатию. В экосистеме Linux задача упаковки файлов решается ежедневно, и Python предоставляет для этого три встроенных инструмента, каждый из которых работает на своем уровне абстракции.

Высокоуровневая упаковка через shutil

Когда требуется просто взять директорию и превратить её в архив без сложной логики фильтрации, модуль shutil предоставляет наиболее лаконичный интерфейс. Он скрывает под капотом работу с файловыми дескрипторами и потоками, предлагая функции make_archive и unpack_archive.

Функция shutil.make_archive(base_name, format, root_dir) принимает базовое имя создаваемого архива (без расширения), желаемый формат (zip, tar, gztar, bztar, xztar) и путь к директории, которую нужно упаковать.

import shutil
from pathlib import Path

backup_dir = Path("/var/backups/myapp")
backup_dir.mkdir(parents=True, exist_ok=True)

source_dir = Path("/opt/myapp/data")

# Создаст файл /var/backups/myapp/data_backup.tar.gz
archive_path = shutil.make_archive(
    base_name=str(backup_dir / "data_backup"),
    format="gztar",
    root_dir=str(source_dir)
)

Этот подход идеален для быстрых бэкапов перед деплоем. Однако у него есть критический недостаток для сложных сценариев автоматизации: подход «всё или ничего». shutil не позволяет исключить конкретные файлы (например, временные .tmp или сокеты), изменить владельца файлов на лету или упаковать файлы из разных директорий в один архив без предварительного копирования их во временную папку. Для гранулярного контроля необходимо спуститься на уровень ниже.

Классика Linux: модуль tarfile

В мире Unix архивация и сжатие исторически разделены. Утилита tar (Tape Archive) изначально создавалась для последовательной записи файлов на магнитную ленту. Её задача — склеить множество файлов, директорий и их метаданных (права доступа, владельцы, временные метки) в один непрерывный поток байтов. Само по себе это не уменьшает размер данных. Сжатие применяется уже поверх этого потока с помощью алгоритмов вроде gzip или xz.

Модуль tarfile в Python полностью реализует эту логику. Для создания архива используется класс TarFile, который работает как контекстный менеджер. Ключевой момент — строка режима открытия. Она состоит из двух частей, разделенных двоеточием: режим:алгоритм.

  • w — создание архива без сжатия (чистый .tar).
  • w:gz — сжатие через gzip (быстро, средняя степень сжатия, стандарт де-факто).
  • w:bz2 — сжатие через bzip2 (медленнее, сжимает лучше).
  • w:xz — сжатие через алгоритм LZMA (очень медленная упаковка, требует много RAM, но обеспечивает максимальное сжатие; распаковка при этом быстрая).

Гранулярный контроль и фильтрация

Главная сила tarfile раскрывается в методе .add(), который принимает аргумент filter. Этот аргумент ожидает функцию, которая принимает объект TarInfo и возвращает либо измененный объект TarInfo, либо None (если файл нужно пропустить).

Объект TarInfo содержит метаданные файла до того, как они будут записаны в архив. Это позволяет модифицировать права доступа или имена файлов «на лету», не трогая оригиналы на диске.

Рассмотрим скрипт, который архивирует конфигурацию веб-сервера, но исключает скрытые файлы (например, .git) и принудительно устанавливает владельцем всех файлов пользователя root, чтобы не скомпрометировать внутренние UID/GID системы при переносе архива на другой сервер.

import tarfile
from pathlib import Path

def sanitize_tarinfo(tarinfo: tarfile.TarInfo) -> tarfile.TarInfo | None:
    # Исключаем скрытые файлы и директории
    if Path(tarinfo.name).name.startswith("."):
        return None

    # Нормализуем владельца и группу для безопасности
    tarinfo.uid = 0
    tarinfo.gid = 0
    tarinfo.uname = "root"
    tarinfo.gname = "root"

    # Сбрасываем права доступа до безопасных (600 для файлов, 700 для директорий)
    if tarinfo.isdir():
        tarinfo.mode = 0o700
    else:
        tarinfo.mode = 0o600

    return tarinfo

config_dir = Path("/etc/nginx")
archive_path = Path("/backup/nginx_secure.tar.gz")

with tarfile.open(archive_path, "w:gz") as tar:
    tar.add(config_dir, arcname="nginx_config", filter=sanitize_tarinfo)

В этом примере параметр arcname меняет корневую директорию внутри архива. Без него файлы легли бы по абсолютному пути /etc/nginx/..., что крайне неудобно при распаковке. arcname="nginx_config" гарантирует, что при извлечении архива появится аккуратная папка nginx_config, содержащая все данные.

Безопасность извлечения: атака Path Traversal

Распаковка tar-архивов, полученных из недоверенных источников, несет серьезную угрозу безопасности. Поскольку tar-архив хранит пути к файлам в виде обычных строк, злоумышленник может создать архив, где имя файла указано как ../../../../etc/shadow или /root/.ssh/authorized_keys.

Если распаковать такой архив наивным вызовом tar.extractall(), процесс Python с правами суперпользователя послушно запишет файл по абсолютному пути или выйдет за пределы целевой директории по относительным ссылкам, перезаписав критические системные файлы. Эта уязвимость известна как «tarbomb» или Path Traversal.

До версии Python 3.12 разработчикам приходилось вручную валидировать каждый путь перед извлечением. Начиная с Python 3.12, в метод extractall добавлен параметр filter, который определяет политику безопасности.

Для безопасного извлечения следует использовать политику 'data', которая запрещает абсолютные пути, переходы на уровни выше (..) и извлечение специальных файлов устройств (символьных и блочных устройств, сокетов).

import tarfile
from pathlib import Path

archive_path = Path("suspicious_update.tar.gz")
extract_dir = Path("/tmp/update_test")
extract_dir.mkdir(exist_ok=True)

with tarfile.open(archive_path, "r:gz") as tar:
    # В Python 3.12+ это предотвратит перезапись системных файлов
    tar.extractall(path=extract_dir, filter='data')

Если код должен работать на старых версиях Python, валидацию необходимо реализовывать вручную, проверяя, что итоговый абсолютный путь распаковываемого файла начинается с абсолютного пути целевой директории, используя метод Path.is_relative_to().

Кроссплатформенность и потоковая запись: модуль zipfile

Формат ZIP фундаментально отличается от TAR. Если TAR — это сплошной поток данных, который сжимается целиком (solid archive), то ZIP сжимает каждый файл индивидуально, а в конце архива хранит центральный каталог (Central Directory) со списком файлов и смещениями к ним.

Такая архитектура делает ZIP менее эффективным при сжатии тысяч мелких файлов (так как метаданные каждого файла не сжимаются вместе с остальными), но позволяет мгновенно извлекать один конкретный файл из многогигабайтного архива без необходимости распаковывать его с самого начала.

В DevOps-практике zipfile используется в двух основных случаях:

  1. Подготовка артефактов для передачи в Windows-окружения (где поддержка .tar.gz исторически слабее).
  2. Формирование архивов в оперативной памяти без касания диска.

Для создания архива используется класс ZipFile. В отличие от tarfile, алгоритм сжатия здесь передается отдельным аргументом compression. По умолчанию используется ZIP_STORED (без сжатия). Для реального сжатия используют ZIP_DEFLATED.

import zipfile
from pathlib import Path

log_dir = Path("/var/log/myapp")
zip_path = Path("/backup/logs.zip")

with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zf:
    for log_file in log_dir.glob("*.log"):
        # arcname обязателен, иначе запишется полный абсолютный путь
        zf.write(log_file, arcname=log_file.name)

Запись напрямую из памяти через writestr

Часто в скриптах автоматизации данные генерируются на лету: например, скрипт опрашивает API облачного провайдера, формирует JSON-отчет и должен отправить его в архив. Сохранять этот JSON во временный файл на диске, архивировать, а затем удалять временный файл — это лишние операции ввода-вывода (I/O) и износ SSD.

Модуль zipfile позволяет писать строковые или байтовые данные напрямую в архив с помощью метода writestr.

import zipfile
import json
from datetime import datetime

# Имитация данных, полученных из API
infrastructure_state = {
    "timestamp": datetime.now().isoformat(),
    "instances": [
        {"id": "i-12345", "status": "running"},
        {"id": "i-67890", "status": "stopped"}
    ]
}

archive_path = Path("/backup/infra_state.zip")

with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
    # Сериализуем словарь в JSON-строку
    json_data = json.dumps(infrastructure_state, indent=2)

    # Записываем строку напрямую в архив под заданным именем
    zf.writestr("state_report.json", json_data)

Этот паттерн критически важен при построении бессерверных функций (например, AWS Lambda), где файловая система доступна только для чтения или сильно ограничена в объеме, а вся обработка должна происходить в оперативной памяти.

Управление памятью при работе с большими файлами

При архивации гигантских дампов баз данных (десятки и сотни гигабайт) возникает вопрос потребления оперативной памяти.

Методы tar.add() и zf.write() реализованы безопасно: они читают исходные файлы с диска небольшими чанками (кусками) и сразу передают их в компрессор, а затем пишут в выходной файл. Потребление памяти остается стабильным и минимальным (обычно несколько мегабайт на буферы), независимо от размера архивируемого файла.

Однако, если используется метод writestr в zipfile, вся передаваемая строка или последовательность байтов должна поместиться в оперативной памяти. Если попытаться передать в writestr результат чтения 10-гигабайтного лога целиком, процесс будет убит операционной системой (OOM Killer) из-за нехватки RAM. Для потоковой обработки гигантских данных без промежуточных файлов часто приходится комбинировать модули сжатия (например, gzip или lzma) с конвейерами процессов (subprocess.Popen), передавая данные через стандартные потоки ввода-вывода на уровне ядра ОС, минуя память Python-скрипта.

Выбор инструмента зависит от контекста. shutil экономит время на написание кода для тривиальных задач. tarfile необходим для сохранения специфичных метаданных Linux и работы с классическими архивами. zipfile выручает при необходимости точечного извлечения файлов и интеграции с системами вне экосистемы Unix.

Сигналы и жизненный цикл процесса: корректное завершение скриптов автоматизации

Представьте ситуацию: ваш Python-скрипт выполняет резервное копирование базы данных размером 50 ГБ. Он уже скачал данные, сжал их и начал загрузку в S3-хранилище. В этот момент администратор замечает высокую нагрузку на сервер и нажимает Ctrl+C в терминале, или systemd решает перезапустить службу при обновлении конфигурации. Скрипт мгновенно обрывается. В результате на диске остается битый временный архив, сетевое соединение с S3 разорвано некорректно, а блокировочный файл (lock-file), запрещающий параллельный запуск бэкапа, не удален. При следующем запуске по расписанию скрипт откажется работать, посчитав, что предыдущая копия еще создается.

Проблема заключается в том, что по умолчанию операционная система завершает процессы жестко. Для создания надежных инструментов автоматизации скрипт должен уметь перехватывать запросы на завершение, корректно закрывать дескрипторы, удалять временные файлы и только после этого отдавать управление ядру.

Природа POSIX-сигналов

В Linux сигналы — это механизм межпроцессного взаимодействия (IPC), позволяющий операционной системе или одним процессам асинхронно уведомлять другие процессы о событиях. Сигнал прерывает нормальный ход выполнения программы, заставляя ее либо выполнить специальную функцию (обработчик), либо завершиться.

Каждый сигнал имеет числовой идентификатор и стандартное имя. В контексте DevOps и автоматизации критически важно различать четыре основных сигнала:

  • SIGINT (2) — сигнал прерывания с клавиатуры (Interrupt). Генерируется терминалом при нажатии Ctrl+C. По умолчанию завершает процесс.
  • SIGTERM (15) — сигнал завершения (Termination). Это вежливая просьба операционной системы завершить работу. Именно его отправляет утилита kill по умолчанию, и именно с него systemd начинает остановку служб. Процесс может перехватить этот сигнал, чтобы выполнить очистку ресурсов.
  • SIGKILL (9) — сигнал безусловного уничтожения. Генерируется командой kill -9 или механизмом ядра OOM Killer (при нехватке оперативной памяти). Этот сигнал невозможно перехватить, проигнорировать или заблокировать. Ядро просто выгружает процесс из памяти.
  • SIGHUP (1) — сигнал обрыва терминала (Hangup). Исторически означал потерю модемного соединения. Сегодня часто используется демонами (например, Nginx) как команда на перечитывание конфигурационных файлов без остановки самого процесса.

Разница между SIGTERM и SIGKILL определяет архитектуру отказоустойчивых служб. Когда система инициализации (например, systemd) останавливает сервис, она отправляет SIGTERM и запускает таймер. Если процесс не завершается за отведенное время (обычно Twait=90T_{wait} = 90 секунд), система отправляет SIGKILL, уничтожая его принудительно. Задача разработчика — уложиться в интервал TwaitT_{wait}, корректно свернув работу.

Перехват сигналов в Python

Для работы с сигналами в стандартной библиотеке Python предусмотрен модуль signal. Он позволяет подменить стандартное поведение программы при получении определенного сигнала на пользовательскую функцию.

Функция-обработчик сигнала в Python всегда должна принимать ровно два аргумента:

  1. signum — номер полученного сигнала (целое число).
  2. frame — объект текущего кадра стека выполнения (stack frame) в момент прерывания. Позволяет узнать, в какой именно строке кода находилась программа, когда пришел сигнал (полезно для отладки).

Базовая регистрация обработчика выглядит так:

import signal
import time
import os

def graceful_shutdown(signum, frame):
    print(f"\n[PID {os.getpid()}] Получен сигнал {signum}. Начинаем очистку...")
    # Здесь должна быть логика очистки
    print("Очистка завершена. Выход.")
    exit(0)

# Регистрируем обработчик для SIGINT (Ctrl+C) и SIGTERM (systemctl stop)
signal.signal(signal.SIGINT, graceful_shutdown)
signal.signal(signal.SIGTERM, graceful_shutdown)

print(f"Скрипт запущен с PID {os.getpid()}. Нажмите Ctrl+C для выхода.")
while True:
    print("Работаем...")
    time.sleep(2)

Если запустить этот код и нажать Ctrl+C, вместо стандартного исключения KeyboardInterrupt и пугающего трейсбека (traceback) на экране появится аккуратное сообщение об очистке.

Паттерны корректного завершения (Graceful Shutdown)

Реализация логики внутри функции graceful_shutdown требует осторожности. Сигналы асинхронны — они могут прервать выполнение скрипта в абсолютно любой момент: посреди записи в файл, во время сетевого запроса или при обновлении глобального словаря.

Если прямо внутри обработчика сигнала начать закрывать файлы или удалять данные, можно столкнуться с состоянием гонки (race condition) или повредить структуры данных. Существует два безопасных архитектурных паттерна для обработки сигналов.

Паттерн 1: Использование глобального флага состояния

Этот подход идеален для скриптов, работающих в цикле (например, обработчиков очередей сообщений, парсеров логов в реальном времени или демонов мониторинга).

Идея заключается в том, что обработчик сигнала не выполняет очистку сам. Он лишь изменяет глобальную переменную (флаг), сигнализируя основному циклу, что пора закругляться. Основной цикл проверяет этот флаг на каждой итерации.

import signal
import time

# Глобальный флаг состояния
shutdown_requested = False

def request_shutdown(signum, frame):
    global shutdown_requested
    print(f"\n[Сигнал {signum}] Запрошена остановка. Завершаем текущую задачу...")
    shutdown_requested = True

signal.signal(signal.SIGINT, request_shutdown)
signal.signal(signal.SIGTERM, request_shutdown)

def process_logs():
    items_processed = 0
    # Основной цикл проверяет флаг
    while not shutdown_requested:
        # Эмуляция полезной работы (например, парсинг 1000 строк лога)
        print("Обработка порции данных...")
        time.sleep(3)
        items_processed += 1

    # Сюда мы попадаем только после того, как текущая итерация полностью завершилась
    print(f"Цикл остановлен. Обработано порций: {items_processed}")
    print("Сохранение прогресса в базу данных...")
    # Безопасная очистка ресурсов
    print("Выход.")

if __name__ == "__main__":
    process_logs()

Преимущество флага: текущая итерация цикла (например, транзакция в БД или запись чанка в файл) завершается штатно, данные не повреждаются. Недостаток: если внутри цикла есть долгая блокирующая операция (например, time.sleep(60) или ожидание ответа от зависшего API), скрипт отреагирует на сигнал только после завершения этой операции.

Паттерн 2: Генерация исключения SystemExit

Если скрипт выполняет длинную линейную последовательность действий (например, сложный бэкап с множеством этапов), расставлять проверки флага if shutdown_requested: после каждой строки нецелесообразно.

В этом случае обработчик сигнала может выбросить встроенное исключение SystemExit. Это заставит Python начать процедуру размотки стека (stack unwinding), при которой гарантированно выполнятся все блоки finally и методы __exit__ контекстных менеджеров with.

import signal
import time
import sys

def exit_gracefully(signum, frame):
    print(f"\nПолучен сигнал {signum}. Инициируем экстренное завершение.")
    # Генерируем исключение, которое перехватывается на верхнем уровне
    sys.exit(143) # 128 + 15 (SIGTERM) - стандартный код выхода в Linux при SIGTERM

signal.signal(signal.SIGINT, exit_gracefully)
signal.signal(signal.SIGTERM, exit_gracefully)

def run_backup():
    # Контекстный менеджер гарантирует закрытие файла при SystemExit
    with open("/tmp/backup.lock", "w") as lock_file:
        lock_file.write("locked")
        print("Lock-файл создан.")

        try:
            print("Архивация файлов (может занять долгое время)...")
            time.sleep(10) # Эмуляция долгой работы
            print("Архивация завершена.")
        finally:
            # Этот блок выполнится даже если time.sleep будет прерван сигналом
            print("Удаление временных файлов...")

run_backup()

В этом сценарии, если нажать Ctrl+C во время time.sleep(10), функция exit_gracefully вызовет sys.exit(). Исполнение прервется, но блок finally отработает, а контекстный менеджер with open(...) закроет файл.

Маршрутизация сигналов в дочерние процессы

В автоматизации часто используется модуль subprocess для вызова системных утилит (tar, rsync, pg_dump). Здесь возникает архитектурная проблема, связанная с деревом процессов.

Если ваш Python-скрипт запустил rsync через subprocess.Popen, и в этот момент Python-скрипт получает SIGTERM от systemd, ядро Linux доставит сигнал только родительскому процессу (Python). Дочерний процесс rsync ничего не узнает о завершении родителя. Когда Python завершится, rsync станет процессом-сиротой (orphan), будет усыновлен процессом инициализации (PID 1) и продолжит потреблять сеть и диск в фоновом режиме.

Чтобы избежать утечки процессов, Python-скрипт обязан перехватить сигнал и явно перенаправить его своим дочерним процессам перед собственным завершением.

import signal
import subprocess
import sys
import time

# Глобальная ссылка на дочерний процесс
child_process = None

def handle_sigterm(signum, frame):
    global child_process
    print("\nПолучен сигнал завершения. Останавливаем дочерние процессы...")
    if child_process is not None and child_process.poll() is None:
        # Отправляем SIGTERM дочернему процессу
        child_process.terminate()
        try:
            # Ждем завершения дочернего процесса максимум 5 секунд
            child_process.wait(timeout=5)
            print("Дочерний процесс корректно завершен.")
        except subprocess.TimeoutExpired:
            # Если утилита зависла, добиваем ее SIGKILL
            print("Дочерний процесс не отвечает. Отправляем SIGKILL.")
            child_process.kill()
    sys.exit(0)

signal.signal(signal.SIGTERM, handle_sigterm)
signal.signal(signal.SIGINT, handle_sigterm)

print("Запуск долгого системного вызова (ping)...")
# Используем Popen для асинхронного запуска
child_process = subprocess.Popen(["ping", "8.8.8.8"])

# Ждем завершения дочернего процесса в штатном режиме
child_process.wait()

В этом примере обработчик сигнала проверяет, запущен ли дочерний процесс (child_process.poll() is None). Если да, он использует метод .terminate() (отправляет SIGTERM утилите ping), дает ей 5 секунд на корректное завершение через .wait(timeout=5), и если она не реагирует — использует .kill() (отправляет SIGKILL). Только после полной остановки дочернего процесса родительский скрипт завершает работу.

Защита от неперехватываемого SIGKILL: атомарные операции

Несмотря на все усилия по обработке SIGTERM, скрипт всегда может быть убит мгновенно сигналом SIGKILL, например, если серверу не хватило оперативной памяти. Поскольку SIGKILL перехватить невозможно, архитектура скрипта должна быть спроектирована так, чтобы внезапная смерть процесса не приводила к повреждению данных.

Этот принцип тесно связан с концепцией идемпотентности, но на уровне файловой системы он реализуется через атомарные операции. Атомарная операция — это действие, которое выполняется целиком и мгновенно с точки зрения наблюдателя; оно не может быть прервано посередине.

Самая частая ошибка при генерации отчетов или бэкапов — запись данных напрямую в целевой файл:

# ОПАСНО: Если скрипт убьют SIGKILL во время записи,
# файл /var/backups/data.tar.gz останется битым.
with open("/var/backups/data.tar.gz", "wb") as f:
    f.write(generate_huge_archive())

Правильный подход (Write-Rename Pattern) заключается в записи данных во временный файл (желательно в той же файловой системе) с последующим переименованием. Системный вызов переименования (rename в C, os.rename или Path.rename в Python) в пределах одной файловой системы POSIX гарантированно атомарен.

import os
import tempfile
from pathlib import Path

target_file = Path("/var/backups/data.tar.gz")
backup_dir = target_file.parent

# 1. Создаем временный файл в той же директории, что и целевой
# delete=False гарантирует, что файл не исчезнет при закрытии
with tempfile.NamedTemporaryFile(dir=backup_dir, delete=False, suffix=".tmp") as tmp:
    temp_path = Path(tmp.name)
    try:
        print("Пишем данные во временный файл...")
        # Эмуляция долгой записи
        tmp.write(b"huge binary data...")
        # Обязательно сбрасываем буферы ОС на диск перед переименованием
        tmp.flush()
        os.fsync(tmp.fileno())
    except Exception:
        # Если произошла обычная ошибка, удаляем мусор
        temp_path.unlink(missing_ok=True)
        raise

# 2. Атомарное переименование
# Если SIGKILL придет ДО этой строки - останется только .tmp файл (целевой не поврежден).
# Если SIGKILL придет ПОСЛЕ - целевой файл уже полностью готов.
# В процессе самого переименования убить процесс невозможно - это атомарная операция ядра.
temp_path.rename(target_file)
print("Бэкап успешно создан.")

Использование os.fsync() здесь критически важно. Метод .flush() сбрасывает данные из внутренних буферов Python в буферы ядра ОС. Но ядро может отложить физическую запись на диск. Вызов fsync принудительно заставляет контроллер диска записать данные. Если сделать атомарное переименование до физической записи, и в этот момент пропадет питание сервера, после перезагрузки файл data.tar.gz может оказаться пустым, так как метаданные (имя файла) обновились, а сами блоки данных не успели записаться.

Понимание жизненного цикла процесса, умение договариваться с операционной системой через сигналы и использование атомарных операций превращают хрупкие скрипты в надежные системные инструменты, способные пережить перезагрузки, нехватку ресурсов и вмешательство администратора без потери данных.

Практикум: Разработка отказоустойчивого скрипта для ротации бэкапов в Linux

Практикум: Разработка отказоустойчивого скрипта для ротации бэкапов в Linux

В 3:00 ночи на сервере базы данных заканчивается свободное место. Причина банальна: скрипт резервного копирования, который исправно работал полгода, упал с ошибкой сети на середине процесса. Он оставил после себя временный файл размером 50 гигабайт. На следующую ночь ситуация повторилась. Встроенная логика удаления старых архивов не сработала, потому что скрипт прерывался до того, как доходил до этапа очистки. В результате DevOps-инженер просыпается от критического алерта системы мониторинга.

Написание скрипта, который просто создает архив — задача тривиальная. Создание отказоустойчивого инструмента, который корректно обрабатывает зависания, не оставляет мусора при внезапном убийстве процесса (SIGKILL/SIGTERM), защищен от параллельного запуска и гарантирует целостность данных — это инженерная задача, требующая интеграции всех подсистем, изученных нами ранее.

В этом практикуме мы спроектируем и напишем production-ready утилиту для резервного копирования PostgreSQL, объединив pathlib, subprocess, управление метаданными и перехват сигналов.

Архитектура отказоустойчивого инструмента

Скрипт автоматизации, претендующий на надежность, должен опираться на три фундаментальных принципа:

  1. Эксклюзивность выполнения (Locking). Если вчерашний бэкап завис и все еще выполняется, сегодняшний запуск cron не должен начать новый процесс. Параллельное чтение базы и запись на диск приведут к деградации производительности (I/O starvation) и исчерпанию места.
  2. Атомарность результата (Write-Rename). Файл резервной копии должен появляться в целевой директории только тогда, когда он полностью и успешно сформирован. Никакая другая утилита (например, скрипт отправки бэкапов в S3) не должна увидеть недописанный файл.
  3. Изоляция сбоев (Cleanup). Любая ошибка (нехватка памяти, обрыв pipe, сигнал от systemd) должна перехватываться, а промежуточные временные файлы — безусловно удаляться.

Рассмотрим реализацию каждого из этих этапов.

Этап 1: Блокировка параллельного запуска

Самый наивный способ предотвратить двойной запуск — проверять наличие процесса в памяти через ps aux | grep script.py. Этот подход хрупок: он подвержен состояниям гонки (race conditions) и может ложно срабатывать на процессы с похожими именами.

Профессиональный стандарт в Linux — использование системного вызова flock (file lock) на уровне файловых дескрипторов. В Python это реализуется через встроенный модуль fcntl.

Механика работы fcntl.flock элегантна: блокировка привязывается не к процессу, а к файловому дескриптору. Если наш скрипт падает с ошибкой или его жестко убивают через kill -9 (SIGKILL), операционная система автоматически закрывает все файловые дескрипторы убитого процесса. В этот момент ядро Linux само снимает блокировку. Нам не нужно писать сложную логику очистки PID-файлов после сбоев.

import fcntl
import sys
from pathlib import Path

LOCK_FILE = Path("/var/run/pg_backup.lock")

def acquire_lock():
    # Открываем файл на запись (создаем, если нет)
    lock_fd = open(LOCK_FILE, "w")
    try:
        # LOCK_EX - эксклюзивная блокировка
        # LOCK_NB - не блокировать выполнение (non-blocking), вернуть ошибку сразу
        fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return lock_fd
    except BlockingIOError:
        print("Ошибка: Другой экземпляр скрипта уже запущен.", file=sys.stderr)
        sys.exit(1)

Сохранение объекта lock_fd в памяти критически важно. Если переменная будет уничтожена сборщиком мусора, файл закроется, и блокировка спадет до завершения работы скрипта.

Этап 2: Конвейер данных и атомарная запись

Наша задача — снять дамп базы данных утилитой pg_dump и сжать его утилитой gzip. Как мы выяснили при изучении конвейеров, сохранять промежуточный дамп на диск неэффективно. Мы свяжем stdout первого процесса со stdin второго на уровне ядра.

Для обеспечения атомарности мы применим паттерн, который обсуждали в контексте сигналов:

  1. Пишем данные в скрытый временный файл (например, .backup_temp.gz).
  2. Принудительно сбрасываем буферы ОС на физический диск (os.fsync).
  3. Переименовываем временный файл в финальное имя (backup_2023.gz).

Системный вызов rename в POSIX-совместимых системах является атомарным, если оба файла находятся на одной файловой системе. Это значит, что для любого стороннего наблюдателя старое имя исчезнет, а новое появится ровно в один такт времени. Состояния «файл наполовину переименован» не существует.

Реализуем функцию создания бэкапа, интегрировав обработку ошибок subprocess:

import subprocess
import os

def create_backup(target_path: Path):
    # Формируем имя временного файла в той же директории
    temp_path = target_path.with_name(f".tmp_{target_path.name}")

    # Открываем временный файл для записи бинарных данных
    with open(temp_path, "wb") as out_file:
        try:
            # Запускаем pg_dump
            p_dump = subprocess.Popen(
                ["pg_dump", "-U", "postgres", "main_db"],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )

            # Запускаем gzip, читая из stdout pg_dump
            p_gzip = subprocess.Popen(
                ["gzip", "-9", "-c"],
                stdin=p_dump.stdout,
                stdout=out_file,
                stderr=subprocess.PIPE
            )

            # Закрываем дескриптор в родительском процессе (защита от deadlock)
            p_dump.stdout.close()

            # Ждем завершения сжатия и читаем возможные ошибки
            _, stderr_data = p_gzip.communicate()

            # Проверяем коды возврата обоих процессов
            if p_dump.wait() != 0 or p_gzip.returncode != 0:
                raise RuntimeError(f"Сбой конвейера: {stderr_data.decode(errors='replace')}")

            # Гарантируем физическую запись на диск
            out_file.flush()
            os.fsync(out_file.fileno())

        except Exception as e:
            # В случае любой ошибки удаляем недописанный мусор
            temp_path.unlink(missing_ok=True)
            raise RuntimeError(f"Ошибка создания бэкапа: {e}")

    # Если мы дошли сюда, файл цел и на диске. Делаем атомарный rename.
    temp_path.rename(target_path)

Обратите внимание на блок except. Если pg_dump упадет из-за потери связи с базой, или gzip завершится с ошибкой из-за нехватки памяти, исключение будет перехвачено, временный файл .tmp_... будет удален через unlink(missing_ok=True), и только после этого ошибка пробросится выше. Диск останется чистым.

Этап 3: Стратегии ротации (Age-based vs Count-based)

Когда новый бэкап успешно создан, необходимо очистить старые, чтобы не исчерпать емкость хранилища. Существует два концептуальных подхода к ротации, и выбор между ними определяет поведение системы в аварийных ситуациях.

Подход 1: На основе возраста (Age-based)

Мы вычисляем разницу между текущим временем и временем модификации файла. Если она больше заданного порога Δt\Delta t, файл удаляется. Математически условие удаления выглядит так: TnowTmtime>N×24×60×60T_{now} - T_{mtime} > N \times 24 \times 60 \times 60 где NN — количество дней.

Уязвимость: Представьте, что база данных сломалась, и pg_dump падает с ошибкой каждый день. Скрипт не создает новые бэкапы, но исправно запускает логику ротации. Через NN дней скрипт удалит последний успешный бэкап, потому что он стал слишком старым. Вы останетесь вообще без резервных копий.

Подход 2: На основе количества (Count-based)

Мы собираем все существующие бэкапы, сортируем их по времени создания (от новых к старым) и оставляем ровно KK штук. Все, что выходит за пределы этого количества (индексы среза от KK до конца), удаляется.

Преимущество: Если новые бэкапы перестают создаваться, ротация просто не происходит. У вас всегда будет KK последних успешных копий, даже если они были сделаны месяц назад. Для критичных систем этот подход значительно безопаснее.

Реализуем логику Count-based ротации с использованием pathlib:

def rotate_backups(backup_dir: Path, keep_count: int = 7):
    # Находим все файлы бэкапов по маске
    backups = list(backup_dir.glob("db_*.sql.gz"))

    if len(backups) <= keep_count:
        return # Удалять нечего

    # Сортируем файлы по времени модификации (st_mtime) по убыванию (новые сверху)
    # Используем stat().st_mtime для получения timestamp
    backups.sort(key=lambda p: p.stat().st_mtime, reverse=True)

    # Файлы, которые нужно удалить - это все элементы списка после индекса keep_count
    files_to_delete = backups[keep_count:]

    for old_file in files_to_delete:
        try:
            old_file.unlink()
            print(f"Ротация: удален старый бэкап {old_file.name}")
        except OSError as e:
            print(f"Ошибка при удалении {old_file.name}: {e}", file=sys.stderr)

Эта функция идемпотентна. Если запустить её несколько раз подряд без создания новых файлов, она ничего не сломает и не удалит лишнего.

Этап 4: Интеграция перехвата сигналов

Мы уже реализовали удаление временного файла в блоке except при сбое конвейера. Но что произойдет, если во время сжатия администратор выполнит systemctl stop backup.service?

Процесс получит сигнал SIGTERM. По умолчанию Python немедленно прерывает выполнение, игнорируя блоки except Exception (так как сигнал не является стандартным исключением Python, наследуемым от Exception). Временный файл останется на диске.

Чтобы этого избежать, мы применим изученный ранее паттерн Graceful Shutdown, конвертируя SIGTERM во встроенное исключение SystemExit. Исключение SystemExit корректно раскручивает стек вызовов и заставляет сработать все блоки finally и контекстные менеджеры with.

Добавим обработчик сигналов в точку входа нашего скрипта.

Финальная сборка SRE-инструмента

Теперь объединим все компоненты в единый монолитный скрипт, добавив базовое логирование и правильную структуру.

#!/usr/bin/env python3
import os
import sys
import time
import fcntl
import signal
import subprocess
from pathlib import Path
from datetime import datetime

BACKUP_DIR = Path("/var/backups/postgres")
LOCK_FILE = Path("/var/run/pg_backup.lock")
KEEP_COUNT = 7

def signal_handler(signum, frame):
    """Конвертирует SIGTERM в SystemExit для корректной очистки ресурсов."""
    print(f"\n[!] Получен сигнал {signum}, инициирована безопасная остановка...", file=sys.stderr)
    sys.exit(143) # Стандартный код выхода для SIGTERM (128 + 15)

def acquire_lock():
    lock_fd = open(LOCK_FILE, "w")
    try:
        fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return lock_fd
    except BlockingIOError:
        print("Скрипт уже выполняется. Выход.", file=sys.stderr)
        sys.exit(0) # Выходим с 0, чтобы cron не слал алерты на нормальное перекрытие

def create_backup(target_path: Path):
    temp_path = target_path.with_name(f".tmp_{target_path.name}")

    # Блок try-finally гарантирует очистку даже при SystemExit (SIGTERM)
    try:
        with open(temp_path, "wb") as out_file:
            p_dump = subprocess.Popen(
                ["pg_dump", "-U", "postgres", "main_db"],
                stdout=subprocess.PIPE, stderr=subprocess.PIPE
            )
            p_gzip = subprocess.Popen(
                ["gzip", "-9", "-c"],
                stdin=p_dump.stdout, stdout=out_file, stderr=subprocess.PIPE
            )
            p_dump.stdout.close()

            _, stderr_data = p_gzip.communicate()

            if p_dump.wait() != 0 or p_gzip.returncode != 0:
                raise RuntimeError(f"Сбой: {stderr_data.decode('utf-8', 'replace')}")

            out_file.flush()
            os.fsync(out_file.fileno())

        temp_path.rename(target_path)
        print(f"Бэкап успешно создан: {target_path.name}")

    finally:
        # Если файл остался (ошибка или сигнал) - удаляем
        if temp_path.exists():
            temp_path.unlink()
            print(f"Очищен временный файл: {temp_path.name}", file=sys.stderr)

def rotate_backups(backup_dir: Path, keep_count: int):
    backups = list(backup_dir.glob("db_*.sql.gz"))
    if len(backups) <= keep_count:
        return

    backups.sort(key=lambda p: p.stat().st_mtime, reverse=True)
    for old_file in backups[keep_count:]:
        old_file.unlink(missing_ok=True)
        print(f"Ротация: удален {old_file.name}")

def main():
    # 1. Перехват сигналов
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

    # 2. Подготовка директории
    BACKUP_DIR.mkdir(parents=True, exist_ok=True)

    # 3. Блокировка
    lock_fd = acquire_lock()

    try:
        # 4. Генерация имени файла на основе текущего времени
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        target_file = BACKUP_DIR / f"db_{timestamp}.sql.gz"

        # 5. Выполнение задач
        create_backup(target_file)
        rotate_backups(BACKUP_DIR, KEEP_COUNT)

    except Exception as e:
        print(f"Критическая ошибка: {e}", file=sys.stderr)
        sys.exit(1)

    finally:
        # 6. Снятие блокировки (произойдет и автоматически при выходе, но явное лучше)
        fcntl.flock(lock_fd, fcntl.LOCK_UN)
        lock_fd.close()

if __name__ == "__main__":
    main()

Нюансы эксплуатации и граничные случаи

Разработанный скрипт решает большинство проблем, но при внедрении в реальную инфраструктуру стоит учитывать несколько дополнительных факторов.

Во-первых, фрагментация времени модификации. Мы используем st_mtime для сортировки файлов при ротации. Если администратор вручную скопирует старый бэкап в директорию с помощью команды cp (без флага -p, сохраняющего атрибуты), st_mtime этого файла обновится до текущего момента. Логика ротации посчитает этот старый файл самым новым и удалит настоящий свежий бэкап. Для защиты от этого некоторые инженеры парсят дату прямо из имени файла (db_20231025_...), используя регулярные выражения, вместо доверия метаданным файловой системы.

Во-вторых, исчерпание inodes. Логика BACKUP_DIR.mkdir(parents=True, exist_ok=True) гарантирует наличие папки. Но если на диске закончились индексные дескрипторы (inodes), даже при наличии свободных гигабайт open(temp_path, "wb") выбросит OSError: [Errno 28] No space left on device. Наш блок except корректно перехватит это, но бэкап создан не будет. Мониторинг должен отслеживать не только объем диска, но и метрику IUse% (которую можно получить системным вызовом os.statvfs, разбиравшимся в первом курсе).

В-третьих, если процесс pg_dump зависнет на уровне сетевого сокета (например, база данных перестала отвечать на TCP-пакеты, но не закрыла соединение), p_gzip.communicate() будет ждать вечно. В идеальной реализации конвейер должен оборачиваться в таймаут (аргумент timeout на уровне subprocess или использование signal.alarm), чтобы принудительно убивать зависшие дочерние процессы и освобождать блокировку flock.

Интеграция системных вызовов, строгий контроль файловых дескрипторов и понимание того, как ядро Linux работает с процессами, отличает хрупкие наброски от надежных инструментов автоматизации. Написанный нами код не просто решает задачу — он спроектирован с ожиданием того, что инфраструктура вокруг него неизбежно будет ломаться.