Skip to content

Instantly share code, notes, and snippets.

@aveao
Forked from richardg867/proxy.py
Last active Oct 4, 2021
Embed
What would you like to do?
Comic Chat fixer - ugly python3 fork with QoL features and optional twitch support
#
# Comic Chat fixer MITM proxy: fixes Comic Chat to (sort of) work with modern
# IRC servers. Tested with Microsoft Chat 2.5 on Windows XP, 8 and 10
#
# This is a fork by ave with additional fixes and python3 support
# Alongside some other QoL features like color nicks being displayed in chat
# https://gist.github.com/aveao/f6d40f41a424b1d9a54c9aba39742b76
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
import getopt
import re
import socket
import sys
import threading
import ssl
# The debug option prints all MITM'd messages. This can help with debugging.
debug = True
def gen_nick_color_index(nick):
# same as thelounge
# https://github.com/thelounge/thelounge/blob/master/client/js/helpers/colorClass.js
nick_hash = 0
for char in nick:
nick_hash += ord(char)
nick_index = 1 + (nick_hash % 32)
# 8, 24: Yellow
# 7: Dark yellow
ugly_color_map = {8: 7, 24: 7}
# hack
nick_index = ugly_color_map.get(nick_index, nick_index)
return nick_index
def format_nick(
nick, brackets=True, color=True, underscore=True, colon=False, space=True
):
# <\x1f{}\x1f>
# "\x0320example red\x03"
out_nick = nick
if underscore:
out_nick = "\x1f{}\x1f".format(out_nick)
if color:
color_index = gen_nick_color_index(nick)
out_nick = "\x03{}{}\x03".format(color_index, out_nick)
if brackets:
out_nick = "<{}>".format(out_nick)
if colon:
out_nick = ":{}".format(out_nick)
if space:
out_nick = "{} ".format(out_nick)
return out_nick
def thread_c2s(client, client_addr, password, host, port, use_ssl):
f = client.makefile()
queued_lines = []
if password:
client.sendall(b":cchat.proxy 464 * :Password incorrect\r\n")
while True:
line = f.readline().rstrip("\r\n")
if line[:12] == "OPER (null) ":
if password == line[12:]:
print("[-] {0}:{1} authenticated successfully".format(*client_addr))
break
else:
print("[!] {0}:{1} failed to authenticate".format(*client_addr))
client.sendall(b":cchat.proxy 464 * :Password incorrect\r\n")
else:
queued_lines.append(line)
irc = socket.create_connection((host, port))
if use_ssl:
irc = ssl.wrap_socket(irc)
for line in queued_lines:
irc.sendall(line + b"\r\n")
t = threading.Thread(target=thread_s2c, args=(client, client_addr, irc))
t.daemon = True
t.start()
try:
while True:
line = f.readline().replace("\n", "\r\n").encode()
if debug:
print("c2s: " + repr(line))
irc.sendall(line)
if len(line) == 0 or line[:5] == b"QUIT ":
break
except KeyboardInterrupt:
sys.exit(1)
except:
pass
try:
irc.close()
except:
pass
try:
client.close()
except:
pass
def thread_s2c(client, client_addr, irc):
f = irc.makefile()
srv_prefix = "@+"
try:
while True:
line = f.readline()
split = line.split(" ")
if len(split) > 2:
if split[0] == "ERROR":
client.sendall(line)
break
elif split[1] == "005":
# Get PREFIX= to fix ranks in the NAMES response
match = re.search(""" PREFIX=\(([^\)]+)\)([^\s]+)""", line)
if match:
srv_prefix = match.group(2)
elif split[1] in "JOIN" and split[2][0] != ":":
# Main purpose of the proxy. Fixes a crash bug with newer
# ircds, which send JOIN confirmations like this:
#
# :nick!user@host JOIN #channel
#
# instead of this:
#
# :nick!user@host JOIN :#channel
#
# CChat expects the channel name to have a : before the
# name. If it doesn't, it will crash, since it somehow
# attempts a stricmp(0).
split[2] = ":" + split[2]
elif split[1] in "PART" and split[2][0] == ":":
# :nick!user@host PART :#channel
# ->
# :nick!user@host PART #channel :""
split[2] = split[2][1:]
split.append(':""')
elif split[1] == "PRIVMSG":
# Prepend nick to PRIVMSGs
nick = split[0].split("!")[0][1:]
is_mschat_prefixed = split[3].startswith(":(#")
is_mschatappear_prefixed = split[3].startswith(":#")
if is_mschatappear_prefixed:
pass
elif is_mschat_prefixed:
# too lazy, just don't prefix on empty msgs
if len(split) > 4:
split[4] = format_nick(nick) + split[4]
else:
split[3] = format_nick(nick, colon=True) + split[3][1:]
elif split[1] == "353":
# Convert additional ranks to regular op
for i in range(5, len(split)):
rank = ""
nick = ""
for char in split[i]:
if char == "+" and rank != "@":
# voice
rank = "+"
elif char in srv_prefix:
# everything unknown to CChat becomes op
rank = "@"
elif char != ":":
# not a rank
nick += char
split[i] = (split[i][0] == ":" and ":" or "") + rank + nick
line = " ".join(split)
# Comic Chat will stop receiving if it receives a line longer than
# 512 bytes, including the trailing CRLF.
tosend = line.rstrip("\r\n")[:510].encode() + b"\r\n"
if debug:
print("s2c: " + repr(tosend))
client.sendall(tosend)
except KeyboardInterrupt:
sys.exit(1)
try:
irc.close()
except:
pass
try:
client.close()
except:
pass
def main():
bind_host = ""
bind_port = 6461
password = None
options, remainder = getopt.getopt(
sys.argv[1:], "h:p:a:", ["bindhost=", "bindport=", "password="]
)
for opt, arg in options:
if opt in ("-h", "--bindhost"):
bind_host = arg
elif opt in ("-p", "--bindport"):
bind_port = int(arg)
elif opt in ("-a", "--password"):
password = arg
if bind_port < 0 or bind_port > 65535 or len(remainder) < 1:
print(
"Usage: proxy.py [-h bindhost] [-p bindport] [-a password] server [[+]port]"
)
sys.exit(1)
host = remainder[0]
if len(remainder) > 1:
if remainder[1][0] == "+":
use_ssl = True
port = int(remainder[1][1:])
else:
use_ssl = False
port = int(remainder[1])
else:
use_ssl = False
port = 6667
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((bind_host, bind_port))
server.listen(5)
print(f"[-] Waiting for connections at {bind_host}:{bind_port}")
try:
while True:
client, client_addr = server.accept()
print("[-] Connection from {0}:{1}".format(*client_addr))
t = threading.Thread(
target=thread_c2s,
args=(client, client_addr, password, host, port, use_ssl),
)
t.daemon = True
t.start()
except KeyboardInterrupt:
server.close()
sys.exit(1)
if __name__ == "__main__":
main()
#
# Comic Chat fixer MITM proxy: fixes Comic Chat to (sort of) work with modern
# IRC servers. Tested with Microsoft Chat 2.5 on Windows XP, 8 and 10
#
# This is a fork by ave with additional fixes and python3 support
# Alongside some other QoL features like color nicks being displayed in chat
# This is the twitch variant of the code.
# https://gist.github.com/aveao/f6d40f41a424b1d9a54c9aba39742b76
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
import getopt
import re
import socket
import sys
import threading
import ssl
# The debug option prints all MITM'd messages. This can help with debugging.
debug = True
twitch_oauth_token = b"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
twitch_username = b"aaaaaa"
def gen_nick_color_index(nick):
# same as thelounge
# https://github.com/thelounge/thelounge/blob/master/client/js/helpers/colorClass.js
nick_hash = 0
for char in nick:
nick_hash += ord(char)
nick_index = 1 + (nick_hash % 32)
# 8, 24: Yellow
# 7: Dark yellow
ugly_color_map = {8: 7, 24: 7}
# hack
nick_index = ugly_color_map.get(nick_index, nick_index)
return nick_index
def format_nick(
nick, brackets=True, color=True, underscore=True, colon=False, space=True
):
# <\x1f{}\x1f>
# "\x0320example red\x03"
out_nick = nick
if underscore:
out_nick = "\x1f{}\x1f".format(out_nick)
if color:
color_index = gen_nick_color_index(nick)
out_nick = "\x03{}{}\x03".format(color_index, out_nick)
if brackets:
out_nick = "<{}>".format(out_nick)
if colon:
out_nick = ":{}".format(out_nick)
if space:
out_nick = "{} ".format(out_nick)
return out_nick
def thread_c2s(client, client_addr, password, host, port, use_ssl):
f = client.makefile()
queued_lines = [b"PASS oauth:" + twitch_oauth_token, b"NICK " + twitch_username]
irc = socket.create_connection((host, port))
if use_ssl:
irc = ssl.wrap_socket(irc)
for line in queued_lines:
irc.sendall(line + b"\r\n")
t = threading.Thread(target=thread_s2c, args=(client, client_addr, irc))
t.daemon = True
t.start()
try:
while True:
line = f.readline().replace("\n", "\r\n")
linesplit = line.split(" ")
if linesplit[0] in ["WHO", "MODE", "NICK", "USER"]:
continue
bline = line.encode()
if debug:
print("c2s: " + repr(bline))
irc.sendall(bline)
if len(bline) == 0 or bline[:5] == b"QUIT ":
break
except KeyboardInterrupt:
sys.exit(1)
except:
pass
try:
irc.close()
except:
pass
try:
client.close()
except:
pass
def thread_s2c(client, client_addr, irc):
f = irc.makefile()
srv_prefix = "@+"
present_users = []
try:
while True:
line = f.readline()
split = line.split(" ")
if len(split) > 2:
if split[0] == "ERROR":
client.sendall(line)
break
elif split[1] == "005":
# Get PREFIX= to fix ranks in the NAMES response
match = re.search(""" PREFIX=\(([^\)]+)\)([^\s]+)""", line)
if match:
srv_prefix = match.group(2)
elif split[1] in "JOIN" and split[2][0] != ":":
# Main purpose of the proxy. Fixes a crash bug with newer
# ircds, which send JOIN confirmations like this:
#
# :nick!user@host JOIN #channel
#
# instead of this:
#
# :nick!user@host JOIN :#channel
#
# CChat expects the channel name to have a : before the
# name. If it doesn't, it will crash, since it somehow
# attempts a stricmp(0).
split[2] = ":" + split[2]
elif split[1] == "PART" and split[2][0] == ":":
# :nick!user@host PART :#channel
# ->
# :nick!user@host PART #channel :""
split[2] = split[2][1:]
split.append(':""')
elif split[1] == "PRIVMSG":
# Prepend nick to PRIVMSGs
nick = split[0].split("!")[0][1:]
if nick not in present_users:
# CURSED
fake_join = ":{}!{}@twitch.tv JOIN :{}\r\n".format(
nick, nick, split[2]
).encode()
print("s2c: " + repr(fake_join))
client.sendall(fake_join)
present_users.append(nick)
is_mschat_prefixed = split[3].startswith(":(#")
is_mschatappear_prefixed = split[3].startswith(":#")
if is_mschatappear_prefixed:
pass
elif is_mschat_prefixed:
# too lazy, just don't prefix on empty msgs
if len(split) > 4:
split[4] = format_nick(nick) + split[4]
else:
split[3] = format_nick(nick, colon=True) + split[3][1:]
elif split[1] == "353":
# Convert additional ranks to regular op
for i in range(5, len(split)):
rank = ""
nick = ""
for char in split[i]:
if char == "+" and rank != "@":
# voice
rank = "+"
elif char in srv_prefix:
# everything unknown to CChat becomes op
rank = "@"
elif char != ":":
# not a rank
nick += char
split[i] = (split[i][0] == ":" and ":" or "") + rank + nick
line = " ".join(split)
# Comic Chat will stop receiving if it receives a line longer than
# 512 bytes, including the trailing CRLF.
tosend_partone = line.rstrip("\r\n")[:510].encode()
tosend = tosend_partone + b"\r\n"
if debug and tosend_partone:
print("s2c: " + repr(tosend))
client.sendall(tosend)
except KeyboardInterrupt:
sys.exit(1)
try:
irc.close()
except:
pass
try:
client.close()
except:
pass
def main():
bind_host = ""
bind_port = 6461
password = None
options, remainder = getopt.getopt(
sys.argv[1:], "h:p:a:", ["bindhost=", "bindport=", "password="]
)
for opt, arg in options:
if opt in ("-h", "--bindhost"):
bind_host = arg
elif opt in ("-p", "--bindport"):
bind_port = int(arg)
elif opt in ("-a", "--password"):
password = arg
if bind_port < 0 or bind_port > 65535 or len(remainder) < 1:
print(
"Usage: proxy.py [-h bindhost] [-p bindport] [-a password] server [[+]port]"
)
sys.exit(1)
host = remainder[0]
if len(remainder) > 1:
if remainder[1][0] == "+":
use_ssl = True
port = int(remainder[1][1:])
else:
use_ssl = False
port = int(remainder[1])
else:
use_ssl = False
port = 6667
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((bind_host, bind_port))
server.listen(5)
print(f"[-] Waiting for connections at {bind_host}:{bind_port}")
try:
while True:
client, client_addr = server.accept()
print("[-] Connection from {0}:{1}".format(*client_addr))
t = threading.Thread(
target=thread_c2s,
args=(client, client_addr, password, host, port, use_ssl),
)
t.daemon = True
t.start()
except KeyboardInterrupt:
server.close()
sys.exit(1)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment