基于Netty的代理网关设计与实现

时间：2022-02-27 02:55:47

基于Netty的代理网关设计与实现

一、问题背景

平台端购置一批裸代理，来做广告异地展现审核。从外部购置的代理，使用方式为：

  • 通过给定的HTTP 的 API 提取代理 IP:PORT，返回的结果会给出代理的有效时长 3~5 分钟，以及代理所属地域;
  • 从提取的代理中，选取指定地域，添加认证信息，请求获取结果;

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

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

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

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

二、技术路线

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

  • 透明代理;
  • 非透明代理;
  • 透明的上游代理;
  • 非透明的上游代理;

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

1.透明代理

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

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

基于Netty的代理网关设计与实现

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

需要注意的点是：

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

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

  1. @Slf4j
  2. public class SimpleTransProxy {
  4. public static void main(String[] args) throws IOException {
  5. int port = 8006;
  6. ServerSocketChannel localServer = ServerSocketChannel.open();
  7. localServer.bind(new InetSocketAddress(port));
  8. Reactor reactor = new Reactor();
  9. // REACTOR线程
  10. GlobalThreadPool.REACTOR_EXECUTOR.submit(reactor::run);
  12. // WORKER单线程调试
  13. while (localServer.isOpen()) {
  14. // 此处阻塞等待连接
  15. SocketChannel remoteClient = localServer.accept();
  17. // 工作线程
  18. GlobalThreadPool.WORK_EXECUTOR.submit(new Runnable() {
  19. @SneakyThrows
  20. @Override
  21. public void run() {
  22. // 代理到远程
  23. SocketChannel remoteServer = new ProxyHandler().proxy(remoteClient);
  25. // 透明传输
  26. reactor.pipe(remoteClient, remoteServer)
  27. .pipe(remoteServer, remoteClient);
  28. }
  29. });
  30. }
  31. }
  32. }
  34. @Data
  35. class ProxyHandler {
  36. private String method;
  37. private String host;
  38. private int port;
  39. private SocketChannel remoteServer;
  40. private SocketChannel remoteClient;
  42. /**
  43. * 原始信息
  44. */
  45. private List buffers = new ArrayList<>();
  46. private StringBuilder stringBuilder = new StringBuilder();
  48. /**
  49. * 连接到远程
  50. * @param remoteClient
  51. * @return
  52. * @throws IOException
  53. */
  54. public SocketChannel proxy(SocketChannel remoteClient) throws IOException {
  55. this.remoteClient = remoteClient;
  56. connect();
  57. return this.remoteServer;
  58. }
  60. public void connect() throws IOException {
  61. // 解析METHOD, HOST和PORT
  62. beforeConnected();
  64. // 链接REMOTE SERVER
  65. createRemoteServer();
  67. // CONNECT请求回应，其他请求WRITE THROUGH
  68. afterConnected();
  69. }
  71. protected void beforeConnected() throws IOException {
  72. // 读取HEADER
  73. readAllHeader();
  75. // 解析HOST和PORT
  76. parseRemoteHostAndPort();
  77. }
  79. /**
  80. * 创建远程连接
  81. * @throws IOException
  82. */
  83. protected void createRemoteServer() throws IOException {
  84. remoteServer = SocketChannel.open(new InetSocketAddress(host, port));
  85. }
  87. /**
  88. * 连接建立后预处理
  89. * @throws IOException
  90. */
  91. protected void afterConnected() throws IOException {
  92. // 当CONNECT请求时，默认写入200到CLIENT
  93. if ("CONNECT".equalsIgnoreCase(method)) {
  94. // CONNECT默认为443端口，根据HOST再解析
  95. remoteClient.write(ByteBuffer.wrap("HTTP/1.0 200 Connection Established\r\nProxy-agent: nginx\r\n\r\n".getBytes()));
  96. } else {
  97. writeThrouth();
  98. }
  99. }
  101. protected void writeThrouth() {
  102. buffers.forEach(byteBuffer -> {
  103. try {
  104. remoteServer.write(byteBuffer);
  105. } catch (IOException e) {
  106. e.printStackTrace();
  107. }
  108. });
  109. }
  111. /**
  112. * 读取请求内容
  113. * @throws IOException
  114. */
  115. protected void readAllHeader() throws IOException {
  116. while (true) {
  117. ByteBuffer clientBuffer = newByteBuffer();
  118. int read = remoteClient.read(clientBuffer);
  119. clientBuffer.flip();
  120. appendClientBuffer(clientBuffer);
  121. if (read < clientBuffer.capacity()) {
  122. break;
  123. }
  124. }
  125. }
  127. /**
  128. * 解析出HOST和PROT
  129. * @throws IOException
  130. */
  131. protected void parseRemoteHostAndPort() throws IOException {
  132. // 读取第一批，获取到METHOD
  133. method = parseRequestMethod(stringBuilder.toString());
  135. // 默认为80端口，根据HOST再解析
  136. port = 80;
  137. if ("CONNECT".equalsIgnoreCase(method)) {
  138. port = 443;
  139. }
  141. this.host = parseHost(stringBuilder.toString());
  143. URI remoteServerURI = URI.create(host);
  144. host = remoteServerURI.getHost();
  146. if (remoteServerURI.getPort() > 0) {
  147. port = remoteServerURI.getPort();
  148. }
  149. }
  151. protected void appendClientBuffer(ByteBuffer clientBuffer) {
  152. buffers.add(clientBuffer);
  153. stringBuilder.append(new String(clientBuffer.array(), clientBuffer.position(), clientBuffer.limit()));
  154. }
  156. protected static ByteBuffer newByteBuffer() {
  157. // buffer必须大于7，保证能读到method
  158. return ByteBuffer.allocate(128);
  159. }
  161. private static String parseRequestMethod(String rawContent) {
  162. // create uri
  163. return rawContent.split("\r\n")[0].split(" ")[0];
  164. }
  166. private static String parseHost(String rawContent) {
  167. String[] headers = rawContent.split("\r\n");
  168. String host = "host:";
  169. for (String header : headers) {
  170. if (header.length() > host.length()) {
  171. String key = header.substring(0, host.length());
  172. String value = header.substring(host.length()).trim();
  173. if (host.equalsIgnoreCase(key)) {
  174. if (!value.startsWith("http://") && !value.startsWith("https://")) {
  175. value = "http://" + value;
  176. }
  177. return value;
  178. }
  179. }
  180. }
  181. return "";
  182. }
  184. }
  186. @Slf4j
  187. @Data
  188. class Reactor {
  190. private Selector selector;
  192. private volatile boolean finish = false;
  194. @SneakyThrows
  195. public Reactor() {
  196. selector = Selector.open();
  197. }
  199. @SneakyThrows
  200. public Reactor pipe(SocketChannel from, SocketChannel to) {
  201. from.configureBlocking(false);
  202. from.register(selector, SelectionKey.OP_READ, new SocketPipe(this, from, to));
  203. return this;
  204. }
  206. @SneakyThrows
  207. public void run() {
  208. try {
  209. while (!finish) {
  210. if (selector.selectNow() > 0) {
  211. Iterator it = selector.selectedKeys().iterator();
  212. while (it.hasNext()) {
  213. SelectionKey selectionKey = it.next();
  214. if (selectionKey.isValid() && selectionKey.isReadable()) {
  215. ((SocketPipe) selectionKey.attachment()).pipe();
  216. }
  217. it.remove();
  218. }
  219. }
  220. }
  221. } finally {
  222. close();
  223. }
  224. }
  226. @SneakyThrows
  227. public synchronized void close() {
  228. if (finish) {
  229. return;
  230. }
  231. finish = true;
  232. if (!selector.isOpen()) {
  233. return;
  234. }
  235. for (SelectionKey key : selector.keys()) {
  236. closeChannel(key.channel());
  237. key.cancel();
  238. }
  239. if (selector != null) {
  240. selector.close();
  241. }
  242. }
  244. public void cancel(SelectableChannel channel) {
  245. SelectionKey key = channel.keyFor(selector);
  246. if (Objects.isNull(key)) {
  247. return;
  248. }
  249. key.cancel();
  250. }
  252. @SneakyThrows
  253. public void closeChannel(Channel channel) {
  254. SocketChannel socketChannel = (SocketChannel)channel;
  255. if (socketChannel.isConnected() && socketChannel.isOpen()) {
  256. socketChannel.shutdownOutput();
  257. socketChannel.shutdownInput();
  258. }
  259. socketChannel.close();
  260. }
  261. }
  263. @Data
  264. @AllArgsConstructor
  265. class SocketPipe {
  267. private Reactor reactor;
  269. private SocketChannel from;
  271. private SocketChannel to;
  273. @SneakyThrows
  274. public void pipe() {
  275. // 取消监听
  276. clearInterestOps();
  278. GlobalThreadPool.PIPE_EXECUTOR.submit(new Runnable() {
  279. @SneakyThrows
  280. @Override
  281. public void run() {
  282. int totalBytesRead = 0;
  283. ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
  284. while (valid(from) && valid(to)) {
  285. byteBuffer.clear();
  286. int bytesRead = from.read(byteBuffer);
  287. totalBytesRead = totalBytesRead + bytesRead;
  288. byteBuffer.flip();
  289. to.write(byteBuffer);
  290. if (bytesRead < byteBuffer.capacity()) {
  291. break;
  292. }
  293. }
  294. if (totalBytesRead < 0) {
  295. reactor.closeChannel(from);
  296. reactor.cancel(from);
  297. } else {
  298. // 重置监听
  299. resetInterestOps();
  300. }
  301. }
  302. });
  303. }
  305. protected void clearInterestOps() {
  306. from.keyFor(reactor.getSelector()).interestOps(0);
  307. to.keyFor(reactor.getSelector()).interestOps(0);
  308. }
  310. protected void resetInterestOps() {
  311. from.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);
  312. to.keyFor(reactor.getSelector()).interestOps(SelectionKey.OP_READ);
  313. }
  315. private boolean valid(SocketChannel channel) {
  316. return channel.isConnected() && channel.isRegistered() && channel.isOpen();
  317. }
  318. }

