150 lines
4.5 KiB
Python
150 lines
4.5 KiB
Python
import os
|
|
import re
|
|
import requests
|
|
import smtplib
|
|
import socket
|
|
import ssl
|
|
import random
|
|
import time
|
|
|
|
from datetime import datetime
|
|
from email.mime.text import MIMEText
|
|
|
|
DEBUG_LOG = os.environ["DEBUG_LOG"].lower() == "true"
|
|
|
|
IRC_SERVER = os.environ["IRC_SERVER"]
|
|
IRC_PORT = int(os.environ["IRC_PORT"])
|
|
USE_TLS = os.environ["IRC_USE_TLS"].lower() == "true"
|
|
IRC_NICK = os.environ["IRC_NICK"]
|
|
IRC_USER = IRC_NICK
|
|
IRC_REALNAME = IRC_NICK
|
|
IRC_CHANNEL = os.environ["IRC_CHANNEL"]
|
|
|
|
TOPIC_REGEX = os.environ["TOPIC_REGEX"]
|
|
|
|
SMTP_SERVER = os.environ["SMTP_SERVER"]
|
|
SMTP_PORT = int(os.environ["SMTP_PORT"])
|
|
SMTP_USER = os.environ["SMTP_USER"]
|
|
SMTP_PASS = os.environ["SMTP_PASS"]
|
|
EMAIL_TO = os.environ["EMAIL_TO"]
|
|
|
|
MIN_WAIT_TIME = int(os.environ["MIN_WAIT_TIME"])
|
|
MAX_WAIT_TIME = int(os.environ["MAX_WAIT_TIME"])
|
|
|
|
MIN_SLEEP_TIME = int(os.environ["MIN_SLEEP_TIME"])
|
|
MAX_SLEEP_TIME = int(os.environ["MAX_SLEEP_TIME"])
|
|
|
|
topic_pattern = re.compile(TOPIC_REGEX)
|
|
|
|
def log(level, msg):
|
|
if level == "." and not DEBUG_LOG:
|
|
return
|
|
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S UTC")
|
|
print(f"[{ts}] [{level}] {msg}", flush=True)
|
|
|
|
def sendmail(subject, message):
|
|
try:
|
|
msg = MIMEText(message)
|
|
msg["Subject"] = subject
|
|
msg["From"] = SMTP_USER
|
|
msg["To"] = EMAIL_TO
|
|
|
|
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
|
|
server.starttls()
|
|
server.login(SMTP_USER, SMTP_PASS)
|
|
server.send_message(msg)
|
|
log("+", f"Email sent to {EMAIL_TO}")
|
|
except Exception as e:
|
|
log("!", f"Failed to send error email: {e}")
|
|
|
|
def connect_to_channel_and_check_topic():
|
|
log("*", f"Connecting to {IRC_SERVER}:{IRC_PORT} {"with TLS" if USE_TLS else "plain"}")
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(60)
|
|
if USE_TLS:
|
|
context = ssl.create_default_context()
|
|
sock = context.wrap_socket(sock, server_hostname=IRC_SERVER)
|
|
|
|
sock.connect((IRC_SERVER, IRC_PORT))
|
|
|
|
# Send NICK and USER
|
|
nick = f"NICK {IRC_NICK}\r\n".encode()
|
|
log(".", nick)
|
|
sock.sendall(nick)
|
|
|
|
irc_user = f"USER {IRC_USER} 0 * :{IRC_REALNAME}\r\n".encode()
|
|
log(".", irc_user)
|
|
sock.sendall(irc_user)
|
|
|
|
topic = None
|
|
joined = False
|
|
|
|
while True:
|
|
data = sock.recv(4096)
|
|
if not data:
|
|
break
|
|
lines = data.decode(errors="ignore").split("\r\n")
|
|
for line in lines:
|
|
if not line:
|
|
continue
|
|
log(".", f"< {line}")
|
|
|
|
# Respond to PING
|
|
if line.startswith("PING"):
|
|
resp = line.replace("PING", "PONG")
|
|
pong = f"{resp}\r\n".encode()
|
|
sock.sendall(pong)
|
|
log(".", pong)
|
|
|
|
# Check for end of MOTD or welcome to join channel
|
|
if " 001 " in line and not joined:
|
|
# Join channel
|
|
join = f"JOIN {IRC_CHANNEL}\r\n".encode()
|
|
sock.sendall(join)
|
|
log(".", join)
|
|
joined = True
|
|
|
|
# Capture topic sent by server upon joining
|
|
# IRC servers send: :server 332 nick #channel :topic here
|
|
if f" 332 {IRC_NICK} " in line:
|
|
parts = line.split(":", 2)
|
|
if len(parts) == 3:
|
|
topic = parts[2]
|
|
log("*", f"Channel topic on join: {topic}")
|
|
|
|
if topic_pattern.search(topic):
|
|
log("+", f"Found topic, sending mail to {EMAIL_TO}")
|
|
sendmail("IRC TOPIC BOT: Found topic", topic)
|
|
|
|
# If we have the topic, we can disconnect after a random delay
|
|
if topic is not None:
|
|
wait_time = random.randint(MIN_WAIT_TIME, MAX_WAIT_TIME)
|
|
log("*", f"Waiting for {wait_time} seconds before disconnecting")
|
|
time.sleep(wait_time)
|
|
bye = b"QUIT :Bye!\r\n"
|
|
sock.sendall(bye)
|
|
log(".", bye)
|
|
sock.close()
|
|
log("*", "Disconnected")
|
|
return
|
|
|
|
def main():
|
|
try:
|
|
res = requests.get("https://api.ipify.org", timeout=10)
|
|
res.raise_for_status()
|
|
ip = res.text.strip()
|
|
log("*", f"Public IP is {ip}")
|
|
except Exception as e:
|
|
log("!", f"Could not get public IP, {e}")
|
|
|
|
while True:
|
|
connect_to_channel_and_check_topic()
|
|
sleep_time = random.randint(MIN_SLEEP_TIME, MAX_SLEEP_TIME)
|
|
log("*", f"Sleeping for {sleep_time} seconds before reconnecting")
|
|
time.sleep(sleep_time)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|