(POC) use python to receive a forwarded port from ssh

recv_ssh.py

import paramiko
import socket
import threading
import sys
import random
from loguru import logger

# https://i.imgur.com/blN1pKc.png

# https://www.kite.com/python/docs/paramiko.ServerInterface.check_port_forward_request
# https://docs.paramiko.org/en/stable/api/keys.html
# https://ismailakkila.medium.com/black-hat-python-ssh-tunnelling-with-paramiko-7df71445cab

# ssh -R 0.0.0.0:9000:localhost:9000 root@1k0.in

host_key = paramiko.RSAKey.generate(2048)
server_address = "0.0.0.0"
server_port = 12379

class Server(paramiko.ServerInterface):
    def __init__(self):
        self.console = None
        self.fake_port = -1

    def check_auth_password(self, username, password):
        return paramiko.AUTH_SUCCESSFUL

    def check_port_forward_request(self, requested_addr, requested_port):
        self.fake_port = requested_port
        if self.console:
            self.console.send(f"sharing port {requested_addr}:{requested_port}")
        return requested_port

    def cancel_port_forward_request(self, addr, port):
        self.fake_port = None
        return True

    def check_channel_shell_request(self, channel):
        return True

    def check_channel_request(self, kind, chanid):
        logger.info("requst for channel {0}", kind)
        if kind in ["forwarded-tcpip", "session"]:
            return paramiko.OPEN_SUCCEEDED
        return paramiko.OPEN_SUCCEEDED

    def check_channel_pty_request(self, channel, term, width, height, pixelwidth, pixelheight, modes):
        # send something to the client
        # channel.send("hi")
        self.console = channel
        return True


@logger.catch()
def client_handler(client_socket):
    session_transport = paramiko.Transport(client_socket)
    session_transport.add_server_key(host_key)
    server = Server()
    
    session_transport.start_server(server=server)

    with session_transport.accept(20) as session_chan:
        if session_chan is None or not session_chan.active:
            logger.error("[!] Failure - SSH channel not active")
            session_transport.close()
            return

        logger.info("[+] Success - SSH channel active")
        
        sender_descriptor = ("1.3.3.7", random.randint(1024, 65535))
        server_descriptor = ("0.0.0.0", server.fake_port)
        
        tunnel_chan = session_transport.open_forwarded_tcpip_channel(sender_descriptor, server_descriptor)

        with tunnel_chan:
            while session_chan.active and tunnel_chan.active:
                data = tunnel_chan.recv(1024)
                if len(data) == 0:
                    break
                logger.info(f"[+] Received data {data} [{len(data)}] bytes via TCP Channel")                
                tunnel_chan.send(data)


server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

try:
    server_socket.bind((server_address, server_port))
except Exception:
    logger.error(f"[!] Failed to bind to {server_address}:{server_port}")
    sys.exit(1)


logger.info(f"[+] ssh server listen on {server_address}:{server_port}")
server_socket.listen(0)

while True:
    client_socket, addr = server_socket.accept()
    logger.info(f"[+] Incoming TCP connection from {addr[0]}:{addr[1]}")
    client_thread = threading.Thread(target=client_handler, args=(client_socket,))
    client_thread.start()

Start a local server:

netcat -lp 13314

Start port forwarding to your server

ssh -R 0.0.0.0:1337:localhost:13314 -t user@server -p 12379

you can then receive the data from the netcat server without exposing your port