以上，借鉴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端口，此时：

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

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

2.非透明代理

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

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

当传输为HTTPS协议时，SocketPipe传输的有效数据为加密数据，并不能透明处理。

另外，无论是传输的HTTP协议还是HTTPS协议，SocketPipe读到的都为非完整的数据，需要做聚批的处理。

SocketPipe聚批问题，可以采用类似BufferedInputStream对InputStream做Decorate的模式来实现，相对比较简单;详细可以参考NETTY的HttpObjectAggregator;

HTTPS原始请求和结果数据的加密和解密的处理，需要实现的NIO的SOCKET CHANNEL;

SslSocketChannel封装原理

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

基于Netty的代理网关设计与实现

可以看出

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

基于Netty的代理网关设计与实现

以下，代码实现 SslSocketChannel

  1. public class SslSocketChannel {
  3. /**
  4. * 握手加解密需要的四个存储
  5. */
  6. protected ByteBuffer myAppData; // 明文
  7. protected ByteBuffer myNetData; // 密文
  8. protected ByteBuffer peerAppData; // 明文
  9. protected ByteBuffer peerNetData; // 密文
  11. /**
  12. * 握手加解密过程中用到的异步执行器
  13. */
  14. protected ExecutorService executor = Executors.newSingleThreadExecutor();
  16. /**
  17. * 原NIO 的 CHANNEL
  18. */
  19. protected SocketChannel socketChannel;
  21. /**
  22. * SSL 引擎
  23. */
  24. protected SSLEngine engine;
  26. public SslSocketChannel(SSLContext context, SocketChannel socketChannel, boolean clientMode) throws Exception {
  27. // 原始的NIO SOCKET
  28. this.socketChannel = socketChannel;
  30. // 初始化BUFFER
  31. SSLSession dummySession = context.createSSLEngine().getSession();
  32. myAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize());
  33. myNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize());
  34. peerAppData = ByteBuffer.allocate(dummySession.getApplicationBufferSize());
  35. peerNetData = ByteBuffer.allocate(dummySession.getPacketBufferSize());
  36. dummySession.invalidate();
  38. engine = context.createSSLEngine();
  39. engine.setUseClientMode(clientMode);
  40. engine.beginHandshake();
  41. }
  43. /**
  44. * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
  45. * 实现的 SSL 的握手协议
  46. * @return
  47. * @throws IOException
  48. */
  49. protected boolean doHandshake() throws IOException {
  50. SSLEngineResult result;
  51. HandshakeStatus handshakeStatus;
  53. int appBufferSize = engine.getSession().getApplicationBufferSize();
  54. ByteBuffer myAppData = ByteBuffer.allocate(appBufferSize);
  55. ByteBuffer peerAppData = ByteBuffer.allocate(appBufferSize);
  56. myNetData.clear();
  57. peerNetData.clear();
  59. handshakeStatus = engine.getHandshakeStatus();
  60. while (handshakeStatus != HandshakeStatus.FINISHED && handshakeStatus != HandshakeStatus.NOT_HANDSHAKING) {
  61. switch (handshakeStatus) {
  62. case NEED_UNWRAP:
  63. if (socketChannel.read(peerNetData) < 0) {
  64. if (engine.isInboundDone() && engine.isOutboundDone()) {
  65. return false;
  66. }
  67. try {
  68. engine.closeInbound();
  69. } catch (SSLException e) {
  70. log.debug("收到END OF STREAM，关闭连接.", e);
  71. }
  72. engine.closeOutbound();
  73. handshakeStatus = engine.getHandshakeStatus();
  74. break;
  75. }
  76. peerNetData.flip();
  77. try {
  78. result = engine.unwrap(peerNetData, peerAppData);
  79. peerNetData.compact();
  80. handshakeStatus = result.getHandshakeStatus();
  81. } catch (SSLException sslException) {
  82. engine.closeOutbound();
  83. handshakeStatus = engine.getHandshakeStatus();
  84. break;
  85. }
  86. switch (result.getStatus()) {
  87. case OK:
  88. break;
  89. case BUFFER_OVERFLOW:
  90. peerAppData = enlargeApplicationBuffer(engine, peerAppData);
  91. break;
  92. case BUFFER_UNDERFLOW:
  93. peerNetData = handleBufferUnderflow(engine, peerNetData);
  94. break;
  95. case CLOSED:
  96. if (engine.isOutboundDone()) {
  97. return false;
  98. } else {
  99. engine.closeOutbound();
  100. handshakeStatus = engine.getHandshakeStatus();
  101. break;
  102. }
  103. default:
  104. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
  105. }
  106. break;
  107. case NEED_WRAP:
  108. myNetData.clear();
  109. try {
  110. result = engine.wrap(myAppData, myNetData);
  111. handshakeStatus = result.getHandshakeStatus();
  112. } catch (SSLException sslException) {
  113. engine.closeOutbound();
  114. handshakeStatus = engine.getHandshakeStatus();
  115. break;
  116. }
  117. switch (result.getStatus()) {
  118. case OK :
  119. myNetData.flip();
  120. while (myNetData.hasRemaining()) {
  121. socketChannel.write(myNetData);
  122. }
  123. break;
  124. case BUFFER_OVERFLOW:
  125. myNetData = enlargePacketBuffer(engine, myNetData);
  126. break;
  127. case BUFFER_UNDERFLOW:
  128. throw new SSLException("加密后消息内容为空，报错");
  129. case CLOSED:
  130. try {
  131. myNetData.flip();
  132. while (myNetData.hasRemaining()) {
  133. socketChannel.write(myNetData);
  134. }
  135. peerNetData.clear();
  136. } catch (Exception e) {
  137. handshakeStatus = engine.getHandshakeStatus();
  138. }
  139. break;
  140. default:
  141. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
  142. }
  143. break;
  144. case NEED_TASK:
  145. Runnable task;
  146. while ((task = engine.getDelegatedTask()) != null) {
  147. executor.execute(task);
  148. }
  149. handshakeStatus = engine.getHandshakeStatus();
  150. break;
  151. case FINISHED:
  152. break;
  153. case NOT_HANDSHAKING:
  154. break;
  155. default:
  156. throw new IllegalStateException("无效的握手状态: " + handshakeStatus);
  157. }
  158. }
  160. return true;
  161. }
  163. /**
  164. * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
  165. * 实现的 SSL 的传输读取协议
  166. * @param consumer
  167. * @throws IOException
  168. */
  169. public void read(Consumer consumer) throws IOException {
  170. // BUFFER初始化
  171. peerNetData.clear();
  172. int bytesRead = socketChannel.read(peerNetData);
  173. if (bytesRead > 0) {
  174. peerNetData.flip();
  175. while (peerNetData.hasRemaining()) {
  176. peerAppData.clear();
  177. SSLEngineResult result = engine.unwrap(peerNetData, peerAppData);
  178. switch (result.getStatus()) {
  179. case OK:
  180. log.debug("收到远程的返回结果消息为：" + new String(peerAppData.array(), 0, peerAppData.position()));
  181. consumer.accept(peerAppData);
  182. peerAppData.flip();
  183. break;
  184. case BUFFER_OVERFLOW:
  185. peerAppData = enlargeApplicationBuffer(engine, peerAppData);
  186. break;
  187. case BUFFER_UNDERFLOW:
  188. peerNetData = handleBufferUnderflow(engine, peerNetData);
  189. break;
  190. case CLOSED:
  191. log.debug("收到远程连接关闭消息.");
  192. closeConnection();
  193. return;
  194. default:
  195. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
  196. }
  197. }
  198. } else if (bytesRead < 0) {
  199. log.debug("收到END OF STREAM，关闭连接.");
  200. handleEndOfStream();
  201. }
  202. }
  204. public void write(String message) throws IOException {
  205. write(ByteBuffer.wrap(message.getBytes()));
  206. }
  208. /**
  209. * 参考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
  210. * 实现的 SSL 的传输写入协议
  211. * @param message
  212. * @throws IOException
  213. */
  214. public void write(ByteBuffer message) throws IOException {
  215. myAppData.clear();
  216. myAppData.put(message);
  217. myAppData.flip();
  218. while (myAppData.hasRemaining()) {
  219. myNetData.clear();
  220. SSLEngineResult result = engine.wrap(myAppData, myNetData);
  221. switch (result.getStatus()) {
  222. case OK:
  223. myNetData.flip();
  224. while (myNetData.hasRemaining()) {
  225. socketChannel.write(myNetData);
  226. }
  227. log.debug("写入远程的消息为: {}", message);
  228. break;
  229. case BUFFER_OVERFLOW:
  230. myNetData = enlargePacketBuffer(engine, myNetData);
  231. break;
  232. case BUFFER_UNDERFLOW:
  233. throw new SSLException("加密后消息内容为空.");
  234. case CLOSED:
  235. closeConnection();
  236. return;
  237. default:
  238. throw new IllegalStateException("无效的握手状态: " + result.getStatus());
  239. }
  240. }
  241. }
  243. /**
  244. * 关闭连接
  245. * @throws IOException
  246. */
  247. public void closeConnection() throws IOException {
  248. engine.closeOutbound();
  249. doHandshake();
  250. socketChannel.close();
  251. executor.shutdown();
  252. }
  254. /**
  255. * END OF STREAM(-1)默认是关闭连接
  256. * @throws IOException
  257. */
  258. protected void handleEndOfStream() throws IOException {
  259. try {
  260. engine.closeInbound();
  261. } catch (Exception e) {
  262. log.error("END OF STREAM 关闭失败.", e);
  263. }
  264. closeConnection();
  265. }
  267. }

