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:devPycat

> echo hooooooi

hooooooi

username@hostame PyCat C:devPyCat

> 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 443npython3 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())

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'

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.