From a7357639f1581500dc60f0469ee5056d977865ce Mon Sep 17 00:00:00 2001 From: Ziliang Liao Date: Fri, 11 Sep 2026 11:19:24 +0800 Subject: [PATCH] feat: forward ucp id and align monitor responses --- .../src/main/proto/monitorSystemgrpc.proto | 2 + .../MonitorSystemServer/pom.xml | 13 +- .../src/main/java/MonitorSystemServer.java | 121 +++++++++++--- .../test/java/MonitorSystemServerTest.java | 113 +++++++++++++ SimpleMonitorSystem/README.md | 2 + .../node/server/MonitorNodeServiceImpl.java | 7 +- .../node/service/IEndorserService.java | 12 ++ .../service/impl/EndorserServiceImpl.java | 47 +++++- .../src/main/proto/monitorSystemgrpc.proto | 2 + .../node/server/MonitorNodeServiceTest.java | 52 ++++++ .../node/service/EndorserServiceTest.java | 157 ++++++++++++++++++ acb-relayer/r-core/pom.xml | 5 + .../UniformCrosschainPacketValidator.java | 6 +- .../UniformCrosschainPacketValidatorTest.java | 139 ++++++++++++++++ .../bridge/ptc/service/IPTCService.java | 9 + .../ptc/committee/CommitteePTCService.java | 14 +- .../network/nodeclient/GrpcNodeClient.java | 10 ++ .../types/network/nodeclient/INodeClient.java | 8 + .../src/main/proto/node.proto | 1 + .../committee/CommitteePTCServiceTest.java | 16 +- 20 files changed, 694 insertions(+), 42 deletions(-) create mode 100644 SimpleMonitorSystem/MonitorSystemServer/src/test/java/MonitorSystemServerTest.java create mode 100644 acb-relayer/r-core/src/test/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidatorTest.java diff --git a/SimpleMonitorSystem/MonitorAPI/src/main/proto/monitorSystemgrpc.proto b/SimpleMonitorSystem/MonitorAPI/src/main/proto/monitorSystemgrpc.proto index c103e05a..55436526 100755 --- a/SimpleMonitorSystem/MonitorAPI/src/main/proto/monitorSystemgrpc.proto +++ b/SimpleMonitorSystem/MonitorAPI/src/main/proto/monitorSystemgrpc.proto @@ -22,11 +22,13 @@ message Empty {} // 监管节点作为客户端向监管系统请求验证ucp的合法性(VerifyCrossChainMessage方法中调用) message VerifyCrossChainMessageInMonitorSystemRequest { bytes rawUcp = 1; + string ucpId = 2; } // 接收无需监管的跨链消息，并转发给课题四监管系统供其分析(VerifyCrossChainMessage方法中调用) message RelayUcpToMonitorSystemRequest { bytes rawUcp = 1; + string ucpId = 2; } message MonitorSystemResponse { diff --git a/SimpleMonitorSystem/MonitorSystemServer/pom.xml b/SimpleMonitorSystem/MonitorSystemServer/pom.xml index 581498f6..27f49910 100755 --- a/SimpleMonitorSystem/MonitorSystemServer/pom.xml +++ b/SimpleMonitorSystem/MonitorSystemServer/pom.xml @@ -23,6 +23,17 @@ MonitorAPI 1.0-SNAPSHOT + + com.alipay.antchain.bridge + antchain-bridge-commons + 1.0.0-SNAPSHOT + + + junit + junit + 4.13.2 + test + @@ -61,4 +72,4 @@ - \ No newline at end of file + diff --git a/SimpleMonitorSystem/MonitorSystemServer/src/main/java/MonitorSystemServer.java b/SimpleMonitorSystem/MonitorSystemServer/src/main/java/MonitorSystemServer.java index bf89be91..df2d2552 100755 --- a/SimpleMonitorSystem/MonitorSystemServer/src/main/java/MonitorSystemServer.java +++ b/SimpleMonitorSystem/MonitorSystemServer/src/main/java/MonitorSystemServer.java @@ -1,6 +1,7 @@ import io.grpc.Server; import io.grpc.stub.StreamObserver; import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; +import com.alipay.antchain.bridge.commons.core.base.UniformCrosschainPacket; import com.alipay.antchain.bridge.ptc.committee.monitor.system.grpc.*; import java.io.File; @@ -20,8 +21,15 @@ public class MonitorSystemServer { private Server server; - // 控制返回成功或失败，默认返回成功 - private volatile boolean verifySuccess = true; + enum VerifyMode { + SUCCESS, + FAILURE, + INTERNAL_ERROR, + UNAVAILABLE + } + + // 控制 verify 的返回结果，默认返回成功 + private volatile VerifyMode verifyMode = VerifyMode.SUCCESS; public void start(int port) throws Exception { server = NettyServerBuilder.forPort(port) @@ -42,32 +50,42 @@ public void start(int port) throws Exception { logInfo("Server shut down."); })); - // 仅在交互式终端中启动控制线程。systemd/nohup 后台运行时没有可用的 stdin， - // 此时保持默认的验证成功行为，避免 Scanner.nextLine() 因 EOF 退出并打印异常。 - if (System.console() != null) { - new Thread(this::startCommandListener, "CommandListenerThread").start(); - } else { - logInfo("No interactive console detected; command listener disabled and verify result defaults to success."); - } + // 仅在交互式终端中启动控制线程。systemd/nohup 后台运行时没有可用的 stdin， + // 此时保持默认的验证成功行为，避免 Scanner.nextLine() 因 EOF 退出并打印异常。 + if (System.console() != null) { + new Thread(this::startCommandListener, "CommandListenerThread").start(); + } else { + logInfo("No interactive console detected; command listener disabled and verify result defaults to success."); + } } private void startCommandListener() { Scanner scanner = new Scanner(System.in); - logInfo("请输入 'success' 或 'fail' 来控制 verifyCrossChainMessageInMonitorSystem 返回结果："); + logInfo("请输入 'success'、'fail'、'500' 或 '503' 来控制 verifyCrossChainMessageInMonitorSystem 返回结果："); while (true) { String input = scanner.nextLine(); if ("success".equalsIgnoreCase(input)) { - verifySuccess = true; + verifyMode = VerifyMode.SUCCESS; logInfo("切换为：返回成功"); } else if ("fail".equalsIgnoreCase(input)) { - verifySuccess = false; + verifyMode = VerifyMode.FAILURE; logInfo("切换为：返回失败"); + } else if ("500".equals(input)) { + verifyMode = VerifyMode.INTERNAL_ERROR; + logInfo("切换为：返回内部错误(code=500)"); + } else if ("503".equals(input)) { + verifyMode = VerifyMode.UNAVAILABLE; + logInfo("切换为：返回服务不可用(code=503)"); } else { - logInfo("无效指令，请输入 'success' 或 'fail'"); + logInfo("无效指令，请输入 'success'、'fail'、'500' 或 '503'"); } } } + void setVerifyMode(VerifyMode verifyMode) { + this.verifyMode = verifyMode; + } + public void stop() { if (server != null) { server.shutdown(); @@ -97,15 +115,44 @@ public void verifyCrossChainMessageInMonitorSystem( VerifyCrossChainMessageInMonitorSystemRequest request, StreamObserver responseObserver) { + MonitorSystemResponse response; + try { + response = buildVerifyResponse(request); + } catch (RuntimeException e) { + logWarning("Unexpected error while processing verify request", e); + response = errorResponse(500, "internal monitor system error: " + errorMessage(e)); + } + + responseObserver.onNext(response); + responseObserver.onCompleted(); + } + + MonitorSystemResponse buildVerifyResponse(VerifyCrossChainMessageInMonitorSystemRequest request) { + byte[] rawUcp = request.getRawUcp().toByteArray(); logInfo("Received verifyCrossChainMessageInMonitorSystem request:"); + logInfo("ucpId: " + request.getUcpId()); logInfo("rawUcp (hex): " + bytesToHex(rawUcp)); + if (verifyMode == VerifyMode.INTERNAL_ERROR) { + return errorResponse(500, "simulated internal monitor system error"); + } + if (verifyMode == VerifyMode.UNAVAILABLE) { + return errorResponse(503, "simulated monitor system unavailable"); + } + + try { + if (UniformCrosschainPacket.decode(rawUcp) == null) { + return errorResponse(400, "failed to decode rawUcp: decoded UCP is null"); + } + } catch (RuntimeException e) { + return errorResponse(400, "failed to decode rawUcp: " + errorMessage(e)); + } + MonitorSystemResponse.Builder responseBuilder = MonitorSystemResponse.newBuilder() .setCode(0) .setErrorMsg(""); - - if (verifySuccess) { + if (verifyMode == VerifyMode.SUCCESS) { responseBuilder.setVerifyCrossChainMessageInMonitorSystemResp( VerifyCrossChainMessageInMonitorSystemResponse.newBuilder() .setResult(0) @@ -116,13 +163,12 @@ public void verifyCrossChainMessageInMonitorSystem( responseBuilder.setVerifyCrossChainMessageInMonitorSystemResp( VerifyCrossChainMessageInMonitorSystemResponse.newBuilder() .setResult(1) - .setMsg("fail") + .setMsg("simulated regulation rejection: matched test rule") .build()); logInfo("返回：失败"); } - responseObserver.onNext(responseBuilder.build()); - responseObserver.onCompleted(); + return responseBuilder.build(); } @Override @@ -130,18 +176,26 @@ public void relayUcpToMonitorSystem( RelayUcpToMonitorSystemRequest request, StreamObserver responseObserver) { - byte[] rawUcp = request.getRawUcp().toByteArray(); - logInfo("Received relayUcpToMonitorSystem request:"); - logInfo("rawUcp (hex): " + bytesToHex(rawUcp)); + responseObserver.onNext(buildRelayResponse(request)); + responseObserver.onCompleted(); + } - MonitorSystemResponse response = MonitorSystemResponse.newBuilder() + MonitorSystemResponse buildRelayResponse(RelayUcpToMonitorSystemRequest request) { + + try { + byte[] rawUcp = request.getRawUcp().toByteArray(); + logInfo("Received relayUcpToMonitorSystem request:"); + logInfo("ucpId: " + request.getUcpId()); + logInfo("rawUcp (hex): " + bytesToHex(rawUcp)); + } catch (RuntimeException e) { + // relay 是旁路留存接口，内部处理失败只记录日志，不能影响跨链流程。 + logWarning("Failed to process relay request, returning success as required", e); + } + + return MonitorSystemResponse.newBuilder() .setCode(0) .setErrorMsg("") .build(); - - responseObserver.onNext(response); - responseObserver.onCompleted(); - } private String bytesToHex(byte[] bytes) { @@ -151,12 +205,27 @@ private String bytesToHex(byte[] bytes) { } return sb.toString(); } + + private MonitorSystemResponse errorResponse(int code, String errorMsg) { + return MonitorSystemResponse.newBuilder() + .setCode(code) + .setErrorMsg(errorMsg) + .build(); + } + + private String errorMessage(RuntimeException e) { + return e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage(); + } } private static void logInfo(String message) { LOGGER.info(message); } + private static void logWarning(String message, Throwable throwable) { + LOGGER.log(Level.WARNING, message, throwable); + } + private static Logger createLogger() { Logger logger = Logger.getLogger(MonitorSystemServer.class.getName()); logger.setUseParentHandlers(false); diff --git a/SimpleMonitorSystem/MonitorSystemServer/src/test/java/MonitorSystemServerTest.java b/SimpleMonitorSystem/MonitorSystemServer/src/test/java/MonitorSystemServerTest.java new file mode 100644 index 00000000..10d7e319 --- /dev/null +++ b/SimpleMonitorSystem/MonitorSystemServer/src/test/java/MonitorSystemServerTest.java @@ -0,0 +1,113 @@ +import com.alipay.antchain.bridge.commons.core.base.CrossChainDomain; +import com.alipay.antchain.bridge.commons.core.base.CrossChainMessage; +import com.alipay.antchain.bridge.commons.core.base.UniformCrosschainPacket; +import com.alipay.antchain.bridge.ptc.committee.monitor.system.grpc.MonitorSystemResponse; +import com.alipay.antchain.bridge.ptc.committee.monitor.system.grpc.RelayUcpToMonitorSystemRequest; +import com.alipay.antchain.bridge.ptc.committee.monitor.system.grpc.VerifyCrossChainMessageInMonitorSystemRequest; +import com.google.protobuf.ByteString; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class MonitorSystemServerTest { + + private MonitorSystemServer server; + + private MonitorSystemServer.MonitorSystemServiceImpl service; + + private byte[] validRawUcp; + + @Before + public void setUp() { + server = new MonitorSystemServer(); + service = server.new MonitorSystemServiceImpl(); + validRawUcp = new UniformCrosschainPacket( + new CrossChainDomain("test-domain"), + CrossChainMessage.createCrossChainMessage( + CrossChainMessage.CrossChainMessageType.DEVELOPER_DESIGN, + 1L, + 1L, + new byte[]{1}, + new byte[]{2}, + new byte[]{3}, + new byte[]{4}, + new byte[]{5} + ), + null + ).encode(); + } + + @Test + public void testVerifySuccessAndFailureResponses() { + VerifyCrossChainMessageInMonitorSystemRequest request = verifyRequest(validRawUcp); + + server.setVerifyMode(MonitorSystemServer.VerifyMode.SUCCESS); + MonitorSystemResponse response = service.buildVerifyResponse(request); + Assert.assertEquals(0, response.getCode()); + Assert.assertEquals("", response.getErrorMsg()); + Assert.assertTrue(response.hasVerifyCrossChainMessageInMonitorSystemResp()); + Assert.assertEquals(0, response.getVerifyCrossChainMessageInMonitorSystemResp().getResult()); + Assert.assertEquals("ok", response.getVerifyCrossChainMessageInMonitorSystemResp().getMsg()); + + server.setVerifyMode(MonitorSystemServer.VerifyMode.FAILURE); + response = service.buildVerifyResponse(request); + Assert.assertEquals(0, response.getCode()); + Assert.assertEquals("", response.getErrorMsg()); + Assert.assertTrue(response.hasVerifyCrossChainMessageInMonitorSystemResp()); + Assert.assertEquals(1, response.getVerifyCrossChainMessageInMonitorSystemResp().getResult()); + Assert.assertFalse(response.getVerifyCrossChainMessageInMonitorSystemResp().getMsg().isEmpty()); + } + + @Test + public void testInvalidRawUcpReturns400WithoutVerifyResponse() { + MonitorSystemResponse response = service.buildVerifyResponse(verifyRequest(new byte[]{1, 2, 3, 4})); + + Assert.assertEquals(400, response.getCode()); + Assert.assertTrue(response.getErrorMsg().contains("failed to decode rawUcp")); + Assert.assertFalse(response.hasVerifyCrossChainMessageInMonitorSystemResp()); + } + + @Test + public void testForced500And503ResponsesDoNotContainVerifyResponse() { + VerifyCrossChainMessageInMonitorSystemRequest request = verifyRequest(validRawUcp); + + server.setVerifyMode(MonitorSystemServer.VerifyMode.INTERNAL_ERROR); + MonitorSystemResponse response = service.buildVerifyResponse(request); + Assert.assertEquals(500, response.getCode()); + Assert.assertFalse(response.getErrorMsg().isEmpty()); + Assert.assertFalse(response.hasVerifyCrossChainMessageInMonitorSystemResp()); + + server.setVerifyMode(MonitorSystemServer.VerifyMode.UNAVAILABLE); + response = service.buildVerifyResponse(request); + Assert.assertEquals(503, response.getCode()); + Assert.assertFalse(response.getErrorMsg().isEmpty()); + Assert.assertFalse(response.hasVerifyCrossChainMessageInMonitorSystemResp()); + } + + @Test + public void testRelayAlwaysReturnsSuccessWithoutVerifyResponse() { + for (byte[] rawUcp : new byte[][]{validRawUcp, new byte[]{1, 2, 3, 4}}) { + MonitorSystemResponse response = service.buildRelayResponse( + RelayUcpToMonitorSystemRequest.newBuilder() + .setRawUcp(ByteString.copyFrom(rawUcp)) + .setUcpId("ucp-id") + .build() + ); + Assert.assertEquals(0, response.getCode()); + Assert.assertEquals("", response.getErrorMsg()); + Assert.assertFalse(response.hasVerifyCrossChainMessageInMonitorSystemResp()); + } + + MonitorSystemResponse responseAfterInternalError = service.buildRelayResponse(null); + Assert.assertEquals(0, responseAfterInternalError.getCode()); + Assert.assertEquals("", responseAfterInternalError.getErrorMsg()); + Assert.assertFalse(responseAfterInternalError.hasVerifyCrossChainMessageInMonitorSystemResp()); + } + + private VerifyCrossChainMessageInMonitorSystemRequest verifyRequest(byte[] rawUcp) { + return VerifyCrossChainMessageInMonitorSystemRequest.newBuilder() + .setRawUcp(ByteString.copyFrom(rawUcp)) + .setUcpId("ucp-id") + .build(); + } +} diff --git a/SimpleMonitorSystem/README.md b/SimpleMonitorSystem/README.md index 206cdd56..1880d8c6 100755 --- a/SimpleMonitorSystem/README.md +++ b/SimpleMonitorSystem/README.md @@ -37,6 +37,8 @@ sudo ./bin/stop.sh systemd模式会把服务文件安装到`/etc/systemd/system/simple-monitor-system.service`。 后台模式没有交互式终端，监管验证结果保持默认的成功状态；前台直接运行Jar时仍可输入`success`或`fail`切换结果。 +前台模式还支持输入`500`或`503`，用于模拟监管系统内部错误和暂时不可用。`verifyCrossChainMessageInMonitorSystem`会先校验`rawUcp`，无法解析时返回`code=400`且不填充判定响应；`relayUcpToMonitorSystem`始终返回`code=0`且不填充判定响应。 + ## MonitorSystemClient 仅用于测试MonitorSystemServer功能是否正常。 diff --git a/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceImpl.java b/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceImpl.java index 567b310e..9491660e 100755 --- a/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceImpl.java +++ b/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceImpl.java @@ -336,11 +336,12 @@ public void verifyCrossChainMessage(VerifyCrossChainMessageRequest request, Stre if (ObjectUtil.isNull(crossChainLane)) { throw new InvalidRequestException("crossChainLane is null"); } + String ucpId = request.getUcpId(); // 为dioxide链定制的逻辑，支持在无实际监管逻辑和PTC逻辑的情形下将跨链信息传递给外部监管系统 if (Objects.equals(crossChainLane.getSenderDomain().getDomain(), "dioxide2")) { log.info("receive crosschain message from Dioxide, relay to monitor system directly without verification"); - MonitorNodeVerifyResult verifyResult = endorserService.relayUcpToMonitorSystem(ucp); + MonitorNodeVerifyResult verifyResult = endorserService.relayUcpToMonitorSystem(ucp, ucpId); responseObserver.onNext( Response.newBuilder() .setCode(0) @@ -372,11 +373,11 @@ public void verifyCrossChainMessage(VerifyCrossChainMessageRequest request, Stre verifyResult = MonitorNodeVerifyResult.approved(endorserService.verifyUcp(crossChainLane, ucp)); } else if (monitorMessage.getMonitorType() == MonitorTypeEnum.MONITOR_OPEN.getCode()) { log.info("crosschain message: need monitor"); - verifyResult = endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp); + verifyResult = endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp, ucpId); } else { // 不监管 把跨链消息发给监管系统即可 log.info("crosschain message: don't need monitor"); - MonitorNodeVerifyResult relayResult = endorserService.relayUcpToMonitorSystem(ucp); + MonitorNodeVerifyResult relayResult = endorserService.relayUcpToMonitorSystem(ucp, ucpId); verifyResult = new MonitorNodeVerifyResult( endorserService.verifyUcp(crossChainLane, ucp), relayResult.getRegulationStatus(), diff --git a/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/IEndorserService.java b/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/IEndorserService.java index 6af2a72d..398ab28b 100755 --- a/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/IEndorserService.java +++ b/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/IEndorserService.java @@ -44,7 +44,19 @@ public interface IEndorserService { MonitorNodeVerifyResult verifyUcpWithMonitorSystem(CrossChainLane crossChainLane, UniformCrosschainPacket ucp); + default MonitorNodeVerifyResult verifyUcpWithMonitorSystem( + CrossChainLane crossChainLane, + UniformCrosschainPacket ucp, + String ucpId + ) { + return verifyUcpWithMonitorSystem(crossChainLane, ucp); + } + MonitorNodeVerifyResult relayUcpToMonitorSystem(UniformCrosschainPacket ucp); + default MonitorNodeVerifyResult relayUcpToMonitorSystem(UniformCrosschainPacket ucp, String ucpId) { + return relayUcpToMonitorSystem(ucp); + } + EndorseBlockStateResp endorseBlockState(CrossChainLane crossChainLane, String receiverDomain, BigInteger height); } diff --git a/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/impl/EndorserServiceImpl.java b/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/impl/EndorserServiceImpl.java index 28f7898f..b28d6e68 100755 --- a/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/impl/EndorserServiceImpl.java +++ b/acb-committeeptc/monitor-node/src/main/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/impl/EndorserServiceImpl.java @@ -303,6 +303,15 @@ public CommitteeNodeProof verifyUcp(CrossChainLane crossChainLane, UniformCrossc @Override public MonitorNodeVerifyResult verifyUcpWithMonitorSystem(CrossChainLane crossChainLane, UniformCrosschainPacket ucp) { + return verifyUcpWithMonitorSystem(crossChainLane, ucp, ""); + } + + @Override + public MonitorNodeVerifyResult verifyUcpWithMonitorSystem( + CrossChainLane crossChainLane, + UniformCrosschainPacket ucp, + String ucpId + ) { var tpbta = endorseServiceRepository.getExactTpBta(crossChainLane); if (ObjectUtil.isNull(tpbta)) { @@ -325,6 +334,7 @@ public MonitorNodeVerifyResult verifyUcpWithMonitorSystem(CrossChainLane crossCh stub -> stub.verifyCrossChainMessageInMonitorSystem( VerifyCrossChainMessageInMonitorSystemRequest.newBuilder() .setRawUcp(ByteString.copyFrom(ucp.encode())) + .setUcpId(StrUtil.nullToEmpty(ucpId)) .build() ) ); @@ -340,11 +350,21 @@ public MonitorNodeVerifyResult verifyUcpWithMonitorSystem(CrossChainLane crossCh if (responseFromMonitorSystem.getCode() != 0) { return MonitorNodeVerifyResult.error( emptySignatureProof(), - String.format("[MonitorSystemGRpcClient] verifyCrossChainMessageInMonitorSystem request failed: %s", - responseFromMonitorSystem.getErrorMsg()) + String.format("[MonitorSystemGRpcClient] verifyCrossChainMessageInMonitorSystem request failed " + + "(code: %d): %s", + responseFromMonitorSystem.getCode(), responseFromMonitorSystem.getErrorMsg()) + ); + } + if (!responseFromMonitorSystem.hasVerifyCrossChainMessageInMonitorSystemResp()) { + return MonitorNodeVerifyResult.error( + emptySignatureProof(), + "[MonitorSystemGRpcClient] verifyCrossChainMessageInMonitorSystem returned code 0 without verify response" ); } - if (responseFromMonitorSystem.getVerifyCrossChainMessageInMonitorSystemResp().getResult() == 0) { + + VerifyCrossChainMessageInMonitorSystemResponse verifyResponse = + responseFromMonitorSystem.getVerifyCrossChainMessageInMonitorSystemResp(); + if (verifyResponse.getResult() == 0) { // 监管通过 流程正常 // log.info("verify ucp with monitor system for domain {}: success", bta.getDomain()); log.info("verify ucp with monitor system for domain {}: success", crossChainLane.getSenderDomain().getDomain()); @@ -360,7 +380,7 @@ public MonitorNodeVerifyResult verifyUcpWithMonitorSystem(CrossChainLane crossCh ).getEncodedToSign() )).build(); return MonitorNodeVerifyResult.approved(proof); - } else { + } else if (verifyResponse.getResult() == 1) { // [监管回滚的v1版本逻辑] // 监管未通过 直接向监管合约发送回滚交易 并且不跑出异常 而是返回一个签名 // 目前是返回一个正确的签名, 保证在监管不通过时系统的稳定运行; 在8~9月开发的最终版本中会返回一个空签名, 实现完整的逻辑 @@ -406,22 +426,35 @@ public MonitorNodeVerifyResult verifyUcpWithMonitorSystem(CrossChainLane crossCh // 返回一个ethereum格式(65字节)的空签名 由目的链的监管合约验证签名时识别为监管失败 构造监管回滚消息 return MonitorNodeVerifyResult.rejected( emptySignatureProof(), - responseFromMonitorSystem.getVerifyCrossChainMessageInMonitorSystemResp().getMsg() + verifyResponse.getMsg() ); // throw new InvalidCrossChainMessageException("[monitor system] illegal crosschain message(block hash: {}): {}", // ucp.getSrcMessage().getProvableData().getBlockHashHex(), responseFromMonitorSystem.getVerifyCrossChainMessageInMonitorSystemResp().getMsg()); } + + return MonitorNodeVerifyResult.error( + emptySignatureProof(), + String.format("[MonitorSystemGRpcClient] verifyCrossChainMessageInMonitorSystem returned " + + "unexpected result %d: %s", + verifyResponse.getResult(), verifyResponse.getMsg()) + ); } @Override public MonitorNodeVerifyResult relayUcpToMonitorSystem(UniformCrosschainPacket ucp) { + return relayUcpToMonitorSystem(ucp, ""); + } + + @Override + public MonitorNodeVerifyResult relayUcpToMonitorSystem(UniformCrosschainPacket ucp, String ucpId) { MonitorSystemResponse responseFromMonitorSystem; try { responseFromMonitorSystem = monitorSystemGrpcClientManager.withStub( stub -> stub.relayUcpToMonitorSystem( RelayUcpToMonitorSystemRequest.newBuilder() .setRawUcp(ByteString.copyFrom(ucp.encode())) + .setUcpId(StrUtil.nullToEmpty(ucpId)) .build() ) ); @@ -436,8 +469,8 @@ public MonitorNodeVerifyResult relayUcpToMonitorSystem(UniformCrosschainPacket u if (responseFromMonitorSystem.getCode() != 0) { return MonitorNodeVerifyResult.error( emptySignatureProof(), - String.format("[MonitorSystemGRpcClient] relayUcpToMonitorSystem request failed: %s", - responseFromMonitorSystem.getErrorMsg()) + String.format("[MonitorSystemGRpcClient] relayUcpToMonitorSystem request failed (code: %d): %s", + responseFromMonitorSystem.getCode(), responseFromMonitorSystem.getErrorMsg()) ); } diff --git a/acb-committeeptc/monitor-node/src/main/proto/monitorSystemgrpc.proto b/acb-committeeptc/monitor-node/src/main/proto/monitorSystemgrpc.proto index be14f86d..6436f1c5 100755 --- a/acb-committeeptc/monitor-node/src/main/proto/monitorSystemgrpc.proto +++ b/acb-committeeptc/monitor-node/src/main/proto/monitorSystemgrpc.proto @@ -22,11 +22,13 @@ message Empty {} // 监管节点作为客户端向监管系统请求验证ucp的合法性(VerifyCrossChainMessage方法中调用) message VerifyCrossChainMessageInMonitorSystemRequest { bytes rawUcp = 1; + string ucpId = 2; } // 接收无需监管的跨链消息，并转发给课题四监管系统供其分析(VerifyCrossChainMessage方法中调用) message RelayUcpToMonitorSystemRequest { bytes rawUcp = 1; + string ucpId = 2; } message MonitorSystemResponse { diff --git a/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceTest.java b/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceTest.java index 65850cab..18cea1d0 100755 --- a/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceTest.java +++ b/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/server/MonitorNodeServiceTest.java @@ -41,6 +41,7 @@ import com.alipay.antchain.bridge.ptc.committee.monitor.node.TestBase; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.BtaWrapper; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.DomainSpaceCertWrapper; +import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.MonitorNodeVerifyResult; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.TpBtaWrapper; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.ValidatedConsensusStateWrapper; import com.alipay.antchain.bridge.ptc.committee.monitor.node.dal.repository.interfaces.IBCDNSRepository; @@ -64,10 +65,14 @@ import org.junit.Test; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.test.util.ReflectionTestUtils; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class MonitorNodeServiceTest extends TestBase { @@ -490,6 +495,53 @@ public void testVerifyCrossChainMessage() { ); } + @Test + @SneakyThrows + public void testForwardUcpIdToEndorserService() { + String ucpId = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + IEndorserService mockEndorserService = mock(IEndorserService.class); + MonitorNodeServiceImpl service = new MonitorNodeServiceImpl(); + ReflectionTestUtils.setField(service, "endorserService", mockEndorserService); + CommitteeNodeProof proof = CommitteeNodeProof.builder() + .nodeId("monitor-node") + .signAlgo(SignAlgoEnum.KECCAK256_WITH_SECP256K1) + .signature(new byte[65]) + .build(); + when(mockEndorserService.verifyUcpWithMonitorSystem(any(), any(), eq(ucpId))) + .thenReturn(MonitorNodeVerifyResult.approved(proof)); + + StreamRecorder verifyResponseObserver = StreamRecorder.create(); + service.verifyCrossChainMessage( + VerifyCrossChainMessageRequest.newBuilder() + .setCrossChainLane(ByteString.copyFrom(crossChainLane.encode())) + .setRawUcp(ByteString.copyFrom(ucp.encode())) + .setUcpId(ucpId) + .build(), + verifyResponseObserver + ); + + Assert.assertTrue(verifyResponseObserver.awaitCompletion(5, TimeUnit.SECONDS)); + Assert.assertEquals(0, verifyResponseObserver.getValues().getFirst().getCode()); + verify(mockEndorserService).verifyUcpWithMonitorSystem(any(), any(), eq(ucpId)); + + reset(mockEndorserService); + when(mockEndorserService.relayUcpToMonitorSystem(any(), eq(""))) + .thenReturn(MonitorNodeVerifyResult.approved(proof)); + CrossChainLane dioxideLane = new CrossChainLane(new CrossChainDomain("dioxide2")); + StreamRecorder relayResponseObserver = StreamRecorder.create(); + service.verifyCrossChainMessage( + VerifyCrossChainMessageRequest.newBuilder() + .setCrossChainLane(ByteString.copyFrom(dioxideLane.encode())) + .setRawUcp(ByteString.copyFrom(ucp.encode())) + .build(), + relayResponseObserver + ); + + Assert.assertTrue(relayResponseObserver.awaitCompletion(5, TimeUnit.SECONDS)); + Assert.assertEquals(0, relayResponseObserver.getValues().getFirst().getCode()); + verify(mockEndorserService).relayUcpToMonitorSystem(any(), eq("")); + } + @Test @SneakyThrows public void testQueryBlockState() { diff --git a/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/EndorserServiceTest.java b/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/EndorserServiceTest.java index 9801c491..b6b77184 100755 --- a/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/EndorserServiceTest.java +++ b/acb-committeeptc/monitor-node/src/test/java/com/alipay/antchain/bridge/ptc/committee/monitor/node/service/EndorserServiceTest.java @@ -17,6 +17,7 @@ package com.alipay.antchain.bridge.ptc.committee.monitor.node.service; import java.math.BigInteger; +import java.util.function.Function; import cn.hutool.core.bean.BeanUtil; import cn.hutool.core.collection.ListUtil; @@ -39,8 +40,10 @@ import com.alipay.antchain.bridge.plugins.spi.ptc.IHeteroChainDataVerifierService; import com.alipay.antchain.bridge.plugins.spi.ptc.core.VerifyResult; import com.alipay.antchain.bridge.ptc.committee.monitor.node.TestBase; +import com.alipay.antchain.bridge.ptc.committee.monitor.node.client.MonitorSystemGrpcClientManager; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.BtaWrapper; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.DomainSpaceCertWrapper; +import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.MonitorNodeVerifyResult; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.TpBtaWrapper; import com.alipay.antchain.bridge.ptc.committee.monitor.node.commons.models.ValidatedConsensusStateWrapper; import com.alipay.antchain.bridge.ptc.committee.monitor.node.dal.repository.interfaces.IBCDNSRepository; @@ -54,9 +57,11 @@ import com.alipay.antchain.bridge.ptc.committee.types.tpbta.NodeEndorseInfo; import com.alipay.antchain.bridge.ptc.committee.types.tpbta.OptionalEndorsePolicy; import com.alipay.antchain.bridge.ptc.committee.types.tpbta.VerifyBtaExtension; +import com.alipay.antchain.bridge.ptc.committee.monitor.system.grpc.*; import jakarta.annotation.Resource; import org.junit.Assert; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.test.mock.mockito.MockBean; @@ -211,6 +216,9 @@ public class EndorserServiceTest extends TestBase { @MockBean private IHcdvsPluginService hcdvsPluginService; + @MockBean + private MonitorSystemGrpcClientManager monitorSystemGrpcClientManager; + @Resource private AbstractCrossChainCertificate ptcCrossChainCert; @@ -356,6 +364,155 @@ public void testVerifyUcp() { ); } + @Test + public void testForwardUcpIdToMonitorSystem() { + String ucpId = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + MonitorSystemServiceGrpc.MonitorSystemServiceBlockingStub stub = + mock(MonitorSystemServiceGrpc.MonitorSystemServiceBlockingStub.class); + when(stub.relayUcpToMonitorSystem(any())).thenReturn( + MonitorSystemResponse.newBuilder().setCode(0).build() + ); + when(stub.verifyCrossChainMessageInMonitorSystem(any())).thenReturn( + MonitorSystemResponse.newBuilder() + .setCode(0) + .setVerifyCrossChainMessageInMonitorSystemResp( + VerifyCrossChainMessageInMonitorSystemResponse.newBuilder() + .setResult(0) + .setMsg("ok") + ).build() + ); + when(monitorSystemGrpcClientManager.withStub(any())).thenAnswer(invocation -> { + Function action = invocation.getArgument(0); + return action.apply(stub); + }); + + when(endorseServiceRepository.getExactTpBta(any())).thenReturn(new TpBtaWrapper(tpbta)); + when(endorseServiceRepository.getBta(anyString(), anyInt())).thenReturn(new BtaWrapper(bta)); + + endorserService.relayUcpToMonitorSystem(ucp, ucpId); + ArgumentCaptor relayCaptor = + ArgumentCaptor.forClass(RelayUcpToMonitorSystemRequest.class); + verify(stub).relayUcpToMonitorSystem(relayCaptor.capture()); + Assert.assertEquals(ucpId, relayCaptor.getValue().getUcpId()); + Assert.assertArrayEquals(ucp.encode(), relayCaptor.getValue().getRawUcp().toByteArray()); + + endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp, ucpId); + ArgumentCaptor verifyCaptor = + ArgumentCaptor.forClass(VerifyCrossChainMessageInMonitorSystemRequest.class); + verify(stub).verifyCrossChainMessageInMonitorSystem(verifyCaptor.capture()); + Assert.assertEquals(ucpId, verifyCaptor.getValue().getUcpId()); + Assert.assertArrayEquals(ucp.encode(), verifyCaptor.getValue().getRawUcp().toByteArray()); + + clearInvocations(stub); + endorserService.relayUcpToMonitorSystem(ucp); + verify(stub).relayUcpToMonitorSystem(relayCaptor.capture()); + Assert.assertEquals("", relayCaptor.getValue().getUcpId()); + } + + @Test + public void testVerifyMonitorSystemResponseContract() { + MonitorSystemServiceGrpc.MonitorSystemServiceBlockingStub stub = prepareMonitorSystemStub(); + when(endorseServiceRepository.getExactTpBta(any())).thenReturn(new TpBtaWrapper(tpbta)); + when(endorseServiceRepository.getBta(anyString(), anyInt())).thenReturn(new BtaWrapper(bta)); + + when(stub.verifyCrossChainMessageInMonitorSystem(any())).thenReturn( + verifyResponse(0, "ok") + ); + MonitorNodeVerifyResult result = endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_APPROVED, result.getRegulationStatus()); + Assert.assertFalse(ArrayUtil.isEmpty(result.getNodeProof().getSig())); + + when(stub.verifyCrossChainMessageInMonitorSystem(any())).thenReturn( + verifyResponse(1, "matched regulation rule") + ); + result = endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_REJECTED, result.getRegulationStatus()); + Assert.assertEquals("matched regulation rule", result.getRegulationReason()); + Assert.assertArrayEquals(new byte[65], result.getNodeProof().getSig()); + + for (int code : new int[]{400, 500, 503}) { + when(stub.verifyCrossChainMessageInMonitorSystem(any())).thenReturn( + MonitorSystemResponse.newBuilder() + .setCode(code) + .setErrorMsg("monitor error " + code) + .build() + ); + result = endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_ERROR, result.getRegulationStatus()); + Assert.assertTrue(result.getRegulationReason().contains(String.valueOf(code))); + Assert.assertTrue(result.getRegulationReason().contains("monitor error " + code)); + Assert.assertArrayEquals(new byte[65], result.getNodeProof().getSig()); + } + + when(stub.verifyCrossChainMessageInMonitorSystem(any())).thenReturn( + MonitorSystemResponse.newBuilder().setCode(0).build() + ); + result = endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_ERROR, result.getRegulationStatus()); + Assert.assertTrue(result.getRegulationReason().contains("without verify response")); + + when(stub.verifyCrossChainMessageInMonitorSystem(any())).thenReturn( + verifyResponse(2, "unknown result") + ); + result = endorserService.verifyUcpWithMonitorSystem(crossChainLane, ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_ERROR, result.getRegulationStatus()); + Assert.assertTrue(result.getRegulationReason().contains("unexpected result 2")); + Assert.assertArrayEquals(new byte[65], result.getNodeProof().getSig()); + } + + @Test + public void testRelayMonitorSystemResponseContract() { + MonitorSystemServiceGrpc.MonitorSystemServiceBlockingStub stub = prepareMonitorSystemStub(); + + when(stub.relayUcpToMonitorSystem(any())).thenReturn( + MonitorSystemResponse.newBuilder().setCode(0).build() + ); + MonitorNodeVerifyResult result = endorserService.relayUcpToMonitorSystem(ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_APPROVED, result.getRegulationStatus()); + + when(stub.relayUcpToMonitorSystem(any())).thenReturn( + MonitorSystemResponse.newBuilder() + .setCode(500) + .setErrorMsg("relay failed") + .build() + ); + result = endorserService.relayUcpToMonitorSystem(ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_ERROR, result.getRegulationStatus()); + Assert.assertTrue(result.getRegulationReason().contains("500")); + Assert.assertTrue(result.getRegulationReason().contains("relay failed")); + + when(stub.relayUcpToMonitorSystem(any())).thenReturn(null); + result = endorserService.relayUcpToMonitorSystem(ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_ERROR, result.getRegulationStatus()); + Assert.assertTrue(result.getRegulationReason().contains("null response")); + + doThrow(new RuntimeException("transport failed")) + .when(monitorSystemGrpcClientManager).withStub(any()); + result = endorserService.relayUcpToMonitorSystem(ucp, "ucp-id"); + Assert.assertEquals(MonitorNodeVerifyResult.STATUS_ERROR, result.getRegulationStatus()); + Assert.assertEquals("transport failed", result.getRegulationReason()); + } + + private MonitorSystemServiceGrpc.MonitorSystemServiceBlockingStub prepareMonitorSystemStub() { + MonitorSystemServiceGrpc.MonitorSystemServiceBlockingStub stub = + mock(MonitorSystemServiceGrpc.MonitorSystemServiceBlockingStub.class); + when(monitorSystemGrpcClientManager.withStub(any())).thenAnswer(invocation -> { + Function action = invocation.getArgument(0); + return action.apply(stub); + }); + return stub; + } + + private MonitorSystemResponse verifyResponse(int result, String msg) { + return MonitorSystemResponse.newBuilder() + .setCode(0) + .setVerifyCrossChainMessageInMonitorSystemResp( + VerifyCrossChainMessageInMonitorSystemResponse.newBuilder() + .setResult(result) + .setMsg(msg) + ).build(); + } + @Test public void testEndorseBlockState() { var currVcs = BeanUtil.copyProperties(currState, ValidatedConsensusStateV1.class); diff --git a/acb-relayer/r-core/pom.xml b/acb-relayer/r-core/pom.xml index addff55a..ab29ae40 100644 --- a/acb-relayer/r-core/pom.xml +++ b/acb-relayer/r-core/pom.xml @@ -56,6 +56,11 @@ junit test + + org.mockito + mockito-core + test + diff --git a/acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidator.java b/acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidator.java index 138d7dda..52456b48 100644 --- a/acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidator.java +++ b/acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidator.java @@ -118,7 +118,8 @@ public void doProcess(UniformCrosschainPacketContext ucpContext) { PTCVerifyCrossChainMessageResult verifyResult = ptcService.verifyCrossChainMessageWithResult( tpBtaDO.getTpbta(), vcs, - ucpContext.getUcp() + ucpContext.getUcp(), + ucpContext.getUcpId() ); ThirdPartyProof tpProof = verifyResult.getThirdPartyProof(); if (ObjectUtil.isNull(tpProof)) { @@ -148,7 +149,8 @@ public void doProcess(UniformCrosschainPacketContext ucpContext) { PTCVerifyCrossChainMessageResult verifyResult = ptcService.verifyCrossChainMessageWithResult( tpBtaOnlyRepresentDioxide, new ValidatedConsensusStateV1(), - ucpContext.getUcp() + ucpContext.getUcp(), + ucpContext.getUcpId() ); platformReportClient.reportRegulation( ucpContext.getUcpId(), diff --git a/acb-relayer/r-core/src/test/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidatorTest.java b/acb-relayer/r-core/src/test/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidatorTest.java new file mode 100644 index 00000000..f5a3cd0d --- /dev/null +++ b/acb-relayer/r-core/src/test/java/com/alipay/antchain/bridge/relayer/core/service/validation/UniformCrosschainPacketValidatorTest.java @@ -0,0 +1,139 @@ +package com.alipay.antchain.bridge.relayer.core.service.validation; + +import java.lang.reflect.Field; + +import com.alipay.antchain.bridge.commons.core.am.AuthMessageV1; +import com.alipay.antchain.bridge.commons.core.base.CrossChainDomain; +import com.alipay.antchain.bridge.commons.core.base.CrossChainIdentity; +import com.alipay.antchain.bridge.commons.core.base.CrossChainLane; +import com.alipay.antchain.bridge.commons.core.base.CrossChainMessage; +import com.alipay.antchain.bridge.commons.core.base.UniformCrosschainPacket; +import com.alipay.antchain.bridge.commons.core.ptc.ThirdPartyBlockchainTrustAnchor; +import com.alipay.antchain.bridge.commons.core.ptc.ThirdPartyProof; +import com.alipay.antchain.bridge.ptc.service.IPTCService; +import com.alipay.antchain.bridge.ptc.types.PtcFeatureDescriptor; +import com.alipay.antchain.bridge.ptc.types.PTCVerifyCrossChainMessageResult; +import com.alipay.antchain.bridge.relayer.commons.model.TpBtaDO; +import com.alipay.antchain.bridge.relayer.commons.model.UniformCrosschainPacketContext; +import com.alipay.antchain.bridge.relayer.core.manager.ptc.PtcManager; +import com.alipay.antchain.bridge.relayer.core.service.report.PlatformReportClient; +import com.alipay.antchain.bridge.relayer.dal.repository.IBlockchainRepository; +import com.alipay.antchain.bridge.relayer.dal.repository.ICrossChainMessageRepository; +import org.junit.Before; +import org.junit.Test; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class UniformCrosschainPacketValidatorTest { + + private static final String UCP_ID = + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + + private ICrossChainMessageRepository crossChainMessageRepository; + + private PtcManager ptcManager; + + private IPTCService ptcService; + + private UniformCrosschainPacketValidator validator; + + @Before + public void setUp() { + crossChainMessageRepository = mock(ICrossChainMessageRepository.class); + ptcManager = mock(PtcManager.class); + ptcService = mock(IPTCService.class); + validator = new UniformCrosschainPacketValidator(); + setField(validator, "crossChainMessageRepository", crossChainMessageRepository); + setField(validator, "blockchainRepository", mock(IBlockchainRepository.class)); + setField(validator, "ptcManager", ptcManager); + setField(validator, "platformReportClient", mock(PlatformReportClient.class)); + setField(validator, "ptcId", "ptc01"); + } + + @Test + public void testPassUcpIdToPtcForEndorsedUcp() { + UniformCrosschainPacketContext context = buildContext("ethereum3"); + CrossChainLane lane = new CrossChainLane(new CrossChainDomain("source.example")); + context.setTpbtaLaneKey(lane.getLaneKey()); + context.setTpbtaVersion(1); + + TpBtaDO tpBtaDO = mock(TpBtaDO.class); + ThirdPartyBlockchainTrustAnchor tpbta = mock(ThirdPartyBlockchainTrustAnchor.class); + when(tpBtaDO.getPtcServiceId()).thenReturn("ptc01"); + when(tpBtaDO.getTpbta()).thenReturn(tpbta); + when(ptcManager.getExactTpBta(any(), eq(1))).thenReturn(tpBtaDO); + when(ptcManager.getPtcService("ptc01")).thenReturn(ptcService); + PtcFeatureDescriptor featureDescriptor = new PtcFeatureDescriptor(); + featureDescriptor.enableStorage(); + when(ptcService.getPtcFeatureDescriptor()).thenReturn(featureDescriptor); + when(ptcService.verifyCrossChainMessageWithResult(any(), isNull(), any(), eq(UCP_ID))) + .thenReturn(new PTCVerifyCrossChainMessageResult(new ThirdPartyProof(), "approved", "")); + + validator.doProcess(context); + + verify(ptcService).verifyCrossChainMessageWithResult( + eq(tpbta), + isNull(), + eq(context.getUcp()), + eq(UCP_ID) + ); + } + + @Test + public void testPassUcpIdToPtcForDioxideUcp() { + UniformCrosschainPacketContext context = buildContext("dioxide2"); + when(ptcManager.getPtcService("ptc01")).thenReturn(ptcService); + when(ptcService.verifyCrossChainMessageWithResult(any(), any(), any(), eq(UCP_ID))) + .thenReturn(new PTCVerifyCrossChainMessageResult(new ThirdPartyProof(), "approved", "")); + + validator.doProcess(context); + + verify(ptcService).verifyCrossChainMessageWithResult( + any(ThirdPartyBlockchainTrustAnchor.class), + any(), + eq(context.getUcp()), + eq(UCP_ID) + ); + } + + private static UniformCrosschainPacketContext buildContext(String product) { + AuthMessageV1 authMessage = new AuthMessageV1(); + authMessage.setIdentity(new CrossChainIdentity(new byte[32])); + authMessage.setUpperProtocol(0); + authMessage.setPayload(new byte[0]); + CrossChainMessage crossChainMessage = CrossChainMessage.createCrossChainMessage( + CrossChainMessage.CrossChainMessageType.AUTH_MSG, + 1L, + 1L, + new byte[]{0x01}, + authMessage.encode(), + new byte[0], + new byte[0], + new byte[]{0x02} + ); + UniformCrosschainPacketContext context = new UniformCrosschainPacketContext(); + context.setUcpId(UCP_ID); + context.setProduct(product); + context.setUcp(new UniformCrosschainPacket( + new CrossChainDomain("source.example"), + crossChainMessage, + null + )); + return context; + } + + private static void setField(Object target, String fieldName, Object value) { + try { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } catch (ReflectiveOperationException e) { + throw new AssertionError("failed to inject field " + fieldName, e); + } + } +} diff --git a/acb-sdk/antchain-bridge-ptc/src/main/java/com/alipay/antchain/bridge/ptc/service/IPTCService.java b/acb-sdk/antchain-bridge-ptc/src/main/java/com/alipay/antchain/bridge/ptc/service/IPTCService.java index a5679c43..84710c66 100644 --- a/acb-sdk/antchain-bridge-ptc/src/main/java/com/alipay/antchain/bridge/ptc/service/IPTCService.java +++ b/acb-sdk/antchain-bridge-ptc/src/main/java/com/alipay/antchain/bridge/ptc/service/IPTCService.java @@ -66,6 +66,15 @@ default PTCVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( ); } + default PTCVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( + ThirdPartyBlockchainTrustAnchor tpbta, + ValidatedConsensusState validatedConsensusState, + UniformCrosschainPacket ucp, + String ucpId + ) { + return verifyCrossChainMessageWithResult(tpbta, validatedConsensusState, ucp); + } + Set querySupportedBlockchainProducts(); BlockState queryCurrVerifiedBlockState(ThirdPartyBlockchainTrustAnchor tpbta); diff --git a/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCService.java b/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCService.java index 514b02ba..a9de3bb8 100644 --- a/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCService.java +++ b/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCService.java @@ -343,6 +343,16 @@ public PTCVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( ThirdPartyBlockchainTrustAnchor tpbta, ValidatedConsensusState validatedConsensusState, UniformCrosschainPacket ucp + ) { + return verifyCrossChainMessageWithResult(tpbta, validatedConsensusState, ucp, ""); + } + + @Override + public PTCVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( + ThirdPartyBlockchainTrustAnchor tpbta, + ValidatedConsensusState validatedConsensusState, + UniformCrosschainPacket ucp, + String ucpId ) { try { @@ -361,7 +371,7 @@ public PTCVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( // "Dioxide"这个product继续通过tpbta中特殊的CrossChainLane来传递 NodeVerifyCrossChainMessageResult nodeResult = monitorNode.getNodeClient() - .verifyCrossChainMessageWithResult(tpbta.getCrossChainLane(), ucp); + .verifyCrossChainMessageWithResult(tpbta.getCrossChainLane(), ucp, ucpId); return new PTCVerifyCrossChainMessageResult( new ThirdPartyProof(), nodeResult.getRegulationStatus(), @@ -390,7 +400,7 @@ public PTCVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( ).map( entry -> (Callable) () -> { log.debug("Verify crosschain msg with node {} {}", entry.getKey(), entry.getValue().getEndpointInfo().getEndpoint().getUrl()); - return entry.getValue().getNodeClient().verifyCrossChainMessageWithResult(tpbta.getCrossChainLane(), ucp); + return entry.getValue().getNodeClient().verifyCrossChainMessageWithResult(tpbta.getCrossChainLane(), ucp, ucpId); } ).collect(Collectors.toList()) ); diff --git a/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/GrpcNodeClient.java b/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/GrpcNodeClient.java index d7649999..868562b0 100644 --- a/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/GrpcNodeClient.java +++ b/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/GrpcNodeClient.java @@ -163,11 +163,21 @@ public CommitteeNodeProof verifyCrossChainMessage(CrossChainLane crossChainLane, public NodeVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( CrossChainLane crossChainLane, UniformCrosschainPacket packet + ) { + return verifyCrossChainMessageWithResult(crossChainLane, packet, ""); + } + + @Override + public NodeVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( + CrossChainLane crossChainLane, + UniformCrosschainPacket packet, + String ucpId ) { Response response = stub.verifyCrossChainMessage( VerifyCrossChainMessageRequest.newBuilder() .setRawUcp(ByteString.copyFrom(packet.encode())) .setCrossChainLane(ByteString.copyFrom(crossChainLane.encode())) + .setUcpId(StrUtil.nullToEmpty(ucpId)) .build() ); if (ObjectUtil.isNull(response)) { diff --git a/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/INodeClient.java b/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/INodeClient.java index 0ff828d8..136c077e 100644 --- a/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/INodeClient.java +++ b/acb-sdk/ptc-services/committee-ptc-core/src/main/java/com/alipay/antchain/bridge/ptc/committee/types/network/nodeclient/INodeClient.java @@ -71,6 +71,14 @@ default NodeVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( ); } + default NodeVerifyCrossChainMessageResult verifyCrossChainMessageWithResult( + CrossChainLane crossChainLane, + UniformCrosschainPacket packet, + String ucpId + ) { + return verifyCrossChainMessageWithResult(crossChainLane, packet); + } + BlockState queryBlockState(CrossChainDomain blockchainDomain); EndorseBlockStateResp endorseBlockState(CrossChainLane tpbtaLane, CrossChainDomain receiverDomain, BigInteger height); diff --git a/acb-sdk/ptc-services/committee-ptc-core/src/main/proto/node.proto b/acb-sdk/ptc-services/committee-ptc-core/src/main/proto/node.proto index 91d07f78..3cca2534 100644 --- a/acb-sdk/ptc-services/committee-ptc-core/src/main/proto/node.proto +++ b/acb-sdk/ptc-services/committee-ptc-core/src/main/proto/node.proto @@ -93,6 +93,7 @@ message CommitConsensusStateResponse { message VerifyCrossChainMessageRequest { bytes crossChainLane = 1; bytes rawUcp = 2; + string ucpId = 3; } message VerifyCrossChainMessageResponse { diff --git a/acb-sdk/ptc-services/committee-ptc-core/src/test/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCServiceTest.java b/acb-sdk/ptc-services/committee-ptc-core/src/test/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCServiceTest.java index 329c5422..a31a186e 100644 --- a/acb-sdk/ptc-services/committee-ptc-core/src/test/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCServiceTest.java +++ b/acb-sdk/ptc-services/committee-ptc-core/src/test/java/com/alipay/antchain/bridge/ptc/committee/CommitteePTCServiceTest.java @@ -65,6 +65,7 @@ import com.google.protobuf.ByteString; import lombok.SneakyThrows; import org.junit.*; +import org.mockito.ArgumentCaptor; import org.mockito.MockedStatic; import static org.junit.Assert.*; @@ -615,6 +616,7 @@ public void testCollectNodeVerifyResultsSkipsFailedOptionalFuture() { public void testVerifyCrossChainMessage() { CommitteePTCService ptcService = new CommitteePTCService(); ptcService.startup(SERVICE_CONF.getBytes()); + String ucpId = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; CommitteeNodeProof nodeProof = CommitteeNodeProof.builder() .nodeId("node1") @@ -637,16 +639,28 @@ public void testVerifyCrossChainMessage() { ).build() ); + clearInvocations(mockStubNode1); PTCVerifyCrossChainMessageResult verifyResult = - ptcService.verifyCrossChainMessageWithResult(tpbta, currVcs, ucp); + ptcService.verifyCrossChainMessageWithResult(tpbta, currVcs, ucp, ucpId); ThirdPartyProof tpProof = verifyResult.getThirdPartyProof(); + ArgumentCaptor requestCaptor = + ArgumentCaptor.forClass(VerifyCrossChainMessageRequest.class); + verify(mockStubNode1).verifyCrossChainMessage(requestCaptor.capture()); + assertEquals(ucpId, requestCaptor.getValue().getUcpId()); + assertArrayEquals(ucp.encode(), requestCaptor.getValue().getRawUcp().toByteArray()); + assertEquals(tpbta.getCrossChainLane().getLaneKey(), tpProof.getTpbtaCrossChainLane().getLaneKey()); assertEquals("approved", verifyResult.getRegulationStatus()); assertEquals("", verifyResult.getRegulationReason()); CommitteeEndorseProof endorseProof = CommitteeEndorseProof.decode(tpProof.getRawProof()); assertEquals(COMMITTEE_ID, endorseProof.getCommitteeId()); assertEquals(1, endorseProof.getSigs().size()); + + clearInvocations(mockStubNode1); + ptcService.verifyCrossChainMessageWithResult(tpbta, currVcs, ucp); + verify(mockStubNode1).verifyCrossChainMessage(requestCaptor.capture()); + assertEquals("", requestCaptor.getValue().getUcpId()); } @Test