"""IRC-related functions and classes to be imported by irc.py"""
import logging
import base64
PchumLog = logging.getLogger("pchumLogger")
class SendIRC:
"""Provides functions for outgoing IRC commands.
Functions are protocol compliant but don't implement all valid uses of certain commands.
"""
def __init__(self):
self.socket = None # INET socket connected with server.
def _send(self, *args: str, text=None):
"""Send a command to the IRC server.
Takes either a string or a list of strings.
The 'text' argument is for the final parameter, which can have spaces.
Since this checks if the socket is alive, it's best to send via this method."""
# Return if disconnected
if not self.socket or self.socket.fileno() == -1:
PchumLog.error(
"Send attempted while disconnected, args: %s, text: %s.", args, text
)
return
command = ""
# Convert command arguments to a single string if passed.
if args:
command += " ".join(args)
# If text is passed, add ':' to imply everything after it is one parameter.
if text:
command += f" :{text}"
# Add characters for end of line in IRC.
command += "\r\n"
# UTF-8 is the prefered encoding in 2023.
outgoing_bytes = command.encode(encoding="utf-8", errors="replace")
try:
PchumLog.debug("Sending: %s", command)
self.socket.send(outgoing_bytes)
except OSError:
PchumLog.exception("Error while sending: '%s'", command.strip())
self.socket.close()
def ping(self, token):
"""Send PING command to server to check for connectivity."""
self._send("PING", text=token)
def pong(self, token):
"""Send PONG command to reply to server PING."""
self._send("PONG", token)
def pass_(self, password):
"""Send a 'connection password' to the server.
Function is 'pass_' because 'pass' is reserved."""
self._send("PASS", text=password)
def nick(self, nick):
"""Send USER command to communicate nick to server."""
self._send("NICK", nick)
def user(self, username, realname):
"""Send USER command to communicate username and realname to server."""
self._send("USER", username, "0", "*", text=realname)
def privmsg(self, target, text):
"""Send PRIVMSG command to send a message."""
for line in text.split("\n"):
self._send("PRIVMSG", target, text=line)
def names(self, channel):
"""Send NAMES command to view channel members."""
self._send("NAMES", channel)
def kick(self, channel, user, reason=""):
"""Send KICK command to force user from channel."""
if reason:
self._send(f"KICK {channel} {user}", text=reason)
else:
self._send(f"KICK {channel} {user}")
def mode(self, target, modestring="", mode_arguments=""):
"""Set or remove modes from target."""
outgoing_mode = " ".join([target, modestring, mode_arguments]).strip()
self._send("MODE", outgoing_mode)
def ctcp(self, target, command, msg=""):
"""Send Client-to-Client Protocol message."""
outgoing_ctcp = " ".join(
[command, msg]
).strip() # Extra spaces break protocol, so strip.
self.privmsg(target, f"\x01{outgoing_ctcp}\x01")
def ctcp_reply(self, target, command, msg=""):
"""Send Client-to-Client Protocol reply message, responding to a CTCP message."""
outgoing_ctcp = " ".join(
[command, msg]
).strip() # Extra spaces break protocol, so strip.
self.notice(target, f"\x01{outgoing_ctcp}\x01")
def metadata(self, target, subcommand, *params):
"""Send Metadata command to get or set metadata.
See IRC metadata draft specification:
https://gist.github.com/k4bek4be/92c2937cefd49990fbebd001faf2b237
"""
self._send("METADATA", target, subcommand, *params)
def cap(self, subcommand, *params):
"""Send IRCv3 CAP command for capability negotiation.
See: https://ircv3.net/specs/extensions/capability-negotiation.html"""
self._send("CAP", subcommand, *params)
def join(self, channel, key=""):
"""Send JOIN command to join a channel/memo.
Keys or joining multiple channels is possible in the specification, but unused.
"""
channel_and_key = " ".join([channel, key]).strip()
self._send("JOIN", channel_and_key)
def part(self, channel):
"""Send PART command to leave a channel/memo.
Providing a reason or leaving multiple channels is possible in the specification.
"""
self._send("PART", channel)
def notice(self, target, text):
"""Send a NOTICE to a user or channel."""
self._send("NOTICE", target, text=text)
def invite(self, nick, channel):
"""Send INVITE command to invite a user to a channel."""
self._send("INVITE", nick, channel)
def away(self, text=None):
"""AWAY command to mark client as away or no longer away.
No 'text' parameter means the client is no longer away."""
if text:
self._send("AWAY", text=text)
else:
self._send("AWAY")
def list(self):
"""Send LIST command to get list of channels."""
self._send("LIST")
def quit(self, reason=""):
"""Send QUIT to terminate connection."""
self._send("QUIT", text=reason)
def authenticate(self, token=None, nick=None, password=None):
"""Authenticate command for SASL.
Send either a token like 'PLAIN' or authenticate with nick and password.
Reference: https://ircv3.net/irc/#account-authentication-and-registration
"""
if token:
self._send("AUTHENTICATE", text=token)
return
if nick and password:
# Authentication identity 'nick', authorization identity 'nick' and password 'password'.
sasl_string = f"{nick}\x00{nick}\x00{password}"
# Encode to use base64, then decode since 'text' only takes str.
sasl_string_bytes = sasl_string.encode(encoding="utf-8", errors="replace")
sasl_string_base64 = base64.b64encode(sasl_string_bytes).decode(
encoding="utf-8"
)
# SASL string needs to be under 400 bytes,
# splitting over multiple messages is in the protocol but not implemented here.
self._send("AUTHENTICATE", text=sasl_string_base64)
def parse_irc_line(line: str):
"""Retrieves tags, prefix, command, and arguments from an unparsed IRC line."""
parts = line.split(" ")
tags = None
prefix = None
if parts[0].startswith(":"):
prefix = parts[0][1:]
command = parts[1]
args = parts[2:]
elif parts[0].startswith("@"):
tags = parts[0] # IRCv3 message tag
prefix = parts[1][1:]
command = parts[2]
args = parts[3:]
else:
command = parts[0]
args = parts[1:]
command = command.casefold()
# If ':' is present the subsequent args are one parameter.
fused_args = []
for idx, arg in enumerate(args):
if arg.startswith(":"):
final_param = " ".join(args[idx:])
fused_args.append(final_param[1:])
break
fused_args.append(arg)
return (tags, prefix, command, fused_args)