代理网关设计与实现(基于NETTY)-阿里云开发者社区

开发者社区> 开发者小助手_LS> 正文

代理网关设计与实现(基于NETTY)

简介: 本文重点在代理网关本身的设计与实现,而非代理资源的管理与维护。
+关注继续查看

image.png

作者 | 新然
来源 | 阿里技术公众号

一 问题背景

  1. 平台端购置一批裸代理,来做广告异地展现审核。从外部购置的代理,使用方式为:
  2. 通过给定的HTTP 的 API 提取代理 IP:PORT,返回的结果会给出代理的有效时长 3~5 分钟,以及代理所属地域;

从提取的代理中,选取指定地域,添加认证信息,请求获取结果;

本文设计实现一个通过的代理网关:

  1. 管理维护代理资源,并做代理的认证鉴权;
  2. 对外暴露统一的代理入口,而非动态变化的代理IP:PORT;
  3. 流量过滤及限流,比如:静态资源不走代理;

本文重点在代理网关本身的设计与实现,而非代理资源的管理与维护。

注:本文包含大量可执行的JAVA代码以解释代理相关的原理

二 技术路线

本文的技术路线。在实现代理网关之前,首先介绍下代理相关的原理及如何实现

  1. 透明代理;
  2. 非透明代理;
  3. 透明的上游代理;
  4. 非透明的上游代理;

最后,本文要构建代理网关,本质上就是一个非透明的上游代理,并给出详细的设计与实现。

1 透明代理

透明代理是代理网关的基础,本文采用JAVA原生的NIO进行详细介绍。在实现代理网关时,实际使用的为NETTY框架。原生NIO的实现对理解NETTY的实现有帮助。

透明代理设计三个交互方,客户端、代理服务、服务端,其原理是:

image.png

  1. 代理服务在收到连接请求时,判定:如果是CONNECT请求,需要回应代理连接成功消息到客户端;
  2. CONNECT请求回应结束后,代理服务需要连接到CONNECT指定的远程服务器,然后直接转发客户端和远程服务通信;
  3. 代理服务在收到非CONNECT请求时,需要解析出请求的远程服务器,然后直接转发客户端和远程服务通信;

需要注意的点是:

  1. 通常HTTPS请求,在通过代理前,会发送CONNECT请求;连接成功后,会在信道上进行加密通信的握手协议;因此连接远程的时机是在CONNECT请求收到时,因为此后是加密数据;
  2. 透明代理在收到CONNECT请求时,不需要传递到远程服务(远程服务不识别此请求);
  3. 透明代理在收到非CONNECT请求时,要无条件转发;

完整的透明代理的实现不到约300行代码,完整摘录如下:

@Slf4j
public class SimpleTransProxy {

    public static void main(String[] args) throws IOException {
        int port = 8006;
        ServerSocketChannel localServer = ServerSocketChannel.open();
        localServer.bind(new InetSocketAddress(port));
        Reactor reactor = new Reactor();
        // REACTOR线程
        GlobalThreadPool.REACTOR_EXECUTOR.submit(reactor::run);

        // WORKER单线程调试
        while (localServer.isOpen()) {
            // 此处阻塞等待连接
            SocketChannel remoteClient = localServer.accept();

            // 工作线程
            GlobalThreadPool.WORK_EXECUTOR.submit(new Runnable() {
                @SneakyThrows
                @Override
                public void run() {
                    // 代理到远程
                    SocketChannel remoteServer = new ProxyHandler().proxy(remoteClient);

                    // 透明传输
                    reactor.pipe(remoteClient, remoteServer)
                            .pipe(remoteServer, remoteClient);
                }
            });
        }
    }
}

@Data
class ProxyHandler {
    private String method;
    private String host;
    private int port;
    private SocketChannel remoteServer;
    private SocketChannel remoteClient;

    /**
     * 原始信息
     */
    private List<ByteBuffer> buffers = new ArrayList<>();
    private StringBuilder stringBuilder = new StringBuilder();

    /**
     * 连接到远程
     * @param remoteClient
     * @return
     * @throws IOException
     */
    public SocketChannel proxy(SocketChannel remoteClient) throws IOException {
        this.remoteClient = remoteClient;
        connect();
        return this.remoteServer;
    }

    public void connect() throws IOException {
        // 解析METHOD, HOST和PORT
        beforeConnected();

        // 链接REMOTE SERVER
        createRemoteServer();

        // CONNECT请求回应,其他请求WRITE THROUGH
        afterConnected();
    }

    protected void beforeConnected() throws IOException {
        // 读取HEADER
        readAllHeader();

        // 解析HOST和PORT
        parseRemoteHostAndPort();
    }

