一、問(wèn)題背景
平臺(tái)端購(gòu)置一批裸代理，來(lái)做廣告異地展現(xiàn)審核。從外部購(gòu)置的代理，使用方式為：
- 通過(guò)給定的HTTP 的 API 提取代理 IP:PORT，返回的結(jié)果會(huì)給出代理的有效時(shí)長(zhǎng) 3~5 分鐘，以及代理所屬地域;
- 從提取的代理中，選取指定地域，添加認(rèn)證信息，請(qǐng)求獲取結(jié)果;
本文設(shè)計(jì)實(shí)現(xiàn)一個(gè)通過(guò)的代理網(wǎng)關(guān)：
- 管理維護(hù)代理資源，并做代理的認(rèn)證鑒權(quán);
- 對(duì)外暴露統(tǒng)一的代理入口，而非動(dòng)態(tài)變化的代理IP:PORT;
- 流量過(guò)濾及限流，比如：靜態(tài)資源不走代理;
本文重點(diǎn)在代理網(wǎng)關(guān)本身的設(shè)計(jì)與實(shí)現(xiàn)，而非代理資源的管理與維護(hù)。
注：本文包含大量可執(zhí)行的JAVA代碼以解釋代理相關(guān)的原理
二、技術(shù)路線
本文的技術(shù)路線。在現(xiàn)代理網(wǎng)關(guān)之前，首先介紹下代理相關(guān)的原理及如何實(shí)現(xiàn)
- 透明代理;
- 非透明代理;
- 透明的上游代理;
- 非透明的上游代理;
最后，本文要構(gòu)建代理網(wǎng)關(guān)，本質(zhì)上就是一個(gè)非透明的上游代理，并給出詳細(xì)的設(shè)計(jì)與實(shí)現(xiàn)。
1.透明代理
透明代理是代理網(wǎng)關(guān)的基礎(chǔ)，本文采用JAVA原生的NIO進(jìn)行詳細(xì)介紹。在實(shí)現(xiàn)代理網(wǎng)關(guān)時(shí)，實(shí)際使用的為NETTY框架。原生NIO的實(shí)現(xiàn)對(duì)理解NETTY的實(shí)現(xiàn)有幫助。
透明代理設(shè)計(jì)三個(gè)交互方，客戶端、代理服務(wù)、服務(wù)端，其原理是：
??
??
- 代理服務(wù)在收到連接請(qǐng)求時(shí)，判定：如果是CONNECT請(qǐng)求，需要回應(yīng)代理連接成功消息到客戶端;
- CONNECT請(qǐng)求回應(yīng)結(jié)束后，代理服務(wù)需要連接到CONNECT指定的遠(yuǎn)程服務(wù)器，然后直接轉(zhuǎn)發(fā)客戶端和遠(yuǎn)程服務(wù)通信;
- 代理服務(wù)在收到非CONNECT請(qǐng)求時(shí)，需要解析出請(qǐng)求的遠(yuǎn)程服務(wù)器，然后直接轉(zhuǎn)發(fā)客戶端和遠(yuǎn)程服務(wù)通信;
需要注意的點(diǎn)是：
- 通常HTTPS請(qǐng)求，在通過(guò)代理前，會(huì)發(fā)送CONNECT請(qǐng)求;連接成功后，會(huì)在信道上進(jìn)行加密通信的握手協(xié)議;因此連接遠(yuǎn)程的時(shí)機(jī)是在CONNECT請(qǐng)求收到時(shí)，因?yàn)榇撕笫羌用軘?shù)據(jù);
- 透明代理在收到CONNECT請(qǐng)求時(shí)，不需要傳遞到遠(yuǎn)程服務(wù)(遠(yuǎn)程服務(wù)不識(shí)別此請(qǐng)求);
- 透明代理在收到非CONNECT請(qǐng)求時(shí)，要無(wú)條件轉(zhuǎn)發(fā);
完整的透明代理的實(shí)現(xiàn)不到約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單線程調(diào)試
while (localServer.isOpen()) {
// 此處阻塞等待連接
SocketChannel remoteClient = localServer.accept();
// 工作線程
GlobalThreadPool.WORK_EXECUTOR.submit(new Runnable() {
@SneakyThrows
@Override
public void run() {
// 代理到遠(yuǎn)程
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();
/**
* 連接到遠(yuǎn)程
* @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請(qǐng)求回應(yīng)，其他請(qǐng)求WRITE THROUGH
afterConnected();
}
protected void beforeConnected() throws IOException {
// 讀取HEADER
readAllHeader();
// 解析HOST和PORT
parseRemoteHostAndPort();
}
/**
* 創(chuàng)建遠(yuǎn)程連接
* @throws IOException
*/
protected void createRemoteServer() throws IOException {
remoteServer = SocketChannel.open(new InetSocketAddress(host, port));
}
/**
* 連接建立后預(yù)處理
* @throws IOException
*/
protected void afterConnected() throws IOException {
// 當(dāng)CONNECT請(qǐng)求時(shí)，默認(rèn)寫(xiě)入200到CLIENT
if ("CONNECT".equalsIgnoreCase(method)) {
// CONNECT默認(rèn)為443端口，根據(jù)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();
}
});
}
/**
* 讀取請(qǐng)求內(nèi)容
* @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());
// 默認(rèn)為80端口，根據(jù)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() {
// 取消監(jiān)聽(tīng)
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 {
// 重置監(jiān)聽(tīng)
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：
- 首先初始化REACTOR線程，然后開(kāi)啟代理監(jiān)聽(tīng)，當(dāng)收到代理請(qǐng)求時(shí)處理。
- 代理服務(wù)在收到代理請(qǐng)求時(shí)，首先做代理的預(yù)處理，然后又SocketPipe做客戶端和遠(yuǎn)程服務(wù)端雙向轉(zhuǎn)發(fā)。
- 代理預(yù)處理，首先讀取第一個(gè)HTTP請(qǐng)求，解析出METHOD, HOST, PORT。
- 如果是CONNECT請(qǐng)求，發(fā)送回應(yīng)Connection Established，然后連接遠(yuǎn)程服務(wù)端，并返回SocketChannel
- 如果是非CONNECT請(qǐng)求，連接遠(yuǎn)程服務(wù)端，寫(xiě)入原始請(qǐng)求，并返回SocketChannel
- SocketPipe在客戶端和遠(yuǎn)程服務(wù)端，做雙向的轉(zhuǎn)發(fā);其本身是將客戶端和服務(wù)端的SocketChannel注冊(cè)到REACTOR
- REACTOR在監(jiān)測(cè)到READABLE的CHANNEL，派發(fā)給SocketPipe做雙向轉(zhuǎn)發(fā)。
測(cè)試
代理的測(cè)試比較簡(jiǎn)單，指向代碼后，代理服務(wù)監(jiān)聽(tīng)8006端口，此時(shí)：
curl -x 'localhost:8006' http://httpbin.org/get測(cè)試HTTP請(qǐng)求
curl -x 'localhost:8006' https://httpbin.org/get測(cè)試HTTPS請(qǐng)求
注意，此時(shí)代理服務(wù)代理了HTTPS請(qǐng)求，但是并不需要-k選項(xiàng)，指示非安全的代理。因?yàn)榇矸?wù)本身并沒(méi)有作為一個(gè)中間人，并沒(méi)有解析出客戶端和遠(yuǎn)程服務(wù)端通信的內(nèi)容。在非透明代理時(shí)，需要解決這個(gè)問(wèn)題。
2.非透明代理
非透明代理，需要解析出客戶端和遠(yuǎn)程服務(wù)端傳輸?shù)膬?nèi)容，并做相應(yīng)的處理。
當(dāng)傳輸為HTTP協(xié)議時(shí)，SocketPipe傳輸?shù)臄?shù)據(jù)即為明文的數(shù)據(jù)，可以攔截后直接做處理。
當(dāng)傳輸為HTTPS協(xié)議時(shí)，SocketPipe傳輸?shù)挠行?shù)據(jù)為加密數(shù)據(jù)，并不能透明處理。
另外，無(wú)論是傳輸?shù)腍TTP協(xié)議還是HTTPS協(xié)議，SocketPipe讀到的都為非完整的數(shù)據(jù)，需要做聚批的處理。
SocketPipe聚批問(wèn)題，可以采用類(lèi)似BufferedInputStream對(duì)InputStream做Decorate的模式來(lái)實(shí)現(xiàn)，相對(duì)比較簡(jiǎn)單;詳細(xì)可以參考NETTY的HttpObjectAggregator;
HTTPS原始請(qǐng)求和結(jié)果數(shù)據(jù)的加密和解密的處理，需要實(shí)現(xiàn)的NIO的SOCKET CHANNEL;
SslSocketChannel封裝原理
考慮到目前JDK自帶的NIO的SocketChannel并不支持SSL;已有的SSLSocket是阻塞的OIO。如圖：
??
??
可以看出
- 每次入站數(shù)據(jù)和出站數(shù)據(jù)都需要 SSL SESSION 做握手;
- 入站數(shù)據(jù)做解密，出站數(shù)據(jù)做加密;
- 握手，數(shù)據(jù)加密和數(shù)據(jù)解密是統(tǒng)一的一套狀態(tài)機(jī);
??
??
以下，代碼實(shí)現(xiàn) SslSocketChannel
public class SslSocketChannel {
/**
* 握手加解密需要的四個(gè)存儲(chǔ)
*/
protected ByteBuffer myAppData; // 明文
protected ByteBuffer myNetData; // 密文
protected ByteBuffer peerAppData; // 明文
protected ByteBuffer peerNetData; // 密文
/**
* 握手加解密過(guò)程中用到的異步執(zhí)行器
*/
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
* 實(shí)現(xiàn)的 SSL 的握手協(xié)議
* @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，關(guān)閉連接.", 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("無(wú)效的握手狀態(tài): " + 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("加密后消息內(nèi)容為空，報(bào)錯(cuò)");
case CLOSED:
try {
myNetData.flip();
while (myNetData.hasRemaining()) {
socketChannel.write(myNetData);
}
peerNetData.clear();
} catch (Exception e) {
handshakeStatus = engine.getHandshakeStatus();
}
break;
default:
throw new IllegalStateException("無(wú)效的握手狀態(tài): " + 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("無(wú)效的握手狀態(tài): " + handshakeStatus);
}
}
return true;
}
/**
* 參考 https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html
* 實(shí)現(xiàn)的 SSL 的傳輸讀取協(xié)議
* @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("收到遠(yuǎn)程的返回結(jié)果消息為：" + 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("收到遠(yuǎn)程連接關(guān)閉消息.");
closeConnection();
return;
default:
throw new IllegalStateException("無(wú)效的握手狀態(tài): " + result.getStatus());
}
}
} else if (bytesRead < 0) {
log.debug("收到END OF STREAM，關(guān)閉連接.");
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
* 實(shí)現(xiàn)的 SSL 的傳輸寫(xiě)入?yún)f(xié)議
* @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("寫(xiě)入遠(yuǎn)程的消息為: {}", message);
break;
case BUFFER_OVERFLOW:
myNetData = enlargePacketBuffer(engine, myNetData);
break;
case BUFFER_UNDERFLOW:
throw new SSLException("加密后消息內(nèi)容為空.");
case CLOSED:
closeConnection();
return;
default:
throw new IllegalStateException("無(wú)效的握手狀態(tài): " + result.getStatus());
}
}
}
/**
* 關(guān)閉連接
* @throws IOException
*/
public void closeConnection() throws IOException {
engine.closeOutbound();
doHandshake();
socketChannel.close();
executor.shutdown();
}
/**
* END OF STREAM(-1)默認(rèn)是關(guān)閉連接
* @throws IOException
*/
protected void handleEndOfStream() throws IOException {
try {
engine.closeInbound();
} catch (Exception e) {
log.error("END OF STREAM 關(guān)閉失敗.", e);
}
closeConnection();
}
}以上：
- 基于 SSL 協(xié)議，實(shí)現(xiàn)統(tǒng)一的握手動(dòng)作;
- 分別實(shí)現(xiàn)讀取的解密，和寫(xiě)入的加密方法;
- 將 SslSocketChannel 實(shí)現(xiàn)為 SocketChannel的Decorator;
SslSocketChannel測(cè)試服務(wù)端
基于以上封裝，簡(jiǎn)單測(cè)試服務(wù)端如下：
@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();
// 注冊(cè)監(jiān)聽(tīng)器
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->{});
// 直接回應(yīng)一個(gè)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("接收新的請(qǐng)求.");
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("握手失敗，關(guān)閉連接.");
}
}
}
以上：
- 基于 SSL 協(xié)議，實(shí)現(xiàn)統(tǒng)一的握手動(dòng)作;
- 分別實(shí)現(xiàn)讀取的解密，和寫(xiě)入的加密方法;
- 將 SslSocketChannel 實(shí)現(xiàn)為 SocketChannel的Decorator;
SslSocketChannel測(cè)試服務(wù)端
基于以上封裝，簡(jiǎn)單測(cè)試服務(wù)端如下:
@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();
// 注冊(cè)監(jiān)聽(tīng)器
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->{});
// 直接回應(yīng)一個(gè)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("接收新的請(qǐng)求.");
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("握手失敗，關(guān)閉連接.");
}
}
}
以上：
由于是NIO，簡(jiǎn)單的測(cè)試需要用到NIO的基礎(chǔ)組件Selector進(jìn)行測(cè)試;
首先初始化ServerSocketChannel，監(jiān)聽(tīng)8006端口;
接收到請(qǐng)求后，將SocketChannel封裝為SslSocketChannel，注冊(cè)到Selector；
接收到數(shù)據(jù)后，通過(guò)SslSocketChannel做read和write;
以上：
- 客戶端的封裝測(cè)試，是為了驗(yàn)證封裝 SSL 協(xié)議雙向都是OK的
- 在后文的非透明上游代理中，會(huì)同時(shí)使用 SslSocketChannel做服務(wù)端和客戶端
- 以上封裝與服務(wù)端封裝類(lèi)似，不同的是初始化 SocketChannel，做connect而非bind
SslSocketChannel測(cè)試客戶端
基于以上服務(wù)端封裝，簡(jiǎn)單測(cè)試客戶端如下：
@Slf4j ：
public class NioSslClient {
public static void main(String[] args) throws Exception {
NioSslClient sslClient = new NioSslClient("httpbin.org", 443);
sslClient.connect();
// 請(qǐng)求 'https://httpbin.org/get'
}
private String remoteAddress;
private int port;
private SSLEngine engine;
private SocketChannel socketChannel;
private SSLContext context;
/**
* 需要遠(yuǎn)程的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()) {
// 通過(guò)REACTOR，不會(huì)出現(xiàn)等待情況
//log.debug("連接中..");
}
SslSocketChannel sslSocketChannel = new SslSocketChannel(context, socketChannel, true);
sslSocketChannel.doHandshake();
// 握手完成后，開(kāi)啟SELECTOR
Selector selector = SelectorProvider.provider().openSelector();
socketChannel.register(selector, SelectionKey.OP_READ, sslSocketChannel);
// 寫(xiě)入請(qǐng)求
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");
// 讀取結(jié)果
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;
}
}
}
}
}
總結(jié)
以上：
- 非透明代理需要拿到完整的請(qǐng)求數(shù)據(jù)，可以通過(guò) Decorator模式，聚批實(shí)現(xiàn);
- 非透明代理需要拿到解密后的HTTPS請(qǐng)求數(shù)據(jù)，可以通過(guò)SslSocketChannel對(duì)原始的SocketChannel做封裝實(shí)現(xiàn);
- 最后，拿到請(qǐng)求后，做相應(yīng)的處理，最終實(shí)現(xiàn)非透明的代理。
3.透明上游代理
透明上游代理相比透明代理要簡(jiǎn)單，區(qū)別是：
- 透明代理需要響應(yīng) CONNECT請(qǐng)求，透明上游代理不需要，直接轉(zhuǎn)發(fā)即可;
.
- 透明的上游代理，只是一個(gè)簡(jiǎn)單的SocketChannel管道;確定下游的代理服務(wù)端，連接轉(zhuǎn)發(fā)請(qǐng)求;
只需要對(duì)透明代理做以上簡(jiǎn)單的修改，即可實(shí)現(xiàn)透明的上游代理。
4.非透明上游代理
非透明的上游代理，相比非透明的代理要復(fù)雜一些。
??
??
以上，分為四個(gè)組件：客戶端，代理服務(wù)(ServerHandler)，代理服務(wù)(ClientHandler)，服務(wù)端
- 如果是HTTP的請(qǐng)求，數(shù)據(jù)直接通過(guò) 客戶端<->ServerHandler<->ClientHandler<->服務(wù)端，代理網(wǎng)關(guān)只需要做簡(jiǎn)單的請(qǐng)求聚批，就可以應(yīng)用相應(yīng)的管理策略;
- 如果是HTTPS請(qǐng)求，代理作為客戶端和服務(wù)端的中間人，只能拿到加密的數(shù)據(jù);因此，代理網(wǎng)關(guān)需要作為HTTPS的服務(wù)方與客戶端通信;然后作為HTTPS的客戶端與服務(wù)端通信;
- 代理作為HTTPS服務(wù)方時(shí)，需要考慮到其本身是個(gè)非透明的代理，需要實(shí)現(xiàn)非透明代理相關(guān)的協(xié)議;
- 代理作為HTTPS客戶端時(shí)，需要考慮到其下游是個(gè)透明的代理，真正的服務(wù)方是客戶端請(qǐng)求的服務(wù)方;
三、設(shè)計(jì)與實(shí)現(xiàn)
本文需要構(gòu)建的是非透明上游代理，以下采用NETTY框架給出詳細(xì)的設(shè)計(jì)實(shí)現(xiàn)。上文將統(tǒng)一代理網(wǎng)關(guān)分為兩大部分，ServerHandler和ClientHandler，以下
- 介紹代理網(wǎng)關(guān)服務(wù)端相關(guān)實(shí)現(xiàn);
- 介紹代理網(wǎng)關(guān)客戶端相關(guān)實(shí)現(xiàn);
1.代理網(wǎng)關(guān)服務(wù)端
主。要包括
- 初始化代理網(wǎng)關(guān)服務(wù)端
- 初始化服務(wù)端處理器
- 服務(wù)端協(xié)議升級(jí)與處理
初始化代理網(wǎng)關(guān)服務(wù)
public void start() {
HookedExecutors.newSingleThreadExecutor().submit(() ->{
log.info("開(kāi)始啟動(dòng)代理服務(wù)器，監(jiān)聽(tīng)端口：{}", 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("代理服務(wù)器被中斷.", e);
Thread.currentThread().interrupt();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
});
}代理網(wǎng)關(guān)初始化相對(duì)簡(jiǎn)單，
bossGroup線程組，負(fù)責(zé)接收請(qǐng)求
workerGroup線程組，負(fù)責(zé)處理接收的請(qǐng)求數(shù)據(jù)，具體處理邏輯封裝在ServerChannelInitializer中。
代理網(wǎng)關(guān)服務(wù)的請(qǐng)求處理器在 ServerChannelInitializer中定義為：
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(new HttpRequestDecoder())
.addLast(new HttpObjectAggregator(auditProxyConfig.getMaxRequestSize()))
.addLast(new ServerChannelHandler(auditProxyConfig));
}
首先解析HTTP請(qǐng)求，然后做聚批的處理，最后ServerChannelHandler實(shí)現(xiàn)代理網(wǎng)關(guān)協(xié)議;
代理網(wǎng)關(guān)協(xié)議：
- 判定是否是CONNECT請(qǐng)求，如果是，會(huì)存儲(chǔ)CONNECT請(qǐng)求;暫停讀取，發(fā)送代理成功的響應(yīng)，并在回應(yīng)成功后，升級(jí)協(xié)議;
- 升級(jí)引擎，本質(zhì)上是采用SslSocketChannel對(duì)原SocketChannel做透明的封裝;
- 最后根據(jù)CONNECT請(qǐng)求連接遠(yuǎn)程服務(wù)端;
詳細(xì)實(shí)現(xiàn)為：
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
FullHttpRequest request = (FullHttpRequest)msg;
try {
if (isConnectRequest(request)) {
// CONNECT 請(qǐng)求，存儲(chǔ)待處理
saveConnectRequest(ctx, request);
// 禁止讀取
ctx.channel().config().setAutoRead(false);
// 發(fā)送回應(yīng)
connectionEstablished(ctx, ctx.newPromise().addListener(future -> {
if (future.isSuccess()) {
// 升級(jí)
if (isSslRequest(request) && !isUpgraded(ctx)) {
upgrade(ctx);
}
// 開(kāi)放消息讀取
ctx.channel().config().setAutoRead(true);
ctx.read();
}
}));
} else {
// 其他請(qǐng)求，判定是否已升級(jí)
if (!isUpgraded(ctx)) {
// 升級(jí)引擎
upgrade(ctx);
}
// 連接遠(yuǎn)程
connectRemote(ctx, request);
}
} finally {
ctx.fireChannelRead(msg);
}
}
2.代理網(wǎng)關(guān)客戶端
代理網(wǎng)關(guān)服務(wù)端需要連接遠(yuǎn)程服務(wù)，進(jìn)入代理網(wǎng)關(guān)客戶端部分。
代理網(wǎng)關(guān)客戶端初始化：
/**
* 初始化遠(yuǎn)程連接
* @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)));
// 動(dòng)態(tài)連接代理
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);
}
以上：
- 復(fù)用代理網(wǎng)關(guān)服務(wù)端的workerGroup線程組;
- 請(qǐng)求和結(jié)果的處理封裝在ClientChannelInitializer;
- 連接的遠(yuǎn)程服務(wù)端的HOST和PORT在服務(wù)端收到的請(qǐng)求中可以解析到。
代理網(wǎng)關(guān)客戶端的處理器的初始化邏輯：
@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));
}
以上：
如果下游是代理，那么會(huì)采用HttpProxyHandler，經(jīng)由下游代理與遠(yuǎn)程服務(wù)端通信;
如果當(dāng)前需要升級(jí)為SSL協(xié)議，會(huì)對(duì)SocketChannel做透明的封裝，實(shí)現(xiàn)SSL通信。
最后，ClientChannelHandler只是簡(jiǎn)單消息的轉(zhuǎn)發(fā);唯一的不同是，由于代理網(wǎng)關(guān)攔截了第一個(gè)請(qǐng)求，此時(shí)需要將攔截的請(qǐng)求，轉(zhuǎn)發(fā)到服務(wù)端。
四、其他問(wèn)題
代理網(wǎng)關(guān)實(shí)現(xiàn)可能面臨的問(wèn)題：
1.內(nèi)存問(wèn)題
代理通常面臨的問(wèn)題是OOM。本文在實(shí)現(xiàn)代理網(wǎng)關(guān)時(shí)保證內(nèi)存中緩存時(shí)當(dāng)前正在處理的HTTP/HTTPS請(qǐng)求體。內(nèi)存使用的上限理論上為實(shí)時(shí)處理的請(qǐng)求數(shù)量*請(qǐng)求體的平均大小，HTTP/HTTPS的請(qǐng)求結(jié)果，直接使用堆外內(nèi)存，零拷貝轉(zhuǎn)發(fā)。
2.性能問(wèn)題
性能問(wèn)題不應(yīng)提早考慮。本文使用NETTY框架實(shí)現(xiàn)的代理網(wǎng)關(guān)，內(nèi)部大量使用堆外內(nèi)存，零拷貝轉(zhuǎn)發(fā)，避免了性能問(wèn)題。
代理網(wǎng)關(guān)一期上線后曾面臨一個(gè)長(zhǎng)連接導(dǎo)致的性能問(wèn)題，
CLIENT和SERVER建立TCP長(zhǎng)連接后(比如，TCP心跳檢測(cè))，通常要么是CLIENT關(guān)閉TCP連接，或者是SERVER關(guān)閉;
如果雙方長(zhǎng)時(shí)間占用TCP連接資源而不關(guān)閉，就會(huì)導(dǎo)致SOCKET資源泄漏;現(xiàn)象是：CPU資源爆滿，處理空閑連接;新連接無(wú)法建立;
使用IdleStateHandler定時(shí)監(jiān)控空閑的TCP連接，強(qiáng)制關(guān)閉;解決了該問(wèn)題。
五、總結(jié)
本文聚焦于統(tǒng)一代理網(wǎng)關(guān)的核心，詳細(xì)介紹了代理相關(guān)的技術(shù)原理。
代理網(wǎng)關(guān)的管理部分，可以在ServerHandler部分維護(hù)，也可以在ClientHandler部分維護(hù);
- ServerHandler可以攔截轉(zhuǎn)換請(qǐng)求
- ClientHanlder可控制請(qǐng)求的出口
注：本文使用Netty的零拷貝;存儲(chǔ)請(qǐng)求以解析處理;但并未實(shí)現(xiàn)對(duì)RESPONSE的處理;也就是RESPONSE是直接通過(guò)網(wǎng)關(guān)，此方面避免了常見(jiàn)的代理實(shí)現(xiàn)，內(nèi)存泄漏OOM相關(guān)問(wèn)題;
最后，本文實(shí)現(xiàn)代理網(wǎng)關(guān)后，針對(duì)代理的資源和流經(jīng)代理網(wǎng)關(guān)的請(qǐng)求做了相應(yīng)的控制，主要包括：
- 當(dāng)遇到靜態(tài)資源的請(qǐng)求時(shí)，代理網(wǎng)關(guān)會(huì)直接請(qǐng)求遠(yuǎn)程服務(wù)端，不會(huì)通過(guò)下游代理
- 當(dāng)請(qǐng)求HEADER中包含地域標(biāo)識(shí)時(shí)，代理網(wǎng)關(guān)會(huì)盡力保證請(qǐng)求打入指定的地域代理，經(jīng)由地域代理訪問(wèn)遠(yuǎn)程服務(wù)端
本文參考https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html實(shí)現(xiàn) SslSocketChannel，以透明處理HTTP和HTTPS協(xié)議。
