Let Me Out of Your Net – Egress Testing

by Posts on malicious.link on July 4, 2020

Use-cases:

IT Admin, Firewall Admin, or Security staff at a company and want to confirm what ports and protocols are allowed of your network.
Pentester that wants to identify ports and protocols that can be used for a pentest to gain C2 outbound.
Purple Team testing ports and protocol detection for C2.

Egress testing is an interesting problem due to the uniqueness of most networks. You may find fully open networks like those found in many Silicon Valley companies or companies attempting to move to a “Beyond Corp” model. Or, you may find a network of a small business that hasn’t put much thought into outbound egress, but follow traditional best practices allowing only specific ports out. Or, you may be up against an enterprise that only allow proxied connections outbound with full protocol filtering and analysis.

I created LetMeOutOfYour.net not only check for ports open but also give easily confirm-able responses that you have reached the correct protocol.

The design idea is that the server listens on all ports multiplexed for the most common 3 protocols (HTTP, HTTPS, and SSH). DNS is setup up in a way that all hostnames and sub-domains also route to the LMO host. On SSH the server host key is used to confirm successful, non-modified connections. HTTP is set up to listen not only on any port, but also any URI. HTTPS has a valid SSL certificate thanks to LetsEncrypt.

For example:

Sponsorships Available

https://randomstring2321.letmeoutofyour.net:3821/return/to/path.wsdl Response: (w00tw00t)
http://test.woot.letmeoutofyour.net:22/cool/stuff.php Response: (w00tw00t)

The following is a script to test a range of ports with HTTP, HTTPS, and SSH and confirm the response”

Just beginning with libraries. Paramiko is a SSH library for Python, and Concurrent.Futures is a method for creating concurrency/threading to speed up the script.

#!/usr/bin/env python3import requestsimport socketimport randomimport stringimport concurrent.futuresfrom paramiko.client import SSHClientimport paramiko

Here are the variable to configure the script:

Ports: Two examples for specifying a list of ports, or doing a range is available.
Domain: Anyone can set up their own LMO server
using this Ansible script: LetMeOutOfYour.net Server Setup.
Verbose: Adds output saying what is being tested at that time.
printOpen and printClosed: These options can help determine how much output you want. If you are in a network that has mostly open connections you may only want things that are closed, or if you are egress testing a more closed network you may only want open. Or you may want everything and just filter it later.
ThreadCount: How many threads to spin up to get through your list.

Finally it randomizes the list so that you aren’t hitting everything in order.

# VARIABLESports = [80,443,445,8080,3389,22,21]#ports = list(range(1,65536))domain = "letmeoutofyour.net"verbose = FalseprintOpen = TrueprintClosed = Truethreadcount = 100random.shuffle(ports)

These functions are just checking if the different verbosity levels are set and printing if they are.

# Verbosity - set to False above if you don't want outputdef vprint(status):  if verbose == True:    print(status)# Print open portsdef print_open(status):  if printOpen == True:    print("[+] " + status)# Print closed portsdef print_closed(status):  if printClosed == True:    print("[-] " + status)

This is the function to test the HTTP/HTTPS connections:

def check_web(base, domain, port):  vprint("Testing: " + base + domain + ":" + str(port))  try:    r = requests.get(base + domain + ":" + str(port), timeout=1)    result = r.text.strip()    if result == "w00tw00t":      print_open("Success! " + base + domain + ":" + str(port))  except requests.exceptions.ConnectionError:    print_closed("Failed! " + base + domain + ":" + str(port))

This is the function to test SSH. It actually is pretty interesting because using Paramiko there isn’t a direct way just ask for remote host key that I’ve found, but you can attempt to connect, catch that it fails because either the host key isn’t in known_hosts or because no user name or password is supplied. Both errors still result in the client object being populated with the remote host’s host key and this is what we use to compare against.

def check_ssh(domain, port):  client = SSHClient()  vprint("Trying SSH to " + domain + " Port: " + str(port))  try:    client.connect(domain, port, timeout=1)  except paramiko.ssh_exception.SSHException:    pass  except socket.timeout:    print_closed("Failed! SSH to " + domain + " Port: " + str(port))    return  key = client.get_transport().get_remote_server_key()  if key.get_base64() == "AAAAC3NzaC1lZDI1NTE5AAAAIIrfkWLMzwGKRliVsJOjm5OJRJo6AZt7NsqAH8bk9tYc":    print_open("Success! SSH to " + domain + " Port: " + str(port))

Here is the “main” portion of the script (yes, I need to actually make a main function…)

with concurrent.futures.ThreadPoolExecutor(threadcount) as executor:  for port in ports:    # Test HTTP    base = "http://"    executor.submit(check_web, base, domain, port)    # Test HTTPS    base = "https://"    executor.submit(check_web, base, domain, port)    # Test SSH    executor.submit(check_ssh, domain, port)

Here is the script all put together:

#!/usr/bin/env python3import requestsimport socketimport randomimport stringimport concurrent.futuresfrom paramiko.client import SSHClientimport paramiko# VARIABLESports = [80,443,445,8080,3389,22,21]#ports = list(range(1,65536))domain = "letmeoutofyour.net"verbose = FalseprintOpen = TrueprintClosed = Truethreadcount = 100random.shuffle(ports)# Verbosity - set to False above if you don't want outputdef vprint(status):  if verbose == True:    print(status)# Print open portsdef print_open(status):  if printOpen == True:    print("[+] " + status)# Print closed portsdef print_closed(status):  if printClosed == True:    print("[-] " + status)def check_web(base, domain, port):  vprint("Testing: " + base + domain + ":" + str(port))  try:    r = requests.get(base + domain + ":" + str(port), timeout=1)    result = r.text.strip()    if result == "w00tw00t":      print_open("Success! " + base + domain + ":" + str(port))  except requests.exceptions.ConnectionError:    print_closed("Failed! " + base + domain + ":" + str(port))def check_ssh(domain, port):  client = SSHClient()  vprint("Trying SSH to " + domain + " Port: " + str(port))  try:    client.connect(domain, port, timeout=1)  except paramiko.ssh_exception.SSHException:    pass  except socket.timeout:    print_closed("Failed! SSH to " + domain + " Port: " + str(port))    return  key = client.get_transport().get_remote_server_key()  if key.get_base64() == "AAAAC3NzaC1lZDI1NTE5AAAAIIrfkWLMzwGKRliVsJOjm5OJRJo6AZt7NsqAH8bk9tYc":    print_open("Success! SSH to " + domain + " Port: " + str(port))with concurrent.futures.ThreadPoolExecutor(threadcount) as executor:  for port in ports:    # Test HTTP    base = "http://"    executor.submit(check_web, base, domain, port)    # Test HTTPS    base = "https://"    executor.submit(check_web, base, domain, port)    # Test SSH    executor.submit(check_ssh, domain, port)