PyCat A small implementation of netcat in Python3.x

Question

Intro

netcat is an all-round tool used with many applicable features

I was playing around with sockets and openssl, and decided I should build my own.

It works as expected, but my code feels a bit smelly.

Features

• ssl
• persistent shell
• cd
• exit

Example

server

$ python pycat.py -lvp 8080 --ssl
[*] Incoming connection from 127.0.0.1:53391
username@hostame PyCat C:\dev\Pycat
> echo hooooooi
hooooooi
username@hostame PyCat C:\dev\PyCat
> cd ../

username@hostame PyCat C:\dev
> exit

client

python pycat.py -i localhost -p 8080 --ssl

Code

import subprocess
import tempfile
import datetime
import socket
import ssl
import sys
import os
import re

import argparse
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.backends import default_backend

class PyCat():
    def __init__(self, host, port, listen, verbose, _ssl):
        self.buffer = b""
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.port = port
        self.host = host if host else '0.0.0.0'
        self.listen = listen
        self.verbose = verbose
        self.ssl = _ssl
        if self.ssl:
            self.context = ssl.create_default_context()
            if self.listen:
                self.key_file, self.cert_file = self.generate_temp_cert()
        self.main_func = self.nc_listen if self.listen else self.nc_connect
        self.main()

    def generate_temp_cert(self):
        key, key_path = tempfile.mkstemp()
        cert, cert_path = tempfile.mkstemp()
        name_attributes = [
            x509.NameAttribute(NameOID.COUNTRY_NAME, "OK"),
            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "OK"),
            x509.NameAttribute(NameOID.LOCALITY_NAME, "OK"),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "OK"),
            x509.NameAttribute(NameOID.COMMON_NAME, "PyCat")
        ]

        key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
            backend=default_backend()
        )

        with open(key_path, "wb") as f:
            f.write(
                key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.TraditionalOpenSSL,
                    encryption_algorithm=serialization.NoEncryption()
                )
            )

        subject = issuer = x509.Name(name_attributes)

        cert = x509.CertificateBuilder()\
                    .subject_name(subject)\
                    .issuer_name(issuer)\
                    .public_key(key.public_key())\
                    .serial_number(x509.random_serial_number())\
                    .not_valid_before(datetime.datetime.utcnow())\
                    .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=365))

        cert = cert.sign(key, hashes.SHA256(), default_backend())
        with open(cert_path, "wb") as f:
            f.write(
                cert.public_bytes(serialization.Encoding.PEM)
            )

        return key_path, cert_path

    def main(self):
        self.main_func()

    def exit(self):
        self.socket.close()
        sys.exit(0)

    def read(self, socket_conn, length=1024):
        data, response = "starting", b""
        while data:
            data = socket_conn.recv(length)
            response += data
            if len(data) < length:
                break
        return response.decode("utf-8").rstrip("\n")

    def handle_command(self, cmd):
        response = b" "
        cd = re.match(r'cd(?:\s+|$)(.*)', cmd)
        if cmd == "exit":
            self.exit()
        elif cd and cd.group(1):
            try:
                os.chdir(cd.group(1))
            except FileNotFoundError:
                pass
        else:
            response = self.exec_command(cmd)
        return response

    def exec_command(self, command):
        try:
            output = subprocess.check_output(command, stderr=subprocess.STDOUT, shell=True)
        except Exception as e:
            output = str(e).encode("utf-8")
        return output

    def nc_connect(self):
        self.socket.connect((self.host, self.port))

        if self.ssl:
            self.context.check_hostname = False
            self.context.verify_mode = ssl.CERT_NONE
            self.socket = self.context.wrap_socket(self.socket)

        while True:
            cmd = self.read(self.socket)
            response = self.handle_command(cmd)
            self.socket.send(response)

    def create_prompt_string(self, client_socket):
        client_socket.send(b"cd")
        pwd = self.read(client_socket)
        client_socket.send(b"whoami")
        whoami = self.read(client_socket)
        client_socket.send(b"hostname")
        hostname = self.read(client_socket)
        return f"{whoami}@{hostname} PyCat {pwd}\n> "

    def client_handler(self, client_socket):
        while True:
            prompt_string = self.create_prompt_string(client_socket)
            buf = input(f"{prompt_string}")
            client_socket.send(buf.encode("utf-8"))

            if buf == "exit":
                self.exit()
                if self.ssl:
                    os.remove(self.cert_file)
                    os.remove(self.key_file)

            print(self.read(client_socket))

    def nc_listen(self):
        self.socket.bind((self.host, self.port))
        self.socket.listen(0)

        if self.ssl:
            self.socket = ssl.wrap_socket(
                self.socket,
                server_side=True, 
                certfile=self.cert_file, 
                keyfile=self.key_file
            )

        client_socket, addr = self.socket.accept()
        if self.verbose:
            ip, port = addr
            print(f"[*] Incoming connection from {ip}:{port}")

        self.client_handler(client_socket)

def parse_arguments():
    parser = argparse.ArgumentParser(usage='%(prog)s [options]',
                                     description='PyCat @Ludisposed',
                                     formatter_class=argparse.RawDescriptionHelpFormatter,
                                     epilog='Examples:\npython3 pycat.py -lvp 443\npython3 pycat.py -i localhost -p 443')
    parser.add_argument('-l', '--listen', action="store_true", help='Listen')
    parser.add_argument('-v', '--verbose', action="store_true", help='Verbose output')
    parser.add_argument('-s', '--ssl', action="store_true", help='Encrypt connection')
    parser.add_argument('-p', '--port', type=int, help='Port to listen on')
    parser.add_argument('-i', '--ip', type=str, help='Ip to connect to')
    args = parser.parse_args()

    if (args.listen or args.ip) and not args.port:
        parser.error('Specify which port to connect to')
    elif not args.listen and not args.ip:
        parser.error('Specify --listen or --ip')
    return args.ip, args.port, args.listen, args.verbose, args.ssl

