lego-monitoring/send_login_alert.py

import asyncio
import logging
import os
import socket
import sys
import traceback

from alerting import alerts
from alerting.delayed import send_alert_delayed
from alerting.enum import AlertType, Severity
from misc.config import get_config


async def main():
    check_config = get_config().checks.login

    try:
        from_where = os.environ["SSH_CLIENT"].split()[0]
    except:
        from_where = "localhost"
        is_local = True
    else:
        is_local = False

    if not is_local and len(sys.argv) > 1 and sys.argv[1] == "local-only":
        return

    try:
        actual_user = os.environ["SUDO_USER"]
    except Exception as exc:
        await alerts.send_alert(
            alerts.Alert(
                alert_type=AlertType.ERROR,
                message=f"Failed to determine username for login from {from_where}: {repr(exc)}, see logs",
                severity=Severity.CRITICAL,
            )
        )
        logging.error(traceback.format_exc())
        return

    if not is_local:
        rdns_result = socket.getnameinfo((from_where, 0), 0)[0]
        message = f"Login from {from_where} as {actual_user} on {check_config.hostname}"
        html_message = f"Login from <code>{from_where}</code> ({rdns_result}) as {actual_user} on <code>{check_config.hostname}</code>"
    else:
        message = f"Login from {from_where} as {actual_user} on {check_config.hostname}"
        html_message = f"Login from {from_where} as {actual_user} on <code>{check_config.hostname}</code>"

    alert = alerts.Alert(alert_type=AlertType.LOGIN, message=message, severity=Severity.INFO, html_message=html_message)
    send_alert_delayed(alert)


if __name__ == "__main__":
    asyncio.run(main())