Как перестать сливать ключи клиентов в ChatGPT: пишем умный буфер обмена для n8n и Python

Каждый интегратор сегодня пишет код и дебажит 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, базовая регулярка посыпалась. Секреты там лежат максимально хитро:

  1. pinData: n8n кэширует реальные ответы от БД прямо в JSON. Это гигабайты мусора и чувствительных данных.

  2. Типы данных: chatId часто идет как число (-100123456). Если заменить его на строковую заглушку HIDDEN, а потом вернуть как строку, n8n выдаст ошибку типизации.

  3. Секреты внутри строк: В нодах 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 = "33[91m"
GREEN = "33[92m"
YELLOW = "33[93m"
RESET = "33[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_.-]+:[^@nrs]+@[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("rn", "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, обсудим архитектуру на языке цифр и сроков.

Автор: chernyaevi

Источник

Оставить комментарий