Melang 之协程并发代理实战

转自本人 CSDN： https://blog.csdn.net/weixin_40960130/article/details/112704999

之前的文章中，给大家介绍了一种新的协程语言——Melang。 今天，给大家带来的是这款语言的企业首战，虽然是个较小的项目，但对于一款新语言的意义无疑是巨大的。并且，利用这款语言，让整个程序结构极为清晰与模块化。 由于笔者公司想搭建一个代理服务供其他网段机器上网之用，因此有了本文的项目。 注意：本文只是用于介绍语言特性和使用，并不鼓励读者违背国家政策法规，请勿将此文内容用于技术讨论外的一切其他用途。

程序结构

在之前的文章中我们介绍过，Melang 的每一个脚本任务都是一个协程，协程之间的调度是抢占式的，协程之间的运行环境是隔离的，且一个协程还可以拉起其他协程。 所以，本文的 socks5 代理将采用协程并发的模式，协程结构如下： 每一个工作协程独立处理一个 TCP 连接。 故此，在访问 web 站点时，会由浏览器发起多个 TCP 到本代理，由主协程完成 TCP 的建立，然后拉起一个独立的工作协程处理该 TCP 上的协议和数据收发。当 TCP 连接断开收尾工作结束后，工作协程退出释放。 注：本文给出的代理目前仅支持 TCP 代理。

Socks5

这里捎带提及一下 socks5 协议。 这个协议是比较简单的，大致流程如下：

1. 建立 TCP 后，代理服务器会收到客户端的握手报文
2. 服务器端验证握手报文中的验证方式，并回复响应报文
3. 客户端收到后验证完会发送本次代理数据的目的地址（可能是域名或 IP ）和端口
4. 代理服务器尝试向该地址建立 TCP，然后给客户端返回响应报文
5. 当前 4 步完成后，就进入了数据透传的阶段

实现

废话再多不如代码上桌。 代码分为两个文件，一个是主协程脚本proxy.mln，另一个是工作协程脚本worker.mln，我们分别给出：

proxy.mln

recvTimeout = 50;
fd = @mln_tcpListen('0.0.0.0', '1080');
@mln_print('Ready');
while (1) {
  sockfd = @mln_tcpAccept(fd);
  if (sockfd) {
    conf = [
      'fd': sockfd,
      'recvTimeout': recvTimeout,
    ];
    @mln_eval('worker.mln', @mln_json_encode(conf));
  } fi
}

可以看到，主协程的任务非常简单：

1. 建立监听套接字
2. 死循环建立 TCP，并为每个 TCP 拉起一个 worker.mln 任务进行处理。

worker.mln

conf = @mln_json_decode(EVAL_DATA);
localFd = @mln_int(conf['fd']);
recvTimeout = @mln_int(conf['recvTimeout']);
state = 1;
localSend = '';
remoteSend = '';
remoteFd = nil;

@closeLocalSock()
{
  @mln_tcpClose(_localFd);
  _localFd = nil;
  _localSend = '';
}

@closeRemoteSock()
{
  @mln_tcpClose(_remoteFd);
  _remoteFd = nil;
  _remoteSend = '';
}

@localRecvHandler()
{
  if (_state == 1) {
    if (@mln_strlen(_remoteSend) < 3 || @mln_bin2int(_remoteSend[0]) != 5) {
      @closeLocalSock();
      return;
    } fi
    n = @mln_bin2int(_remoteSend[1]);
    if (@mln_strlen(_remoteSend) < n+2) {
      @closeLocalSock();
      return;
    } fi
    for (i = 0; i < n; ++i) {
      if (@mln_bin2int(_remoteSend[2+i]) == 0) {
        break;
      } fi
    }
    if (i >= n) {
      @closeLocalSock();
      return;
    } fi
    ret = @mln_tcpSend(_localFd, @mln_int2bin(5)[-1]+@mln_int2bin(0)[-1]);
    if (!ret) {
      @closeLocalSock();
      return;
    } fi
    _remoteSend = @mln_split(_remoteSend, n+2);
    _state = 2;
  } else if (_state == 2) {
    arr = [5, 7, 0, 1, 0, 0, 0, 0, 0, 0];
    err = '';
    for (i = 0; i < @mln_size(arr); ++i) {
      err += @mln_int2bin(arr[i])[-1];
    }
    len = @mln_strlen(_remoteSend);
    if (len < 8 || @mln_bin2int(_remoteSend[0]) != 5 || @mln_bin2int(_remoteSend[1]) != 1 || @mln_bin2int(_remoteSend[2]) != 0) {
      goto fail;
    } fi
    type = @mln_bin2int(_remoteSend[3]);
    addr = '';
    if (type == 1) {
      if (len < 10) {
        goto fail;
      } fi
      for (i = 0; i < 4; ++i) {
        addr += @mln_str(@mln_bin2int(_remoteSend[4+i]));
        if (i < 3) {
          addr += '.';
        } fi
      }
      n = 8;
    } else if (type == 3) {
      n = 5+@mln_bin2int(_remoteSend[4]);
      if (len < n+2) {
        goto fail;
      } fi
      addr = @mln_split(_remoteSend, 5, @mln_bin2int(_remoteSend[4]));
    } else if (type == 4) {
      if (len < 22) {
        goto fail;
      } fi
      for (i = 0; i < 8; ++i) {
        addr += @mln_bin2hex(_remoteSend[4+i*2]);
        addr += @mln_bin2hex(_remoteSend[4+i*2+1]);
        if (i < 7) {
          addr += ':';
        } fi
      }
      n = 20;
    } else {
      goto fail;
    }
    if (len < n + 2) {
      goto fail;
    } fi
    port = (@mln_bin2int(_remoteSend[n])<<8)|@mln_bin2int(_remoteSend[n+1]);

    @mln_print('connect['+addr+']');
    ret = @mln_tcpConnect(addr, @mln_str(port), 30000);
    if (!ret) {
      goto fail;
    } fi
    _remoteFd = ret;

    ret = ''+@mln_int2bin(5)[-1]+@mln_int2bin(0)[-1]+@mln_int2bin(0)[-1]+_remoteSend[3];
    ret += @mln_split(_remoteSend, 4, n - 2);
    ret = @mln_tcpSend(_localFd, ret);
    if (!ret) {
      @closeRemoteSock();
      @closeLocalSock();
      return;
    } fi
    _remoteSend = @mln_split(_remoteSend, n+2);
    _state = 3;
  } else {
    ret = @mln_tcpSend(_remoteFd, _remoteSend);
    if (!ret) {
      @closeRemoteSock();
    } else {
      _remoteSend = '';
    }
  }
  return;

fail:
  @mln_tcpSend(_localFd, err);
  @closeLocalSock();
  return;
}