以上：

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

SslSocketChannel测试服务端

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

  1. @Slf4j
  2. public class NioSslServer {
  4. public static void main(String[] args) throws Exception {
  5. NioSslServer sslServer = new NioSslServer("127.0.0.1", 8006);
  6. sslServer.start()
  7. // 使用 curl -vv -k 'https://localhost:8006' 连接
  8. }
  10. private SSLContext context;
  12. private Selector selector;
  14. public NioSslServer(String hostAddress, int port) throws Exception {
  15. // 初始化SSL Context
  16. context = serverSSLContext();
  18. // 注册监听器
  19. selector = SelectorProvider.provider().openSelector();
  20. ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  21. serverSocketChannel.configureBlocking(false);
  22. serverSocketChannel.socket().bind(new InetSocketAddress(hostAddress, port));
  23. serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  24. }
  26. public void start() throws Exception {
  28. log.debug("等待连接中.");
  30. while (true) {
  31. selector.select();
  32. Iterator selectedKeys = selector.selectedKeys().iterator();
  33. while (selectedKeys.hasNext()) {
  34. SelectionKey key = selectedKeys.next();
  35. selectedKeys.remove();
  36. if (!key.isValid()) {
  37. continue;
  38. }
  39. if (key.isAcceptable()) {
  40. accept(key);
  41. } else if (key.isReadable()) {
  42. ((SslSocketChannel)key.attachment()).read(buf->{});
  43. // 直接回应一个OK
  44. ((SslSocketChannel)key.attachment()).write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nOK\r\n\r\n");
  45. ((SslSocketChannel)key.attachment()).closeConnection();
  46. }
  47. }
  48. }
  49. }
  51. private void accept(SelectionKey key) throws Exception {
  52. log.debug("接收新的请求.");
  54. SocketChannel socketChannel = ((ServerSocketChannel)key.channel()).accept();
  55. socketChannel.configureBlocking(false);
  57. SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, false);
  58. if (sslSocketChannel.doHandshake()) {
  59. socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
  60. } else {
  61. socketChannel.close();
  62. log.debug("握手失败，关闭连接.");
  63. }
  64. }
  65. }