    /**
     * 创建远程连接
     * @throws IOException
     */
    protected void createRemoteServer() throws IOException {
        remoteServer = SocketChannel.open(new InetSocketAddress(host, port));
    }

    /**
     * 连接建立后预处理
     * @throws IOException
     */
    protected void afterConnected() throws IOException {
        // 当CONNECT请求时,默认写入200到CLIENT
        if ("CONNECT".equalsIgnoreCase(method)) {
            // CONNECT默认为443端口,根据HOST再解析
            remoteClient.write(ByteBuffer.wrap("HTTP/1.0 200 Connection Established\r\nProxy-agent: nginx\r\n\r\n".getBytes()));
        } else {
            writeThrouth();
        }
    }

    protected void writeThrouth() {
        buffers.forEach(byteBuffer -> {
            try {
                remoteServer.write(byteBuffer);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 读取请求内容
     * @throws IOException
     */
    protected void readAllHeader() throws IOException {
        while (true) {
            ByteBuffer clientBuffer = newByteBuffer();
            int read = remoteClient.read(clientBuffer);
            clientBuffer.flip();
            appendClientBuffer(clientBuffer);
            if (read < clientBuffer.capacity()) {
                break;
            }
        }
    }

    /**
     * 解析出HOST和PROT
     * @throws IOException
     */
    protected void parseRemoteHostAndPort() throws IOException {
        // 读取第一批,获取到METHOD
        method = parseRequestMethod(stringBuilder.toString());

        // 默认为80端口,根据HOST再解析
        port = 80;
        if ("CONNECT".equalsIgnoreCase(method)) {
            port = 443;
        }

        this.host = parseHost(stringBuilder.toString());

        URI remoteServerURI = URI.create(host);
        host = remoteServerURI.getHost();

        if (remoteServerURI.getPort() > 0) {
            port = remoteServerURI.getPort();
        }
    }

    protected void appendClientBuffer(ByteBuffer clientBuffer) {
        buffers.add(clientBuffer);
        stringBuilder.append(new String(clientBuffer.array(), clientBuffer.position(), clientBuffer.limit()));
    }

    protected static ByteBuffer newByteBuffer() {
        // buffer必须大于7,保证能读到method
        return ByteBuffer.allocate(128);
    }

    private static String parseRequestMethod(String rawContent) {
        // create uri
        return rawContent.split("\r\n")[0].split(" ")[0];
    }

    private static String parseHost(String rawContent) {
        String[] headers = rawContent.split("\r\n");
        String host = "host:";
        for (String header : headers) {
            if (header.length() > host.length()) {
                String key = header.substring(0, host.length());
                String value = header.substring(host.length()).trim();
                if (host.equalsIgnoreCase(key)) {
                    if (!value.startsWith("http://") && !value.startsWith("https://")) {
                        value = "http://" + value;
                    }
                    return value;
                }
            }
        }
        return "";
    }

}

@Slf4j
@Data
class Reactor {

    private Selector selector;

    private volatile boolean finish = false;

    @SneakyThrows
    public Reactor() {
        selector = Selector.open();
    }

    @SneakyThrows
    public Reactor pipe(SocketChannel from, SocketChannel to) {
        from.configureBlocking(false);
        from.register(selector, SelectionKey.OP_READ, new SocketPipe(this, from, to));
        return this;
    }

    @SneakyThrows
    public void run() {
        try {
            while (!finish) {
                if (selector.selectNow() > 0) {
                    Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                    while (it.hasNext()) {
                        SelectionKey selectionKey = it.next();
                        if (selectionKey.isValid() && selectionKey.isReadable()) {
                            ((SocketPipe) selectionKey.attachment()).pipe();
                        }
                        it.remove();
                    }
                }
            }
        } finally {
            close();
        }
    }

    @SneakyThrows
    public synchronized void close() {
        if (finish) {
            return;
        }
        finish = true;
        if (!selector.isOpen()) {
            return;
        }
        for (SelectionKey key : selector.keys()) {
            closeChannel(key.channel());
            key.cancel();
        }
        if (selector != null) {
            selector.close();
        }
    }

    public void cancel(SelectableChannel channel) {
        SelectionKey key = channel.keyFor(selector);
        if (Objects.isNull(key)) {
            return;
        }
        key.cancel();
    }

    @SneakyThrows
    public void closeChannel(Channel channel) {
        SocketChannel socketChannel = (SocketChannel)channel;
        if (socketChannel.isConnected() && socketChannel.isOpen()) {
            socketChannel.shutdownOutput();
            socketChannel.shutdownInput();
        }
        socketChannel.close();
    }
}

@Data
@AllArgsConstructor
class SocketPipe {

    private Reactor reactor;

    private SocketChannel from;

    private SocketChannel to;

    @SneakyThrows
    public void pipe() {
        // 取消监听
        clearInterestOps();

        GlobalThreadPool.PIPE_EXECUTOR.submit(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                int totalBytesRead = 0;
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                while (valid(from) && valid(to)) {
                    byteBuffer.clear();
                    int bytesRead = from.read(byteBuffer);
                    totalBytesRead = totalBytesRead + bytesRead;
                    byteBuffer.flip();
                    to.write(byteBuffer);
                    if (bytesRead < byteBuffer.capacity()) {
                        break;
                    }
                }
                if (totalBytesRead < 0) {
                    reactor.closeChannel(from);
                    reactor.cancel(from);
                } else {
                    // 重置监听
                    resetInterestOps();
                }
            }
        });
    }

    protected void clearInterestOps() {
        from.keyFor(reactor.getSelector()).interestOps(0);
        to.keyFor(reactor.getSelector()).interestOps(0);
    }

    protected void resetInterestOps() {
        from.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);
        to.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);
    }

    private boolean valid(SocketChannel channel) {
        return channel.isConnected() && channel.isRegistered() && channel.isOpen();
    }
}

以上,借鉴NETTY:

  1. 首先初始化REACTOR线程,然后开启代理监听,当收到代理请求时处理。
  2. 代理服务在收到代理请求时,首先做代理的预处理,然后又SocketPipe做客户端和远程服务端双向转发。
  3. 代理预处理,首先读取第一个HTTP请求,解析出METHOD, HOST, PORT。
  4. 如果是CONNECT请求,发送回应Connection Established,然后连接远程服务端,并返回SocketChannel
  5. 如果是非CONNECT请求,连接远程服务端,写入原始请求,并返回SocketChannel
  6. SocketPipe在客户端和远程服务端,做双向的转发;其本身是将客户端和服务端的SocketChannel注册到REACTOR
  7. REACTOR在监测到READABLE的CHANNEL,派发给SocketPipe做双向转发。

测试

代理的测试比较简单,指向代码后,代理服务监听8006端口,此时:

curl -x 'localhost:8006' http://httpbin.org/get测试HTTP请求

curl -x 'localhost:8006' https://httpbin.org/get测试HTTPS请求

注意,此时代理服务代理了HTTPS请求,但是并不需要-k选项,指示非安全的代理。因为代理服务本身并没有作为一个中间人,并没有解析出客户端和远程服务端通信的内容。在非透明代理时,需要解决这个问题。

2 非透明代理

非透明代理,需要解析出客户端和远程服务端传输的内容,并做相应的处理。

当传输为HTTP协议时,SocketPipe传输的数据即为明文的数据,可以拦截后直接做处理。

当传输为HTTPS协议时,SocketPipe传输的有效数据为加密数据,并不能透明处理。
另外,无论是传输的HTTP协议还是HTTPS协议,SocketPipe读到的都为非完整的数据,需要做聚批的处理。

  1. SocketPipe聚批问题,可以采用类似BufferedInputStream对InputStream做Decorate的模式来实现,相对比较简单;详细可以参考NETTY的HttpObjectAggregator;
  2. HTTPS原始请求和结果数据的加密和解密的处理,需要实现的NIO的SOCKET CHANNEL;

SslSocketChannel封装原理

考虑到目前JDK自带的NIO的SocketChannel并不支持SSL;已有的SSLSocket是阻塞的OIO。如图:

image.png

可以看出

  1. 每次入站数据和出站数据都需要 SSL SESSION 做握手;
  2. 入站数据做解密,出站数据做加密;
  3. 握手,数据加密和数据解密是统一的一套状态机;

image.png

以下,代码实现 SslSocketChannel

public class SslSocketChannel {

    /**
     * 握手加解密需要的四个存储
     */
    protected ByteBuffer myAppData; // 明文
    protected ByteBuffer myNetData; // 密文
    protected ByteBuffer peerAppData; // 明文
    protected ByteBuffer peerNetData; // 密文

    /**
     * 握手加解密过程中用到的异步执行器
     */
    protected ExecutorService executor = Executors.newSingleThreadExecutor();

    /**
     * 原NIO 的 CHANNEL
     */
    protected SocketChannel socketChannel;

    /**
     * SSL 引擎
     */
    protected SSLEngine engine;

    public SslSocketChannel(SSLContext context, SocketChannel socketChannel, boolean clientMode) throws Exception {
        // 原始的NIO SOCKET
        this.socketChannel = socketChannel;

        // 初始化BUFFER
        SSLSession dummySession = context.createSSLEngine().getSession();
        myAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize());
        myNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize());
        peerAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize());
        peerNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize());
        dummySession.invalidate();

        engine = context.createSSLEngine();
        engine.setUseClientMode(clientMode);
        engine.beginHandshake();
    }

    /**
     * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
     * 实现的 SSL 的握手协议
     * @return
     * @throws IOException
     */
    protected boolean doHandshake() throws IOException {
        SSLEngineResult result;
        HandshakeStatus handshakeStatus;

        int appBufferSize = engine.getSession().getApplicationBufferSize();
        ByteBuffer myAppData = ByteBuffer.allocate(appBufferSize);
        ByteBuffer peerAppData = ByteBuffer.allocate(appBufferSize);
        myNetData.clear();
        peerNetData.clear();

        handshakeStatus = engine.getHandshakeStatus();
        while (handshakeStatus != HandshakeStatus.FINISHED && handshakeStatus != HandshakeStatus.NOT_HANDSHAKING) {
            switch (handshakeStatus) {
                case NEED_UNWRAP:
                    if (socketChannel.read(peerNetData) < 0) {
                        if (engine.isInboundDone() && engine.isOutboundDone()) {
                            return false;
                        }
                        try {
                            engine.closeInbound();
                        } catch (SSLException e) {
                            log.debug("收到END OF STREAM,关闭连接.", e);
                        }
                        engine.closeOutbound();
                        handshakeStatus = engine.getHandshakeStatus();
                        break;
                    }
                    peerNetData.flip();
                    try {
                        result = engine.unwrap(peerNetData, peerAppData);
                        peerNetData.compact();
                        handshakeStatus = result.getHandshakeStatus();
                    } catch (SSLException sslException) {
                        engine.closeOutbound();
                        handshakeStatus = engine.getHandshakeStatus();
                        break;
                    }
                    switch (result.getStatus()) {
                        case OK:
                            break;
                        case BUFFER_OVERFLOW:
                            peerAppData = enlargeApplicationBuffer(engine, peerAppData);
                            break;
                        case BUFFER_UNDERFLOW:
                            peerNetData = handleBufferUnderflow(engine, peerNetData);
                            break;
                        case CLOSED:
                            if (engine.isOutboundDone()) {
                                return false;
                            } else {
                                engine.closeOutbound();
                                handshakeStatus = engine.getHandshakeStatus();
                                break;
                            }
                        default:
                            throw new IllegalStateException("无效的握手状态: " + result.getStatus());
                    }
                    break;
                case NEED_WRAP:
                    myNetData.clear();
                    try {
                        result = engine.wrap(myAppData, myNetData);
                        handshakeStatus = result.getHandshakeStatus();
                    } catch (SSLException sslException) {
                        engine.closeOutbound();
                        handshakeStatus = engine.getHandshakeStatus();
                        break;
                    }
                    switch (result.getStatus()) {
                        case OK :
                            myNetData.flip();
                            while (myNetData.hasRemaining()) {
                                socketChannel.write(myNetData);
                            }
                            break;
                        case BUFFER_OVERFLOW:
                            myNetData = enlargePacketBuffer(engine, myNetData);
                            break;
                        case BUFFER_UNDERFLOW:
                            throw new SSLException("加密后消息内容为空,报错");
                        case CLOSED:
                            try {
                                myNetData.flip();
                                while (myNetData.hasRemaining()) {
                                    socketChannel.write(myNetData);
                                }
                                peerNetData.clear();
                            } catch (Exception e) {
                                handshakeStatus = engine.getHandshakeStatus();
                            }
                            break;
                        default:
                            throw new IllegalStateException("无效的握手状态: " + result.getStatus());
                    }
                    break;
                case NEED_TASK:
                    Runnable task;
                    while ((task = engine.getDelegatedTask()) != null) {
                        executor.execute(task);
                    }
                    handshakeStatus = engine.getHandshakeStatus();
                    break;
                case FINISHED:
                    break;
                case NOT_HANDSHAKING:
                    break;
                default:
                    throw new IllegalStateException("无效的握手状态: " + handshakeStatus);
            }
        }

        return true;
    }

    /**
     * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
     * 实现的 SSL 的传输读取协议
     * @param consumer
     * @throws IOException
     */
    public void read(Consumer<ByteBuffer> consumer) throws IOException {
        // BUFFER初始化
        peerNetData.clear();
        int bytesRead = socketChannel.read(peerNetData);
        if (bytesRead > 0) {
            peerNetData.flip();
            while (peerNetData.hasRemaining()) {
                peerAppData.clear();
                SSLEngineResult result = engine.unwrap(peerNetData, peerAppData);
                switch (result.getStatus()) {
                    case OK:
                        log.debug("收到远程的返回结果消息为:" + new String(peerAppData.array(), 0, peerAppData.position()));
                        consumer.accept(peerAppData);
                        peerAppData.flip();
                        break;
                    case BUFFER_OVERFLOW:
                        peerAppData = enlargeApplicationBuffer(engine, peerAppData);
                        break;
                    case BUFFER_UNDERFLOW:
                        peerNetData = handleBufferUnderflow(engine, peerNetData);
                        break;
                    case CLOSED:
                        log.debug("收到远程连接关闭消息.");
                        closeConnection();
                        return;
                    default:
                        throw new IllegalStateException("无效的握手状态: " + result.getStatus());
                }
            }
        } else if (bytesRead < 0) {
            log.debug("收到END OF STREAM,关闭连接.");
            handleEndOfStream();
        }
    }

    public void write(String message) throws IOException {
        write(ByteBuffer.wrap(message.getBytes()));
    }

    /**
     * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
     * 实现的 SSL 的传输写入协议
     * @param message
     * @throws IOException
     */
    public void write(ByteBuffer message) throws IOException {
        myAppData.clear();
        myAppData.put(message);
        myAppData.flip();
        while (myAppData.hasRemaining()) {
            myNetData.clear();
            SSLEngineResult result = engine.wrap(myAppData, myNetData);
            switch (result.getStatus()) {
                case OK:
                    myNetData.flip();
                    while (myNetData.hasRemaining()) {
                        socketChannel.write(myNetData);
                    }
                    log.debug("写入远程的消息为: {}", message);
                    break;
                case BUFFER_OVERFLOW:
                    myNetData = enlargePacketBuffer(engine, myNetData);
                    break;
                case BUFFER_UNDERFLOW:
                    throw new SSLException("加密后消息内容为空.");
                case CLOSED:
                    closeConnection();
                    return;
                default:
                    throw new IllegalStateException("无效的握手状态: " + result.getStatus());
            }
        }
    }

    /**
     * 关闭连接
     * @throws IOException
     */
    public void closeConnection() throws IOException  {
        engine.closeOutbound();
        doHandshake();
        socketChannel.close();
        executor.shutdown();
    }

    /**
     * END OF STREAM(-1)默认是关闭连接
     * @throws IOException
     */
    protected void handleEndOfStream() throws IOException  {
        try {
            engine.closeInbound();
        } catch (Exception e) {
            log.error("END OF STREAM 关闭失败.", e);
        }
        closeConnection();
    }

}

以上:

  1. 基于 SSL 协议,实现统一的握手动作;
  2. 分别实现读取的解密,和写入的加密方法;
  3. 将 SslSocketChannel 实现为 SocketChannel的Decorator;

SslSocketChannel测试服务端

基于以上封装,简单测试服务端如下

@Slf4j
public class NioSslServer {

    public static void main(String[] args) throws Exception {
        NioSslServer sslServer = new NioSslServer("127.0.0.1", 8006);
        sslServer.start();
        // 使用 curl -vv -k 'https://localhost:8006' 连接
    }

    private SSLContext context;

    private Selector selector;

    public NioSslServer(String hostAddress, int port) throws Exception {
        // 初始化SSL Context
        context = serverSSLContext();

        // 注册监听器
        selector = SelectorProvider.provider().openSelector();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().bind(new InetSocketAddress(hostAddress, port));
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    public void start() throws Exception {

        log.debug("等待连接中.");

        while (true) {
            selector.select();
            Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selectedKeys.remove();
                if (!key.isValid()) {
                    continue;
                }
                if (key.isAcceptable()) {
                    accept(key);
                } else if (key.isReadable()) {
                    ((SslSocketChannel)key.attachment()).read(buf->{});
                    // 直接回应一个OK
                    ((SslSocketChannel)key.attachment()).write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nOK\r\n\r\n");
                    ((SslSocketChannel)key.attachment()).closeConnection();
                }
            }
        }
    }

    private void accept(SelectionKey key) throws Exception {
        log.debug("接收新的请求.");

        SocketChannel socketChannel = ((ServerSocketChannel)key.channel()).accept();
        socketChannel.configureBlocking(false);

        SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, false);
        if (sslSocketChannel.doHandshake()) {
            socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
        } else {
            socketChannel.close();
            log.debug("握手失败,关闭连接.");
        }
    }
}