//@mln_print(''+localFd);
while (1) {
  if (localFd) {
      if (state == 3 && !remoteFd && !localSend) {
        @closeLocalSock();
      } else {
        res = @mln_tcpRecv(localFd, recvTimeout);
        if (res) {
          if (@mln_isBool(res)) {
            @closeLocalSock();
          } else {
            remoteSend += res;
          }
        } else if (@mln_isBool(res)) {
          @closeLocalSock();
        } fi
      }
  } fi
  if (remoteFd) {
      if (state == 3 && !localFd && !remoteSend) {
        @closeRemoteSock();
      } else {
        res = @mln_tcpRecv(remoteFd, recvTimeout);
        if (res) {
          if (@mln_isBool(res)) {
            @closeRemoteSock();
          } else {
            localSend += res;
          }
        } else if (@mln_isBool(res)) {
            @closeRemoteSock();
        } fi
      }
  } fi
  if (@mln_isNil(localFd) && @mln_isNil(remoteFd)) {
    break;
  } fi
  if (remoteSend) {
    @localRecvHandler();
  } fi
  if (localSend) {
    ret = @mln_tcpSend(_localFd, _localSend);
    _localSend = '';
    if (!ret) {
      @closeLocalSock();
    } fi
  } fi
}
@mln_print('quit');

worker 协程不足 200 行，简单说一下 ：

• conf 是从主协程传过来的配置；
• locaFd 就是刚刚建立的套接字；
• recvTimeout 是 tcp 接收数据的超时时间（毫秒），由于 Melang 的代码是同步代码（底层异步执行），因此需要有防止长时间阻塞在一个函数上的机制。但由于是同步模式代码，因此执行流程是非常清晰的；
• state 是用来标记 socks5 的状态：1-处理握手报文阶段，2-处理目的地址报文并建立连接阶段，3-数据透传阶段；
• remoteFd 是 state 为 2 时向目的地址建立的 TCP 套接字；
• localSend 和 remoteSend 是两个发送缓冲区，localSend 是发送给 localFd 的数据，而 remoteSend 是发送给 remoteFd 的数据（ state 不为 3 时不会发给 remoteFd ）；
• closeLocalSock 函数就是用来关闭与客户端通信的 TCP 连接并清空发送缓冲区；
• closeRemoteSock 函数就是用来关闭与目的地址通信的 TCP 连接并清空发送缓冲区；
• localRecvHandler 函数用来处理与客户端通信的套接字数据的，由于有两个阶段的 socks5 相关处理，因此相对冗长一些；
• while 死循环，这个循环就是用来接收两个 TCP 上数据，然后处理，然后再由指定的 TCP 发送出去；

额外说明，在函数中可以看到一些对全局变量名前加下划线的变量，这样的变量在本例中依旧是指全局变量。如不加下划线，那么变量名只会在当前函数作用域内搜寻，因此无法获取调用栈上层的变量值，而加了前置下划线的变量则会延调用栈顺序由内向外查询变量。

结尾

感兴趣的读者可以去到 Melang 的 Github repo （https://github.com/Water-Melon/Melang）上按照 README 的内容下载并安装 Melang 进行尝试运行。 感谢阅读，欢迎大家留言评论或私信交流。