fizzgate/fizz-gateway-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dedicated line crypto

Browse Source
This commit is contained in:
hongqiaowei
2021-11-20 16:41:10 +08:00
parent afd31a1a4a
commit 26d8808e03
18 changed files with 645 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fizz-bootstrap/pom.xml
View File

@@ -34,7 +34,7 @@
<commons-codec.version>1.15</commons-codec.version>
<commons-pool2.version>2.11.1</commons-pool2.version>
<gson.version>2.8.8</gson.version>
<netty-tcnative.version>2.0.45.Final</netty-tcnative.version>
<netty-tcnative.version>2.0.46.Final</netty-tcnative.version>
<spring-cloud.version>2.2.9.RELEASE</spring-cloud.version>
<resilience4j.version>1.7.1</resilience4j.version>
</properties>

27
fizz-bootstrap/src/main/resources/application.yml
View File

@@ -122,19 +122,20 @@ fizz:
enable: false
port: 8601
request:
timeliness: 300 # default 300 sec
timeout: 0 # default no timeout
retry-count: 0 # default no retry
retry-interval: 0 # default no retry interval
service-registration:
# type: eureka # service registration type, can be eureka or nacos
# application: ax # register the name of this application to server
# port: 8601 # the port to be registered
# service-url: http://6.6.6.6:6600/eureka/ # server address
type: nacos
application: ax
port: 8601
service-url: 6.6.6.6:8848
timeliness: 300 # default 300 sec
timeout: 0 # default no timeout
retry-count: 0 # default no retry
retry-interval: 0 # default no retry interval
secret-key: # if key configured, client will encrypt request body and decrypt response body
# service-registration:
# type: eureka # service registration type, can be eureka or nacos
# application: ax # register the name of this application to server
# port: 8601 # the port to be registered
# service-url: http://6.6.6.6:6600/eureka/ # server address
# type: nacos
# application: ax
# port: 8601
# service-url: 6.6.6.6:8848
fizz-trace-id:
header: X-Trace-Id

5
fizz-common/pom.xml
View File

@@ -105,6 +105,11 @@
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-crypto</artifactId>
</dependency>
</dependencies>
<repositories>

30
fizz-common/src/main/java/we/spring/http/server/reactive/ext/FizzServerHttpRequestDecorator.java
View File

