Webhook у Harbor или как я оповещения о пушах docker images нашей команды делал часть — 1

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

И вот решил я разработать инструмент который слал бы сообщения о пуша\удаления и проверках в наш harbor. Оповещения должны были слаться в телеграм чаты, в том числе в разные топики для удобства координации наших команд. В общем создали мы топик "Releases and Updates" и начал я думать как реализовать, первым делом подумал про API, но как я понял оно там несколько для других целей. Проще всего было это организовать через webhook'и, оно к слову в harbor из коробки, как раз для тех целей которые были нужны.

Вот следующая последовательность действий для настройки этого добра:

1. Откройте Telegram и найдите пользователя BotFather.

2. Создайте нового бота, отправив команду /newbot и следуйте инструкциям.

3. Запишите ваш токен бота, который вы получите в результате.

4. Добавьте бота в канал\группу.

5. У бота должны быть админские права, чтобы он мог слать оповещения.

На этом пока отложим telegram и перейдем к настройкам harbor.

1. Зайдите в harbor.

2. Перейдите в настройки репозитория.

3. Настройте новый Webhook, указав URL вашего сервера, который будет принимать уведомления (например, http://your-server-ip:5000/webhook).

Убедитесь, что вы выбрали необходимое событие, push.

Далее переходим к тонкостям отправки сообщений в telegram, а именно к отправки сообщений в конкретный топик. А нюанс вот в чем:

• Чат id можно скопировал через ссылку на сообщение по типу - https://t.me/c/123xxxx567/17/16

  (Вот расшифровка ссылки https://t.me/c/{chat_id}/{message_thread_id}/{message_id})

• Для использования chat_id в API надо добавить -100, получите -100123xxxx567

Теперь переходим к разработке, я решил использовать flask для этих целей.

Примерный вид POST сообщения о пуше от harbor -

INFO:werkzeug:127.0.0.1 - - [30/Apr/2025 10:21:33] "POST /webhook HTTP/1.1" 200 -

вот тело сообщения:

{
  "type": "PUSH_ARTIFACT",
  "occur_at": 1746008487,
  "operator": "sergey.akhmineev",
  "event_data": {
    "resources": [
      {
        "digest": "sha256:9sdfgsdfgsdfgsdfgsgdfgsfgsdfgsdfgsdfgsdfgsdfgsfdgs",
        "tag": "1.9.0",
        "resource_url": "your-server-ip/your-repo/your-images:1.9.0"
      }
    ],
    "repository": {
      "date_created": 1733465413,
      "name": "your-images",
      "namespace": "your-repo",
      "repo_full_name": "your-repo/your-images",
      "repo_type": "public"
    }
  }
}

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

Переходим к разработке, вот минимум для отправки сообщения в telegram

# Извлечение токена Telegram и Chat ID из конфигурации

TELEGRAM_API_URL = f'https://api.telegram.org/bot{conf_dict["telegram"]["bot_token"]}/sendMessage'
CHAT_ID = conf_dict["telegram"]["chat_id"]
MESSAGE_THREAD_ID = conf_dict.get("telegram", {}).get("message_thread_id")

# Отправка сообщения 

def send_message_to_telegram(message):
    payload = {
        'chat_id': CHAT_ID,
        'text': message,
        'parse_mode': 'Markdown',  # Для форматирования сообщения
    }
    if MESSAGE_THREAD_ID:
        payload['message_thread_id'] = MESSAGE_THREAD_ID

    try:
        response = requests.post(TELEGRAM_API_URL, json=payload)
        response.raise_for_status()  # Проверка на HTTP-ошибки
        logging.info("Сообщение успешно отправлено в Telegram.")
    except requests.exceptions.RequestException as e:
        logging.error(f"Ошибка при отправке сообщения: {str(e)}")

А вот и само приложение для того чтобы слушать webhook:

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    logging.info(f"Получены данные вебхука: {data}")

    if data and data.get('type') == 'PUSH_ARTIFACT':
        event_data = data.get('event_data', {})
        repository_info = event_data.get('repository', {})
        resources = event_data.get('resources', [])

        repo_full_name = repository_info.get('repo_full_name', 'неизвестный репозиторий')
        repo_full_name = escape_markdown(repo_full_name)  # Экранирование

        if not resources:
            logging.warning("Нет ресурсов для обработки.")
            return 'No resources to process', 200
        # Формарование сообщения для отправки
        messages = []
        for resource in resources:
            tag = escape_markdown(resource.get('tag', 'неизвестный тег'))  # Экранирование тега
            resource_url = resource.get('resource_url', 'неизвестный URL')
            operator = escape_markdown(data.get('operator', 'неизвестный'))  # Экранирование оператора
            message = (
                f"📦 *Новый пуш в Harbor*\n"
                f"*Репозиторий:* {repo_full_name}\n"
                f"*Тег:* {tag}\n"
                f"*URL:* `{resource_url}`\n"
                f"*Автор пуша:* {operator}"
            )
            messages.append(message)

        # Отправка всех сообщений последовательно
        for msg in messages:
            logging.info(f"Отправка сообщения: {msg}")
            send_message_to_telegram(msg)
    else:
        logging.warning("Получены некорректные данные вебхука или отсутствует тип 'PUSH_ARTIFACT'.")

    return 'OK', 200

В целом остается сделать экранирование спецсимволов -

def escape_markdown(text):
    """
    Экранирует специальные символы Markdown в тексте.
    """
    escape_chars = ['\\', '_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '!']
    for char in escape_chars:
        text = text.replace(char, f'\\{char}')
    return text

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