open-gateway/open-gateway-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

T1.3: StreamingForwarder 流式转发（AsyncContext+HttpClient 零缓冲透传；剥 hop-by-hop 与伪造 X-Caller-App-Key、签名头四件原样；超限裸413/连接失败裸502/超时裸504）

Browse Source
This commit is contained in:
chenxing
2026-09-21 09:22:07 +08:00
parent 58a1bb5358
commit cdc4396ec2
2 changed files with 485 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/yunhai/platform/gateway/forward/StreamingForwarder.java
+257
View File
@@ -0,0 +1,257 @@
package com.yunhai.platform.gateway.forward;
import com.yunhai.platform.gateway.envelope.EnvelopeWriter;
import org.apache.http.HttpEntity;
import org.apache.http.Header;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.client.methods.RequestBuilder;
import org.apache.http.entity.InputStreamEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.SocketTimeoutException;
import java.net.URI;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* 流式转发器（T1.3）：网关 → 提供方的 HTTP 转发，请求/响应 body 均<b>流式透传</b>
* （不聚合、不落盘、不改写响应体），经异步 Servlet（{@link AsyncContext}）+ Apache HttpClient 执行，
* 请求线程不阻塞等待后端。
* <ul>
* <li>转发 URL 取 T0.12 {@code TargetRouter.Routing.forwardUrl}（换主机不换路径）；</li>
* <li>读超时默认 30s（{@link #DEFAULT_READ_TIMEOUT_MS}；route.timeout_ms 覆盖，本期注册表不配 = 统一默认）；</li>
* <li>请求头：保留 {@code X-App-Key / X-Timestamp / X-Signature / X-Nonce} 原样透传
* （UserBase 自验签依赖，R3）；剥除客户端伪造的 {@code X-Caller-App-Key}（身份头唯一合法来源是
* 网关，T1.4 注入）；剥除 hop-by-hop 头（Connection/Keep-Alive/Transfer-Encoding/Content-Length
* 由客户端库重算）；</li>
* <li>body 上限 route.request_body_limit_mb（默认 10MB），超限<b>裸 413</b>；</li>
* <li>响应 status/headers/body 原样透传（不经 EnvelopeWriter）；连接失败 → <b>裸 502</b>，
* 超时 → <b>裸 504</b>。</li>
* </ul>
*/
@Component
public class StreamingForwarder {
/** 读超时默认 30s（route.timeout_ms 未配时的统一默认） */
public static final int DEFAULT_READ_TIMEOUT_MS = 30_000;
/** body 上限默认 10MB（route.request_body_limit_mb 未配时的统一默认） */
public static final long DEFAULT_BODY_LIMIT_BYTES = 10L * 1024 * 1024;
/** 连接建立超时（容器网内建连应极快；非卡参数，防御挂死的 SYN） */
private static final int CONNECT_TIMEOUT_MS = 5_000;
/** 身份头：转发时剥除客户端伪造值，注入权属网关（T1.4） */
public static final String HEADER_CALLER_APP_KEY = "X-Caller-App-Key";
private static final Logger log = LoggerFactory.getLogger(StreamingForwarder.class);
/** RFC 2616 hop-by-hop + 由客户端库重算的头 */
private static final Set<String> REQ_SKIP;
private static final Set<String> RES_SKIP;
static {
Set<String> s = new HashSet<>();
Collections.addAll(s, "connection", "keep-alive", "proxy-connection", "proxy-authenticate",
"proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade",
"content-length", "host", HEADER_CALLER_APP_KEY.toLowerCase());
REQ_SKIP = s;
Set<String> r = new HashSet<>();
Collections.addAll(r, "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
"te", "trailer", "transfer-encoding", "upgrade");
RES_SKIP = r;
}
private final CloseableHttpClient client;
private final Executor executor;
private final ExecutorService ownedPool;
/** Spring 装配：默认连接池 client + 异步执行线程池（IO 密集，缓存池） */
public StreamingForwarder() {
this.client = HttpClients.custom()
.disableRedirectHandling()
.setMaxConnTotal(400)
.setMaxConnPerRoute(200)
.build();
this.ownedPool = Executors.newCachedThreadPool(r -> {
Thread t = new Thread(r, "gw-forward");
t.setDaemon(true);
return t;
});
this.executor = ownedPool;
}
/** 测试/显式装配入口（同 TargetRouter.setProperties 惯例）：注入 client 与执行器 */
StreamingForwarder(CloseableHttpClient client, Executor executor) {
this.client = client;
this.executor = executor;
this.ownedPool = null;
}
@PreDestroy
public void shutdown() throws IOException {
client.close();
if (ownedPool != null) {
ownedPool.shutdown();
try {
ownedPool.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
/**
* 转发并原样回写。声明长度超限在<b>建连之前</b>拦（裸 413）；否则转异步：
* 工作线程流式收发，完成后 {@link AsyncContext#complete()}。
*/
public void forward(HttpServletRequest req, HttpServletResponse res, String forwardUrl,
int readTimeoutMs, long bodyLimitBytes) throws IOException {
if (req.getContentLengthLong() > bodyLimitBytes) {
EnvelopeWriter.writeBare(res, 413);
return;
}
final AsyncContext ctx = req.startAsync(req, res);
executor.execute(() -> {
try {
doForward(req, res, forwardUrl, readTimeoutMs, bodyLimitBytes);
} catch (BodyTooLargeException e) {
writeBareGuard(res, 413); // 未声明长度的 chunked 请求在流式读取中超限
} catch (SocketTimeoutException e) {
writeBareGuard(res, 504);
} catch (IOException e) {
if (hasBodyTooLargeCause(e)) {
writeBareGuard(res, 413);
} else {
log.warn("[GatewayForward] backend io failure url={}", forwardUrl, e);
writeBareGuard(res, 502); // 连接失败/协议断流
}
} catch (RuntimeException e) {
log.error("[GatewayForward] unexpected forward error url={}", forwardUrl, e);
writeBareGuard(res, 502);
} finally {
ctx.complete();
}
});
}
private void doForward(HttpServletRequest req, HttpServletResponse res, String forwardUrl,
int readTimeoutMs, long bodyLimitBytes) throws IOException {
RequestBuilder rb = RequestBuilder.create(req.getMethod())
.setUri(URI.create(forwardUrl))
.setConfig(RequestConfig.custom()
.setConnectTimeout(CONNECT_TIMEOUT_MS)
.setSocketTimeout(readTimeoutMs)
.setConnectionRequestTimeout(CONNECT_TIMEOUT_MS)
.setRedirectsEnabled(false)
.build());
Enumeration<String> names = req.getHeaderNames();
while (names != null && names.hasMoreElements()) {
String name = names.nextElement();
if (REQ_SKIP.contains(name.toLowerCase())) {
continue;
}
Enumeration<String> values = req.getHeaders(name);
while (values.hasMoreElements()) {
rb.addHeader(name, values.nextElement());
}
}
String method = req.getMethod().toUpperCase();
if (!"GET".equals(method) && !"HEAD".equals(method)
&& !"OPTIONS".equals(method) && !"TRACE".equals(method)) {
// 长度未知 → chunked 交由 HttpClient 重算；流式读取、超上限即断
rb.setEntity(new InputStreamEntity(new LimitedInputStream(req.getInputStream(), bodyLimitBytes), -1));
}
HttpUriRequest request = rb.build();
try (CloseableHttpResponse up = client.execute(request)) {
res.setStatus(up.getStatusLine().getStatusCode());
for (Header h : up.getAllHeaders()) {
if (!RES_SKIP.contains(h.getName().toLowerCase())) {
res.addHeader(h.getName(), h.getValue());
}
}
HttpEntity entity = up.getEntity();
if (entity != null) {
entity.writeTo(res.getOutputStream());
}
}
}
private static void writeBareGuard(HttpServletResponse res, int status) {
try {
if (!res.isCommitted()) {
EnvelopeWriter.writeBare(res, status);
}
} catch (IOException ignored) {
// 裸码写出失败无从补救（响应面已坏）
}
}
private static boolean hasBodyTooLargeCause(Throwable e) {
for (Throwable t = e; t != null; t = t.getCause()) {
if (t instanceof BodyTooLargeException) {
return true;
}
}
return false;
}
/** 流式 body 上限哨兵异常 */
static class BodyTooLargeException extends IOException {
BodyTooLargeException() {
super("request body exceeds limit");
}
}
/** 计数透传流：超过 limit 立即抛哨兵异常（绝不缓冲整体） */
private static final class LimitedInputStream extends FilterInputStream {
private final long limit;
private long count;
LimitedInputStream(InputStream in, long limit) {
super(in);
this.limit = limit;
}
private void counted(long n) throws BodyTooLargeException {
if (n > 0) {
count += n;
if (count > limit) {
throw new BodyTooLargeException();
}
}
}
@Override
public int read() throws IOException {
int b = super.read();
if (b >= 0) {
counted(1);
}
return b;
}
@Override
public int read(byte[] b, int off, int len) throws IOException {
int n = super.read(b, off, len);
counted(n);
return n;
}
}
}
src/test/java/com/yunhai/platform/gateway/forward/StreamingForwarderTest.java
+228
View File
@@ -0,0 +1,228 @@
package com.yunhai.platform.gateway.forward;
import com.sun.net.httpserver.HttpServer;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.mock.web.MockHttpServletResponse;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import static org.assertj.core.api.Assertions.assertThat;
/**
* StreamingForwarder 五用例（T1.3 TDD，假后端 = JDK 自带 com.sun.net.httpserver.HttpServer，
* 零新增测试依赖）：GET 转发 forwardUrl 逐字符 + 响应透传；POST body 流式透传；
* 客户端伪造 X-Caller-App-Key 不出现在转发请求；签名头四件原样到达；
* 超时裸 504 / 连接拒绝裸 502（另附 body 超限裸 413——实现要点契约）。
*/
class StreamingForwarderTest {
private static HttpServer server;
private static int port;
/** 假后端最近一次收到的请求快照 */
private static volatile Map<String, String> lastHeaders;
private static volatile String lastMethod;
private static volatile String lastUri;
private static volatile byte[] lastBody;
/** 假后端行为控制 */
private static volatile int respStatus = 201;
private static volatile String respHeader = "X-Upstream:v1"; // name:value
private static volatile byte[] respBody = "hello-gateway".getBytes(StandardCharsets.UTF_8);
private static volatile long respSleepMs = 0;
private static volatile int hitCount = 0;
private static volatile boolean echoBody = false;
@BeforeAll
static void startFakeBackend() throws IOException {
server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.setExecutor(Executors.newFixedThreadPool(4));
server.createContext("/", exchange -> {
hitCount++;
lastMethod = exchange.getRequestMethod();
lastUri = exchange.getRequestURI().toString();
lastHeaders = new ConcurrentHashMap<>();
for (Map.Entry<String, java.util.List<String>> e : exchange.getRequestHeaders().entrySet()) {
lastHeaders.put(e.getKey().toLowerCase(), String.join(",", e.getValue()));
}
ByteArrayOutputStream bos = new ByteArrayOutputStream();
byte[] buf = new byte[8192];
int n;
while ((n = exchange.getRequestBody().read(buf)) > 0) {
bos.write(buf, 0, n);
}
lastBody = bos.toByteArray();
if (respSleepMs > 0) {
try {
Thread.sleep(respSleepMs);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
}
}
byte[] body = echoBody ? lastBody : respBody;
if (respHeader != null) {
String[] kv = respHeader.split(":", 2);
exchange.getResponseHeaders().add(kv[0], kv[1]);
}
exchange.sendResponseHeaders(respStatus, body.length == 0 ? -1 : body.length);
try (OutputStream out = exchange.getResponseBody()) {
out.write(body);
}
exchange.close();
});
server.start();
port = server.getAddress().getPort();
}
@AfterAll
static void stopFakeBackend() {
server.stop(0);
}
@BeforeEach
@AfterEach
void resetFakeBackend() {
respStatus = 201;
respHeader = "X-Upstream:v1";
respBody = "hello-gateway".getBytes(StandardCharsets.UTF_8);
respSleepMs = 0;
hitCount = 0;
echoBody = false;
lastHeaders = null;
lastMethod = null;
lastUri = null;
lastBody = null;
}
/** 同线程执行器：转发逻辑同步完成，测试无需起线程等待 */
private StreamingForwarder forwarder() {
CloseableHttpClient client = HttpClients.custom().disableRedirectHandling().build();
return new StreamingForwarder(client, Runnable::run);
}
private MockHttpServletRequest request(String method, String uri) {
MockHttpServletRequest req = new MockHttpServletRequest(method, uri);
req.setAsyncSupported(true); // 异步 Servlet 契约（T1.6 链上 filter 需 async-supported）
return req;
}
@Test
void GET转发_forwardUrl逐字符且响应原样透传() throws Exception {
MockHttpServletRequest req = request("GET", "/userbase/oapi/x");
req.setQueryString("a=1&b=2");
MockHttpServletResponse res = new MockHttpServletResponse();
String forwardUrl = "http://127.0.0.1:" + port + "/userbase/oapi/x?a=1&b=2";
forwarder().forward(req, res, forwardUrl, 5000, 1024 * 1024);
assertThat(lastUri).as("换主机不换路径：URI+query 逐字符送达").isEqualTo("/userbase/oapi/x?a=1&b=2");
assertThat(lastMethod).isEqualTo("GET");
assertThat(res.getStatus()).as("响应 status 透传").isEqualTo(201);
assertThat(res.getHeader("X-Upstream")).as("响应 header 透传").isEqualTo("v1");
assertThat(res.getContentAsString()).as("响应 body 原样透传").isEqualTo("hello-gateway");
}
@Test
void POST_body流式透传() throws Exception {
echoBody = true;
respStatus = 200;
byte[] payload = new byte[100 * 1024];
for (int i = 0; i < payload.length; i++) {
payload[i] = (byte) (i % 251);
}
MockHttpServletRequest req = request("POST", "/usermsg/oapi/msg/sendUserMsg");
req.setContent(payload);
MockHttpServletResponse res = new MockHttpServletResponse();
forwarder().forward(req, res, "http://127.0.0.1:" + port + "/usermsg/oapi/msg/sendUserMsg",
5000, 4 * 1024 * 1024);
assertThat(lastBody).as("请求 body 完整到达假后端").isEqualTo(payload);
assertThat(res.getContentAsByteArray()).as("响应 body 回显一致（不改写）").isEqualTo(payload);
}
@Test
void 客户端伪造XCallerAppKey不出现在转发请求() throws Exception {
MockHttpServletRequest req = request("GET", "/userbase/oapi/x");
req.addHeader("X-Caller-App-Key", "hack");
req.addHeader("Connection", "close"); // hop-by-hop 也不得透传
MockHttpServletResponse res = new MockHttpServletResponse();
forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/x", 5000, 1024 * 1024);
assertThat(res.getStatus()).isEqualTo(201);
assertThat(lastHeaders).as("身份头唯一合法来源是网关（T1.4）").doesNotContainKey("x-caller-app-key");
}
@Test
void 签名头四件原样到达假后端() throws Exception {
MockHttpServletRequest req = request("GET", "/userbase/oapi/x");
req.addHeader("X-App-Key", "100617");
req.addHeader("X-Timestamp", "1758441600000");
req.addHeader("X-Signature", "Regw08O602Tp61nhH6oG/5VIvyfmU9ZbWs42P7OHzj4=");
req.addHeader("X-Nonce", "c7ce1f3e-9a1d-4b3e-8f2a-0b1c2d3e4f5a");
MockHttpServletResponse res = new MockHttpServletResponse();
forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/x", 5000, 1024 * 1024);
assertThat(lastHeaders.get("x-app-key")).isEqualTo("100617");
assertThat(lastHeaders.get("x-timestamp")).isEqualTo("1758441600000");
assertThat(lastHeaders.get("x-signature")).isEqualTo("Regw08O602Tp61nhH6oG/5VIvyfmU9ZbWs42P7OHzj4=");
assertThat(lastHeaders.get("x-nonce")).as("R3：UserBase 自验签依赖").isEqualTo("c7ce1f3e-9a1d-4b3e-8f2a-0b1c2d3e4f5a");
}
@Test
void 假后端超时_裸504() throws Exception {
respSleepMs = 600;
MockHttpServletRequest req = request("GET", "/userbase/oapi/slow");
MockHttpServletResponse res = new MockHttpServletResponse();
forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/slow", 100, 1024 * 1024);
assertThat(res.getStatus()).isEqualTo(504);
assertThat(res.getContentAsByteArray()).as("裸码无信封").isEmpty();
}
@Test
void 连接拒绝_裸502() throws Exception {
int deadPort;
try (ServerSocket ss = new ServerSocket(0)) {
deadPort = ss.getLocalPort();
}
MockHttpServletRequest req = request("GET", "/userbase/oapi/x");
MockHttpServletResponse res = new MockHttpServletResponse();
forwarder().forward(req, res, "http://127.0.0.1:" + deadPort + "/userbase/oapi/x", 5000, 1024 * 1024);
assertThat(res.getStatus()).isEqualTo(502);
assertThat(res.getContentAsByteArray()).isEmpty();
}
@Test
void body超限_裸413且不连后端() throws Exception {
MockHttpServletRequest req = request("POST", "/userbase/oapi/x");
req.setContent("0123456789".getBytes(StandardCharsets.UTF_8)); // 10B > 4B
MockHttpServletResponse res = new MockHttpServletResponse();
forwarder().forward(req, res, "http://127.0.0.1:" + port + "/userbase/oapi/x", 5000, 4);
assertThat(res.getStatus()).isEqualTo(413);
assertThat(res.getContentAsByteArray()).as("裸码").isEmpty();
assertThat(hitCount).as("上限拦截在连接之前").isZero();
}
}