Netcat

Netcat can establish connections and will forward everything on the stardard input (e.g. your keyboard) to the socket and the other way around.

Start a UDP server on port 8000:

nc -u -l localhost -p 8000

Start a client:

nc -u localhost 8000

TCP version

Just omit the -u:

nc -l localhost -p 8000
nc localhost 8000

But I came for Python!

Netcat is an elegant classic. But what if we want something more “automatic”?

Server

#!/usr/bin/python3

# (C) 2022 Massimo Girondi - CC BY-NC-SA 4.0

import socket
import sys
if len(sys.argv) !=2:
  print("Usage:", sys.argv[0], "address:port")
  exit(1)

addr, port = sys.argv[1].split(":")
port = int(port)

server = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
server.bind((addr, port))
while True:
	recv = server.recvfrom(1024)
	m = recv[0].decode()
	add = recv[1]
	print("Message %s from %s" % (m, add))
	if m == "ping":
		server.sendto(str.encode("pong"), add)

Client

#!/usr/bin/python3

# (C) 2022 Massimo Girondi - CC BY-NC-SA 4.0

import socket
import sys
if len(sys.argv) !=2:
  print("Usage:", sys.argv[0], "address:port")
  exit(1)

addr, port = sys.argv[1].split(":")
port = int(port)

message = str.encode("ping")

client = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)

client.sendto( message, (addr, port))
try:
	# Wait at most 2 seconds for the response
	client.settimeout(2)
	recv = client.recvfrom(1024)
	m = recv[0].decode()
	add = recv[1]
	print("Message %s from %s" % (m, add))
	if m == "pong":
		exit(0)
	else:
		exit(1)
except socket.timeout:
	exit(1)

And for TCP?

The scripts above could be easily changed to use a TCP socket. But let’s say that our firewall does HTTP inspection, and multiple connections are needed.

HTTPServer, the replacement for the classic SimpleHTTPServer works nicely, but needs a bit of tweaking to reach our goal: we need a multi-thread server (explotiing ThreadMixIn) and to handle all the different HTTP (chaniging the HTTPRequestHandler).

#!/usr/bin/python3
# (C) 2022 Massimo Girondi - CC BY-NC-SA 4.0

from http.server import HTTPServer
from http.server import BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
import socket
import threading
import sys


class RequestHandler(BaseHTTPRequestHandler):

    # Generate responses with some content
    def response(self,method = "GET", code=200):
        self.send_response(code)
        self.end_headers()
        cl_a, cl_p = self.client_address
        response = "Hi "+ cl_a + ":" + str(cl_p) + "!\n"
        response += "You asked " + method +" at " + self.path +".\n"
        response += "My job here is finished.\n"
        self.wfile.write(response.encode())

    # Implement here the methods needed
    def do_GET(self):
        self.response("GET")
    def do_POST(self):
        self.response("POST")
    def do_PUT(self):
        self.response("PUT")
    def do_UPDATE(self):
        self.response("UPDATE")
    def do_DELETE(self):
        self.response("DELETE")
    def do_INSERT(self):
        self.response("INSERT")
    def do_HEAD(self):
        self.response("HEAD")
    def do_TRACE(self):
        self.response("TRACE")
    def do_CONNECT(self):
        self.response("CONNECT")
    def do_OPTIONS(self):
        self.response("OPTIONS")


# This is the class that will spawn our threads
class MultiThreadServer(ThreadingMixIn, HTTPServer):
    pass

# And here the  startup logic
"""
Call it as
http_server.py -> server will listen on 0.0.0.0:8080
http_server.py 8081 -> server will listen on 0.0.0.0:8081
http_server.py 127.0.0.1 8081 -> server will listen on 127.0.0.1:8081
http_server.py 127.0.0.1:8081 -> server will listen on 127.0.0.1:8081

"""
if __name__ == '__main__':
    address = "0.0.0.0"
    port = "8080"
    if len(sys.argv) == 3:
      address = sys.argv[1]
      port = sys.argv[2]
    elif len(sys.argv) == 2 and ":" in sys.argv[1]:
      address, port = sys.argv[1].split(":")
    elif len(sys.argv) == 2:
      port = sys.argv[1]
    print(f"Server will listen on {address}:{port}")
    server = MultiThreadServer((address, int(port)), RequestHandler)
    server.serve_forever()

The above could be tested with curl -X POST localhost:8000, where POST is the HTTP method to test.

The files are available here:

Small Python Servers