绕过SSH服务器的端口转发限制

0x00 背景

在某些场景下SSH服务器会禁用掉端口转发的能力，以降低安全风险。这会导致很多依赖SSH端口转发的工具无法正常工作。

这里主要是修改了/etc/ssh/sshd_config文件中以下几项实现的：

#AllowAgentForwarding yes
#AllowTcpForwarding yes
#X11Forwarding yes
  COPY

此时，SSH服务器基本就变成了只能执行shell命令的工具，无法用于建立通信通道。

是否有办法可以绕过这一限制呢？答案是肯定的。

0x01 借尸还魂

SSH最常用的能力就是交互式命令行，所谓交互式命令行，就是允许用户进行实时输入，并将输出实时展示出来。也就是说：交互式命令行本身就是一个双向通信的通道。因此，可以编写一个程序，它会在初始化时与指定的服务器端口建立Socket连接，然后将所有stdin读到的数据实时发送给Socket，并将Socket接收到的数据写到stdout中，stderr则用于输出控制信息和日志等。

根据上面的分析，这个程序其实跟telnet命令非常相似，但又不完全相同。因此用GO写了下面这个程序：

package main
import (
    "bufio"
    "fmt"
    "net"
    "os"
    "os/signal"
    "strconv"
    "sync"
    "time"
)
func telnet(host string, port int) int {
    conn, err := net.DialTimeout("tcp", fmt.Sprintf("%s:%d", host, port), 10*time.Second)
    if err != nil {
        fmt.Fprintln(os.Stderr, fmt.Sprintf("[FAIL] Connect %s:%d failed: %s", host, port, err))
        return -1
    }
    defer conn.Close()
    fmt.Fprintln(os.Stderr, "[OKAY]")
    var wg sync.WaitGroup
    wg.Add(2)
    go handleWrite(conn, &wg)
    go handleRead(conn, &wg)
    wg.Wait()
    return 0
}
func handleRead(conn net.Conn, wg *sync.WaitGroup) int {
    defer wg.Add(-2)
    reader := bufio.NewReader(conn)
    buff := make([]byte, 4096)
    for {
        var bytes int
        var err error
        bytes, err = reader.Read(buff)
        if err != nil {
            fmt.Fprintln(os.Stderr, "Error to read from upstream because of", err)
            return -1
        }
        _, err = os.Stdout.Write(buff[:bytes])
        if err != nil {
            fmt.Fprintln(os.Stderr, "Error to write to stdout because of", err)
            return -1
        }
    }
}
func handleWrite(conn net.Conn, wg *sync.WaitGroup) int {
    defer wg.Add(-2)
    reader := bufio.NewReader(os.Stdin)
    buff := make([]byte, 4096)
    for {
        var bytes int
        var err error
        bytes, err = reader.Read(buff)
        if err != nil {
            fmt.Fprintln(os.Stderr, "Error to read from stdin because of", err)
            return -1
        }
        _, err = conn.Write(buff[:bytes])
        if err != nil {
            fmt.Fprintln(os.Stderr, "Error to write to upstream because of", err)
            return -1
        }
    }
}
func main() {
    if len(os.Args) < 3 {
        fmt.Fprintln(os.Stderr, "Usage: telnet host port")
        os.Exit(-1)
    }
    host := os.Args[1]
    port, _ := strconv.Atoi(os.Args[2])
    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt)
    go func(){
        for sig := range c {
            // sig is a ^C, handle it
            fmt.Fprintln(os.Stderr, "Signal", sig)
            os.Exit(0)
        }
    }()
    os.Exit(telnet(host, port))
}
  COPY

完整的代码可以参考：telnet-go。

0x02 暗度陈仓

要使用telnet-go提供的通信通道，需要与Paramiko或ASyncSSH之类的SSH库进行集成才行。下面是使用ASyncSSH进行集成的核心逻辑：

class SSHProcessTunnel(SSHTunnel):
    """SSH Tunnel Over Process StdIn and StdOut"""
    def __init__(self, tunnel, url, address):
        super(SSHProcessTunnel, self).__init__(tunnel, url, address)
        self._process = None
    @classmethod
    def has_cache(cls, url):
        return False
    async def _log_stderr(self):
        while not self.closed():
            error_line = await self._process.stderr.readline()
            error_line = error_line.strip()
            utils.logger.warn(
                "[%s][stderr] %s" % (self.__class__.__name__, error_line.decode())
            )
            await asyncio.sleep(0.5)
        self._process = None
    async def connect(self):
        ssh_conn = await self.create_ssh_conn()
        if not ssh_conn:
            return False
        bin_path = self._url.path
        cmdline = "%s %s %d" % (bin_path, self._addr, self._port)
        self._process = await ssh_conn.create_process(cmdline, encoding=None)
        await asyncio.sleep(0.5)
        if self._process.exit_status is not None and self._process.exit_status != 0:
            utils.logger.error(
                "[%s] Create process %s failed: [%d]%s"
                % (
                    self.__class__.__name__,
                    cmdline,
                    self._process.exit_status,
                    await self._process.stderr.read(),
                )
            )
            return False
        status_line = await self._process.stderr.readline()
        if status_line.startswith(b"[OKAY]"):
            utils.safe_ensure_future(self._log_stderr())
            return True
        elif status_line.startswith(b"[FAIL]"):
            utils.logger.warn(
                "[%s] Connect %s:%d failed: %s"
                % (
                    self.__class__.__name__,
                    self._addr,
                    self._port,
                    status_line.decode(),
                )
            )
            return False
        else:
            raise RuntimeError("Unexpected stderr: %s" % status_line.decode())
    async def read(self):
        if self._process:
            buffer = await self._process.stdout.read(4096)
            if buffer:
                return buffer
        raise utils.TunnelClosedError()
    async def write(self, buffer):
        if self._process:
            return self._process.stdin.write(buffer)
        else:
            raise utils.TunnelClosedError()
    def closed(self):
        return self._process is None or self._process.exit_status is not None
    def close(self):
        if self._process:
            self._process.stdin.write(b"\x03")
  COPY

完整的代码可以参考：turbo-tunnel。

turbo-tunnel中可以使用以下方法将流量转发给SSH服务器：

turbo-tunnel -l http://:8080/ -t ssh+process://root:password@1.1.1.1:2222/usr/local/bin/telnet
  COPY

/usr/local/bin/telnet是telnet-go在服务器上的路径，需要设置好可执行权限。

然后，本地通过http://127.0.0.1:8080代理访问的流量都会转发到ssh服务器上，从而实现了通过ssh服务器进行端口转发的目的。

0x03 总结

利用进程的实时输入输出，可以解决SSH服务器不支持端口转发的问题，从而绕过服务器限制，建立通信通道。这种方式应用场景更广，也更加隐蔽，只是使用上需要提前将一个文件拷贝到SSH服务器上，这里可能少数场景会有些阻碍（例如删除了chmod命令），需要寻找绕过这些限制的方法。

不过总的来说，使用这种方法，大大提升了建立SSH隧道的成功率，具有较大的实际应用价值。