@@ -36,6 +36,7 @@ import we.util.NettyDataBufferUtils;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -60,7 +61,9 @@ public class FizzServerHttpRequestDecorator extends ServerHttpRequestDecorator {
@Nullable
private MultiValueMap<String, HttpCookie> cookies;
private Flux<DataBuffer> body = Flux.empty();
private Flux<DataBuffer> body = Flux.empty();
private byte[] bodyBytes;
public FizzServerHttpRequestDecorator(ServerHttpRequest delegate) {
super(delegate);
@@ -128,16 +131,20 @@ public class FizzServerHttpRequestDecorator extends ServerHttpRequestDecorator {
}
public void setEmptyBody() {
this.body = Flux.empty();
body = Flux.empty();
bodyBytes = null;
}
public void setBody(DataBuffer body) {
if (body instanceof PooledDataBuffer) {
byte[] bytes = new byte[body.readableByteCount()];
body.read(bytes);
byte[] bytes = NettyDataBufferUtils.copyBytes(body);
setBody(bytes);
} else {
this.body = Flux.just(body);
ByteBuffer byteBuffer = body.asByteBuffer();
if (byteBuffer.hasArray()) {
bodyBytes = byteBuffer.array();
}
}
}
@@ -147,7 +154,8 @@ public class FizzServerHttpRequestDecorator extends ServerHttpRequestDecorator {
}
public void setBody(byte[] body) {
NettyDataBuffer from = NettyDataBufferUtils.from(body);
bodyBytes = body;
NettyDataBuffer from = NettyDataBufferUtils.from(bodyBytes);
this.body = Flux.just(from);
}
@@ -156,13 +164,7 @@ public class FizzServerHttpRequestDecorator extends ServerHttpRequestDecorator {
return body;
}
// public DataBuffer getRawBody() {
// final DataBuffer[] raw = {null};
// body.subscribe(
// dataBuffer -> {
// raw[0] = dataBuffer;
// }
// );
// return raw[0];
// }
public byte[] getBodyBytes() {
return bodyBytes;
}
}

2
fizz-common/src/main/java/we/spring/http/server/reactive/ext/FizzServerHttpResponseDecorator.java
View File

@@ -46,7 +46,7 @@ public abstract class FizzServerHttpResponseDecorator extends ServerHttpResponse
if (body != NettyDataBufferUtils.EMPTY_DATA_BUFFER) {
if (body instanceof PooledDataBuffer) {
try {
b = NettyDataBufferUtils.from(body.asByteBuffer());
b = NettyDataBufferUtils.copy2heap(body);
} finally {
NettyDataBufferUtils.release(body);
}

16
fizz-common/src/main/java/we/util/NettyDataBufferUtils.java
View File

@@ -54,15 +54,25 @@ public abstract class NettyDataBufferUtils extends org.springframework.core.io.b
return (NettyDataBuffer) dataBufferFactory.wrap(bytes);
}
public static NettyDataBuffer from(ByteBuffer byteBuffer) {
/*public static NettyDataBuffer from(ByteBuffer byteBuffer) {
return dataBufferFactory.wrap(byteBuffer);
}
public static NettyDataBuffer from(ByteBuf byteBuf) {
return dataBufferFactory.wrap(byteBuf);
}*/
public static byte[] copyBytes(DataBuffer dataBuffer) {
byte[] bytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(bytes);
return bytes;
}
public static boolean release(@Nullable String traceId, @Nullable DataBuffer dataBuffer) {
public static DataBuffer copy2heap(DataBuffer dataBuffer) {
return from(copyBytes(dataBuffer));
}
/*public static boolean release(@Nullable String traceId, @Nullable DataBuffer dataBuffer) {
if (dataBuffer instanceof PooledDataBuffer) {
PooledDataBuffer pooledDataBuffer = (PooledDataBuffer) dataBuffer;
if (pooledDataBuffer.isAllocated()) {
@@ -81,5 +91,5 @@ public abstract class NettyDataBufferUtils extends org.springframework.core.io.b
}
}
return false;
}
}*/
}

107
fizz-common/src/main/java/we/util/SymmetricCryptoUtils.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* Copyright (C) 2020 the original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package we.util;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.KeyUtil;
import cn.hutool.crypto.Padding;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
import javax.crypto.Cipher;
import javax.crypto.SecretKey;
import javax.crypto.spec.IvParameterSpec;
import javax.crypto.spec.PBEParameterSpec;
import java.security.InvalidAlgorithmParameterException;
import java.security.InvalidKeyException;
import java.security.spec.AlgorithmParameterSpec;
/**
* @author hongqiaowei
*/
public abstract class SymmetricCryptoUtils {
private SymmetricCryptoUtils() {
}
public static Cipher createCipher(SymmetricAlgorithm algorithm, String key, int mode) {
byte[] keyBytes = SecureUtil.decode(key);
String algorithmName = algorithm.getValue();
SecretKey secretKey = KeyUtil.generateKey(algorithmName, keyBytes);
Cipher cipher = SecureUtil.createCipher(algorithmName);
byte[] iv = cipher.getIV();
AlgorithmParameterSpec parameterSpec = null;
if (StrUtil.startWithIgnoreCase(algorithmName, "AES")) {
if (iv != null) {
parameterSpec = new IvParameterSpec(iv);
}
} else if (StrUtil.startWithIgnoreCase(algorithmName, "PBE")) {
if (null == iv) {
iv = RandomUtil.randomBytes(8);
}
parameterSpec = new PBEParameterSpec(iv, 100);
}
try {
if (null == parameterSpec) {
cipher.init(mode, secretKey);
} else {
cipher.init(mode, secretKey, parameterSpec);
}
} catch (InvalidAlgorithmParameterException | InvalidKeyException e) {
throw new RuntimeException(e);
}
return cipher;
}
public static boolean isZeroPadding(SymmetricAlgorithm algorithm) {
String algorithmName = algorithm.getValue();
return algorithmName.contains(Padding.ZeroPadding.name());
}
public static byte[] paddingWith0(byte[] data, int blockSize) {
int length = data.length;
int remainLength = length % blockSize;
if (remainLength > 0) {
return ArrayUtil.resize(data, length + blockSize - remainLength);
} else {
return data;
}
}
public static byte[] removePadding(byte[] data, int blockSize) {
if (blockSize > 0) {
int length = data.length;
int remainLength = length % blockSize;
if (remainLength == 0) {
int i = length - 1;
while (i >= 0 && 0 == data[i]) {
i--;
}
return ArrayUtil.resize(data, i + 1);
}
}
return data;
}
}

72
fizz-common/src/main/java/we/util/SymmetricDecryptor.java Normal file
View File

@@ -0,0 +1,72 @@
/*
* Copyright (C) 2020 the original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package we.util;
import cn.hutool.core.util.CharsetUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
import javax.crypto.BadPaddingException;
import javax.crypto.Cipher;
import javax.crypto.IllegalBlockSizeException;
/**
* @author hongqiaowei
*/
public class SymmetricDecryptor {
public final SymmetricAlgorithm algorithm;
public final String secretKey;
private final Cipher cipher;
private final boolean isZeroPadding;
public SymmetricDecryptor(SymmetricAlgorithm algorithm, String key) {
this.algorithm = algorithm;
secretKey = key;
cipher = SymmetricCryptoUtils.createCipher(algorithm, key, Cipher.DECRYPT_MODE);
isZeroPadding = SymmetricCryptoUtils.isZeroPadding(algorithm);
}
/**
* @param data can be hex or base64 string
*/
public String decrypt(String data) {
byte[] decode = SecureUtil.decode(data);
return StrUtil.str(decrypt(decode), CharsetUtil.CHARSET_UTF_8);
}
public byte[] decrypt(byte[] data) {
int blockSize = cipher.getBlockSize();
byte[] decryptData;
try {
decryptData = cipher.doFinal(data);
} catch (IllegalBlockSizeException | BadPaddingException e) {
throw new RuntimeException(e);
}
if (isZeroPadding) {
return SymmetricCryptoUtils.removePadding(decryptData, blockSize);
} else {
return decryptData;
}
}
}

73
fizz-common/src/main/java/we/util/SymmetricEncryptor.java Normal file
View File

@@ -0,0 +1,73 @@
/*
* Copyright (C) 2020 the original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package we.util;
import cn.hutool.core.codec.Base64;
import cn.hutool.core.util.CharsetUtil;
import cn.hutool.core.util.HexUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
import javax.crypto.BadPaddingException;
import javax.crypto.Cipher;
import javax.crypto.IllegalBlockSizeException;
/**
* @author hongqiaowei
*/
public class SymmetricEncryptor {
public final SymmetricAlgorithm algorithm;
public final String secretKey;
private final Cipher cipher;
private final boolean isZeroPadding;
public SymmetricEncryptor(SymmetricAlgorithm algorithm, String key) {
this.algorithm = algorithm;
secretKey = key;
cipher = SymmetricCryptoUtils.createCipher(algorithm, key, Cipher.ENCRYPT_MODE);
isZeroPadding = SymmetricCryptoUtils.isZeroPadding(algorithm);
}
public byte[] encrypt(byte[] data) {
if (isZeroPadding) {
data = SymmetricCryptoUtils.paddingWith0(data, cipher.getBlockSize());
}
try {
return cipher.doFinal(data);
} catch (IllegalBlockSizeException | BadPaddingException e) {
throw new RuntimeException(e);
}
}
public byte[] encrypt(String data) {
return encrypt(StrUtil.bytes(data, CharsetUtil.CHARSET_UTF_8));
}
public String base64encrypt(String data) {
return Base64.encode(encrypt(data));
}
public String hexEncrypt(String data) {
return HexUtil.encodeHexStr(encrypt(data));
}
}

11
fizz-common/src/main/java/we/util/Utils.java
View File

@@ -132,4 +132,15 @@ public abstract class Utils {
return new Throwable(msg, null, false, false) {
};
}
public static String getMessage(Throwable t) {
String message = t.getMessage();
if (message == null) {
StringBuilder b = new StringBuilder();
b.append(t.toString()).append(Consts.S.LF);
b.append("at ").append(t.getStackTrace()[0].toString());
message = b.toString();
}
return message;
}
}

24
fizz-common/src/test/java/we/util/SymmetricCryptoTests.java Normal file
View File

@@ -0,0 +1,24 @@
package we.util;
import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class SymmetricCryptoTests {
@Test
public void encryptTest() {
String secretKey = "1gG1dVcEaQz8JyifTHeEnQ==";
SymmetricEncryptor symmetricEncryptor = new SymmetricEncryptor(SymmetricAlgorithm.AES, secretKey);
String encrypt = symmetricEncryptor.base64encrypt("abc");
SymmetricDecryptor symmetricDecryptor = new SymmetricDecryptor(SymmetricAlgorithm.AES, secretKey);
String decrypt = symmetricDecryptor.decrypt(encrypt);
Assertions.assertEquals("abc", decrypt);
byte[] encryptBytes = symmetricEncryptor.encrypt("123".getBytes());
byte[] decryptBytes = symmetricDecryptor.decrypt(encryptBytes);
Assertions.assertEquals("123", new String(decryptBytes));
}
}

7
fizz-core/src/main/java/we/config/SystemConfig.java
View File

@@ -114,6 +114,9 @@ public class SystemConfig {
@Value("${fizz.dedicated-line.client.request.retry-interval:0}")
private int fizzDedicatedLineClientRequestRetryInterval = 0; // mills
@Value("${fizz.dedicated-line.client.request.secret-key:}")
private String fizzDedicatedLineClientRequestSecretkey;
public int fizzDedicatedLineClientRequestTimeout() {
return fizzDedicatedLineClientRequestTimeout;
}
@@ -130,6 +133,10 @@ public class SystemConfig {
return fizzDedicatedLineClientRequestTimeliness;
}
public String fizzDedicatedLineClientRequestSecretkey() {
return fizzDedicatedLineClientRequestSecretkey;
}
public String fizzTraceIdHeader() {

168
fizz-core/src/main/java/we/dedicated_line/DedicatedLineHttpHandler.java
View File

@@ -17,11 +17,15 @@
package we.dedicated_line;
import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.web.reactive.context.ReactiveWebServerApplicationContext;
import org.springframework.core.NestedExceptionUtils;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.PooledDataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.server.reactive.HttpHandler;
@@ -35,12 +39,11 @@ import org.springframework.web.server.adapter.DefaultServerWebExchange;
import org.springframework.web.server.adapter.ForwardedHeaderTransformer;
import org.springframework.web.server.i18n.LocaleContextResolver;
import org.springframework.web.server.session.WebSessionManager;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import we.config.SystemConfig;
import we.proxy.FizzWebClient;
import we.util.Consts;
import we.util.ThreadContext;
import we.util.WebUtils;
import we.util.*;
import java.net.URI;
import java.util.Arrays;
@@ -61,6 +64,10 @@ class DedicatedLineHttpHandler implements HttpHandler {
private static final Set<String> disconnected_client_exceptions = new HashSet<>(Arrays.asList("AbortedException", "ClientAbortException", "EOFException", "EofException"));
private static final String symmetricEncryptor = "symmEncpT";
private static final String symmetricDecryptor = "symmDecpT";
private WebSessionManager sessionManager;
private ServerCodecConfigurer serverCodecConfigurer;
private LocaleContextResolver localeContextResolver;
@@ -69,7 +76,7 @@ class DedicatedLineHttpHandler implements HttpHandler {
private SystemConfig systemConfig;
private FizzWebClient fizzWebClient;
private DedicatedLineInfoService dedicatedLineInfoService;
private DedicatedLineInfoService dedicatedLineInfoService;
public DedicatedLineHttpHandler(ReactiveWebServerApplicationContext applicationContext, WebSessionManager sessionManager, ServerCodecConfigurer codecConfigurer,
LocaleContextResolver localeContextResolver, ForwardedHeaderTransformer forwardedHeaderTransformer) {
@@ -108,36 +115,25 @@ class DedicatedLineHttpHandler implements HttpHandler {
DedicatedLineInfo dedicatedLineInfo = dedicatedLineInfoService.get(service);
if (dedicatedLineInfo == null) {
log.warn("{}{} service no dedicated line info", logPrefix, service);
return WebUtils.response(response, HttpStatus.FORBIDDEN, null, service + " service no dedicated line info").then(response.setComplete());
return WebUtils.response(response, HttpStatus.FORBIDDEN, null, logPrefix + ' ' + service + " service no dedicated line info");
}
StringBuilder b = ThreadContext.getStringBuilder();
b.append(dedicatedLineInfo.url).append(path);
String qry = requestURI.getQuery();
if (StringUtils.hasText(qry)) {
if (org.apache.commons.lang3.StringUtils.indexOfAny(qry, Consts.S.LEFT_BRACE, Consts.S.FORWARD_SLASH, Consts.S.HASH) > 0) {
qry = requestURI.getRawQuery();
}
b.append(Consts.S.QUESTION).append(qry);
}
String targetUrl = b.toString();
String pairCodeId = dedicatedLineInfo.pairCodeId;
String secretKey = dedicatedLineInfo.secretKey;
String timestamp = String.valueOf(System.currentTimeMillis());
String sign = DedicatedLineUtils.sign(pairCodeId, timestamp, secretKey);
HttpHeaders writableHttpHeaders = HttpHeaders.writableHttpHeaders(request.getHeaders());
writableHttpHeaders.set(SystemConfig.FIZZ_DL_ID, pairCodeId);
writableHttpHeaders.set(SystemConfig.FIZZ_DL_TS, timestamp);
writableHttpHeaders.set(SystemConfig.FIZZ_DL_SIGN, sign);
String targetUrl = constructTargetUrl(requestURI, path, dedicatedLineInfo.url);
HttpHeaders writableHttpHeaders = signAndSetHeaders(request.getHeaders(), dedicatedLineInfo.pairCodeId, dedicatedLineInfo.secretKey);
int requestTimeout = systemConfig.fizzDedicatedLineClientRequestTimeout();
int retryCount = systemConfig.fizzDedicatedLineClientRequestRetryCount();
int retryInterval = systemConfig.fizzDedicatedLineClientRequestRetryInterval();
try {
// TODO: 如果有请求体，则对请求体加密
Mono<ClientResponse> remoteResponseMono = fizzWebClient.send( request.getId(), request.getMethod(), targetUrl, writableHttpHeaders, request.getBody(),
Flux<DataBuffer> dataBufferFlux = request.getBody();
Flux<DataBuffer> bodyFlux = dataBufferFlux;
if (StringUtils.hasLength(systemConfig.fizzDedicatedLineClientRequestSecretkey()) && request.getMethod() != HttpMethod.GET) {
bodyFlux = encrypt(dataBufferFlux);
writableHttpHeaders.remove(HttpHeaders.CONTENT_LENGTH);
}
Mono<ClientResponse> remoteResponseMono = fizzWebClient.send( request.getId(), request.getMethod(), targetUrl, writableHttpHeaders, bodyFlux,
requestTimeout, retryCount, retryInterval );
Mono<Void> respMono = remoteResponseMono.flatMap(
@@ -151,24 +147,124 @@ class DedicatedLineHttpHandler implements HttpHandler {
WebUtils.response2stringBuilder(logPrefix, remoteResp, sb);
log.debug(sb.toString());
}
// TODO: 如果有响应体，则对响应体解密；响应可能是页面、表单、文件上传的结果、图片等
return response.writeWith ( remoteResp.body(BodyExtractors.toDataBuffers()) )
.doOnError ( throwable -> cleanup(remoteResp) )
.doOnCancel( () -> cleanup(remoteResp) );
Flux<DataBuffer> remoteRespBody = remoteResp.body(BodyExtractors.toDataBuffers());
if (StringUtils.hasLength(systemConfig.fizzDedicatedLineClientRequestSecretkey())) {
respHeaders.remove(HttpHeaders.CONTENT_LENGTH);
return response.writeWith (decrypt(remoteRespBody));
} else {
return response.writeWith (remoteRespBody)
.doOnError ( throwable -> cleanup(remoteResp) )
.doOnCancel( () -> cleanup(remoteResp) );
}
}
);
return respMono.doOnSuccess ( v -> logResponse(exchange) )
.onErrorResume( t -> handleUnresolvedError(exchange, t) )
.then ( Mono.defer(response::setComplete) );
.onErrorResume( t -> handleUnresolvedError(exchange, t) );
//.then ( Mono.defer(response::setComplete) );
} catch (Throwable t) {
log.error(logPrefix + "500 Server Error for " + formatRequest(request), t);
response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
return response.setComplete();
return WebUtils.response(response, HttpStatus.INTERNAL_SERVER_ERROR, null, logPrefix + ' ' + Utils.getMessage(t));
}
}
private Flux<DataBuffer> encrypt(Flux<DataBuffer> bodyFlux) {
return NettyDataBufferUtils.join(bodyFlux).defaultIfEmpty(NettyDataBufferUtils.EMPTY_DATA_BUFFER)
.flatMap(
body -> {
if (body == NettyDataBufferUtils.EMPTY_DATA_BUFFER) {
return Mono.empty();
} else {
byte[] bytes = null;
if (body instanceof PooledDataBuffer) {
try {
bytes = NettyDataBufferUtils.copyBytes(body);
} finally {
NettyDataBufferUtils.release(body);
}
} else {
bytes = body.asByteBuffer().array();
}
String cryptoKey = systemConfig.fizzDedicatedLineClientRequestSecretkey();
SymmetricEncryptor encryptor = (SymmetricEncryptor) ThreadContext.get(symmetricEncryptor);
if (encryptor == null) {
encryptor = new SymmetricEncryptor(SymmetricAlgorithm.AES, cryptoKey);
ThreadContext.set(symmetricEncryptor, encryptor);
} else {
if (!encryptor.secretKey.equals(cryptoKey)) {
encryptor = new SymmetricEncryptor(SymmetricAlgorithm.AES, cryptoKey);
ThreadContext.set(symmetricEncryptor, encryptor);
}
}
DataBuffer from = NettyDataBufferUtils.from(encryptor.encrypt(bytes));
return Mono.just(from);
}
}
)
.flux();
}
private Flux<DataBuffer> decrypt(Flux<DataBuffer> bodyFlux) {
return NettyDataBufferUtils.join(bodyFlux).defaultIfEmpty(NettyDataBufferUtils.EMPTY_DATA_BUFFER)
.flatMap(
body -> {
if (body == NettyDataBufferUtils.EMPTY_DATA_BUFFER) {
return Mono.empty();
} else {
byte[] bytes = null;
if (body instanceof PooledDataBuffer) {
try {
bytes = NettyDataBufferUtils.copyBytes(body);
} finally {
NettyDataBufferUtils.release(body);
}
} else {
bytes = body.asByteBuffer().array();
}
String cryptoKey = systemConfig.fizzDedicatedLineClientRequestSecretkey();
SymmetricDecryptor decryptor = (SymmetricDecryptor) ThreadContext.get(symmetricDecryptor);
if (decryptor == null) {
decryptor = new SymmetricDecryptor(SymmetricAlgorithm.AES, cryptoKey);
ThreadContext.set(symmetricDecryptor, decryptor);
} else {
if (!decryptor.secretKey.equals(cryptoKey)) {
decryptor = new SymmetricDecryptor(SymmetricAlgorithm.AES, cryptoKey);
ThreadContext.set(symmetricDecryptor, decryptor);
}
}
DataBuffer from = NettyDataBufferUtils.from(decryptor.decrypt(bytes));
return Mono.just(from);
}
}
)
.flux();
}
private String constructTargetUrl(URI requestURI, String path, String serverAddress) {
StringBuilder b = ThreadContext.getStringBuilder();
b.append(serverAddress).append(path);
String qry = requestURI.getQuery();
if (StringUtils.hasText(qry)) {
if (org.apache.commons.lang3.StringUtils.indexOfAny(qry, Consts.S.LEFT_BRACE, Consts.S.FORWARD_SLASH, Consts.S.HASH) > 0) {
qry = requestURI.getRawQuery();
}
b.append(Consts.S.QUESTION).append(qry);
}
return b.toString();
}
private HttpHeaders signAndSetHeaders(HttpHeaders headers, String pairCodeId, String secretKey) {
String timestamp = String.valueOf(System.currentTimeMillis());
String sign = DedicatedLineUtils.sign(pairCodeId, timestamp, secretKey);
HttpHeaders writableHttpHeaders = HttpHeaders.writableHttpHeaders(headers);
writableHttpHeaders.set(SystemConfig.FIZZ_DL_ID, pairCodeId);
writableHttpHeaders.set(SystemConfig.FIZZ_DL_TS, timestamp);
writableHttpHeaders.set(SystemConfig.FIZZ_DL_SIGN, sign);
return writableHttpHeaders;
}
private void cleanup(ClientResponse clientResponse) {
if (clientResponse != null) {
clientResponse.bodyToMono(Void.class).subscribe();
@@ -200,7 +296,7 @@ class DedicatedLineHttpHandler implements HttpHandler {
if (response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR)) {
log.error(logPrefix + "500 Server Error for " + formatRequest(request), t);
return Mono.empty();
return WebUtils.response(response, null, null, logPrefix + ' ' + Utils.getMessage(t));
} else if (isDisconnectedClientError(t)) {
if (lostClientLog.isTraceEnabled()) {
@@ -208,7 +304,7 @@ class DedicatedLineHttpHandler implements HttpHandler {
} else if (lostClientLog.isDebugEnabled()) {
lostClientLog.debug(logPrefix + "Client went away: " + t + " (stacktrace at TRACE level for '" + disconnected_client_log_category + "')");
}
return Mono.empty();
return WebUtils.response(response, null, null, logPrefix + ' ' + Utils.getMessage(t));
} else {
// After the response is committed, propagate errors to the server...

4
fizz-core/src/main/java/we/filter/CallbackFilter.java
View File

@@ -103,10 +103,8 @@ public class CallbackFilter extends FizzWebFilter {
DataBuffer body = null;
if (b != NettyDataBufferUtils.EMPTY_DATA_BUFFER) {
if (b instanceof PooledDataBuffer) {
byte[] bytes = new byte[b.readableByteCount()];
try {
b.read(bytes);
body = NettyDataBufferUtils.from(bytes);
body = NettyDataBufferUtils.copy2heap(b);
} finally {
NettyDataBufferUtils.release(b);
}

3
fizz-plugin/src/main/java/we/plugin/dedicatedline/auth/DedicatedLineApiAuthPluginFilter.java
View File

@@ -27,6 +27,7 @@ import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import we.dedicated_line.DedicatedLineService;
import we.flume.clients.log4j2appender.LogService;
import we.plugin.FizzPluginFilter;
import we.plugin.FizzPluginFilterChain;
import we.util.ReactorUtils;
@@ -76,7 +77,7 @@ public class DedicatedLineApiAuthPluginFilter implements FizzPluginFilter {
return WebUtils.response(exchange, HttpStatus.UNAUTHORIZED, null, respJson);
}
} catch (Exception e) {
log.error("{} {} exception", traceId, DEDICATED_LINE_API_AUTH_PLUGIN_FILTER, e);
log.error("{} {} exception", traceId, DEDICATED_LINE_API_AUTH_PLUGIN_FILTER, LogService.BIZ_ID, traceId, e);
String respJson = WebUtils.jsonRespBody(HttpStatus.INTERNAL_SERVER_ERROR.value(),
HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase(), traceId);
return WebUtils.response(exchange, HttpStatus.INTERNAL_SERVER_ERROR, null, respJson);

164
fizz-plugin/src/main/java/we/plugin/dedicatedline/codec/DedicatedLineCodecPluginFilter.java
View File

@@ -17,11 +17,9 @@
package we.plugin.dedicatedline.codec;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import javax.annotation.Resource;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
import cn.hutool.crypto.symmetric.SymmetricCrypto;
import org.apache.commons.lang3.StringUtils;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
@@ -35,97 +33,123 @@ import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
import cn.hutool.crypto.symmetric.SymmetricCrypto;
import reactor.core.publisher.Mono;
import we.config.SystemConfig;
import we.dedicated_line.DedicatedLineService;
import we.flume.clients.log4j2appender.LogService;
import we.plugin.FizzPluginFilterChain;
import we.plugin.auth.App;
import we.plugin.auth.AppService;
import we.plugin.requestbody.RequestBodyPlugin;
import we.spring.http.server.reactive.ext.FizzServerHttpRequestDecorator;
import we.spring.http.server.reactive.ext.FizzServerHttpResponseDecorator;
import we.util.Consts;
import we.util.NettyDataBufferUtils;
import we.util.WebUtils;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Map;
/**
*
* @author Francis Dong
*
*/
@ConditionalOnBean(DedicatedLineService.class)
@Component(DedicatedLineCodecPluginFilter.DEDICATED_LINE_CODEC_PLUGIN_FILTER)
public class DedicatedLineCodecPluginFilter extends RequestBodyPlugin {
private static final Logger log = LoggerFactory.getLogger(DedicatedLineCodecPluginFilter.class);
private static final Logger log = LoggerFactory.getLogger(DedicatedLineCodecPluginFilter.class);
public static final String DEDICATED_LINE_CODEC_PLUGIN_FILTER = "dedicatedLineCodecPlugin";
public static final String DEDICATED_LINE_CODEC_PLUGIN_FILTER = "dedicatedLineCodecPlugin";
@Resource
private SystemConfig systemConfig;
@Resource
private SystemConfig systemConfig;
@Resource
private DedicatedLineService dedicatedLineService;
@Resource
private DedicatedLineService dedicatedLineService;
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public Mono<Void> doFilter(ServerWebExchange exchange, Map<String, Object> config) {
String traceId = WebUtils.getTraceId(exchange);
try {
LogService.setBizId(traceId);
String dedicatedLineId = WebUtils.getDedicatedLineId(exchange);
String secretKey = dedicatedLineService.getPairCodeSecretKey(dedicatedLineId);
@SuppressWarnings({"unchecked", "rawtypes"})
@Override
public Mono<Void> doFilter(ServerWebExchange exchange, Map<String, Object> config) {
String traceId = WebUtils.getTraceId(exchange);
try {
LogService.setBizId(traceId);
String dedicatedLineId = WebUtils.getDedicatedLineId(exchange);
// String secretKey = dedicatedLineService.getPairCodeSecretKey(dedicatedLineId);
FizzServerHttpRequestDecorator request = (FizzServerHttpRequestDecorator) exchange.getRequest();
return request.getBody().defaultIfEmpty(NettyDataBufferUtils.EMPTY_DATA_BUFFER).single().flatMap(body -> {
String reqBody = body.toString(StandardCharsets.UTF_8);
request.setBody(decrypt(reqBody, secretKey));
FizzServerHttpRequestDecorator request = (FizzServerHttpRequestDecorator) exchange.getRequest();
return request.getBody().defaultIfEmpty(NettyDataBufferUtils.EMPTY_DATA_BUFFER).single().flatMap(body -> {
/*String reqBody = body.toString(StandardCharsets.UTF_8);
request.setBody(decrypt(reqBody, secretKey));*/
ServerHttpResponse original = exchange.getResponse();
FizzServerHttpResponseDecorator fizzServerHttpResponseDecorator = new FizzServerHttpResponseDecorator(
original) {
@Override
public Publisher<? extends DataBuffer> writeWith(DataBuffer remoteResponseBody) {
String respBody = remoteResponseBody.toString(StandardCharsets.UTF_8);
HttpHeaders headers = getDelegate().getHeaders();
headers.setContentType(MediaType.TEXT_PLAIN);
headers.remove(HttpHeaders.CONTENT_LENGTH);
NettyDataBuffer from = NettyDataBufferUtils.from(encrypt(respBody, secretKey));
return Mono.just(from);
}
};
ServerWebExchange build = exchange.mutate().response(fizzServerHttpResponseDecorator).build();
return FizzPluginFilterChain.next(build);
});
String cryptoKey = systemConfig.fizzDedicatedLineClientRequestSecretkey();
if (body != NettyDataBufferUtils.EMPTY_DATA_BUFFER && StringUtils.isNotBlank(cryptoKey)) {
byte[] bodyBytes = request.getBodyBytes();
request.setBody(decrypt(bodyBytes, cryptoKey));
request.getHeaders().remove(HttpHeaders.CONTENT_LENGTH);
}
} catch (Exception e) {
log.error("{} {} Exception", traceId, DEDICATED_LINE_CODEC_PLUGIN_FILTER, e, LogService.BIZ_ID, traceId);
String respJson = WebUtils.jsonRespBody(HttpStatus.INTERNAL_SERVER_ERROR.value(),
HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase(), traceId);
return WebUtils.response(exchange, HttpStatus.INTERNAL_SERVER_ERROR, null, respJson);
}
}
ServerHttpResponse original = exchange.getResponse();
FizzServerHttpResponseDecorator fizzServerHttpResponseDecorator = new FizzServerHttpResponseDecorator(original) {
@Override
public Publisher<? extends DataBuffer> writeWith(DataBuffer remoteResponseBody) {
/*String respBody = remoteResponseBody.toString(StandardCharsets.UTF_8);
HttpHeaders headers = getDelegate().getHeaders();
headers.setContentType(MediaType.TEXT_PLAIN);
headers.remove(HttpHeaders.CONTENT_LENGTH);
NettyDataBuffer from = NettyDataBufferUtils.from(encrypt(respBody, secretKey));
return Mono.just(from);*/
if (remoteResponseBody == NettyDataBufferUtils.EMPTY_DATA_BUFFER) {
return Mono.empty();
} else {
if (StringUtils.isNotBlank(cryptoKey)) {
getDelegate().getHeaders().remove(HttpHeaders.CONTENT_LENGTH);
byte[] bytes = remoteResponseBody.asByteBuffer().array();
NettyDataBuffer from = NettyDataBufferUtils.from(encrypt(bytes, cryptoKey));
return Mono.just(from);
} else {
return Mono.just(remoteResponseBody);
}
}
}
};
ServerWebExchange build = exchange.mutate().response(fizzServerHttpResponseDecorator).build();
return FizzPluginFilterChain.next(build);
});
public String encrypt(String data, String secretKey) {
if (StringUtils.isBlank(data)) {
return data;
}
byte[] key = SecureUtil.decode(secretKey);
SymmetricCrypto symmetric = new SymmetricCrypto(SymmetricAlgorithm.AES, key);
return symmetric.encryptBase64(data);
}
} catch (Exception e) {
log.error("{} {} Exception", traceId, DEDICATED_LINE_CODEC_PLUGIN_FILTER, LogService.BIZ_ID, traceId, e);
String respJson = WebUtils.jsonRespBody(HttpStatus.INTERNAL_SERVER_ERROR.value(),
HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase(), traceId);
return WebUtils.response(exchange, HttpStatus.INTERNAL_SERVER_ERROR, null, respJson);
}
}
public String decrypt(String data, String secretKey) {
if (StringUtils.isBlank(data)) {
return data;
}
byte[] key = SecureUtil.decode(secretKey);
SymmetricCrypto symmetric = new SymmetricCrypto(SymmetricAlgorithm.AES, key);
return symmetric.decryptStr(data);
}
public String encrypt(String data, String secretKey) {
if (StringUtils.isBlank(data)) {
return data;
}
byte[] key = SecureUtil.decode(secretKey);
SymmetricCrypto symmetric = new SymmetricCrypto(SymmetricAlgorithm.AES, key);
return symmetric.encryptBase64(data);
}
public byte[] encrypt(byte[] data, String secretKey) {
byte[] key = SecureUtil.decode(secretKey);
SymmetricCrypto symmetric = new SymmetricCrypto(SymmetricAlgorithm.AES, key);
return symmetric.encrypt(data);
}
public String decrypt(String data, String secretKey) {
if (StringUtils.isBlank(data)) {
return data;
}
byte[] key = SecureUtil.decode(secretKey);
SymmetricCrypto symmetric = new SymmetricCrypto(SymmetricAlgorithm.AES, key);
return symmetric.decryptStr(data);
}
public byte[] decrypt(byte[] data, String secretKey) {
byte[] key = SecureUtil.decode(secretKey);
SymmetricCrypto symmetric = new SymmetricCrypto(SymmetricAlgorithm.AES, key);
return symmetric.decrypt(data);
}
}

150
fizz-plugin/src/main/java/we/plugin/dedicatedline/pairing/DedicatedLinePairingPluginFilter.java
View File

@@ -17,10 +17,6 @@
package we.plugin.dedicatedline.pairing;
import java.util.Map;
import javax.annotation.Resource;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,105 +25,101 @@ import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import we.config.SystemConfig;
import we.dedicated_line.DedicatedLineService;
import we.flume.clients.log4j2appender.LogService;
import we.plugin.FizzPluginFilter;
import we.plugin.FizzPluginFilterChain;
import we.plugin.auth.App;
import we.plugin.auth.AppService;
import we.util.DigestUtils;
import we.util.ReactorUtils;
import we.util.WebUtils;
import javax.annotation.Resource;
import java.util.Map;
/**
*
* @author Francis Dong
*
*/
@ConditionalOnBean(DedicatedLineService.class)
@Component(DedicatedLinePairingPluginFilter.DEDICATED_LINE_PAIRING_PLUGIN_FILTER)
public class DedicatedLinePairingPluginFilter implements FizzPluginFilter {
private static final Logger log = LoggerFactory.getLogger(DedicatedLinePairingPluginFilter.class);
private static final Logger log = LoggerFactory.getLogger(DedicatedLinePairingPluginFilter.class);
public static final String DEDICATED_LINE_PAIRING_PLUGIN_FILTER = "dedicatedLinePairingPlugin";
public static final String DEDICATED_LINE_PAIRING_PLUGIN_FILTER = "dedicatedLinePairingPlugin";
@Resource
private SystemConfig systemConfig;
@Resource
private SystemConfig systemConfig;
@Resource
private DedicatedLineService dedicatedLineService;
@Resource
private DedicatedLineService dedicatedLineService;
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public Mono<Void> filter(ServerWebExchange exchange, Map<String, Object> config) {
String traceId = WebUtils.getTraceId(exchange);
try {
LogService.setBizId(traceId);
String dedicatedLineId = WebUtils.getDedicatedLineId(exchange);
String secretKey = dedicatedLineService.getPairCodeSecretKey(dedicatedLineId);
String ts = WebUtils.getDedicatedLineTimestamp(exchange);
String sign = WebUtils.getDedicatedLineSign(exchange);
if (validateSign(dedicatedLineId, ts, sign, secretKey)) {
// Go to next plugin
Mono next = FizzPluginFilterChain.next(exchange);
return next.defaultIfEmpty(ReactorUtils.NULL).flatMap(nil -> {
doAfter();
return Mono.empty();
});
} else {
// Auth failed
ServerHttpResponse response = exchange.getResponse();
response.setStatusCode(HttpStatus.UNAUTHORIZED);
response.getHeaders().setCacheControl("no-store");
response.getHeaders().setExpires(0);
String respJson = WebUtils.jsonRespBody(HttpStatus.UNAUTHORIZED.value(),
HttpStatus.UNAUTHORIZED.getReasonPhrase(), traceId);
return WebUtils.response(exchange, HttpStatus.UNAUTHORIZED, null, respJson);
}
} catch (Exception e) {
log.error("{} {} Exception", traceId, DEDICATED_LINE_PAIRING_PLUGIN_FILTER, e, LogService.BIZ_ID, traceId);
String respJson = WebUtils.jsonRespBody(HttpStatus.INTERNAL_SERVER_ERROR.value(),
HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase(), traceId);
return WebUtils.response(exchange, HttpStatus.INTERNAL_SERVER_ERROR, null, respJson);
}
}
@SuppressWarnings({"unchecked", "rawtypes"})
@Override
public Mono<Void> filter(ServerWebExchange exchange, Map<String, Object> config) {
String traceId = WebUtils.getTraceId(exchange);
try {
LogService.setBizId(traceId);
String dedicatedLineId = WebUtils.getDedicatedLineId(exchange);
String secretKey = dedicatedLineService.getPairCodeSecretKey(dedicatedLineId);
String ts = WebUtils.getDedicatedLineTimestamp(exchange);
String sign = WebUtils.getDedicatedLineSign(exchange);
if (validateSign(dedicatedLineId, ts, sign, secretKey)) {
// Go to next plugin
Mono next = FizzPluginFilterChain.next(exchange);
return next.defaultIfEmpty(ReactorUtils.NULL).flatMap(nil -> {
doAfter();
return Mono.empty();
});
} else {
// Auth failed
ServerHttpResponse response = exchange.getResponse();
response.setStatusCode(HttpStatus.UNAUTHORIZED);
response.getHeaders().setCacheControl("no-store");
response.getHeaders().setExpires(0);
String respJson = WebUtils.jsonRespBody(HttpStatus.UNAUTHORIZED.value(),
HttpStatus.UNAUTHORIZED.getReasonPhrase(), traceId);
return WebUtils.response(exchange, HttpStatus.UNAUTHORIZED, null, respJson);
}
} catch (Exception e) {
log.error("{} {} Exception", traceId, DEDICATED_LINE_PAIRING_PLUGIN_FILTER, LogService.BIZ_ID, traceId, e);
String respJson = WebUtils.jsonRespBody(HttpStatus.INTERNAL_SERVER_ERROR.value(),
HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase(), traceId);
return WebUtils.response(exchange, HttpStatus.INTERNAL_SERVER_ERROR, null, respJson);
}
}
private boolean validateSign(String dedicatedLineId, String ts, String sign, String secretkey) {
if (StringUtils.isBlank(dedicatedLineId) || StringUtils.isBlank(ts) || StringUtils.isBlank(sign)
|| StringUtils.isBlank(secretkey)) {
return false;
}
private boolean validateSign(String dedicatedLineId, String ts, String sign, String secretkey) {
if (StringUtils.isBlank(dedicatedLineId) || StringUtils.isBlank(ts) || StringUtils.isBlank(sign)
|| StringUtils.isBlank(secretkey)) {
return false;
}
// SHA256(dedicatedLineId+_+ts+_+secretkey)
String data = dedicatedLineId + "_" + ts + "_" + secretkey;
if (!DigestUtils.sha256Hex(data).equals(sign)) {
return false;
}
// SHA256(dedicatedLineId+_+ts+_+secretkey)
String data = dedicatedLineId + "_" + ts + "_" + secretkey;
if (!DigestUtils.sha256Hex(data).equals(sign)) {
return false;
}
// validate timestamp
long t = 0;
try {
t = Long.valueOf(ts).longValue();
} catch (Exception e) {
log.warn("invalid timestamp: {}", ts);
return false;
}
long now = System.currentTimeMillis();
long offset = 5 * 60 * 1000;
if (t < now - offset || t > now + offset) {
log.warn("timestamp expired: {}", ts);
return false;
}
// validate timestamp
long t = 0;
try {
t = Long.valueOf(ts).longValue();
} catch (Exception e) {
log.warn("invalid timestamp: {}", ts);
return false;
}
long now = System.currentTimeMillis();
long offset = 5 * 60 * 1000;
if (t < now - offset || t > now + offset) {
log.warn("timestamp expired: {}", ts);
return false;
}
return true;
}
public void doAfter() {
}
return true;
}
public void doAfter() {
}
}

2
pom.xml
View File

@@ -22,7 +22,7 @@
<r2dbc-mysql.version>0.8.2</r2dbc-mysql.version>
<reflections.version>0.9.11</reflections.version>
<commons-pool2.version>2.11.1</commons-pool2.version>
<netty-tcnative.version>2.0.45.Final</netty-tcnative.version>
<netty-tcnative.version>2.0.46.Final</netty-tcnative.version>
<spring-cloud.version>2.2.9.RELEASE</spring-cloud.version>
<resilience4j.version>1.7.1</resilience4j.version>
</properties>