diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index 7c1e94e..63ccf96 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -44,9 +44,11 @@ import org.lfenergy.shapeshifter.core.service.handler.UftpPayloadHandler; import org.lfenergy.shapeshifter.core.service.participant.ParticipantResolutionService; import org.lfenergy.shapeshifter.core.service.receiving.UftpReceivedMessageService; +import org.lfenergy.shapeshifter.core.service.receiving.response.UftpValidationResponseCreator; import org.lfenergy.shapeshifter.core.service.sending.UftpSendMessageService; import org.lfenergy.shapeshifter.core.service.serialization.UftpSerializer; import org.lfenergy.shapeshifter.core.service.validation.UftpValidationService; +import org.lfenergy.shapeshifter.core.service.validation.model.ValidationResult; import org.openremote.container.timer.TimerService; import org.openremote.container.web.CORSConfig; import org.openremote.container.web.WebApplication; @@ -551,8 +553,9 @@ protected long calculatePower(FlexRequestISPType requestIsp) { /** * Only act on flex messages whose congestion point matches this handler's contracted EAN. Returns - * false (and logs a warning) for out-of-scope messages so they are dropped before any asset mutation - * or outbound response. See issue #28 for full per-contract/role scoping via the V3 contracts endpoint. + * false (and logs a warning) for out-of-scope messages so they are rejected (with a rejection + * response, but without asset mutation or FlexOffer). See issue #28 for full per-contract/role + * scoping via the V3 contracts endpoint. */ protected boolean isWithinContractedScope(String messageType, String conversationId, String congestionPoint) { if (Objects.equals(toCongestionPoint(contractedEAN), toCongestionPoint(congestionPoint))) { @@ -594,11 +597,20 @@ protected void processRawMessage(String transportXml) { // resolves any participant domain, so a validly signed flex message from a participant outside // this handler's contracted EAN would otherwise be applied to the asset. Full per-contract/role // scoping via the V3 contracts endpoint is tracked in issue #28. - // Out-of-scope messages are intentionally dropped here: the transport call still returns 200 - // (signed envelope accepted) but no FlexRequestResponse/FlexOffer is sent in reply. + // Out-of-scope messages are not applied to the asset and get no FlexOffer, but UFTP still + // requires a response: reject with a reason instead of leaving the DSO's conversation dangling. if (payloadMessage instanceof FlexMessageType flexMessage && !isWithinContractedScope(payloadMessage.getClass().getSimpleName(), payloadMessage.getConversationID(), flexMessage.getCongestionPoint())) { + PayloadMessageType rejection = UftpValidationResponseCreator.getResponseForMessage( + payloadMessage, ValidationResult.rejection("CongestionPoint not within contracted scope")); + UftpParticipant sender = new UftpParticipant(signedMessage); + UftpParticipant responder = new UftpParticipant(payloadMessage.getRecipientDomain(), + UftpRoleInformation.getRecipientRoleBySenderRole(sender.role())); + // Send delayed so the HTTP 200 on the transport call goes out first, like the accepted path + scheduledExecutorService.schedule(() -> + notifyNewOutgoingMessage(OutgoingUftpMessage.create(responder, rejection)), + this.responseDelaySeconds, TimeUnit.SECONDS); return; } diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy index 50d9df1..095732b 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy @@ -224,7 +224,10 @@ class GOPACSHandlerHttpTest extends Specification implements ManagerContainerTra } } - def "a validly-signed FlexRequest for a different congestion point is accepted but produces no broker delivery"() { + def "a validly-signed FlexRequest for a different congestion point is rejected via the broker without a FlexOffer"() { + given: "polling for the asynchronous outbound delivery" + def conditions = new PollingConditions(timeout: 10, delay: 0.2) + when: "a signed out-of-scope FlexRequest (different EAN) is POSTed" Response response = postSignedAsDso(flexRequestXml("ean.999999999999999999")) @@ -232,10 +235,22 @@ class GOPACSHandlerHttpTest extends Specification implements ManagerContainerTra response.statusInfo.family == Response.Status.Family.SUCCESSFUL response.close() - and: "no message is ever delivered to the broker" - // The drop happens before any reply is scheduled; allow the (zero-delay) scheduler to run first. + and: "a rejected FlexRequestResponse with a reason is delivered to the broker as signed XML" + conditions.eventually { + wireMock.verify(postRequestedFor(urlPathEqualTo(BROKER_PATH)) + .withHeader("Authorization", equalTo("Bearer " + ACCESS_TOKEN)) + .withRequestBody(containing("SignedMessage"))) + assert decodedBrokerPayloads().any { + it.contains("FlexRequestResponse") && it.contains('Result="Rejected"') && it.contains("RejectionReason=") + } + } + + and: "no FlexOffer is ever delivered" + // Replies are scheduled with zero delay in this test; give any (erroneous) FlexOffer time to arrive. Thread.sleep(1000) - wireMock.verify(0, postRequestedFor(urlPathEqualTo(BROKER_PATH))) + def payloads = decodedBrokerPayloads() + payloads.size() == 1 + !payloads.any { it.contains("FlexOffer") } } def "malformed transport XML is rejected with 400"() { diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index bc145d4..f14046d 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -183,14 +183,41 @@ class GOPACSHandlerTest extends Specification { ((FlexOrderResponse) handler.sent[0]).result == AcceptedRejectedType.ACCEPTED } - def "a validly-signed FlexRequest for a different congestion point is dropped with no mutation and no reply"() { + def "a validly-signed FlexRequest for a different congestion point is rejected without mutation or FlexOffer"() { when: "a signed out-of-scope FlexRequest (different EAN) is processed" signAndProcess(flexRequestXml("ean.999999999999999999")) - then: "the asset is not mutated and nothing is sent back" + then: "the asset is not mutated" 0 * assetPredictedDatapointService.updateValues(_, _, _) 0 * assetProcessingService.sendAttributeEvent(_, _) - handler.sent.isEmpty() + + and: "only a rejected FlexRequestResponse is sent back, no FlexOffer" + handler.sent.size() == 1 + handler.sent[0] instanceof FlexRequestResponse + def response = (FlexRequestResponse) handler.sent[0] + response.result == AcceptedRejectedType.REJECTED + !response.rejectionReason.isBlank() + response.flexRequestMessageID == FLEX_REQUEST_MESSAGE_ID + response.conversationID == CONVERSATION_ID + response.senderDomain == AGR_DOMAIN + response.recipientDomain == DSO_DOMAIN + } + + def "a validly-signed FlexOrder for a different congestion point is rejected without mutation"() { + when: "a signed out-of-scope FlexOrder (different EAN) is processed" + signAndProcess(flexOrderXml("ean.999999999999999999", [4000, 8000])) + + then: "the asset is not mutated" + 0 * assetPredictedDatapointService.updateValues(_, _, _) + 0 * assetProcessingService.sendAttributeEvent(_, _) + + and: "only a rejected FlexOrderResponse is sent back" + handler.sent.size() == 1 + handler.sent[0] instanceof FlexOrderResponse + def response = (FlexOrderResponse) handler.sent[0] + response.result == AcceptedRejectedType.REJECTED + !response.rejectionReason.isBlank() + response.conversationID == CONVERSATION_ID } def "toCongestionPoint canonicalises an EAN to the GOPACS ean. format (#input -> #expected)"() {