Каждый интегратор сегодня пишет код и дебажит JSON-воркфлоу с помощью нейросетей. И каждый хоть раз ловил холодный пот, когда понимал, что только что скормил в чат ИИ боевой токен от базы данных клиента или API-ключ продакшена.
В этой статье я покажу, как мы решили проблему утечки данных (NDA) при работе со стеком n8n и Python. Мы напишем легковесный фоновый демон, который на лету перехватывает буфер обмена Mac/Linux, вырезает из кода все секреты, токены и IP-адреса, а когда ИИ возвращает исправленный код — автоматически подставляет реальные ключи обратно из локального сейфа. Итог: 100% безопасность коммерческой тайны, сохранение типов данных для n8n и ноль рутины.
Просто жесть и ручная цензура
Когда ты собираешь сложную автоматизацию в n8n, твой воркфлоу — это огромный JSON. Внутри этого JSON зашито всё: строки подключения к PostgreSQL, токены Telegram-ботов, вебхуки, реальные ID чатов и кэш ответов от API.
Раньше процесс дебага выглядел так: копируешь ноду, вставляешь в блокнот, руками меняешь пароль на ***, отправляешь в ChatGPT. ИИ возвращает код, ты копируешь его обратно в n8n, забываешь вернуть реальный пароль — и весь прод падает. Вот такая вот байда. Тратить на это время — непозволительная роскошь, а слить данные клиента — прямое нарушение NDA.
Почесал репу и решил: нужен скрипт, который будет делать это за меня.
Разбор ошибок: Эволюция решения и наши «Грабли»
Написать регулярку для замены пароля — дело пяти минут. Но когда мы начали внедрять это в реальный рабочий процесс, мы собрали все возможные грабли.
Грабли 1. От ручного скрипта к фоновому демону
Сначала мы сделали скрипт, который нужно было дергать руками (python vault.py hide). Это оказалось жутко неудобно.
Решение: Переписали логику на бесконечный цикл while True с time.sleep(0.5) и проверкой pyperclip.paste(). Скрипт стал висеть в фоне, потребляя <0.1% CPU, и реагировать только на изменение содержимого буфера.
Грабли 2. Ошибка вставки кода (Обрезание строк)
При попытке вставить длинный сгенерированный код через редактор nano в терминале, строки обрезались. Python выплевывал SyntaxError: EOL while scanning string literal.
Решение: Ушли от текстовых редакторов. Использовали команду cat << 'EOF' > file.py, которая пишет код в файл напрямую из терминала байт в байт.
Грабли 3. Баг с регулярками и f-строками (Python-специфика)
Я честно в афиге был с этого бага. Регулярка rf"...([^'\"]{4,})" в упор не находила секреты.
Причина: Из-за префикса f (f-строка) Python воспринял {4,} не как синтаксис регулярного выражения («от 4 символов»), а как Python-кортеж! Он превратил его в текст (4,). Скрипт буквально искал символы (4,) в паролях.
Решение: Экранировали фигурные скобки двойным написанием: {{4,}}.
Грабли 4. Специфика n8n (Самое жесткое мясо)
Когда пошли реальные JSON-воркфлоу из n8n, базовая регулярка посыпалась. Секреты там лежат максимально хитро:
pinData: n8n кэширует реальные ответы от БД прямо в JSON. Это гигабайты мусора и чувствительных данных.
Типы данных: chatId часто идет как число (-100123456). Если заменить его на строковую заглушку HIDDEN, а потом вернуть как строку, n8n выдаст ошибку типизации.
Секреты внутри строк: В нодах Code (поле jsCode) токены и IP зашиты прямо внутри куска JavaScript-кода, который сам по себе является строкой внутри JSON.
Решение (Гибридный парсер):
Скрипт научился отличать обычный текст от JSON. Если это JSON, он делает рекурсивный обход дерева:
Жестко обнуляет pinData ➔ {}.
Ищет конкретные ключи (chatId, webhookId) и маскирует их, запоминая тип данных (int, bool, str), чтобы при восстановлении вернуть число числом.
Текстовые значения дополнительно прогоняет через пачку регулярок на IP, email, токены TG и строки подключения к БД.
Грабли 5. Баг «Уроборос» (Скрипт сожрал сам себя)
При попытке скопировать финальный код самого скрипта, старый процесс, висящий в фоне, увидел в коде слова KEY и PASSWORD и заменил их на заглушки прямо в буфере обмена. В файл записался битый код.
Решение: Дави спокуху. Просто останавливаем старый процесс (Ctrl+C) перед копированием нового кода.
Инженерное решение: Финальный код
Вот итоговый, отполированный гибридный парсер, который решает все вышеописанные задачи.
Как установить с нуля:
codeBash
mkdir ~/clipboard_tool && cd ~/clipboard_tool python3 -m venv venv source venv/bin/activate pip install pyperclip
Сам скрипт (копировать целиком и вставить в терминал):
codeBash
Запуск:
codeBash
cat << 'EOF' > clipboard_monitor.py import re import json import os import time import pyperclip VAULT_FILE = "secrets_vault.json" # Кэш для предотвращения повторного скрытия только что восстановленного текста RECENTLY_RESTORED = [] # ANSI-цвета RED = "\033[91m" GREEN = "\033[92m" YELLOW = "\033[93m" RESET = "\033[0m" # Список чувствительных ключей в JSON SENSITIVE_KEYS = { "chatid", "chat_id", "webhookid", "webhook_id", "documentid", "document_id", "spreadsheetid", "spreadsheet_id", "sheetname", "sheet_name", "cachedresulturl", "bottoken", "bot_token", "userid", "user_id", "channelid", "channel_id", "phone", "email", "password", "passwd", "secret", "private_key", "privatekey" } # Регулярные выражения для поиска секретов IP_PATTERN = r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b" EMAIL_PATTERN = r"\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b" PHONE_PATTERN = r"\b(?:\+?7|8)[\s-]?\(?\d{3}\)?[\s-]?\d{3}[\s-]?\d{2}[\s-]?\d{2}\b" TG_BOT_TOKEN_PATTERN = r"\b[0-9]+:[a-zA-Z0-9_-]{35}\b" DB_CONN_PATTERN = r"(?i)(mongodb\+srv|postgres|postgresql|mysql|mongodb)://[a-zA-Z0-9_.-]+:[^@\n\r\s]+@[a-zA-Z0-9_.-]+(?::\d+)?/[a-zA-Z0-9_.-]*" WEBHOOK_URL_PATTERN = r"https?://[a-zA-Z0-9.:-]+/webhook(?:-test)?/[a-zA-Z0-9-]+" TOKEN_PATTERN = r"\b(?![0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b)[a-zA-Z0-9_-]{24,64}\b" def load_vault(): if os.path.exists(VAULT_FILE): with open(VAULT_FILE, "r", encoding="utf-8") as f: try: return json.load(f) except json.JSONDecodeError: return {} return {} def save_vault(vault): with open(VAULT_FILE, "w", encoding="utf-8") as f: json.dump(vault, f, ensure_ascii=False, indent=4) def normalize(text): """Приводит переносы строк к единому стандарту и убирает лишние пробелы""" return text.replace("\r\n", "\n").strip() def redact_value(key, val, vault): val_str = str(val) if val_str.startswith("__HIDDEN_"): return val placeholder = f"__HIDDEN_{key.upper()}_{abs(hash(val_str)) % 10000}__" vault[placeholder] = { "value": val, "type": type(val).__name__ } return placeholder def redact_text_secrets(text, vault): if not isinstance(text, str): return text redacted = text def apply_redaction(pattern, label, current_text): for m in re.finditer(pattern, current_text): val = m.group(0) if val.startswith("__HIDDEN_"): continue placeholder = f"__HIDDEN_{label.upper()}_{abs(hash(val)) % 10000}__" vault[placeholder] = {"value": val, "type": "str"} current_text = current_text.replace(val, placeholder) return current_text redacted = apply_redaction(DB_CONN_PATTERN, "DB_CONN", redacted) redacted = apply_redaction(TG_BOT_TOKEN_PATTERN, "TG_TOKEN", redacted) redacted = apply_redaction(WEBHOOK_URL_PATTERN, "WEBHOOK_URL", redacted) redacted = apply_redaction(IP_PATTERN, "IP_ADDR", redacted) redacted = apply_redaction(EMAIL_PATTERN, "EMAIL", redacted) redacted = apply_redaction(PHONE_PATTERN, "PHONE", redacted) redacted = apply_redaction(TOKEN_PATTERN, "TOKEN", redacted) return redacted def redact_json_recursive(data, vault, stats): if isinstance(data, dict): new_dict = {} for k, v in data.items(): if k == "pinData": new_dict[k] = {} stats["pinData_cleared"] = True continue k_lower = k.lower() if k_lower in SENSITIVE_KEYS and v: new_dict[k] = redact_value(k, v, vault) stats["keys_redacted"] += 1 elif isinstance(v, str): redacted_str = redact_text_secrets(v, vault) if redacted_str != v: stats["text_secrets_redacted"] += 1 new_dict[k] = redacted_str else: new_dict[k] = redact_json_recursive(v, vault, stats) return new_dict elif isinstance(data, list): return [redact_json_recursive(item, vault, stats) for item in data] return data def restore_json_recursive(data, vault, stats): """Рекурсивно восстанавливает секреты (регистронезависимо)""" if isinstance(data, dict): new_dict = {} for k, v in data.items(): if isinstance(v, str): v_lower = v.lower() matched_placeholder = None for placeholder in vault.keys(): if placeholder.lower() == v_lower: matched_placeholder = placeholder break if matched_placeholder: entry = vault[matched_placeholder] real_val = entry["value"] if entry["type"] == "int": new_dict[k] = int(real_val) elif entry["type"] == "float": new_dict[k] = float(real_val) elif entry["type"] == "bool": new_dict[k] = bool(real_val) else: new_dict[k] = real_val stats["restored"] += 1 else: new_dict[k] = restore_text_secrets(v, vault, stats) else: new_dict[k] = restore_json_recursive(v, vault, stats) return new_dict elif isinstance(data, list): return [restore_json_recursive(item, vault, stats) for item in data] return data def restore_text_secrets(text, vault, stats): """Восстанавливает секреты в тексте (регистронезависимо)""" restored = text for placeholder, entry in vault.items(): pattern = re.escape(placeholder) if re.search(pattern, restored, re.IGNORECASE): restored = re.sub(pattern, lambda m: str(entry["value"]), restored, flags=re.IGNORECASE) stats["restored"] += 1 return restored def process_text(text): global RECENTLY_RESTORED vault = load_vault() normalized_text = normalize(text) # 1. Если этот текст мы только что восстановили — игнорируем его, чтобы не зациклиться if normalized_text in RECENTLY_RESTORED: return text, None # 2. Проверка на ВОССТАНОВЛЕНИЕ (регистронезависимая) text_lower = text.lower() has_placeholders = any(p.lower() in text_lower for p in vault.keys()) if has_placeholders and vault: stats = {"restored": 0} try: data = json.loads(text) restored_data = restore_json_recursive(data, vault, stats) new_text = json.dumps(restored_data, ensure_ascii=False, indent=2) msg = f"{GREEN}[Восстановление (JSON)]{RESET} Вернул {stats['restored']} секрет(ов) с сохранением типов." except json.JSONDecodeError: new_text = restore_text_secrets(text, vault, stats) msg = f"{GREEN}[Восстановление (Текст)]{RESET} Вернул {stats['restored']} секрет(ов)." # Добавляем в список недавно восстановленных RECENTLY_RESTORED.append(normalize(new_text)) if len(RECENTLY_RESTORED) > 10: RECENTLY_RESTORED.pop(0) return new_text, msg # 3. Проверка на СКРЫТИЕ stats = {"keys_redacted": 0, "text_secrets_redacted": 0, "pinData_cleared": False} try: data = json.loads(text) redacted_data = redact_json_recursive(data, vault, stats) new_text = json.dumps(redacted_data, ensure_ascii=False, indent=2) details = [] if stats["keys_redacted"] > 0: details.append(f"ключей: {stats['keys_redacted']}") if stats["text_secrets_redacted"] > 0: details.append(f"в коде: {stats['text_secrets_redacted']}") if stats["pinData_cleared"]: details.append("очищен pinData") if details: save_vault(vault) return new_text, f"{YELLOW}[Скрытие (JSON)]{RESET} Обработан JSON ({', '.join(details)})." except json.JSONDecodeError: new_text = redact_text_secrets(text, vault) if new_text != text: save_vault(vault) return new_text, f"{YELLOW}[Скрытие (Текст)]{RESET} Скрыты секреты по регуляркам." return text, None def main(): print(f"{GREEN}=== Гибридный монитор буфера обмена запущен ==={RESET}") last_text = "" while True: try: current_text = pyperclip.paste() if normalize(current_text) != normalize(last_text) and current_text.strip(): new_text, action_message = process_text(current_text) if action_message: pyperclip.copy(new_text) last_text = new_text print(action_message) else: last_text = current_text time.sleep(0.5) except KeyboardInterrupt: print(f"\n{YELLOW}Остановлен.{RESET}") break except Exception: time.sleep(0.5) if __name__ == "__main__": main() EOF
Точка Б и профит
100% безопасность: Ни один токен, IP-адрес или пароль клиента больше не улетает на сервера OpenAI.
Скорость: Я просто жму Cmd+C в n8n, вставляю код в ChatGPT, получаю ответ, копирую его, и скрипт сам возвращает все ключи на место.
Стабильность: Благодаря сохранению типов данных (int/str), воркфлоу n8n не ломаются при обратном импорте.
P.S. В эпоху ИИ-ассистентов безопасность данных — это не паранойя, а базовая гигиена. Этот скрипт спас нас от десятков случайных сливов клиентских доступов.
Если вашему бизнесу нужна автоматизация на n8n, разработка парсеров или интеграция с нейросетями, при которой ваши коммерческие данные и ключи гарантированно не утекут в сеть — пишите мне в Telegram, обсудим архитектуру на языке цифр и сроков.
