from alerts.utils import plural_sectors def temperature(device, value=None): if value is not None: return f"\u26c3 {device:<8} {value}\u00b0C" return f"\u26c3 {device:<8}" def bad_blocks(device, value=None): if value is not None: return f"\u26c3 {device:<8} {value} {plural_sectors(value)}" return f"\u26c3 {device:<8}" def smart(device, value=None): return f"\u26c3 {device:<8}" def read_errors(device, value=None): if value is not None: return f"\u26c3 {device:<8} {value}" return f"\u26c3 {device:<8}" def zfs_pool(device, value=None): return f"\U0001F3DB \u041e\u043b\u0438\u043c\u043f ({device})" def backup_outdated(device, value=None): return f"\U0001F4E6 {device:<8}" def backup_success(info): lines = [] for b in info.get("backups", []): lines.append(f"\U0001F4E6 {b['backup_id']}") lines.append(f"\U0001F5C3 {b['size']}") if b.get("duration"): lines.append(f"\u23F1 {b['duration']}") lines.append("") lines.append(f"\u23F0 {info.get('time', 'unknown')}") if info.get("total_size"): lines.append(f"\U0001F5C3 \u041e\u0431\u0449\u0438\u0439 \u0440\u0430\u0437\u043c\u0435\u0440: {info['total_size']}") return "\n".join(lines) def backup_failed(info): lines = [] for b in info.get("backups", []): icon = "\u2705" if b["status"] == "ok" else "\u274C" lines.append(f"{icon} {b['backup_id']}") lines.append(f"\U0001F5C3 {b['size']}") lines.append("") lines.append(f"\u23F0 {info.get('time', 'unknown')}") if info.get("total_size"): lines.append(f"\U0001F5C3 \u041e\u0431\u0449\u0438\u0439 \u0440\u0430\u0437\u043c\u0435\u0440: {info['total_size']}") return "\n".join(lines) def crowdsec_ban(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") scope = alert_info.get("scope", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F4CD \u0426\u0435\u043b\u044c: {scope}:{value}") lines.append(f"\u26A1 \u0414\u0435\u0439\u0441\u0442\u0432\u0438\u0435: \u0417\u0430\u0431\u043b\u043e\u043a\u0438\u0440\u043e\u0432\u0430\u043d") return "\n".join(lines) def crowdsec_alert(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") scope = alert_info.get("scope", "") lines.append(f"\U0001F5A5 Host: {host}") if scope and value: lines.append(f"\U0001F4CD \u0426\u0435\u043b\u044c: {scope}:{value}") return "\n".join(lines) def crowdsec_mass_attack(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F4CA \u041a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u0431\u0430\u043d\u043e\u0432: {value}") lines.append(f"\u26A1 \u0421\u0442\u0430\u0442\u0443\u0441: \u041a\u0440\u0438\u0442\u0438\u0447\u0435\u0441\u043a\u0430\u044f \u0441\u0438\u0442\u0443\u0430\u0446\u0438\u044f") return "\n".join(lines) def crowdsec_service_disabled(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F6E1 \u0421\u0438\u0441\u0442\u0435\u043c\u0430: {value}") lines.append(f"\u26A1 \u0421\u0442\u0430\u0442\u0443\u0441: \u041e\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d\u0430") return "\n".join(lines) def crowdsec_sudo(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") reason = alert_info.get("reason", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F464 \u041f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u0435\u043b\u044c: {value}") lines.append(f"\U0001F4DD \u041a\u043e\u043c\u0430\u043d\u0434\u0430: {reason}") return "\n".join(lines) def crowdsec_mass_attack(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F4CA \u041a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u0431\u0430\u043d\u043e\u0432: {value}") lines.append(f"\u26A1 \u0421\u0442\u0430\u0442\u0443\u0441: \u041a\u0440\u0438\u0442\u0438\u0447\u0435\u0441\u043a\u0430\u044f \u0441\u0438\u0442\u0443\u0430\u0446\u0438\u044f") return "\n".join(lines) def crowdsec_file_change(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F5C2\uFE0F \u0424\u0430\u0439\u043b: {value}") return "\n".join(lines) def crowdsec_package(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F9F0 \u041f\u0430\u043a\u0435\u0442: {value}") return "\n".join(lines) def crowdsec_service(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F6E0 \u0421\u0435\u0440\u0432\u0438\u0441: {value}") return "\n".join(lines) def crowdsec_connection(alert_info): lines = [] host = alert_info.get("host", "unknown") value = alert_info.get("value", "") lines.append(f"\U0001F5A5 Host: {host}") lines.append(f"\U0001F310 \u041f\u043e\u0434\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u0435: {value}") return "\n".join(lines) FORMATTERS = { "temperature": temperature, "bad_blocks": bad_blocks, "smart": smart, "read_errors": read_errors, "zfs_pool": zfs_pool, "backup_outdated": backup_outdated, "backup_success": backup_success, "backup_failed": backup_failed, "crowdsec_ban": crowdsec_ban, "crowdsec_alert": crowdsec_alert, "crowdsec_mass_attack": crowdsec_mass_attack, "crowdsec_service_disabled": crowdsec_service_disabled, "crowdsec_sudo": crowdsec_sudo, "crowdsec_mass_attack": crowdsec_mass_attack, "crowdsec_file_change": crowdsec_file_change, "crowdsec_package": crowdsec_package, "crowdsec_service": crowdsec_service, "crowdsec_connection": crowdsec_connection, } def get_formatter(name): return FORMATTERS.get(name)