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

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

https://github.com/sergey-akhmineev/webhook-harbor-telegram