以上：

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

SslSocketChannel测试服务端

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

  1. @Slf4j
  2. public class NioSslServer {
  4. public static void main(String[] args) throws Exception {
  5. NioSslServer sslServer = new NioSslServer("127.0.0.1", 8006);
  6. sslServer.start();
  7. // 使用 curl -vv -k 'https://localhost:8006' 连接
  8. }
  10. private SSLContext context;
  12. private Selector selector;
  14. public NioSslServer(String hostAddress, int port) throws Exception {
  15. // 初始化SSL Context
  16. context = serverSSLContext();
  18. // 注册监听器
  19. selector = SelectorProvider.provider().openSelector();
  20. ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  21. serverSocketChannel.configureBlocking(false);
  22. serverSocketChannel.socket().bind(new InetSocketAddress(hostAddress, port));
  23. serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  24. }
  26. public void start() throws Exception {
  28. log.debug("等待连接中.");
  30. while (true) {
  31. selector.select();
  32. Iterator selectedKeys = selector.selectedKeys().iterator();
  33. while (selectedKeys.hasNext()) {
  34. SelectionKey key = selectedKeys.next();
  35. selectedKeys.remove();
  36. if (!key.isValid()) {
  37. continue;
  38. }
  39. if (key.isAcceptable()) {
  40. accept(key);
  41. } else if (key.isReadable()) {
  42. ((SslSocketChannel)key.attachment()).read(buf->{});
  43. // 直接回应一个OK
  44. ((SslSocketChannel)key.attachment()).write("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nOK\r\n\r\n");
  45. ((SslSocketChannel)key.attachment()).closeConnection();
  46. }
  47. }
  48. }
  49. }
  51. private void accept(SelectionKey key) throws Exception {
  52. log.debug("接收新的请求.");
  54. SocketChannel socketChannel = ((ServerSocketChannel)key.channel()).accept();
  55. socketChannel.configureBlocking(false);
  57. SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, false);
  58. if (sslSocketChannel.doHandshake()) {
  59. socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
  60. } else {
  61. socketChannel.close();
  62. log.debug("握手失败，关闭连接.");
  63. }
  64. }
  65. }

