225 lines
6.3 KiB
Python
225 lines
6.3 KiB
Python
#!/usr/bin/env python
|
|
# meshtastic irc relay - developed by acidvegas in python (https://git.acid.vegas/meshtastic)
|
|
|
|
import argparse
|
|
import asyncio
|
|
import logging
|
|
import ssl
|
|
import time
|
|
|
|
# 0xEF576MkXA3aEURbCfNn6p0FfZdua4I
|
|
|
|
# Formatting Control Characters / Color Codes
|
|
bold = '\x02'
|
|
italic = '\x1D'
|
|
underline = '\x1F'
|
|
reverse = '\x16'
|
|
reset = '\x0f'
|
|
white = '00'
|
|
black = '01'
|
|
blue = '02'
|
|
green = '03'
|
|
red = '04'
|
|
brown = '05'
|
|
purple = '06'
|
|
orange = '07'
|
|
yellow = '08'
|
|
light_green = '09'
|
|
cyan = '10'
|
|
light_cyan = '11'
|
|
light_blue = '12'
|
|
pink = '13'
|
|
grey = '14'
|
|
light_grey = '15'
|
|
|
|
|
|
# Logging Configuration
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(funcName)s - %(message)s')
|
|
|
|
|
|
def color(msg: str, foreground: str, background: str = None) -> str:
|
|
'''
|
|
Color a string with the specified foreground and background colors.
|
|
|
|
:param msg: The string to color.
|
|
:param foreground: The foreground color to use.
|
|
:param background: The background color to use.
|
|
'''
|
|
|
|
return f'\x03{foreground},{background}{msg}{reset}' if background else f'\x03{foreground}{msg}{reset}'
|
|
|
|
|
|
class Bot():
|
|
def __init__(self):
|
|
self.nickname = 'MESHTASTIC'
|
|
self.reader = None
|
|
self.writer = None
|
|
self.last = time.time()
|
|
|
|
|
|
async def action(self, chan: str, msg: str):
|
|
'''
|
|
Send an ACTION to the IRC server.
|
|
|
|
:param chan: The channel to send the ACTION to.
|
|
:param msg: The message to send to the channel.
|
|
'''
|
|
|
|
await self.sendmsg(chan, f'\x01ACTION {msg}\x01')
|
|
|
|
|
|
async def raw(self, data: str):
|
|
'''
|
|
Send raw data to the IRC server.
|
|
|
|
:param data: The raw data to send to the IRC server. (512 bytes max including crlf)
|
|
'''
|
|
|
|
await self.writer.write(data[:510].encode('utf-8') + b'\r\n')
|
|
|
|
|
|
async def sendmsg(self, target: str, msg: str):
|
|
'''
|
|
Send a PRIVMSG to the IRC server.
|
|
|
|
:param target: The target to send the PRIVMSG to. (channel or user)
|
|
:param msg: The message to send to the target.
|
|
'''
|
|
|
|
await self.raw(f'PRIVMSG {target} :{msg}')
|
|
|
|
|
|
async def connect(self):
|
|
'''Connect to the IRC server.'''
|
|
|
|
while True:
|
|
try:
|
|
options = {
|
|
'host' : args.server,
|
|
'port' : args.port,
|
|
'limit' : 1024,
|
|
'ssl' : ssl._create_unverified_context() if args.ssl else None, # TODO: Do not use the args variable here
|
|
'family' : 2, # AF_INET = 2, AF_INET6 = 10
|
|
'local_addr' : None
|
|
}
|
|
|
|
self.reader, self.writer = await asyncio.wait_for(asyncio.open_connection(**options), 15)
|
|
|
|
await self.raw(f'USER MESHT 0 * :git.acid.vegas/meshtastic') # Static for now
|
|
await self.raw('NICK ' + self.nickname)
|
|
|
|
while not self.reader.at_eof():
|
|
data = await asyncio.wait_for(self.reader.readuntil(b'\r\n'), 300)
|
|
await self.handle(data.decode('utf-8').strip())
|
|
|
|
except Exception as ex:
|
|
logging.error(f'failed to connect to {args.server} ({str(ex)})')
|
|
|
|
finally:
|
|
await asyncio.sleep(15)
|
|
|
|
|
|
async def eventPRIVMSG(self, data: str):
|
|
'''
|
|
Handle the PRIVMSG event.
|
|
|
|
:param data: The data received from the IRC server.
|
|
'''
|
|
|
|
parts = data.split()
|
|
|
|
ident = parts[0][1:]
|
|
nick = parts[0].split('!')[0][1:]
|
|
target = parts[2]
|
|
msg = ' '.join(parts[3:])[1:]
|
|
|
|
if target == args.channel: # TODO: Don't use the args variable here
|
|
if msg.startswith('!'):
|
|
if time.time() - self.last < 3:
|
|
if not self.slow:
|
|
self.slow = True
|
|
await self.sendmsg(target, color('Slow down nerd!', red))
|
|
else:
|
|
self.slow = False
|
|
parts = msg.split()
|
|
if parts[0] == '!meshage' and len(parts) > 1:
|
|
message = ' '.join(parts[1:])
|
|
if len(message) > 255:
|
|
await self.sendmsg(target, color('Message exceeds 255 bytes nerd!', red))
|
|
# TODO: Send a meshtastic message (We have to ensure our outbounds from IRC don't loop back into IRC)
|
|
|
|
self.last = time.time() # Update the last command time if it starts with ! character to prevent command flooding
|
|
|
|
|
|
async def handle(self, data: str):
|
|
'''
|
|
Handle the data received from the IRC server.
|
|
|
|
:param data: The data received from the IRC server.
|
|
'''
|
|
|
|
logging.info(data)
|
|
|
|
try:
|
|
parts = data.split()
|
|
|
|
if parts[0] == 'PING':
|
|
await self.raw('PONG ' + parts[1])
|
|
|
|
elif parts[1] == '001': # RPL_WELCOME
|
|
await self.raw(f'MODE {self.nickname} +B')
|
|
await self.sendmsg('NickServ', f'IDENTIFY {self.nickname} simps0nsfan420')
|
|
await asyncio.sleep(10) # Wait for NickServ to identify or any channel join delays
|
|
await self.raw(f'JOIN {args.channel} {args.key if args.key else ""}')
|
|
|
|
elif parts[1] == '433': # ERR_NICKNAMEINUSE
|
|
self.nickname += '_' # revamp this to be more unique
|
|
await self.raw('NICK ' + self.nickname)
|
|
|
|
elif parts[1] == 'INVITE':
|
|
target = parts[2]
|
|
chan = parts[3][1:]
|
|
if target == self.nickname and chan == args.channel:
|
|
await self.raw(f'JOIN {chan}')
|
|
|
|
elif parts[1] == 'KICK':
|
|
chan = parts[2]
|
|
kicked = parts[3]
|
|
if kicked == self.nickname and chan == args.channel:
|
|
await asyncio.sleep(3)
|
|
await self.raw(f'JOIN {args.channel} {args.key if args.key else ""}')
|
|
|
|
elif parts[1] == 'PRIVMSG':
|
|
await self.eventPRIVMSG(data) # We put this in a separate function since it will likely be the most used/handled event
|
|
|
|
except (UnicodeDecodeError, UnicodeEncodeError):
|
|
pass
|
|
|
|
except Exception as ex:
|
|
logging.exception(f'Unknown error has occured! ({ex})')
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description='Connect to an IRC server.')
|
|
parser.add_argument('server', help='The IRC server address.')
|
|
parser.add_argument('channel', help='The IRC channel to join.')
|
|
parser.add_argument('--port', type=int, help='The port number for the IRC server.')
|
|
parser.add_argument('--ssl', action='store_true', help='Use SSL for the connection.')
|
|
parser.add_argument('--key', default='', help='The key (password) for the IRC channel, if required.')
|
|
args = parser.parse_args()
|
|
|
|
if not args.channel.startswith('#'):
|
|
channel = '#' + args.channel
|
|
|
|
if not args.port:
|
|
args.port = 6697 if args.ssl else 6667
|
|
elif args.port < 1 or args.port > 65535:
|
|
raise ValueError('Port must be between 1 and 65535.')
|
|
|
|
print(f'Connecting to {args.server}:{args.port} (SSL: {args.ssl}) and joining {args.channel} (Key: {args.key or 'None'})')
|
|
|
|
bot = Bot()
|
|
|
|
asyncio.run(bot.connect())
|