lego-monitoring/send_login_alert.py

import asyncio
import os
import socket
import sys

from alerting import alerts
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 = os.ttyname(sys.stdout.fileno())
        is_local = True
    else:
        is_local = False

    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}, see logs",
                severity=Severity.CRITICAL,
            )
        )
        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)
    await alerts.send_alert(alert)


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