以上：

由于是NIO，简单的测试需要用到NIO的基础组件Selector进行测试;

首先初始化ServerSocketChannel，监听8006端口;

接收到请求后，将SocketChannel封装为SslSocketChannel，注册到Selector；

接收到数据后，通过SslSocketChannel做read和write;

以上：

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

SslSocketChannel测试客户端

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

  1. @Slf4j ：
  2. public class NioSslClient {
  4. public static void main(String[] args) throws Exception {
  5. NioSslClient sslClient = new NioSslClient("httpbin.org", 443);
  6. sslClient.connect();
  7. // 请求 'https://httpbin.org/get'
  8. }
  10. private String remoteAddress;
  12. private int port;
  14. private SSLEngine engine;
  16. private SocketChannel socketChannel;
  18. private SSLContext context;
  20. /**
  21. * 需要远程的HOST和PORT
  22. * @param remoteAddress
  23. * @param port
  24. * @throws Exception
  25. */
  26. public NioSslClient(String remoteAddress, int port) throws Exception {
  27. this.remoteAddress = remoteAddress;
  28. this.port = port;
  30. context = clientSSLContext();
  31. engine = context.createSSLEngine(remoteAddress, port);
  32. engine.setUseClientMode(true);
  33. }
  35. public boolean connect() throws Exception {
  36. socketChannel = SocketChannel.open();
  37. socketChannel.configureBlocking(false);
  38. socketChannel.connect(new InetSocketAddress(remoteAddress, port));
  39. while (!socketChannel.finishConnect()) {
  40. // 通过REACTOR，不会出现等待情况
  41. //log.debug("连接中..");
  42. }
  44. SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, true);
  45. sslSocketChannel.doHandshake();
  47. // 握手完成后，开启SELECTOR
  48. Selector selector = SelectorProvider.provider().openSelector();
  49. socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
  51. // 写入请求
  52. sslSocketChannel.write("GET /get HTTP/1.1\r\n"
  53. + "Host: httpbin.org:443\r\n"
  54. + "User-Agent: curl/7.62.0\r\n"
  55. + "Accept: */*\r\n"
  56. + "\r\n");
  58. // 读取结果
  59. while (true) {
  60. selector.select();
  61. Iterator selectedKeys = selector.selectedKeys().iterator();
  62. while (selectedKeys.hasNext()) {
  63. SelectionKey key = selectedKeys.next();
  64. selectedKeys.remove();
  65. if (key.isValid() && key.isReadable()) {
  66. ((SslSocketChannel)key.attachment()).read(buf->{
  67. log.info("{}", new String(buf.array(), 0, buf.position()));
  68. });
  69. ((SslSocketChannel)key.attachment()).closeConnection();
  70. return true;
  71. }
  72. }
  73. }
  74. }
  75. }