以上:

  1. 由于是NIO,简单的测试需要用到NIO的基础组件Selector进行测试;
  2. 首先初始化ServerSocketChannel,监听8006端口;
  3. 接收到请求后,将SocketChannel封装为SslSocketChannel,注册到Selector
  4. 接收到数据后,通过SslSocketChannel做read和write;

SslSocketChannel测试客户端

基于以上服务端封装,简单测试客户端如下

@Slf4j
public class NioSslClient {

    public static void main(String[] args) throws Exception {
        NioSslClient sslClient = new NioSslClient("httpbin.org", 443);
        sslClient.connect();
        // 请求 'https://httpbin.org/get'
    }

    private String remoteAddress;

    private int port;

    private SSLEngine engine;

    private SocketChannel socketChannel;

    private SSLContext context;

    /**
     * 需要远程的HOST和PORT
     * @param remoteAddress
     * @param port
     * @throws Exception
     */
    public NioSslClient(String remoteAddress, int port) throws Exception {
        this.remoteAddress = remoteAddress;
        this.port = port;

        context = clientSSLContext();
        engine = context.createSSLEngine(remoteAddress, port);
        engine.setUseClientMode(true);
    }

    public boolean connect() throws Exception {
        socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false);
        socketChannel.connect(new InetSocketAddress(remoteAddress, port));
        while (!socketChannel.finishConnect()) {
            // 通过REACTOR,不会出现等待情况
            //log.debug("连接中..");
        }

        SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, true);
        sslSocketChannel.doHandshake();

        // 握手完成后,开启SELECTOR
        Selector selector = SelectorProvider.provider().openSelector();
        socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);

        // 写入请求
        sslSocketChannel.write("GET /get HTTP/1.1\r\n"
            + "Host: httpbin.org:443\r\n"
            + "User-Agent: curl/7.62.0\r\n"
            + "Accept: */*\r\n"
            + "\r\n");

        // 读取结果
        while (true) {
            selector.select();
            Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selectedKeys.remove();
                if (key.isValid() && key.isReadable()) {
                    ((SslSocketChannel)key.attachment()).read(buf->{
                        log.info("{}", new String(buf.array(), 0, buf.position()));
                    });
                    ((SslSocketChannel)key.attachment()).closeConnection();
                    return true;
                }
            }
        }
    }
}

以上:

  1. 客户端的封装测试,是为了验证封装 SSL 协议双向都是OK的,
  2. 在后文的非透明上游代理中,会同时使用 SslSocketChannel做服务端和客户端
  3. 以上封装与服务端封装类似,不同的是初始化 SocketChannel,做connect而非bind

总结

以上:

  1. 非透明代理需要拿到完整的请求数据,可以通过 Decorator模式,聚批实现;
  2. 非透明代理需要拿到解密后的HTTPS请求数据,可以通过SslSocketChannel对原始的SocketChannel做封装实现;
  3. 最后,拿到请求后,做相应的处理,最终实现非透明的代理。

3 透明上游代理

透明上游代理相比透明代理要简单,区别是

  1. 透明代理需要响应 CONNECT请求,透明上游代理不需要,直接转发即可;
  2. 透明代理需要解析CONNECT请求中的HOST和PORT,并连接服务端;透明上游代理只需要连接下游代理的IP:PORT,直接转发请求即可;
  3. 透明的上游代理,只是一个简单的SocketChannel管道;确定下游的代理服务端,连接转发请求;

只需要对透明代理做以上简单的修改,即可实现透明的上游代理。

4 非透明上游代理

非透明的上游代理,相比非透明的代理要复杂一些

image.png

以上,分为四个组件:客户端,代理服务(ServerHandler),代理服务(ClientHandler),服务端

  1. 如果是HTTP的请求,数据直接通过 客户端<->ServerHandler<->ClientHandler<->服务端,代理网关只需要做简单的请求聚批,就可以应用相应的管理策略;
  2. 如果是HTTPS请求,代理作为客户端和服务端的中间人,只能拿到加密的数据;因此,代理网关需要作为HTTPS的服务方与客户端通信;然后作为HTTPS的客户端与服务端通信;
  3. 代理作为HTTPS服务方时,需要考虑到其本身是个非透明的代理,需要实现非透明代理相关的协议;
  4. 代理作为HTTPS客户端时,需要考虑到其下游是个透明的代理,真正的服务方是客户端请求的服务方;

三 设计与实现

本文需要构建的是非透明上游代理,以下采用NETTY框架给出详细的设计实现。上文将统一代理网关分为两大部分,ServerHandler和ClientHandler,以下

  1. 介绍代理网关服务端相关实现;
  2. 介绍代理网关客户端相关实现;

1 代理网关服务端

主要包括

  1. 初始化代理网关服务端
  2. 初始化服务端处理器
  3. 服务端协议升级与处理