if __name__ == '__main__':
    PyCat(*parse_arguments())

Answer 1

I/O problems

I trust that this code works on your machine, but it did not work for me (Homebrew Python 3.7.1 on macOS 10.14.2). As soon as the client connects, both sides crash. The client's stack trace:

$ python3 pycat.py -i localhost -p 8080 --ssl
Traceback (most recent call last):
  File "pycat.py", line 188, in <module>
    PyCat(*parse_arguments())
  File "pycat.py", line 31, in __init__
    self.main()
  File "pycat.py", line 78, in main
    self.main_func()
  File "pycat.py", line 125, in nc_connect
    self.socket.send(response)
  File "/usr/local/Cellar/python/3.7.1/Frameworks/Python.framework/Versions/3.7/lib/python3.7/ssl.py", line 984, in send
    return self._sslobj.write(data)
ssl.SSLEOFError: EOF occurred in violation of protocol (_ssl.c:2324)

If I try to launch both the server and client without --ssl, then after the server reports that it accepted the incoming connection, both sides just sit there and do nothing, no matter what I type.

I suspect that the default I/O buffering behavior in Windows is different from Unix.

Object-oriented design

In OOP, classes are nouns. A constructor should initialize an object. The constructor should not also start actively doing things. The action happens when someone calls a method (a verb) on the object.

I don't think that the PyCat class is a good one. Its behavior completely switches depending on self.main_func = self.nc_listen if self.listen else self.nc_connect! Consider the calling chains for the server mode compared to the client mode:

Server mode:

• PyCat.__init__()
  • self.generate_temp_cert()
  • self.main()
  • self.nc_listen()
    • self.socket.…
    • self.client_handler(…)
    • self.create_prompt_string(…)
      • self.read(…)
    • self.read(…)

Client mode:

• PyCat.__init__()
  • self.main()
  • self.nc_connect()
    • self.socket.connect(…)
    • self.context.…
    • self.read(…)
    • self.handle_command(…)
    • self.exec_command(…)
    • self.socket.send(…)

There is very little in common between the two modes. There is, therefore, no advantage to cramming them into the same class. All that the two modes have in common, really, are some socket-creation parameters and the ability to read and write. That should be a class used in some composition, or perhaps a base class in some inheritance hierarchy.

Nitpicks

Imports should be listed in alphabetical order. The argparse import belongs with the first group of imports.

Having parse_arguments() return a tuple, whose elements are in the same order that the PyCat constructor expects, is fragile. Rather, parse_arguments should return a dictionary, and you can call PyCat(**parse_arguments()). Note that what the argument parser calls an "ip" is what the PyCat class inconsistently calls the "host". Also, the _ssl parameter is ugly. If you're worried about name collisions, you can rename the parameter to use_ssl, or change the import statement in one of these ways:

from ssl import CERT_NONE, create_default_context, wrap_socket
import ssl as ssllib

A shorter way to write

self.host = host if host else '0.0.0.0'

would be

self.host = host or '0.0.0.0'

Answer 2

self.host = host if host else '0.0.0.0'

Since this seems to be optional, perhaps you should given the corresponding __init__ argument a default value (None?). Same with _ssl, if it's an object. If it's a boolean, rename it to _use_ssl or something; otherwise its purpose is not clear.

This:

self.main_func = self.nc_listen if self.listen else self.nc_connect
self.main()
# ...

def main(self):
    self.main_func()

can be simplified to:

self.main = self.nc_listen if self.listen else self.nc_connect
self.main()

However, I advise that you don't call main() from the constructor. Init is for init. The entire app shouldn't run from the constructor. Call main() from the top level after construction.

One nicety would be to write a static method PyCat.from_command_line which returns a new instance of PyCat based on arg parsing, which would also be a static method. That way, the class contains more of its business logic and is more useful on its own.

generate_temp_cert can - and should - be a @staticmethod, as well. It doesn't reference self.

You should turn the class into a context manager, since you have a socket that needs to be closed. For more information do some reading here - https://docs.python.org/3/library/stdtypes.html#typecontextmanager - and then use a with at the top level.

Your Python file, since it's executable, is missing a shebang at the top:

#!/usr/bin/env python3

Your little hack to initialize data to starting to work around the first loop iteration is not advisable. There's also a bug where you don't actually check data directly after your recv, so you could get back an unexpected None. Instead, do something like:

response = b''
while True:
    data = socket_conn.recv(length)
    if not data:
        break
    # ...

The concatenation you're doing on response might also be problematic. If you're in Python 2 it's still (strangely) the best thing to do; however, if you're in Python 3, this kind of concatenation can become very inefficient, and you should be using a BytesIO instead.

You should check if cmd == "exit" before doing your regex, because you might just throw the results away anyway.

subprocess.check_output(command, stderr=subprocess.STDOUT, shell=True)

Be very careful about this. shell=True on its own is a security risk. Allowing a network-connected, unauthenticated application to run arbitrary shell commands is a gaping security hole.