Update server.py
All checks were successful
/ check (push) Successful in 26s

This commit is contained in:
Nova Cat 2025-01-29 20:17:03 -08:00
parent c9e44686b3
commit 4ac5292d2c

861
server.py
View file

@ -231,6 +231,8 @@ def session(connection, client, ip, isssl=False):
CAPEND = False CAPEND = False
clident = None clident = None
pendingCommands = "" # list of commands that were executed before verification pendingCommands = "" # list of commands that were executed before verification
unfinished = False
textt = ""
try: try:
print("Connected to client IP: {}".format(client)) print("Connected to client IP: {}".format(client))
connection.sendall(bytes(f":{server} NOTICE * :*** Looking for your hostname...\r\n","UTF-8")) connection.sendall(bytes(f":{server} NOTICE * :*** Looking for your hostname...\r\n","UTF-8"))
@ -270,456 +272,457 @@ def session(connection, client, ip, isssl=False):
break break
print("Received data: {}".format(data)) print("Received data: {}".format(data))
try: try:
textt = data.decode() textt += data.decode()
for text in textt.replace("\r", "").split("\n"): if textt[-1] == "\n":
for i in socketListeners: for text in textt.replace("\r", "").split("\n"):
if "onSocket" in dir(i):
i.onSocket(socket=connection, ip=client[0], value=text, cachedNick=pending if pending != "*" else None, validated=finished)
command = text.split(" ")[0].upper()
try:
args = text.split(" ")[1:]
except:
pass
if command == "NICK" and not finished:
pending = text.split(" ")[1]
if pending[0] == ":": pending = pending[1:]
if "!" in pending or ":" in pending or "#" in pending or "*" in pending:
connection.sendall(bytes(f":{server} 432 * {pending} :Erroneus nickname\r\n","UTF-8"))
pending = "*"
elif pending.lower() in lower_nicks:
connection.sendall(bytes(f":{server} 433 * {pending} :Nickname is already in use.\r\n","UTF-8"))
pending = "*"
else:
already_set = True
print(f"User {pending} set nick")
elif command == "USER":
if not ready:
username = text.split(" ")[1]
realname = " ".join(text.split(" ")[4:])[1:]
ready = True
elif command == "CAP":
#usesIRCv3 = True
if args[0].upper() == "LS":
connection.sendall(bytes(f":{server} CAP * LS :ircat.xyz/foo\r\n", "UTF-8"))
elif args[0].upper() == "REQ":
if args[1].lower() == ":sasl":
pass
#connection.sendall(f":{server} CAP * ACK :sasl")
elif args[0].upper() == "END":
CAPEND = True
elif command == "WEBIRC" and not finished:
try:
if args[0] == data2["webirc_pass"]:
hostname = args[2]
client = (args[3], client[1])
connection.sendall(bytes(f":{server} NOTICE * :*** WebIRC detected, welcome to IRC!\r\n", "UTF-8"))
if hostname != client[0]:
connection.sendall(bytes(f":{server} NOTICE * :*** Got WebIRC hostname! {hostname}\r\n", "UTF-8"))
except:
print(traceback.format_exc())
break
elif (ready and already_set) and (CAPEND if usesIRCv3 else True) and not finished:
cleanup_manual()
print(f"User {pending} successfully logged in.")
nickname_list[pending] = connection
property_list[pending] = {"host": hostname, "username": clident if clident != None else f"~{username }", "realname": realname, "modes": "iw", "last_ping": time.time(), "ping_pending": False, "away": False, "identified": False, "ssl": isssl}
lower_nicks[pending.lower()] = pending
for i in socketListeners: for i in socketListeners:
if "onValidate" in dir(i): if "onSocket" in dir(i):
i.onValidate(socket=connection, ip=client[0]) i.onSocket(socket=connection, ip=client[0], value=text, cachedNick=pending if pending != "*" else None, validated=finished)
threading.Thread(target=pinger, args=[pending, connection]).start() command = text.split(" ")[0].upper()
if clident == None:
rident = f"~{username}"
connection.sendall(bytes(f":{server} 001 {pending} :Welcome to the {displayname} Internet Relay Chat Network {pending}\r\n", "UTF-8"))
connection.sendall(bytes(f":{server} 002 {pending} :Your host is {server}[{ip}/6667], running version IRCat-v{__version__}\r\n", "UTF-8"))
connection.sendall(bytes(f":{server} 004 {pending} {server} IRCat-{__version__} iow ovmsitnlbkq\r\n", "UTF-8"))
connection.sendall(bytes(f":{server} 005 {pending} CHANMODES=bq,k,l,irmnpst CHANTYPES=# NETWORK={displayname} :are supported by this server\r\n", "UTF-8"))
# connection.sendall(bytes(f":{server} 251 {pending} :There are {allusers} users and {allinvis} invisible in {servers} servers\r\n", "UTF-8")) Not supported as there isn't multi-server capability (yet)
ops = 0 # Placeholder, will replace with caclulating how much people have +o
connection.sendall(bytes(f":{server} 252 {pending} {ops} :IRC Operators online\r\n", "UTF-8"))
connection.sendall(bytes(f":{server} 253 {pending} 0 :unknown connection(s)\r\n", "UTF-8")) # Replace 0 with a variable of not setup clients.
chans = len(channels_list)
connection.sendall(bytes(f":{server} 254 {pending} {chans} :channels formed\r\n", "UTF-8"))
cleints = len(nickname_list)
servers = 1
connection.sendall(bytes(f":{server} 255 {pending} :I have {cleints} clients and {servers} servers\r\n", "UTF-8"))
# Start the MOTD
if motd_file != None:
motd = open(motd_file).read()
connection.sendall(bytes(f":{server} 375 {pending} :- {server} Message of the Day -\r\n", "UTF-8"))
for i in motd.rstrip().split("\n"):
connection.sendall(bytes(f":{server} 372 {pending} :- {i}\r\n", "UTF-8"))
connection.sendall(bytes(f":{server} 376 {pending} :End of /MOTD command\r\n", "UTF-8"))
# End the MOTD
connection.sendall(bytes(f":{pending} MODE {pending} +iw\r\n","UTF-8"))
finished = True
elif command == "PING":
try: try:
e = text.split(" ")[1] args = text.split(" ")[1:]
e = f":{e}" if e[0] != ":" else e
connection.sendall(bytes(f":{server} PONG {server} {e}\r\n","UTF-8"))
except: except:
connection.sendall(bytes(f":{server} PONG {server}\r\n","UTF-8")) pass
elif command == "MOTD": if command == "NICK" and not finished:
if motd_file != None: pending = text.split(" ")[1]
motd = open(motd_file).read() if pending[0] == ":": pending = pending[1:]
connection.sendall(bytes(f":{server} 375 {pending} :- {server} Message of the Day -\r\n", "UTF-8")) if "!" in pending or ":" in pending or "#" in pending or "*" in pending:
for i in motd.rstrip().split("\n"): connection.sendall(bytes(f":{server} 432 * {pending} :Erroneus nickname\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 372 {pending} :- {i}\r\n", "UTF-8")) pending = "*"
connection.sendall(bytes(f":{server} 376 {pending} :End of /MOTD command\r\n", "UTF-8")) elif pending.lower() in lower_nicks:
elif finished: connection.sendall(bytes(f":{server} 433 * {pending} :Nickname is already in use.\r\n","UTF-8"))
pendingCommands += text pending = "*"
for comd in pendingCommands.split("\r\n"): else:
command = comd.split(" ")[0].upper() already_set = True
args = comd.split(" ")[1:] print(f"User {pending} set nick")
text = comd elif command == "USER":
processedExternally = False if not ready:
for i in commandProviders: username = text.split(" ")[1]
cmdrun = i.command(command=command, args=args, nick=pending, ip=client[0], user=property_list[pending], connection=connection) realname = " ".join(text.split(" ")[4:])[1:]
if cmdrun["success"]: ready = True
if "identify" in cmdrun: elif command == "CAP":
if cmdrun["identify"] == "logout": #usesIRCv3 = True
if "o" in property_list[pending]["modes"]: if args[0].upper() == "LS":
connection.sendall(bytes(f":{pending} MODE {pending} -o\r\n","UTF-8")) connection.sendall(bytes(f":{server} CAP * LS :ircat.xyz/foo\r\n", "UTF-8"))
if not "i" in property_list[pending]["modes"]: elif args[0].upper() == "REQ":
connection.sendall(bytes(f":{pending} MODE {pending} +i\r\n","UTF-8")) if args[1].lower() == ":sasl":
if not "w" in property_list[pending]["modes"]:
connection.sendall(bytes(f":{pending} MODE {pending} +w\r\n","UTF-8"))
property_list[pending]["modes"] = "iw"
property_list[pending]["identified"] = False
else:
property_list[pending]["identified"] = True
property_list[pending]["identusername"] = cmdrun["identify"][0]
temp_mode = cmdrun["identify"][1]
property_list[pending]["modes"] = temp_mode
connection.sendall(bytes(f":{pending} MODE {pending} +{temp_mode}\r\n","UTF-8"))
processedExternally = True
break
if processedExternally:
pass
elif command == "JOIN":
channels = text.split(" ")[1]
if channels[0] == ":":
channels = channels[1:]
for channelt in channels.split(","):
channel = channelt.strip()
if channel.lower() in lower_chans:
channel = lower_chans[channel.lower()]
success = True
if channel in channels_list:
if pending in channels_list[channel]:
success=False
print(f"{pending} is already in {channel} , ignoring JOIN request.")
if success:
try:
if channel in channels_list:
channels_list[channel].append(pending)
else:
channels_list[channel] = [pending]
lower_chans[channel.lower()] = channel
topic_list[channel] = "Topic is not implemented."
except:
connection.sendall(bytes(f":{server} NOTICE * :*** Could not join {channel}\r\n","UTF-8"))
print(channels_list)
for i in channels_list[channel]:
try:
nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} JOIN {channel}\r\n","UTF-8"))
except:
pass
# Code re-used in the NAMES command
if channel in channels_list:
if pending in channels_list[channel]:
users = " ".join(channels_list[channel])
connection.sendall(bytes(f":{server} 353 {pending} = {channel} :{users}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 366 {pending} {channel} :End of /NAMES list.\r\n","UTF-8"))
print("Successfully pre-loaded /NAMES list")
elif command == "LIST":
connection.sendall(bytes(f":{server} 321 {pending} Channel :Users Name\r\n","UTF-8"))
for key, value in topic_list.items():
usersin = len(channels_list[key])
connection.sendall(bytes(f":{server} 322 {pending} {key} {usersin} :{value}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 323 {pending} :End of /LIST\r\n","UTF-8"))
elif command == "PONG":
e = text.split(" ")[1]
if e == server or e == f":{server}":
print(pending + " replied to PING.")
property_list[pending]["last_ping"] = time.time()
property_list[pending]["ping_pending"] = False
elif command == "NICK":
if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
elif text.split(" ")[1] == pending:
pass pass
else: #connection.sendall(f":{server} CAP * ACK :sasl")
pending2 = text.split(" ")[1] elif args[0].upper() == "END":
if pending2[0] == ":": pending2 = pending2[1:] CAPEND = True
if "!" in pending2 or ":" in pending2 or "#" in pending2 or "*" in pending2: elif command == "WEBIRC" and not finished:
connection.sendall(bytes(f":{server} 432 {pending} {pending2} :Erroneus nickname\r\n","UTF-8")) try:
elif pending2.lower() in lower_nicks: if args[0] == data2["webirc_pass"]:
connection.sendall(bytes(f":{server} 433 {pending} {pending2} :Nickname is already in use.\r\n","UTF-8")) hostname = args[2]
else: client = (args[3], client[1])
print("Sending nickname change...") connection.sendall(bytes(f":{server} NOTICE * :*** WebIRC detected, welcome to IRC!\r\n", "UTF-8"))
connection.sendall(bytes(f":{pending}!{rident}@{hostname} NICK {pending2}\r\n","UTF-8")) if hostname != client[0]:
# Broadcast the nickname change connection.sendall(bytes(f":{server} NOTICE * :*** Got WebIRC hostname! {hostname}\r\n", "UTF-8"))
done = [] except:
for i, users in channels_list.items(): print(traceback.format_exc())
if pending in users: break
for j in users: elif (ready and already_set) and (CAPEND if usesIRCv3 else True) and not finished:
if j != pending and j != pending2 and not j in done: cleanup_manual()
print("Broadcasting on " + j) print(f"User {pending} successfully logged in.")
nickname_list[j].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8")) nickname_list[pending] = connection
done.append(j) property_list[pending] = {"host": hostname, "username": clident if clident != None else f"~{username }", "realname": realname, "modes": "iw", "last_ping": time.time(), "ping_pending": False, "away": False, "identified": False, "ssl": isssl}
# Replace the nickname lower_nicks[pending.lower()] = pending
try: for i in socketListeners:
print("Changing on " + i) if "onValidate" in dir(i):
channels_list[i].remove(pending) i.onValidate(socket=connection, ip=client[0])
channels_list[i].append(pending2) threading.Thread(target=pinger, args=[pending, connection]).start()
except: if clident == None:
print(traceback.format_exc()) rident = f"~{username}"
print("Moving config...") connection.sendall(bytes(f":{server} 001 {pending} :Welcome to the {displayname} Internet Relay Chat Network {pending}\r\n", "UTF-8"))
property_list[pending2] = property_list.pop(pending) connection.sendall(bytes(f":{server} 002 {pending} :Your host is {server}[{ip}/6667], running version IRCat-v{__version__}\r\n", "UTF-8"))
nickname_list[pending2] = nickname_list.pop(pending) connection.sendall(bytes(f":{server} 004 {pending} {server} IRCat-{__version__} iow ovmsitnlbkq\r\n", "UTF-8"))
del lower_nicks[pending.lower()] connection.sendall(bytes(f":{server} 005 {pending} CHANMODES=bq,k,l,irmnpst CHANTYPES=# NETWORK={displayname} :are supported by this server\r\n", "UTF-8"))
lower_nicks[pending2.lower()] = pending2 # connection.sendall(bytes(f":{server} 251 {pending} :There are {allusers} users and {allinvis} invisible in {servers} servers\r\n", "UTF-8")) Not supported as there isn't multi-server capability (yet)
print("starting pinger...") ops = 0 # Placeholder, will replace with caclulating how much people have +o
pending = pending2 connection.sendall(bytes(f":{server} 252 {pending} {ops} :IRC Operators online\r\n", "UTF-8"))
property_list[pending2]["ping_pending"] = False connection.sendall(bytes(f":{server} 253 {pending} 0 :unknown connection(s)\r\n", "UTF-8")) # Replace 0 with a variable of not setup clients.
property_list[pending2]["last_ping"] = time.time() chans = len(channels_list)
threading.Thread(target=pinger, args=[pending, connection]).start() connection.sendall(bytes(f":{server} 254 {pending} {chans} :channels formed\r\n", "UTF-8"))
print(f"User {pending} set nick") cleints = len(nickname_list)
print("Broadcasting nickname change...") servers = 1
elif command == "PART": connection.sendall(bytes(f":{server} 255 {pending} :I have {cleints} clients and {servers} servers\r\n", "UTF-8"))
if len(args) == 0: # Start the MOTD
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8")) if motd_file != None:
else: motd = open(motd_file).read()
channel = text.split(" ")[1] connection.sendall(bytes(f":{server} 375 {pending} :- {server} Message of the Day -\r\n", "UTF-8"))
for i in channels_list[channel]: for i in motd.rstrip().split("\n"):
try: connection.sendall(bytes(f":{server} 372 {pending} :- {i}\r\n", "UTF-8"))
nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8")) connection.sendall(bytes(f":{server} 376 {pending} :End of /MOTD command\r\n", "UTF-8"))
except: # End the MOTD
pass connection.sendall(bytes(f":{pending} MODE {pending} +iw\r\n","UTF-8"))
try: finished = True
channels_list[channel].remove(pending) elif command == "PING":
except: try:
print(traceback.format_exc()) e = text.split(" ")[1]
elif command == "AWAY": e = f":{e}" if e[0] != ":" else e
if len(args) == 0: connection.sendall(bytes(f":{server} PONG {server} {e}\r\n","UTF-8"))
property_list[pending]["away"] = False except:
connection.sendall(bytes(f":{server} 305 {pending} :You are no longer marked as being away\r\n","UTF-8")) connection.sendall(bytes(f":{server} PONG {server}\r\n","UTF-8"))
else: elif command == "MOTD":
reasons = " ".join(args) if motd_file != None:
if reasons[0] == ":": reasons = reasons[1:] motd = open(motd_file).read()
property_list[pending]["away"] = True connection.sendall(bytes(f":{server} 375 {pending} :- {server} Message of the Day -\r\n", "UTF-8"))
property_list[pending]["reason"] = reasons for i in motd.rstrip().split("\n"):
connection.sendall(bytes(f":{server} 306 {pending} :You have been marked as being away\r\n","UTF-8")) connection.sendall(bytes(f":{server} 372 {pending} :- {i}\r\n", "UTF-8"))
elif command == "WHO": connection.sendall(bytes(f":{server} 376 {pending} :End of /MOTD command\r\n", "UTF-8"))
if len(args) == 0: elif finished:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8")) pendingCommands += text
else: for comd in pendingCommands.split("\r\n"):
channel = text.split(" ")[1] command = comd.split(" ")[0].upper()
if channel in channels_list: args = comd.split(" ")[1:]
for i in channels_list[channel]: text = comd
who_host = property_list[i]["host"] processedExternally = False
who_user = property_list[i]["username"] for i in commandProviders:
who_realname = property_list[i]["realname"] cmdrun = i.command(command=command, args=args, nick=pending, ip=client[0], user=property_list[pending], connection=connection)
who_away = "G" if property_list[i]["away"] else "H" if cmdrun["success"]:
connection.sendall(bytes(f":{server} 352 {pending} {channel} {who_user} {who_host} {server} {i} {who_away} :0 {who_realname}\r\n","UTF-8")) if "identify" in cmdrun:
elif channel in nickname_list: if cmdrun["identify"] == "logout":
who_host = property_list[channel]["host"] if "o" in property_list[pending]["modes"]:
who_user = property_list[channel]["username"] connection.sendall(bytes(f":{pending} MODE {pending} -o\r\n","UTF-8"))
who_realname = property_list[channel]["realname"] if not "i" in property_list[pending]["modes"]:
who_away = "G" if property_list[channel]["away"] else "H" connection.sendall(bytes(f":{pending} MODE {pending} +i\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 352 {pending} * {who_user} {who_host} {server} {channel} {who_away} :0 {who_realname}\r\n","UTF-8")) if not "w" in property_list[pending]["modes"]:
connection.sendall(bytes(f":{pending} MODE {pending} +w\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 315 {pending} {channel} :End of /WHO list.\r\n","UTF-8")) property_list[pending]["modes"] = "iw"
elif command == "WHOIS": property_list[pending]["identified"] = False
if len(args) == 0: else:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8")) property_list[pending]["identified"] = True
else: property_list[pending]["identusername"] = cmdrun["identify"][0]
target = text.split(" ")[1] temp_mode = cmdrun["identify"][1]
if target.lower() in lower_nicks: property_list[pending]["modes"] = temp_mode
target = lower_nicks[target.lower()] connection.sendall(bytes(f":{pending} MODE {pending} +{temp_mode}\r\n","UTF-8"))
if target in property_list: processedExternally = True
who_user = property_list[target]["username"] break
who_realname = property_list[target]["realname"] if processedExternally:
who_host = property_list[target]["host"] pass
who_identified = property_list[target]["identified"] elif command == "JOIN":
who_ssl = property_list[target]["ssl"] channels = text.split(" ")[1]
if who_identified: if channels[0] == ":":
who_identifying = property_list[target]["identusername"] channels = channels[1:]
else: for channelt in channels.split(","):
who_identifying = None channel = channelt.strip()
try: if channel.lower() in lower_chans:
who_flags = property_list[target]["modes"] channel = lower_chans[channel.lower()]
except: success = True
who_flags = None if channel in channels_list:
connection.sendall(bytes(f":{server} 311 {pending} {target} {who_user} {who_host} * :{who_realname}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 312 {pending} {target} {server} :{identifier}\r\n","UTF-8"))
if "o" in who_flags: connection.sendall(bytes(f":{server} 313 {pending} {target} :is an IRC operator\r\n","UTF-8"))
who_away = property_list[target]["away"]
if who_away:
who_reason = who_away = property_list[target]["reason"]
connection.sendall(bytes(f":{server} 301 {pending} {target} :{who_reason}\r\n","UTF-8"))
if who_identified:
connection.sendall(bytes(f":{server} 330 {pending} {target} {who_identifying} :is logged in as\r\n","UTF-8"))
if who_ssl:
connection.sendall(bytes(f":{server} 671 {pending} {target} :is using a secure connection\r\n","UTF-8"))
#connection.sendall(bytes(f":{server} 317 {pending} {target} {time} :seconds idle\r\n","UTF-8")) # I haven't implemented idle time yet.
if who_flags != None and who_flags != "iw":
connection.sendall(bytes(f":{server} 379 {pending} {target} :Is using modes +{who_flags}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 318 {pending} {target} :End of /WHOIS list\r\n","UTF-8"))
else:
connection.sendall(bytes(f":{server} 401 {pending} {target} :No such nick/channel\r\n","UTF-8"))
elif command == "NAMES":
if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
else:
channel = text.split(" ")[1]
if channel in channels_list:
if pending in channels_list[channel]: if pending in channels_list[channel]:
users = " ".join(channels_list[channel]) success=False
connection.sendall(bytes(f":{server} 353 {pending} = {channel} :{users}\r\n","UTF-8")) print(f"{pending} is already in {channel} , ignoring JOIN request.")
connection.sendall(bytes(f":{server} 366 {pending} {channel} :End of /NAMES list.\r\n","UTF-8")) if success:
elif command == "NOTICE": try:
if len(args) >= 2: if channel in channels_list:
target = text.split(" ")[1] channels_list[channel].append(pending)
if target.lower() in lower_nicks: else:
target = lower_nicks[target.lower()] channels_list[channel] = [pending]
if target in channels_list: lower_chans[channel.lower()] = channel
if pending in channels_list[target]: topic_list[channel] = "Topic is not implemented."
except:
connection.sendall(bytes(f":{server} NOTICE * :*** Could not join {channel}\r\n","UTF-8"))
print(channels_list)
for i in channels_list[channel]: for i in channels_list[channel]:
try: try:
if i != pending: nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} JOIN {channel}\r\n","UTF-8"))
nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
except: except:
pass pass
elif target in nickname_list: # Code re-used in the NAMES command
nickname_list[target].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8")) if channel in channels_list:
if pending in channels_list[channel]:
users = " ".join(channels_list[channel])
connection.sendall(bytes(f":{server} 353 {pending} = {channel} :{users}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 366 {pending} {channel} :End of /NAMES list.\r\n","UTF-8"))
print("Successfully pre-loaded /NAMES list")
elif command == "LIST":
connection.sendall(bytes(f":{server} 321 {pending} Channel :Users Name\r\n","UTF-8"))
for key, value in topic_list.items():
usersin = len(channels_list[key])
connection.sendall(bytes(f":{server} 322 {pending} {key} {usersin} :{value}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 323 {pending} :End of /LIST\r\n","UTF-8"))
elif command == "PONG":
e = text.split(" ")[1]
if e == server or e == f":{server}":
print(pending + " replied to PING.")
property_list[pending]["last_ping"] = time.time()
property_list[pending]["ping_pending"] = False
elif command == "NICK":
if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
elif text.split(" ")[1] == pending:
pass
else: else:
connection.sendall(bytes(f":{server} 401 {pending} {target} :No such nick/channel\r\n","UTF-8")) pending2 = text.split(" ")[1]
else: if pending2[0] == ":": pending2 = pending2[1:]
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8")) if "!" in pending2 or ":" in pending2 or "#" in pending2 or "*" in pending2:
elif command == "QUIT": connection.sendall(bytes(f":{server} 432 {pending} {pending2} :Erroneus nickname\r\n","UTF-8"))
# Parse the quit message. elif pending2.lower() in lower_nicks:
done = [] connection.sendall(bytes(f":{server} 433 {pending} {pending2} :Nickname is already in use.\r\n","UTF-8"))
if len(text.split(" ")) == 1: else:
msg = "Client Quit" print("Sending nickname change...")
else: connection.sendall(bytes(f":{pending}!{rident}@{hostname} NICK {pending2}\r\n","UTF-8"))
msg = text.split(" ")[1:] # Broadcast the nickname change
if msg[0][0] == ":": done = []
msg[0]=msg[0][1:] for i, users in channels_list.items():
if len(msg) > 0: if pending in users:
mse = " ".join(msg) for j in users:
msg = f"Quit: {mse}" if j != pending and j != pending2 and not j in done:
text = f"QUIT :{msg}" print("Broadcasting on " + j)
# Broadcast all users in the joined channels that the person quit. nickname_list[j].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
for i, users in channels_list.items(): done.append(j)
if pending in users: # Replace the nickname
for j in users: try:
if j != pending and not j in done: print("Changing on " + i)
nickname_list[j].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8")) channels_list[i].remove(pending)
done.append(j) channels_list[i].append(pending2)
# Remove the quitting user from the channel. except:
print(traceback.format_exc())
print("Moving config...")
property_list[pending2] = property_list.pop(pending)
nickname_list[pending2] = nickname_list.pop(pending)
del lower_nicks[pending.lower()]
lower_nicks[pending2.lower()] = pending2
print("starting pinger...")
pending = pending2
property_list[pending2]["ping_pending"] = False
property_list[pending2]["last_ping"] = time.time()
threading.Thread(target=pinger, args=[pending, connection]).start()
print(f"User {pending} set nick")
print("Broadcasting nickname change...")
elif command == "PART":
if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
else:
channel = text.split(" ")[1]
for i in channels_list[channel]:
try:
nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
except:
pass
try: try:
channels_list[i].remove(pending) channels_list[channel].remove(pending)
except: except:
print(traceback.format_exc()) print(traceback.format_exc())
# Confirm QUIT and close the socket. elif command == "AWAY":
try: if len(args) == 0:
connection.sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8")) property_list[pending]["away"] = False
connection.sendall(bytes(f"ERROR :Closing Link: {hostname} ({msg})\r\n","UTF-8")) connection.sendall(bytes(f":{server} 305 {pending} :You are no longer marked as being away\r\n","UTF-8"))
finally: else:
connection.close() reasons = " ".join(args)
safe_quit = True if reasons[0] == ":": reasons = reasons[1:]
break property_list[pending]["away"] = True
elif command == "MODE": property_list[pending]["reason"] = reasons
target = args[0] connection.sendall(bytes(f":{server} 306 {pending} :You have been marked as being away\r\n","UTF-8"))
if len(args) == 0: elif command == "WHO":
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8")) if len(args) == 0:
elif len(args) == 1: connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
if args[0] == pending: else:
yourmodes = property_list[pending]["modes"] channel = text.split(" ")[1]
connection.sendall(bytes(f":{server} 221 {pending} +{yourmodes}\r\n","UTF-8")) if channel in channels_list:
elif args[0] in channels_list: for i in channels_list[channel]:
if args[0] in property_list: who_host = property_list[i]["host"]
if "modes" in property_list[args[0]]: who_user = property_list[i]["username"]
# Get the modes + parameters, then print them out. who_realname = property_list[i]["realname"]
modes = property_list[args[0]]["modes"] who_away = "G" if property_list[i]["away"] else "H"
params = property_list[args[0]]["params"] connection.sendall(bytes(f":{server} 352 {pending} {channel} {who_user} {who_host} {server} {i} {who_away} :0 {who_realname}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 221 {pending} {target} +{modes} {params}\r\n","UTF-8")) elif channel in nickname_list:
who_host = property_list[channel]["host"]
who_user = property_list[channel]["username"]
who_realname = property_list[channel]["realname"]
who_away = "G" if property_list[channel]["away"] else "H"
connection.sendall(bytes(f":{server} 352 {pending} * {who_user} {who_host} {server} {channel} {who_away} :0 {who_realname}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 315 {pending} {channel} :End of /WHO list.\r\n","UTF-8"))
elif command == "WHOIS":
if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
else:
target = text.split(" ")[1]
if target.lower() in lower_nicks:
target = lower_nicks[target.lower()]
if target in property_list:
who_user = property_list[target]["username"]
who_realname = property_list[target]["realname"]
who_host = property_list[target]["host"]
who_identified = property_list[target]["identified"]
who_ssl = property_list[target]["ssl"]
if who_identified:
who_identifying = property_list[target]["identusername"]
else:
who_identifying = None
try:
who_flags = property_list[target]["modes"]
except:
who_flags = None
connection.sendall(bytes(f":{server} 311 {pending} {target} {who_user} {who_host} * :{who_realname}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 312 {pending} {target} {server} :{identifier}\r\n","UTF-8"))
if "o" in who_flags: connection.sendall(bytes(f":{server} 313 {pending} {target} :is an IRC operator\r\n","UTF-8"))
who_away = property_list[target]["away"]
if who_away:
who_reason = who_away = property_list[target]["reason"]
connection.sendall(bytes(f":{server} 301 {pending} {target} :{who_reason}\r\n","UTF-8"))
if who_identified:
connection.sendall(bytes(f":{server} 330 {pending} {target} {who_identifying} :is logged in as\r\n","UTF-8"))
if who_ssl:
connection.sendall(bytes(f":{server} 671 {pending} {target} :is using a secure connection\r\n","UTF-8"))
#connection.sendall(bytes(f":{server} 317 {pending} {target} {time} :seconds idle\r\n","UTF-8")) # I haven't implemented idle time yet.
if who_flags != None and who_flags != "iw":
connection.sendall(bytes(f":{server} 379 {pending} {target} :Is using modes +{who_flags}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 318 {pending} {target} :End of /WHOIS list\r\n","UTF-8"))
else:
connection.sendall(bytes(f":{server} 401 {pending} {target} :No such nick/channel\r\n","UTF-8"))
elif command == "NAMES":
if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
else:
channel = text.split(" ")[1]
if channel in channels_list:
if pending in channels_list[channel]:
users = " ".join(channels_list[channel])
connection.sendall(bytes(f":{server} 353 {pending} = {channel} :{users}\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 366 {pending} {channel} :End of /NAMES list.\r\n","UTF-8"))
elif command == "NOTICE":
if len(args) >= 2:
target = text.split(" ")[1]
if target.lower() in lower_nicks:
target = lower_nicks[target.lower()]
if target in channels_list:
if pending in channels_list[target]:
for i in channels_list[channel]:
try:
if i != pending:
nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
except:
pass
elif target in nickname_list:
nickname_list[target].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
else:
connection.sendall(bytes(f":{server} 401 {pending} {target} :No such nick/channel\r\n","UTF-8"))
else:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
elif command == "QUIT":
# Parse the quit message.
done = []
if len(text.split(" ")) == 1:
msg = "Client Quit"
else:
msg = text.split(" ")[1:]
if msg[0][0] == ":":
msg[0]=msg[0][1:]
if len(msg) > 0:
mse = " ".join(msg)
msg = f"Quit: {mse}"
text = f"QUIT :{msg}"
# Broadcast all users in the joined channels that the person quit.
for i, users in channels_list.items():
if pending in users:
for j in users:
if j != pending and not j in done:
nickname_list[j].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
done.append(j)
# Remove the quitting user from the channel.
try:
channels_list[i].remove(pending)
except:
print(traceback.format_exc())
# Confirm QUIT and close the socket.
try:
connection.sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
connection.sendall(bytes(f"ERROR :Closing Link: {hostname} ({msg})\r\n","UTF-8"))
finally:
connection.close()
safe_quit = True
break
elif command == "MODE":
target = args[0]
if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
elif len(args) == 1:
if args[0] == pending:
yourmodes = property_list[pending]["modes"]
connection.sendall(bytes(f":{server} 221 {pending} +{yourmodes}\r\n","UTF-8"))
elif args[0] in channels_list:
if args[0] in property_list:
if "modes" in property_list[args[0]]:
# Get the modes + parameters, then print them out.
modes = property_list[args[0]]["modes"]
params = property_list[args[0]]["params"]
connection.sendall(bytes(f":{server} 221 {pending} {target} +{modes} {params}\r\n","UTF-8"))
else:
# Default channel mode
connection.sendall(bytes(f":{server} 324 {pending} {target} +n\r\n","UTF-8"))
else: else:
# Default channel mode # Default channel mode
connection.sendall(bytes(f":{server} 324 {pending} {target} +n\r\n","UTF-8")) connection.sendall(bytes(f":{server} 324 {pending} {target} +n\r\n","UTF-8"))
else: else:
# Default channel mode if args[0][0] == "#":
connection.sendall(bytes(f":{server} 324 {pending} {target} +n\r\n","UTF-8")) connection.sendall(bytes(f":{server} 403 {pending} {target} :No such channel\r\n","UTF-8"))
else: else:
if args[0][0] == "#": connection.sendall(bytes(f":{server} 505 {pending} :Cant change mode for other users\r\n","UTF-8"))
connection.sendall(bytes(f":{server} 403 {pending} {target} :No such channel\r\n","UTF-8"))
else:
connection.sendall(bytes(f":{server} 505 {pending} :Cant change mode for other users\r\n","UTF-8"))
elif command == "CATSERV" or (command == "PRIVMSG" and args[0].lower() == "catserv"): elif command == "CATSERV" or (command == "PRIVMSG" and args[0].lower() == "catserv"):
if command == "PRIVMSG": if command == "PRIVMSG":
args = args[1:] args = args[1:]
if args[0][0] == ":": if args[0][0] == ":":
args[0] = args[0][1:] args[0] = args[0][1:]
if len(args) == 0: if len(args) == 0:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8")) connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
elif args[0].upper() == "PULL": elif args[0].upper() == "PULL":
updater = subprocess.run(["git", "pull"], stdout=subprocess.PIPE) updater = subprocess.run(["git", "pull"], stdout=subprocess.PIPE)
if updater.stdout.decode().strip() == "Already up to date.": if updater.stdout.decode().strip() == "Already up to date.":
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :Codename IRCat is already up-to-date.\r\n","UTF-8")) connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :Codename IRCat is already up-to-date.\r\n","UTF-8"))
else:
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :Done, it is recommended to use /RESTART if you're an IRC op\r\n","UTF-8"))
elif args[0].upper() == "VERSION":
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :Codename IRCat version {__version__}\r\n","UTF-8"))
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :This is Codename IRCat's integrated services.\r\n","UTF-8"))
else: else:
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :Done, it is recommended to use /RESTART if you're an IRC op\r\n","UTF-8")) connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :CatServ Usage:\r\n","UTF-8"))
elif args[0].upper() == "VERSION": connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :PULL - Pulls the latest version of Codename IRCat\r\n","UTF-8"))
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :Codename IRCat version {__version__}\r\n","UTF-8")) connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :VERSION - Gets the version number of this service.\r\n","UTF-8"))
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :This is Codename IRCat's integrated services.\r\n","UTF-8")) elif command == "RESTART":
else: if "o" in property_list[pending]["modes"]:
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :CatServ Usage:\r\n","UTF-8")) global opened
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :PULL - Pulls the latest version of Codename IRCat\r\n","UTF-8")) opened = False
connection.sendall(bytes(f":CatServ!Meow@IRCatCore NOTICE {pending} :VERSION - Gets the version number of this service.\r\n","UTF-8"))
elif command == "RESTART":
if "o" in property_list[pending]["modes"]:
global opened
opened = False
else:
connection.sendall(bytes(f":{server} 481 {pending} :Permission Denied- You're not an IRC operator\r\n","UTF-8"))
elif command == "PRIVMSG":
if len(args) >= 2:
target = text.split(" ")[1]
if target.lower() in lower_nicks:
target = lower_nicks[target.lower()]
if target in channels_list:
print("Sending to "+ target + " Channel")
if pending in channels_list[target]:
print(channels_list[target])
for i in channels_list[target]:
try:
if i != pending:
print(i)
print(f":{pending}!{rident}@{hostname} {text}\r\n")
nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
else:
print(i + " Is the current user!")
except:
print(traceback.format_exc)
elif target in nickname_list:
nickname_list[target].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
else: else:
connection.sendall(bytes(f":{server} 401 {pending} {target} :No such nick/channel\r\n","UTF-8")) connection.sendall(bytes(f":{server} 481 {pending} :Permission Denied- You're not an IRC operator\r\n","UTF-8"))
elif command == "PRIVMSG":
if len(args) >= 2:
target = text.split(" ")[1]
if target.lower() in lower_nicks:
target = lower_nicks[target.lower()]
if target in channels_list:
print("Sending to "+ target + " Channel")
if pending in channels_list[target]:
print(channels_list[target])
for i in channels_list[target]:
try:
if i != pending:
print(i)
print(f":{pending}!{rident}@{hostname} {text}\r\n")
nickname_list[i].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
else:
print(i + " Is the current user!")
except:
print(traceback.format_exc)
elif target in nickname_list:
nickname_list[target].sendall(bytes(f":{pending}!{rident}@{hostname} {text}\r\n","UTF-8"))
else:
connection.sendall(bytes(f":{server} 401 {pending} {target} :No such nick/channel\r\n","UTF-8"))
else:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8"))
# Ignore empty text
elif text.split(" ")[0] == "":
pass
else: else:
connection.sendall(bytes(f":{server} 461 {pending} {command} :Not enough parameters\r\n","UTF-8")) # Unknown command
# Ignore empty text cmd = text.split(" ")[0]
elif text.split(" ")[0] == "": connection.sendall(bytes(f":{server} 421 {pending} {cmd} :Unknown command\r\n","UTF-8"))
pass pendingCommands = ""
else: else:
# Unknown command pendingCommands += text
cmd = text.split(" ")[0]
connection.sendall(bytes(f":{server} 421 {pending} {cmd} :Unknown command\r\n","UTF-8"))
pendingCommands = ""
else:
pendingCommands += text
except ssl.SSLEOFError: except ssl.SSLEOFError:
print("EOF occured...") print("EOF occured...")
except Exception as ex: except Exception as ex: