0x00 背景
在某些场景下SSH服务器会禁用掉端口转发的能力，以降低安全风险。这会导致很多依赖SSH端口转发的工具无法正常工作。
这里主要是修改了/etc/ssh/sshd_config文件中以下几项实现的：
#AllowAgentForwarding yes
#AllowTcpForwarding yes
#X11Forwarding yes
此时，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))
}
完整的代码可以参考：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")
完整的代码可以参考：turbo-tunnel。
turbo-tunnel中可以使用以下方法将流量转发给SSH服务器：
turbo-tunnel -l http://:8080/ -t ssh+process://root:password@1.1.1.1:2222/usr/local/bin/telnet
/usr/local/bin/telnet是telnet-go在服务器上的路径，需要设置好可执行权限。
然后，本地通过http://127.0.0.1:8080代理访问的流量都会转发到ssh服务器上，从而实现了通过ssh服务器进行端口转发的目的。
0x03 总结
利用进程的实时输入输出，可以解决SSH服务器不支持端口转发的问题，从而绕过服务器限制，建立通信通道。这种方式应用场景更广，也更加隐蔽，只是使用上需要提前将一个文件拷贝到SSH服务器上，这里可能少数场景会有些阻碍（例如删除了chmod命令），需要寻找绕过这些限制的方法。
不过总的来说，使用这种方法，大大提升了建立SSH隧道的成功率，具有较大的实际应用价值。