总结

以上：

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

3.透明上游代理

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

  • 透明代理需要响应 CONNECT请求，透明上游代理不需要，直接转发即可;

  • .

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

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

4.非透明上游代理

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

基于Netty的代理网关设计与实现

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

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

三、设计与实现

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

  • 介绍代理网关服务端相关实现;
  • 介绍代理网关客户端相关实现;

1.代理网关服务端

主。要包括

  • 初始化代理网关服务端
  • 初始化服务端处理器
  • 服务端协议升级与处理

初始化代理网关服务

  1. public void start() {
  2. HookedExecutors.newSingleThreadExecutor().submit(() ->{
  3. log.info("开始启动代理服务器，监听端口：{}", auditProxyConfig.getProxyServerPort());
  4. EventLoopGroup bossGroup = new NioEventLoopGroup(auditProxyConfig.getBossThreadCount());
  5. EventLoopGroup workerGroup = new NioEventLoopGroup(auditProxyConfig.getWorkThreadCount());
  6. try {
  7. ServerBootstrap b = new ServerBootstrap();
  8. b.group(bossGroup, workerGroup)
  9. .channel(NioServerSocketChannel.class)
  10. .handler(new LoggingHandler(LogLevel.DEBUG))
  11. .childHandler(new ServerChannelInitializer(auditProxyConfig))
  12. .bind(auditProxyConfig.getProxyServerPort()).sync().channel().closeFuture().sync();
  13. } catch (InterruptedException e) {
  14. log.error("代理服务器被中断.", e);
  15. Thread.currentThread().interrupt();
  16. } finally {
  17. bossGroup.shutdownGracefully();
  18. workerGroup.shutdownGracefully();
  19. }
  20. });
  21. }

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

bossGroup线程组，负责接收请求

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

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

  1. @Override
  2. protected void initChannel(SocketChannel ch) throws Exception {
  3. ch.pipeline()
  4. .addLast(new HttpRequestDecoder())
  5. .addLast(new HttpObjectAggregator(auditProxyConfig.getMaxRequestSize()))
  6. .addLast(new ServerChannelHandler(auditProxyConfig));
  7. }

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

代理网关协议：

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

详细实现为：

  1. @Override
  2. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  3. FullHttpRequest request = (FullHttpRequest)msg;
  5. try {
  6. if (isConnectRequest(request)) {
  7. // CONNECT 请求，存储待处理
  8. saveConnectRequest(ctx, request);
  10. // 禁止读取
  11. ctx.channel().config().setAutoRead(false);
  13. // 发送回应
  14. connectionEstablished(ctx, ctx.newPromise().addListener(future -> {
  15. if (future.isSuccess()) {
  16. // 升级
  17. if (isSslRequest(request) && !isUpgraded(ctx)) {
  18. upgrade(ctx);
  19. }
  21. // 开放消息读取
  22. ctx.channel().config().setAutoRead(true);
  23. ctx.read();
  24. }
  25. }));
  27. } else {
  28. // 其他请求，判定是否已升级
  29. if (!isUpgraded(ctx)) {
  31. // 升级引擎
  32. upgrade(ctx);
  33. }
  35. // 连接远程
  36. connectRemote(ctx, request);
  37. }
  38. } finally {
  39. ctx.fireChannelRead(msg);
  40. }
  41. }