初始化代理网关服务

    public void start() {
        HookedExecutors.newSingleThreadExecutor().submit(() ->{
            log.info("开始启动代理服务器,监听端口:{}", auditProxyConfig.getProxyServerPort());
            EventLoopGroup bossGroup = new NioEventLoopGroup(auditProxyConfig.getBossThreadCount());
            EventLoopGroup workerGroup = new NioEventLoopGroup(auditProxyConfig.getWorkThreadCount());
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .childHandler(new ServerChannelInitializer(auditProxyConfig))
                    .bind(auditProxyConfig.getProxyServerPort()).sync().channel().closeFuture().sync();
            } catch (InterruptedException e) {
                log.error("代理服务器被中断.", e);
                Thread.currentThread().interrupt();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        });
    }

代理网关初始化相对简单,

  1. bossGroup线程组,负责接收请求
  2. workerGroup线程组,负责处理接收的请求数据,具体处理逻辑封装在ServerChannelInitializer中。

代理网关服务的请求处理器在 ServerChannelInitializer中定义为

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
            .addLast(new HttpRequestDecoder())
            .addLast(new HttpObjectAggregator(auditProxyConfig.getMaxRequestSize()))
            .addLast(new ServerChannelHandler(auditProxyConfig));
    }

首先解析HTTP请求,然后做聚批的处理,最后ServerChannelHandler实现代理网关协议;

代理网关协议:

  1. 判定是否是CONNECT请求,如果是,会存储CONNECT请求;暂停读取,发送代理成功的响应,并在回应成功后,升级协议;
  2. 升级引擎,本质上是采用SslSocketChannel对原SocketChannel做透明的封装;
  3. 最后根据CONNECT请求连接远程服务端;

详细实现为:

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        FullHttpRequest request = (FullHttpRequest)msg;

        try {
            if (isConnectRequest(request)) {
                // CONNECT 请求,存储待处理
                saveConnectRequest(ctx, request);

                // 禁止读取
                ctx.channel().config().setAutoRead(false);

                // 发送回应
                connectionEstablished(ctx, ctx.newPromise().addListener(future -> {
                    if (future.isSuccess()) {
                        // 升级
                        if (isSslRequest(request) && !isUpgraded(ctx)) {
                            upgrade(ctx);
                        }

                        // 开放消息读取
                        ctx.channel().config().setAutoRead(true);
                        ctx.read();
                    }
                }));

            } else {
                // 其他请求,判定是否已升级
                if (!isUpgraded(ctx)) {

                    // 升级引擎
                    upgrade(ctx);
                }

                // 连接远程
                connectRemote(ctx, request);
            }
        } finally {
            ctx.fireChannelRead(msg);
        }
    }

2 代理网关客户端

代理网关服务端需要连接远程服务,进入代理网关客户端部分。

代理网关客户端初始化:

    /**
     * 初始化远程连接
     * @param ctx
     * @param httpRequest
     */
    protected void connectRemote(ChannelHandlerContext ctx, FullHttpRequest httpRequest) {
        Bootstrap b = new Bootstrap();
        b.group(ctx.channel().eventLoop()) // use the same EventLoop
            .channel(ctx.channel().getClass())
            .handler(new ClientChannelInitializer(auditProxyConfig, ctx, safeCopy(httpRequest)));

        // 动态连接代理
        FullHttpRequest originRequest = ctx.channel().attr(CONNECT_REQUEST).get();
        if (originRequest == null) {
            originRequest = httpRequest;
        }
        ChannelFuture cf = b.connect(new InetSocketAddress(calculateHost(originRequest), calculatePort(originRequest)));
        Channel cch = cf.channel();
        ctx.channel().attr(CLIENT_CHANNEL).set(cch);    
    }

以上:

  1. 复用代理网关服务端的workerGroup线程组;
  2. 请求和结果的处理封装在ClientChannelInitializer;
  3. 连接的远程服务端的HOST和PORT在服务端收到的请求中可以解析到。

代理网关客户端的处理器的初始化逻辑:

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        SocketAddress socketAddress = calculateProxy();
        if (!Objects.isNull(socketAddress)) {
            ch.pipeline().addLast(new HttpProxyHandler(calculateProxy(), auditProxyConfig.getUserName(), auditProxyConfig
                .getPassword()));
        }
        if (isSslRequest()) {
            String host = host();
            int port = port();
            if (StringUtils.isNoneBlank(host) && port > 0) {
                ch.pipeline().addLast(new SslHandler(sslEngine(host, port)));
            }
        }
        ch.pipeline().addLast(new ClientChannelHandler(clientContext, httpRequest));
    }

