mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
import logging
|
||
import requests
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
SEVERITY_ORDER = ['ok', 'info', 'warning', 'critical']
|
||
SEVERITY_ICONS = {'critical': '🔴', 'warning': '🟡', 'ok': '🟢', 'info': 'ℹ️'}
|
||
|
||
|
||
def notify_telegram(profile, run, findings):
|
||
"""
|
||
Send a Telegram message if any finding meets or exceeds notify_on_severity.
|
||
"""
|
||
threshold_idx = SEVERITY_ORDER.index(profile.notify_on_severity)
|
||
flagged = [f for f in findings if SEVERITY_ORDER.index(f.severity) >= threshold_idx]
|
||
|
||
if not flagged:
|
||
return
|
||
|
||
finished_str = run.finished_at.strftime('%Y-%m-%d %H:%M') if run.finished_at else 'unknown'
|
||
lines = [
|
||
f'🔒 *NetScan Alert* — {profile.name}',
|
||
f'Run \\#{run.pk} finished at {finished_str}',
|
||
f'Summary: {run.summary}',
|
||
'',
|
||
]
|
||
|
||
for f in flagged[:10]:
|
||
icon = SEVERITY_ICONS.get(f.severity, '•')
|
||
lines.append(f'{icon} *{f.title}*\n {f.detail[:120]}')
|
||
|
||
if len(flagged) > 10:
|
||
lines.append(f'_...and {len(flagged) - 10} more findings_')
|
||
|
||
text = '\n'.join(lines)
|
||
url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage'
|
||
resp = requests.post(url, json={
|
||
'chat_id': profile.telegram_chat_id,
|
||
'text': text,
|
||
'parse_mode': 'Markdown',
|
||
}, timeout=10)
|
||
resp.raise_for_status()
|
||
logger.info(f'Telegram notification sent for run #{run.pk}')
|
||
|
||
|
||
def send_test_telegram(profile) -> dict:
|
||
"""Send a test message. Returns {'ok': True} or {'ok': False, 'error': str}."""
|
||
if not profile.telegram_bot_token or not profile.telegram_chat_id:
|
||
return {'ok': False, 'error': 'Telegram bot token or chat ID not configured.'}
|
||
try:
|
||
url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage'
|
||
resp = requests.post(url, json={
|
||
'chat_id': profile.telegram_chat_id,
|
||
'text': f'✅ *NetScan test message* from profile _{profile.name}_. Notifications are working.',
|
||
'parse_mode': 'Markdown',
|
||
}, timeout=10)
|
||
resp.raise_for_status()
|
||
return {'ok': True}
|
||
except Exception as e:
|
||
return {'ok': False, 'error': str(e)}
|