2.代理网关客户端

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

代理网关客户端初始化：

  1. /**
  2. * 初始化远程连接
  3. * @param ctx
  4. * @param httpRequest
  5. */
  6. protected void connectRemote(ChannelHandlerContext ctx, FullHttpRequest httpRequest) {
  7. Bootstrap b = new Bootstrap();
  8. b.group(ctx.channel().eventLoop()) // use the same EventLoop
  9. .channel(ctx.channel().getClass())
  10. .handler(new ClientChannelInitializer(auditProxyConfig, ctx, safeCopy(httpRequest)));
  12. // 动态连接代理
  13. FullHttpRequest originRequest = ctx.channel().attr(CONNECT_REQUEST).get();
  14. if (originRequest == null) {
  15. originRequest = httpRequest;
  16. }
  17. ChannelFuture cf = b.connect(new InetSocketAddress(calculateHost(originRequest), calculatePort(originRequest)));
  18. Channel cch = cf.channel();
  19. ctx.channel().attr(CLIENT_CHANNEL).set(cch);
  20. }

以上：

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

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

  1. @Override
  2. protected void initChannel(SocketChannel ch) throws Exception {
  3. SocketAddress socketAddress = calculateProxy();
  4. if (!Objects.isNull(socketAddress)) {
  5. ch.pipeline().addLast(new HttpProxyHandler(calculateProxy(), auditProxyConfig.getUserName(), auditProxyConfig
  6. .getPassword()));
  7. }
  8. if (isSslRequest()) {
  9. String host = host();
  10. int port = port();
  11. if (StringUtils.isNoneBlank(host) && port > 0) {
  12. ch.pipeline().addLast(new SslHandler(sslEngine(host, port)));
  13. }
  14. }
  15. ch.pipeline().addLast(new ClientChannelHandler(clientContext, httpRequest));
  16. }

以上：

如果下游是代理，那么会采用HttpProxyHandler，经由下游代理与远程服务端通信;

如果当前需要升级为SSL协议，会对SocketChannel做透明的封装，实现SSL通信。

最后，ClientChannelHandler只是简单消息的转发;唯一的不同是，由于代理网关拦截了第一个请求，此时需要将拦截的请求，转发到服务端。

四、其他问题

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

1.内存问题

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

2.性能问题

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

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

CLIENT和SERVER建立TCP长连接后(比如，TCP心跳检测)，通常要么是CLIENT关闭TCP连接，或者是SERVER关闭;

如果双方长时间占用TCP连接资源而不关闭，就会导致SOCKET资源泄漏;现象是：CPU资源爆满，处理空闲连接;新连接无法建立;

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

五、总结

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

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

  • ServerHandler可以拦截转换请求
  • ClientHanlder可控制请求的出口

注：本文使用Netty的零拷贝;存储请求以解析处理;但并未实现对RESPONSE的处理;也就是RESPONSE是直接通过网关，此方面避免了常见的代理实现，内存泄漏OOM相关问题;

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

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

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

原文链接：https://zhuanlan.51cto.com/art/202111/692258.htm

标签：

相关文章