OTPのSSH Moduleを使ってポート転送用Channelを実装しました。

ポート転送用Channelについて

（正しくはRFC4254をご覧ください。）
おなじみのSSHポート転送は以下のように実行できます。

ssh -L <ローカルIP>:<ローカルポート>:<転送先IP>:<転送先ポート> <SSH接続先>

# <例>
ssh -L 127.0.0.1:10080:192.168.122.111:80 192.168.122.61

実行すると<ローカルIP>:<ローカルポート>でListenします。
そして、Listenしたポートに接続すると、SSHのChannelが作成され、Channelを通してTCP通信が<転送先IP>:<転送先ポート>に転送されます。

今回はこのChannelを扱う仕組みを実装してみます。

環境

Ubuntu 20.04
Erlang OTP 22.2
- apt-getで入る。


$ erl
Erlang/OTP 22 [erts-10.6.4] [source] [64-bit] [smp:1:1] [ds:1:1:10] [async-threads:1]

Eshell V10.6.4  (abort with ^G)
1>

実装

ssh_client_channelModuleを使って、Channel管理用プロセスを作ります。
直感に反して、ssh_client_channel:startでの起動は想定していません。

ChannelをOpenした後は、以下の動作をします。

direct_tcpip_channel:sendでChannelにデータを送信する。
direct_tcpip_channel:recvでChannelからデータを受信する。。

direct_tcpip_channel.erl

-module(direct_tcpip_channel).

-export([start/3, recv/1, send/2, close/1]).
-export([start_direct_tcpip/4]).
-export([init/1, handle_ssh_msg/2, handle_msg/2, handle_call/3]).

-behavior(ssh_client_channel).

-record(state, {
        parent, %% spawnを実行したプロセス
        chanid, %% Channel ID
        conref, %% Connection
        msg_queue, %% SSH受信データキュー
        req_queue %% Recv要求キュー
    }).

%% 起動用関数。
start(ConRef, Remote, Local) ->
    Pid = proc_lib:spawn(?MODULE, start_direct_tcpip,
                         [ConRef, Remote, Local, self()]),
    receive
        {up, Pid} -> {ok, Pid};
        {error, Reason} -> {error, Reason}
    end.

%% Channel管理用プロセスを開始する。spawnされる。
start_direct_tcpip(ConRef, Remote, Local, Parent) ->
    case open_direct_tcpip(ConRef, Remote, Local) of
        {open, ChanID} ->
             {ok, State} = ssh_client_channel:init([[
                               {channel_cb, ?MODULE},
                               {init_args, #state{
                                              chanid=ChanID,conref=ConRef,
                                              msg_queue=queue:new(),req_queue=queue:new(),
                                              parent=Parent}},
                               {cm, ConRef},
                               {channel_id, ChanID}]]),
             ssh_client_channel:enter_loop(State);
        _ ->
             Parent ! {error, error}
    end.

%% ChannelをOpenする。
open_direct_tcpip(ConRef, {RHost, RPort}, {LHost, LPort}) ->
    RHostAddr = list_to_binary(RHost),
    LHostAddr = list_to_binary(LHost),
    Data = << (size(RHostAddr)):32/unsigned-integer, RHostAddr/binary,
              RPort:32/unsigned-integer,
              (size(LHostAddr)):32/unsigned-integer, LHostAddr/binary,
              LPort:32/unsigned-integer >>,
    ssh_connection_handler:open_channel(ConRef, "direct-tcpip", Data, 65536, 655360, infinity).


%% 以下Callback。
init(State) ->
    {ok, State}.

