This script is basically a proxy service; except it doesn't require an active client connection. In fact it expects the client connection to connect before, and disconnect after every transaction. Once running the proxy will buffer all incoming bytes, and will dump those bytes to the client upon request. Before I go any further and begin working on the IRC client intended to be used with this, I am posting here to persuade any feedback regarding the project in its current state from readers here:


#! /usr/bin/python3

from socket import socket
from threading import Thread as thread
from sys import getdefaultencoding as enc
from sys import byteorder
from sys import argv

enc = enc()
CNCT, SEND, RECV, DISC, STAT = 0, 1, 2, 3, 4

class sox:
        def recv_t(self, s):
                while self.run:
                        try: self.buff += s.recv(8192)
                        except: continue
        def process(self, lo):
                try: bs = lo[0].recv(8192)
                except: return
                op, data = bs[0], bs[1:]
                if op is CNCT:
                        data = str(data, enc)
                        host, port = data.split(":")
                        self.sock = socket()
                        self.sock.connect((host, int(port)))
                        self.run = True
                        thread(target=self.recv_t, args=(self.sock, )).start()
                elif op is SEND: self.sock.send(data)
                elif op is RECV: lo[0].send(self.buff); self.buff = bytes()
                elif op is DISC: self.run = False; self.sock.close()
                elif op is STAT: bytes("error: %s" % (str(self.serr), ), enc)
        def server(self, port, host):
                ss = socket()
                ss.bind((host, int(port)))
                while True:
        def __init__(self, argv):
                self.sock = None
                self.buff = bytes()
                self.serr = False
                self.server(argv[1], argv[2])

if __name__ == "__main__":
        sox([argv + ["", ], argv][len(argv) == 3])


#! /usr/bin/python3

from soxxy import *

lo_host = None
lo_port = None
remote_host = None
remote_port = None

def send(bs):
        s = socket()
        s.connect((lo_host, int(lo_port)))
        x = s.recv(8192)
        return x

if __name__ == "__main__":
        lo_host, lo_port, remote_host, remote_port = argv[1:]
        send(bytes([CNCT, ] + [ord(c) for c in remote_host + ":" + remote_port]))
        from time import sleep
        print(str(send(bytes([RECV, ])), enc))
        send(bytes([DISC, ]))

Test case

$ cat test.sh 
./soxxy.py 8000 & sleep 2 && \
./irc 8000 irc.gnu.org 6667
$ sh test.sh 
:morgan.freenode.net NOTICE * :*** Looking up your hostname...
:morgan.freenode.net NOTICE * :*** Checking Ident
:morgan.freenode.net NOTICE * :*** Found your hostname
:morgan.freenode.net NOTICE * :*** No Ident response

1 Answer 1


I believe enc = enc() is considered an anti-pattern, because the semantics of enc changes. I'd suggest renaming the variable to encoding.

Rather than a long if..elif..else block you could use a simple handle_operation(op) call.

Since there's effectively no exception handling it will be difficult to diagnose errors.

Using argparse would make it possible to understand how to use the code without reading it. As a bonus it adds -h/--help support by default so users can see what the expected call structure is.

sleeping to wait for an end point to be available is an anti-pattern. To guarantee that an end point is up and that you get started processing as quickly as possible it's common to poll the end point until it responds correctly or fail if there's a timeout.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.