pew

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 Colby
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,37 @@
+# IRC Bot
+
+This is an IRC bot written in Python. It connects to an IRC server, authenticates using SASL if desired, and responds to various commands. It is also able to join and save channels that it is invited to.
+
+## Requirements
+
+- Python 3
+- A configuration file `config.json` containing the following variables:
+  - `BSERVER`: the IRC server to connect to
+  - `BPORT`: the port to use for the connection (can be preceded with a `+` to specify a secure connection)
+  - `BNICK`: the desired bot nickname
+  - `BIDENT`: the bot's ident
+  - `BNAME`: the bot's real name
+  - `UseSASL`: a boolean indicating whether or not to use SASL authentication
+  - `SANICK`: the bot's SASL account name (if using SASL)
+  - `SAPASS`: the bot's SASL password (if using SASL)
+  
+## Usage
+
+To run the bot, simply execute the script with `python3 elitebot.py config.json`. The bot will connect to the specified server and authenticate if necessary. It will then listen for commands and respond accordingly.
+
+## Commands
+
+The following commands are recognized by the bot:
+
+- `&moo`: makes the bot moo.
+- `&join`: makes the bot join the specified channel.
+- `&part`: makes the bot part the current channel if a channel is not specified.
+- `&quit`: makes the bot quit.
+
+## Saving Channels
+
+The bot is able to save a list of channels that it should automatically join upon connecting to the server. 
+
+## Contributing
+
+If you have any suggestions or improvements for the bot, feel free to create a pull request.

config.json

@@ -0,0 +1,18 @@
+{
+    "Connection": {
+        "Hostname": "irc.example.net",
+        "Port": "+6697",
+        "Nick": "EliteBot",
+        "Ident": "EliteBot",
+        "Name": "EliteBot",
+        "BindHost": "0.0.0.0"
+    },
+    "SASL": {
+        "UseSASL": false,
+        "SASLNick": "EliteBot",
+        "SASLPassword": "password"
+    },
+    "Logging": {
+        "Console": true
+    }
+}

config.yaml

@@ -0,0 +1,13 @@
+Connection:
+  Hostname: irc.rizon.net
+  Port: "+6697"
+  Nick: EliteBot
+  Ident: EliteBot
+  Name: EliteBot
+  BindHost: 0.0.0.0
+SASL:
+  UseSASL: false
+  SASLNick: EliteBot
+  SASLPassword: password
+Logging:
+  Console: true

data/channels.json

@@ -0,0 +1 @@
+["#EliteBot"]