以上:

  1. 如果下游是代理,那么会采用HttpProxyHandler,经由下游代理与远程服务端通信;
  2. 如果当前需要升级为SSL协议,会对SocketChannel做透明的封装,实现SSL通信。
  3. 最后,ClientChannelHandler只是简单消息的转发;唯一的不同是,由于代理网关拦截了第一个请求,此时需要将拦截的请求,转发到服务端。

四 其他问题

代理网关实现可能面临的问题:

1 内存问题

代理通常面临的问题是OOM。本文在实现代理网关时保证内存中缓存时当前正在处理的HTTP/HTTPS请求体。内存使用的上限理论上为实时处理的请求数量*请求体的平均大小,HTTP/HTTPS的请求结果,直接使用堆外内存,零拷贝转发。

2 性能问题

性能问题不应提早考虑。本文使用NETTY框架实现的代理网关,内部大量使用堆外内存,零拷贝转发,避免了性能问题。

代理网关一期上线后曾面临一个长连接导致的性能问题,

  1. CLIENT和SERVER建立TCP长连接后(比如,TCP心跳检测),通常要么是CLIENT关闭TCP连接,或者是SERVER关闭;
  2. 如果双方长时间占用TCP连接资源而不关闭,就会导致SOCKET资源泄漏;现象是:CPU资源爆满,处理空闲连接;新连接无法建立;

使用IdleStateHandler定时监控空闲的TCP连接,强制关闭;解决了该问题。

五 总结

本文聚焦于统一代理网关的核心,详细介绍了代理相关的技术原理。

代理网关的管理部分,可以在ServerHandler部分维护,也可以在ClientHandler部分维护;

  1. ServerHandler可以拦截转换请求
  2. ClientHanlder可控制请求的出口
注:本文使用Netty的零拷贝;存储请求以解析处理;但并未实现对RESPONSE的处理;也就是RESPONSE是直接通过网关,此方面避免了常见的代理实现,内存泄漏OOM相关问题;

最后,本文实现代理网关后,针对代理的资源和流经代理网关的请求做了相应的控制,主要包括:

  1. 当遇到静态资源的请求时,代理网关会直接请求远程服务端,不会通过下游代理
  2. 当请求HEADER中包含地域标识时,代理网关会尽力保证请求打入指定的地域代理,经由地域代理访问远程服务端

本文参考https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html实现 SslSocketChannel,以透明处理HTTP和HTTPS协议。


基于ELK+Flink日志全观测最佳实践

点击这里,查看详情!

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
8827 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,大概有三种登录方式:
2865 0
阿里云服务器ECS远程登录用户名密码查询方法
阿里云服务器ECS远程连接登录输入用户名和密码,阿里云没有默认密码,如果购买时没设置需要先重置实例密码,Windows用户名是administrator,Linux账号是root,阿小云来详细说下阿里云服务器远程登录连接用户名和密码查询方法
11065 0
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
10647 0
使用OpenApi弹性释放和设置云服务器ECS释放
云服务器ECS的一个重要特性就是按需创建资源。您可以在业务高峰期按需弹性的自定义规则进行资源创建,在完成业务计算的时候释放资源。本篇将提供几个Tips帮助您更加容易和自动化的完成云服务器的释放和弹性设置。
11960 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
12475 0
windows server 2008阿里云ECS服务器安全设置
最近我们Sinesafe安全公司在为客户使用阿里云ecs服务器做安全的过程中,发现服务器基础安全性都没有做。为了为站长们提供更加有效的安全基础解决方案,我们Sinesafe将对阿里云服务器win2008 系统进行基础安全部署实战过程! 比较重要的几部分 1.
8923 0
腾讯云服务器 设置ngxin + fastdfs +tomcat 开机自启动
在tomcat中新建一个可以启动的 .sh 脚本文件 /usr/local/tomcat7/bin/ export JAVA_HOME=/usr/local/java/jdk7 export PATH=$JAVA_HOME/bin/:$PATH export CLASSPATH=.
4575 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,云吞铺子总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系统盘、创建快照、配置安全组等操作如何登录ECS云服务器控制台? 1、先登录到阿里云ECS服务器控制台 2、点击顶部的“控制台” 3、通过左侧栏,切换到“云服务器ECS”即可,如下图所示 通过ECS控制台的远程连接来登录到云服务器 阿里云ECS云服务器自带远程连接功能,使用该功能可以登录到云服务器,简单且方便,如下图:点击“远程连接”,第一次连接会自动生成6位数字密码,输入密码即可登录到云服务器上。
21768 0
313
文章
1008
问答
来源圈子
更多
+ 订阅
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载