%% 受信データを要求。
handle_call(recv, From, #state{msg_queue=Q,req_queue=W}=S) ->
    case queue:out(Q) of
        {empty, _} ->
            {noreply, S#state{req_queue=queue:in(From,W)}};
        {{value, Msg}, Rest} -> 
            {reply, {ok, Msg}, S#state{msg_queue=Rest}}
    end;

%% Channelにデータ送信。
handle_call({send, Data}, _, #state{conref=ConRef,chanid=ChanID}=S) ->
    {reply, ssh_connection:send(ConRef, ChanID, Data), S};

handle_call(close, _, #state{conref=ConRef,chanid=ChanID}=S) ->
    {reply, ssh_connection:send_eof(ConRef,ChanID), S}.

%% Channelにデータ受信。
handle_ssh_msg({ssh_cm, _, {data, _, _, Data}}, #state{req_queue=W,msg_queue=Q}=S) ->
    case queue:out(W) of
        {empty, _} ->
            {ok, S#state{msg_queue=queue:in(Data,Q)}};
        {{value, Pid}, Rest} ->
            ssh_client_channel:reply(Pid, {ok, Data}),
            {ok, S#state{req_queue=Rest}}
    end;

handle_ssh_msg({ssh_cm, _, {eof, _}}, #state{conref=ConRef,chanid=ChanID}=S) ->
    ssh_connection:send_eof(ConRef, ChanID),
    {ok, S}.

handle_msg({ssh_channel_up, _, _}, S) ->
    S#state.parent ! {up, self()},
    {ok, S}.

recv(ChanRef) ->
    ssh_client_channel:call(ChanRef, recv).

send(ChanRef, Data) ->
    ssh_client_channel:call(ChanRef, {send, Data}).

close(ChanRef) ->
    ssh_client_channel:call(ChanRef, close).

特徴的な部分について記載します。

Channel Open

SSHのプロトコル的には肝の部分です。

%% ChannelをOpenする。
open_direct_tcpip(ConRef, {RHost, RPort}, {LHost, LPort}) ->
    RHostAddr = list_to_binary(RHost),
    LHostAddr = list_to_binary(LHost),
    Data = << (size(RHostAddr)):32/unsigned-integer, RHostAddr/binary,
              RPort:32/unsigned-integer,
              (size(LHostAddr)):32/unsigned-integer, LHostAddr/binary,
              LPort:32/unsigned-integer >>,
    ssh_connection_handler:open_channel(ConRef, "direct-tcpip", Data, 65536, 655360, infinity).

ssh_connection_handler:open_channelでChannelをOpenします。
本来このModuleはマニュアル等には載っていない裏方のModuleで、使うべきではないのでしょうが、
他に方法が無いためやむを得ず使います。

Dataは規定通りに実装します。

（この部分のコードはOTPのSSH実装を強く参考にしています。）

ssh_client_channelのstart

Channel管理プロセスを開始します。

%% 起動用関数。
start(ConRef, Remote, Local) ->
    Pid = proc_lib:spawn(?MODULE, start_direct_tcpip,
                         [ConRef, Remote, Local, self()]),
    receive
        {up, Pid} -> {ok, Pid};
        {error, Reason} -> {error, Reason}
    end.

%% Channel管理用プロセスを開始する。spawnされる。
start_direct_tcpip(ConRef, Remote, Local, Parent) ->
    case open_direct_tcpip(ConRef, Remote, Local) of
        {open, ChanID} ->
             {ok, State} = ssh_client_channel:init([[
                               {channel_cb, ?MODULE},
                               {init_args, #state{
                                              chanid=ChanID,conref=ConRef,
                                              msg_queue=queue:new(),req_queue=queue:new(),
                                              parent=Parent}},
                               {cm, ConRef},
                               {channel_id, ChanID}]]),
             ssh_client_channel:enter_loop(State);
        _ ->
             Parent ! {error, error}
    end.

Callback抜粋

handle_msg({ssh_channel_up, _, _}, S) ->
    S#state.parent ! {up, self()},
    {ok, S}.

ssh_client_channelModuleを使っているので、初めはssh_client_channel:startで開始するつもりでしたが…

startの呼び出し時にChannel IDが必要。
Channel IDはChannelをOpenした時に取得できる。
ChannelへのデータはConnection管理プロセスから、ChannelをOpenしたプロセスへ送信される。

このような動作になっていて、ChannelをOpenしてstartすると、Channel管理プロセスではデータを受信することができませんでした。

そのため、ChannelをOpenしたプロセスをenter_loopでそのままChannel管理プロセスにして、データを受信できるようにしています。

データ送受信

recv

recv(ChanRef) ->
    ssh_client_channel:call(ChanRef, recv).

Callback抜粋

%% 受信データを要求。
handle_call(recv, From, #state{msg_queue=Q,req_queue=W}=S) ->
    case queue:out(Q) of
        {empty, _} ->
            {noreply, S#state{req_queue=queue:in(From,W)}};
        {{value, Msg}, Rest} -> 
            {reply, {ok, Msg}, S#state{msg_queue=Rest}}
    end;

recv関数でChannelが受信したデータを取得します。
msg_queueには受信したデータが格納されていて、recv呼び出し時はこれを返します。
msg_queueにデータがない場合は、req_queueに呼び出し元のPidを格納し、noreplyでCallbackを終えます。（callが返りません。）

Callback抜粋

%% Channelにデータ受信。
handle_ssh_msg({ssh_cm, _, {data, _, _, Data}}, #state{req_queue=W,msg_queue=Q}=S) ->
    case queue:out(W) of
        {empty, _} ->
            {ok, S#state{msg_queue=queue:in(Data,Q)}};
        {{value, Pid}, Rest} ->
            ssh_client_channel:reply(Pid, {ok, Data}),
            {ok, S#state{req_queue=Rest}}
    end;

Channelでデータを受信したら、msg_queueにデータを格納します。
req_queueにPidが格納されている場合、msg_queueにはデータを格納せずreplyします。

send

send(ChanRef, Data) ->
    ssh_client_channel:call(ChanRef, {send, Data}).

Callback抜粋

%% Channelにデータ送信。
handle_call({send, Data}, _, #state{conref=ConRef,chanid=ChanID}=S) ->
    {reply, ssh_connection:send(ConRef, ChanID, Data), S};

send関数でChannelにデータを送信します。
ChanRefはPidで、起動用関数startの戻り値として取得できます。

使う

使ってみます。
事前にSSHのConnectionを開始しておきます。

1> ssh:start().
ok
2> {ok, Con} = ssh:connect("192.168.122.61", 22, []).
ssh password: 
{ok,<0.93.0>}

直接データを送る

Shellで直接データを送ってみます。

3> {ok, Pid} = direct_tcpip_channel:start(Con, {"192.168.122.111",80}, {"127.0.0.1",54321}).
{ok,<0.96.0>}
4> direct_tcpip_channel:send(Pid, "GET /\n").
ok
5> Self = self().
<0.78.0>
6> spawn(fun() -> Self ! direct_tcpip_channel:recv(Pid) end).
<0.101.0>
7> direct_tcpip_channel:close(Pid).
ok
8> flush().
Shell got {ok,<<"<!DOCTYPE HTML PUBLIC \"-//W3C//DTD HTML 4.01//EN\" \"http://www.w3.org/TR/html4/strict.dtd\">\n<html>...<略>">>}
ok

HTTPの通信ができました。
通信はSSH Channelで転送されるため、WebサーバからはSSH接続先のサーバと通信しているように見えます。

サーバログ

192.168.122.61 - - [21/Sep/2020 17:18:32] "GET /" 200 -

ローカルポート転送（`-L`）

OpenSSHの-Lオプションでおなじみのポート転送を実装してみます。
指定のポートでListenして、接続をAcceptするとデータ送受信プロセスをspawnします。

受信プロセスはメッセージを受け取り、Socketに送信します。
送信プロセスはSocketからデータを受け取り、メッセージを送信します。

static_forward.erl

-module(static_forward).

-export([start/5]).
-export([recv_loop/2,send_loop/2]).


start(Con, LHost, LPort, RHost, RPort) ->
    {ok, LAddr} = inet:parse_address(LHost),
    {ok, LSock} = gen_tcp:listen(LPort, [{active, false},{mode, binary},
                                         {ip, LAddr}]),
    server(LSock, Con, RHost, RPort).

server(LSock, Con, RHost, RPort) ->
    {ok, Sock} = gen_tcp:accept(LSock),
    {ok, {LHostAddr,LPort}} = inet:peername(Sock),
    LHost = inet:ntoa(LHostAddr),
    {ok,ChanRef} = direct_tcpip_channel:start(Con,{RHost,RPort},{LHost,LPort}),
    spawn(?MODULE, recv_loop, [Sock, ChanRef]),
    spawn(?MODULE, send_loop, [Sock, ChanRef]),
    server(LSock, Con, RHost, RPort).

recv_loop(Sock, Pid) ->
    case direct_tcpip_channel:recv(Pid) of
        {ok, Data} ->
            gen_tcp:send(Sock, Data),
            recv_loop(Sock, Pid);
        {error, _} ->
            gen_tcp:close(Sock)
    end.

send_loop(Sock, Pid) ->
    case gen_tcp:recv(Sock, 0) of
        {ok, Data} ->
            direct_tcpip_channel:send(Pid ,Data),
            send_loop(Sock, Pid);
        {error, _} ->
            direct_tcpip_channel:close(Pid)
    end.

shellから起動します。

9> static_forward:start(Con, "127.0.0.1", 10022, "127.0.0.1", 22).

127.0.0.1:10022を、SSH接続先から見て127.0.0.1:22に転送しました。
SSHでログインしてみます。

$ ssh -p 10022 127.0.0.1
testing@127.0.0.1's password: 
Last login: Sun Sep 20 23:53:10 2020 from localhost
[testing@remote ~]$ 
[testing@remote ~]$ hostname -I
192.168.122.61 192.168.124.1 
[testing@remote ~]$

できました。転送されているようです。

Dynamicポート転送（`-D`）

動的なポート転送で、OpenSSHでは-Dオプションです。
OpenSSH等では指定したポートでSocksを受付け、Socksで要求されたポートへ通信を転送します。

これにならってSocksで実装してみます。

socks.erl

-module(socks).
-export([start_server/3]).
-export([recv_loop/2, send_loop/2]).


start_server(LHost, LPort, Con) ->
    {ok, LAddr} = inet:parse_address(LHost),
    {ok, LSock} = gen_tcp:listen(LPort, [{active, false},{mode, binary},
                                         {ip, LAddr}]),
    server(LSock, Con).

server(LSock, Con) ->
    {ok, Sock} = gen_tcp:accept(LSock),

    ok = socks_auth(Sock, 0, []),
    {ok, {RHost, RPort}, Req} = socks_wait_request(Sock),

    {ok, {LHostAddr, LPort}} = inet:peername(Sock),
    LHost = inet:ntoa(LHostAddr),

    case direct_tcpip_channel:start(Con,{RHost,RPort},{LHost,LPort}) of
        {ok,ChanRef} ->
            spawn(?MODULE, recv_loop, [Sock, ChanRef]),
            spawn(?MODULE, send_loop, [Sock, ChanRef]),
            socks_success(Sock,Req);
        {error,_} ->
            socks_failure(Sock,Req)
    end,

    server(LSock, Con).

socks_auth(Sock, Method, AuthOpts) ->
    {ok, <<5,N,Methods:N/binary>>} = gen_tcp:recv(Sock, 0),

    case lists:member(Method, binary_to_list(Methods)) of
        true ->
            ok = gen_tcp:send(Sock, <<5,Method>>),
            authenticate(Sock, Method, AuthOpts);
        false ->
            ok = gen_tcp:send(Sock, <<5,255>>),
            error
    end.

authenticate(_, 0, _) ->
    ok.

socks_wait_request(Sock) ->
    {ok, Req} = gen_tcp:recv(Sock, 0),
    <<5,1,0,ATYP,Rest/binary>> = Req,

    case ATYP of
        1 ->
            <<A1,A2,A3,A4,Port:16>> = Rest,
            Host = inet:ntoa({A1,A2,A3,A4}),
            {ok, {Host,Port}, Req};
        3 ->
            <<L,BinHost:L/binary,Port:16>> = Rest,
            Host = binary_to_list(BinHost),
            {ok, {Host,Port}, Req};
        Other ->
            gen_tcp:send(Sock, <<5,1,Other,ATYP,Rest>>),
            error
    end.

socks_success(Sock, Req) ->
    <<5,_,0,Rest/binary>> = Req,
    gen_tcp:send(Sock, <<5,0,0,Rest/binary>>),
    ok.

socks_failure(Sock, Req) ->
    <<5,_,0,Rest/binary>> = Req,
    gen_tcp:send(Sock, <<5,1,0,Rest/binary>>),
    gen_tcp:close(Sock),
    ok.

recv_loop(Sock, Pid) ->
    case direct_tcpip_channel:recv(Pid) of
        {ok, Data} ->
            gen_tcp:send(Sock, Data),
            recv_loop(Sock, Pid);
        {error, _} ->
            gen_tcp:close(Sock)
    end.

send_loop(Sock, Pid) ->
    case gen_tcp:recv(Sock, 0) of
        {ok, Data} ->
            direct_tcpip_channel:send(Pid ,Data),
            send_loop(Sock, Pid);
        {error, _} ->
            direct_tcpip_channel:close(Pid)
    end.

シェルから起動します。

3> socks:start_server("127.0.0.1", 1080, Con).

Socks対応のクライアントで試してみます。

$ curl --SOCKS5 127.0.0.1:1080 192.168.122.111:80
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01//EN" "http://www.w3.org/TR/html4/strict.dtd">
<html>
<head>
# 以下略

$ curl --SOCKS5 127.0.0.1:1080 https://qiita.com
<!DOCTYPE html><html><head><meta charset="utf-8" /><title>Qiita</title>
# 以下略

動作しているようです。

おわりに

ポート転送用Channelを動かすことができました。

（ちなみに）OTP SSHのポート転送対応について

OTP22ではありませんが、新しいものではポート転送が提供されています。
以下の関数です。

ssh:tcpip_tunnel_to_server/4
ssh:tcpip_tunnel_to_server/5

ただ、これらの関数はローカルでListenするところまで含めて実装されています。
そのためデータを直接送り込んだり、Dynamicポート転送を実装することはできないようです。

コードからsend/recvできるように作ってくれたら…

参考

Erlang SSH https://erlang.org/doc/apps/ssh/users_guide.html
RFC4254 "The Secure Shell (SSH) Connection Protocol" https://tools.ietf.org/html/rfc4254
RFC1928 "SOCKS Protocol Version 5" https://tools.ietf.org/html/rfc1928