elitebot.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+
+import sys
+from src.bot import Bot
+
+def main():
+    if len(sys.argv) < 2:
+        print("Usage: python elitebot.py <config_file>")
+        sys.exit(1)
+
+    config_file = sys.argv[1]
+    try:
+        bot = Bot(config_file)
+    except FileNotFoundError as e:
+        print(f"Config file not found: {e}")
+        sys.exit(1)
+    except Exception as e:
+        print(f"Error loading bot: {e}")
+        sys.exit(1)
+
+    try:
+        print("EliteBot started successfully!")
+        bot.start()  
+    except Exception as e:
+        print(f"Error starting EliteBot: {e}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        print("\nEliteBot has been stopped.")
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")

plugins/commands.py

@@ -0,0 +1,59 @@
+from src.plugin_base import PluginBase
+from src.channel_manager import ChannelManager
+import sys
+
+class Plugin(PluginBase):
+        
+    def handle_message(self, source_nick, channel, message):
+        message_parts = message.split()
+        self.channel_manager = ChannelManager()  
+        if message_parts[0] == '!hello':
+            self.bot.ircsend(f'PRIVMSG {channel} :Hello, {source_nick}!')
+   
+        elif message_parts[0] == '!join':
+            if len(message_parts) == 0:
+                self.bot.ircsend(f'PRIVMSG {channel} :Please specify a channel to join')
+                return
+            else:
+                self.channel_manager.save_channel(message_parts[1])
+                self.bot.ircsend(f'JOIN {message_parts[1]}')
+
+        elif message_parts[0] == '!part':
+            if len(message_parts) == 1:
+                self.bot.ircsend(f'PART {channel}')
+                self.channel_manager.remove_channel(channel)
+            else:
+                self.bot.ircsend(f'PART {message_parts[1]}')
+                self.channel_manager.remove_channel(message_parts[1])
+
+
+        elif message_parts[0] == "!quit":
+            if len(message_parts) == 0:
+                quit_message = 'EliteBot!'
+            else:
+                quit_message = message[len(message_parts[0])+1:]
+            self.bot.ircsend(f'QUIT :{quit_message}')
+            self.bot.ircsock.close()
+            self.bot.connected = False
+            sys.exit()
+
+        elif message_parts[0] == "!raw":
+            if len(message_parts) > 1:
+                if message_parts[1].upper() == "PRIVMSG" and len(message_parts) > 3:
+                    raw_command = ' '.join(message_parts[1:3]) + " :" + ' '.join(message_parts[3:])
+                else:
+                    raw_command = ' '.join(message_parts[1:])
+                self.bot.ircsend(raw_command)
+        
+        elif message_parts[0] == '!me':
+            if len(message_parts) > 1:
+                action_message = ' '.join(message_parts[1:])
+                self.bot.ircsend(f'PRIVMSG {channel} :\x01ACTION {action_message}\x01')
+            else:
+                self.bot.ircsend(f'PRIVMSG {channel} :Please specify an action')
+
+        elif message_parts[0] == '!ping':
+            if len(message_parts) > 1:
+                self.bot.ircsend(f'PRIVMSG {channel} :Pinging {message_parts[1]}')
+            else:
+                self.bot.ircsend(f'PRIVMSG {channel} :Please specify a nick to ping')

src/__init__.py

@@ -0,0 +1 @@
+

src/bot.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+
+import socket
+import ssl
+import time
+import json
+import yaml
+import inspect
+import sys
+import os
+import importlib.util
+from src.channel_manager import ChannelManager
+from src.logger import Logger
+from src.plugin_base import PluginBase
+from src.sasl import handle_sasl, handle_authenticate, handle_903
+
+class Bot:
+    def __init__(self, config_file):
+        self.config = self.load_config(config_file)
+        self.channel_manager = ChannelManager()
+        self.logger = Logger('logs/elitebot.log')
+        self.connected = False
+        self.ircsock = None
+        self.running = True
+        self.plugins = []
+        self.load_plugins()
+        self.config = self.load_config(config_file)
+        self.validate_config(self.config)
+
+    def validate_config(self, config):
+        required_fields = [
+            ["Connection", "Port"],
+            ["Connection", "Hostname"],
+            ["Connection", "Nick"],
+            ["Connection", "Ident"],
+            ["Connection", "Name"],
+            ["SASL", "UseSASL"]
+        ]
+
+        for field in required_fields:
+            if not self.get_nested_config_value(config, field):
+                raise ValueError(f'Missing required config field: {" -> ".join(field)}')
+            
+    def get_nested_config_value(self, config, keys):
+        value = config
+        for key in keys:
+            value = value.get(key)
+            if value is None:
+                return None
+        return value
+
+    def load_plugins(self):
+        self.plugins = []
+        plugin_folder = "./plugins"
+        for filename in os.listdir(plugin_folder):
+            if filename.endswith('.py'):
+                filepath = os.path.join(plugin_folder, filename)
+                spec = importlib.util.spec_from_file_location("module.name", filepath)
+                module = importlib.util.module_from_spec(spec)
+                spec.loader.exec_module(module)
+                for name, obj in inspect.getmembers(module):
+                    if inspect.isclass(obj) and issubclass(obj, PluginBase) and obj is not PluginBase:
+                        plugin_instance = obj(self)
+                        self.plugins.append(plugin_instance)
+
+    def load_config(self, config_file):
+        _, ext = os.path.splitext(config_file)
+        try:
+            with open(config_file, 'r') as file:
+                if ext == '.json':
+                    config = json.load(file)
+                elif ext == '.yaml' or ext == '.yml':
+                    config = yaml.safe_load(file)
+                else:
+                    raise ValueError(f'Unsupported file extension: {ext}')
+        except FileNotFoundError as e:
+            self.logger.error(f'Error loading config file: {e}')
+            raise
+        except (json.JSONDecodeError, yaml.YAMLError) as e:
+            self.logger.error(f'Error parsing config file: {e}')
+            raise
+        return config
+
+    def decode(self, bytes):
+        for encoding in ['utf-8', 'latin1', 'iso-8859-1', 'cp1252']:
+            try:
+                return bytes.decode(encoding)
+            except UnicodeDecodeError:
+                continue
+        self.logger.error('Could not decode byte string with any known encoding')
+        return bytes.decode('utf-8', 'ignore')  # Ignore errors and return as much as possible
+
+    def ircsend(self, msg):
+        try:
+            if msg != '':
+                self.logger.info(f'Sending command: {msg}')
+                self.ircsock.send(bytes(f'{msg}\r\n','UTF-8'))
+        except Exception as e:
+            self.logger.error(f'Error sending IRC message: {e}')
+            raise
+
+    def parse_message(self, message):
+        parts = message.split()
+        if not parts:
+            return None, None, []
+        source = parts[0][1:] if parts[0].startswith(':') else None
+        command = parts[1] if source else parts[0]
+        args_start = 2 if source else 1
+        args = []
+        trailing_arg_start = None
+        for i, part in enumerate(parts[args_start:], args_start):
+            if part.startswith(':'):
+                trailing_arg_start = i
+                break
+            else:
+                args.append(part)
+        if trailing_arg_start is not None:
+            args.append(' '.join(parts[trailing_arg_start:])[1:])
+        return source, command, args
+
+    def connect(self):
+        try:
+            self.ircsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+            if str(self.config["Connection"].get("Port"))[:1] == '+':
+                context = ssl.create_default_context()
+                self.ircsock = context.wrap_socket(self.ircsock, server_hostname=self.config["Connection"].get("Hostname"))
+                port = int(self.config['Connection'].get('Port')[1:])
+            else:
+                port = int(self.config['Connection'].get('Port'))
+
+            if 'BindHost' in self.config:
+                self.ircsock.bind((self.config['Connection'].get('BindHost'), 0))
+
+            self.ircsock.connect_ex((self.config['Connection'].get('Hostname'), port))
+            self.ircsend(f'NICK {self.config["Connection"].get("Nick")}')
+            self.ircsend(f'USER {self.config["Connection"].get("Ident")} * * :{self.config["Connection"].get("Name")}')
+            if self.config["SASL"].get("UseSASL"):
+                self.ircsend('CAP REQ :sasl')
+        except Exception as e:
+            self.logger.error(f"Error establishing connection: {e}")
+            self.connected = False
+            return
+
+    def start(self):
+        while True:
+            if not self.connected:
+                try:
+                    self.connect()
+                    self.connected = True
+                except Exception as e:
+                    self.logger.error(f'Connection error: {e}')
+                    time.sleep(60)
+                    continue
+
+            try:
+                recvText = self.ircsock.recv(2048)
+                if not recvText:
+                    self.connected = False
+                    continue
+
+                ircmsg = self.decode(recvText)
+                source, command, args = self.parse_message(ircmsg)
+                self.logger.debug(f'Received: source: {source} | command: {command} | args: {args}')
+
+                # Handle PING requests immediately
+                if command == 'PING':
+                    nospoof = args[0][1:] if args[0].startswith(':') else args[0]
+                    self.ircsend(f'PONG :{nospoof}')
+                    continue
+
+                # Process other messages
+                if command == 'PRIVMSG':
+                    channel, message = args[0], args[1]
+                    source_nick = source.split('!')[0]
+                    if message.startswith('&'):
+                        cmd, *cmd_args = message[1:].split()
+                        self.handle_command(source_nick, channel, cmd, cmd_args)
+                    for plugin in self.plugins:
+                        plugin.handle_message(source_nick, channel, message)
+
+                elif command == 'CAP' and args[1] == 'ACK' and 'sasl' in args[2]:
+                    handle_sasl(self.config, self.ircsend)
+
+                elif command == 'AUTHENTICATE':
+                    handle_authenticate(args, self.config, self.ircsend)
+
+                elif command == '903':
+                    handle_903(self.ircsend)
+
+                if command == 'PRIVMSG' and args[1].startswith('\x01VERSION\x01'):
+                    source_nick = source.split('!')[0]
+                    self.ircsend(f'NOTICE {source_nick} :\x01VERSION EliteBot 0.1\x01')
+
+                if command == '001':
+                    for channel in self.channel_manager.get_channels():
+                        self.ircsend(f'JOIN {channel}')
+
+                if command == 'INVITE':
+                    channel = args[1]
+                    self.ircsend(f'JOIN {channel}')
+                    self.channel_manager.save_channel(channel)
+
+                if command == 'VERSION':
+                    self.ircsend('NOTICE', f'{source_nick} :I am a bot version 1.0.0')
+
+            except Exception as e:
+                self.logger.error(f'General error: {e}')
+                self.connected = False
+
+if __name__ == '__main__':
+    try:
+        bot = Bot(sys.argv[1])
+        bot.start()
+    except KeyboardInterrupt:
+        print('\nEliteBot has been stopped.')
+    except Exception as e:
+        print(f'An unexpected error occurred: {e}')

src/channel_manager.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+
+import json
+import os
+from os import path
+
+class ChannelManager:
+    def __init__(self):
+        self.channels = self._load_channels()
+
+    def _load_channels(self):
+        os.makedirs("data", exist_ok=True)
+        if not path.exists('data/channels.json'):
+            with open('data/channels.json', 'w') as f:
+                json.dump([], f)
+            return []
+        try:
+            with open('data/channels.json', 'r') as f:
+                return json.load(f)
+        except json.JSONDecodeError as e:
+            print(f"Error decoding JSON: {e}")
+            return []
+        except Exception as e:
+            print(f"Error loading channels: {e}")
+            return []
+
+    def save_channel(self, channel):
+        channel = channel.lstrip(':')  
+        if channel not in self.channels:
+            self.channels.append(channel)
+            self._write_channels()
+
+    def remove_channel(self, channel):
+        channel = channel.lstrip(':') 
+        if channel in self.channels:
+            self.channels.remove(channel)
+            self._write_channels()
+
+    def _write_channels(self):
+        os.makedirs("data", exist_ok=True)
+        try:
+            with open('data/channels.json', 'w') as f:
+                json.dump(self.channels, f)
+        except Exception as e:
+            print(f"Error saving channels: {e}")
+
+    def get_channels(self):
+        return self.channels

src/logger.py

@@ -0,0 +1,40 @@
+import os
+from datetime import datetime
+
+import colorama
+
+
+class Logger:
+    def __init__(self, log_file: str, datefmt: str = '%m/%d/%Y %I:%M:%S %p'):
+        colorama.init()
+        self.log_file = log_file
+        self.datefmt = datefmt
+
+        os.makedirs(os.path.dirname(log_file), exist_ok=True)
+
+    def log(self, level, message):
+        asctime = datetime.now().strftime(self.datefmt)
+
+        match level:
+            case 'debug':
+                print(f'\033[92m[{asctime}] - {message}\033[39m')
+            case 'info':
+                print(f'\033[96m[{asctime}] - {message}\033[39m')
+            case 'warn' | 'warning':
+                print(f'\033[93m[{asctime}] - {message}\033[39m')
+            case 'error':
+                print(f'\033[91m[{asctime}] - {message}\033[39m')
+            case _:
+                pass  # just ignore it.
+
+    def debug(self, message):
+        self.log('debug', message)
+
+    def info(self, message):
+        self.log('info', message)
+
+    def warning(self, message):
+        self.log('warning', message)
+
+    def error(self, message):
+        self.log('error', message)

src/plugin_base.py

@@ -0,0 +1,31 @@
+class PluginBase:
+    def __init__(self, bot_instance):
+        """
+        Constructor for the base plugin.
+
+        :param bot_instance: Reference to the main bot instance
+        """
+        self.bot = bot_instance
+        self.commands = {}
+
+    def handle_message(self, source_nick, channel, message):
+        """
+        Called when a message is received.
+
+        :param source_nick: Nickname of the user who sent the message
+        :param channel: Channel where the message was sent
+        :param message: Content of the message
+        """
+        pass
+
+    def on_connect(self):
+        """
+        Called when the bot connects to the server.
+        """
+        pass
+
+    def on_disconnect(self):
+        """
+        Called when the bot disconnects from the server.
+        """
+        pass

src/sasl.py

@@ -0,0 +1,41 @@
+# sasl.py
+import base64
+
+NULL_BYTE = '\x00'
+ENCODING = 'UTF-8'
+
+def handle_sasl(config, ircsend):
+    """
+    Handles SASL authentication by sending an AUTHENTICATE command.
+
+    Parameters:
+    config (dict): Configuration dictionary
+    ircsend (function): Function to send IRC commands
+    """
+    ircsend('AUTHENTICATE PLAIN')
+
+def handle_authenticate(args, config, ircsend):
+    """
+    Handles the AUTHENTICATE command response.
+
+    Parameters:
+    args (list): List of arguments from the AUTHENTICATE command
+    config (dict): Configuration dictionary
+    ircsend (function): Function to send IRC commands
+    """
+    if args[0] == '+':
+        if "SASLNick" in config['SASL'] and "SASLPassword" in config['SASL']:
+            authpass = f"{config['SASL']['SASLNick']}{NULL_BYTE}{config['SASL']['SASLNick']}{NULL_BYTE}{config['SASL']['SASLPassword']}"
+            ap_encoded = base64.b64encode(authpass.encode(ENCODING)).decode(ENCODING)
+            ircsend(f'AUTHENTICATE {ap_encoded}')
+        else:
+            raise KeyError("SASLNICK and/or SASLPASS not found in config")
+
+def handle_903(ircsend):
+    """
+    Handles the 903 command by sending a CAP END command.
+
+    Parameters:
+    ircsend (function): Function to send IRC commands
+    """
+    ircsend('CAP END')

test.json

@@ -0,0 +1,18 @@
+{
+    "Connection": {
+        "Hostname": "irc.rizon.net",
+        "Port": "+6697",
+        "Nick": "EliteBot",
+        "Ident": "EliteBot",
+        "Name": "EliteBot",
+        "BindHost": "0.0.0.0"
+    },
+    "SASL": {
+        "UseSASL": true,
+        "SASLNick": "EliteBot",
+        "SASLPassword": "789abc//"
+    },
+    "Logging": {
+        "Console": true
+    }
+}