Commit 72a6dc6b authored by Hema Vishnu Pola [Microsoft]'s avatar Hema Vishnu Pola [Microsoft]
Browse files

Merge branch 'extractor_azure_impl' into 'master'

Pubsub Azure Implementation.

See merge request !12
parents bda02143 c939a272
Pipeline #7305 passed with stages
in 25 minutes and 58 seconds
// Copyright © Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package org.opengroup.osdu.notification.provider.azure.models;
import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class HandshakeRequestData {
private String validationCode;
private String validationUrl;
}
// Copyright © Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package org.opengroup.osdu.notification.provider.azure.models;
import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Map;
@Data
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class NotificationData {
private String data;
private Map<String,String> attributes;
private String messageId;
}
// Copyright © Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package org.opengroup.osdu.notification.provider.azure.models;
import com.google.gson.JsonObject;
import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class NotificationRequest {
private String id;
private String eventType;
private String subject;
private JsonObject data;
private String eventTime;
}
// Copyright © Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package org.opengroup.osdu.notification.provider.azure.pubsub;
import com.google.gson.JsonObject;
import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.notification.provider.interfaces.IPubsubHandshakeHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
@Component
public class EventGridHandshakeHandler implements IPubsubHandshakeHandler {
@Autowired
private EventGridRequestBodyExtractor eventGridRequestBodyExtractor;
/**
* Extract Handshake response string form Handshake request.
* TODO: Check if there is a need to verify subscription name with
* registration service, before verifying the endpoint.
* @return validation string
*/
@Override
public String getHandshakeResponse() {
String response;
try {
String validationCode = this.eventGridRequestBodyExtractor.getValidationCodeForHandshake();
JsonObject jsonResponse = new JsonObject();
jsonResponse.addProperty("ValidationResponse", validationCode);
response = jsonResponse.toString();
} catch (Exception exception) {
throw new AppException(HttpStatus.BAD_REQUEST.value(), "Request payload parsing error",
"Unable to parse request payload.", exception);
}
return response;
}
}
// Copyright © Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package org.opengroup.osdu.notification.provider.azure.pubsub;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.notification.provider.azure.models.HandshakeRequestData;
import org.opengroup.osdu.notification.provider.azure.models.NotificationData;
import org.opengroup.osdu.notification.provider.azure.models.NotificationRequest;
import org.opengroup.osdu.notification.provider.interfaces.IPubsubRequestBodyExtractor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import org.springframework.web.context.annotation.RequestScope;
import javax.servlet.http.HttpServletRequest;
import java.io.BufferedReader;
import java.util.Base64;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
// TODO : Make error messages as constants.
// TODO : Add TracePoints to exception logs so that monitoring is done well.
@Component
@RequestScope
public class EventGridRequestBodyExtractor implements IPubsubRequestBodyExtractor {
private static final String INVALID_EVENTGRID_MESSAGE = "Invalid Event Grid Message";
private static final String SUBSCRIPTION_ID = "Aeg-Subscription-Name";
private static final String EVENTGRID_VALIDATION_EVENT = "Microsoft.EventGrid.SubscriptionValidationEvent";
private static final Gson GSON = new Gson();
private static final ObjectMapper objectMapper = new ObjectMapper();
private JsonObject root = null;
private HttpServletRequest httpServletRequest;
private JaxRsDpsLog log;
private NotificationRequest notificationRequest;
private NotificationData notificationData;
private HandshakeRequestData handshakeRequestData;
private boolean isHandshakeRequest;
@Autowired
public EventGridRequestBodyExtractor(HttpServletRequest httpServletRequest, JaxRsDpsLog log) {
this.httpServletRequest = httpServletRequest;
this.log = log;
this.notificationRequest = extractNotificationRequestFromHttpRequest();
}
/**
* Extracts the attributes from the request that are filled in by publisher of the message.
*
* @throws AppException
* @return Request Attributes Map
*/
public Map<String, String> extractAttributesFromRequestBody() {
if(isHandshakeRequest) {
return null;
}
return this.notificationData.getAttributes();
}
/**
* Extracts the data from the request that are filled in by publisher of the message,
*
* @throws AppException
* @return Request Data String
*/
public String extractDataFromRequestBody() {
if(isHandshakeRequest) {
return null;
}
return new String(Base64.getDecoder().decode(notificationData.getData()));
}
/**
* Extracts the notificationId from the request that are filled in by EventGrid.
*
* @throws AppException
* @return Request NotificationId String.
*/
public String extractNotificationIdFromRequestBody() {
String subscriptionId = httpServletRequest.getHeader(SUBSCRIPTION_ID);
if (Strings.isNullOrEmpty(subscriptionId)) {
throw new AppException(HttpStatus.BAD_REQUEST.value(), "Invalid Event Grid Message", "Subscription ID not found");
}
return subscriptionId;
}
/**
* Checks if the request is for handshake.
*
* @throws AppException
* @return Request Type Boolean
*/
public boolean isHandshakeRequest() {
return this.isHandshakeRequest ;
}
/**
* Return ValidationCode
*
* @throws AppException
* @return Request Type Boolean
*/
public String getValidationCodeForHandshake() {
if(!isHandshakeRequest) {
return null;
}
return this.handshakeRequestData.getValidationCode();
}
/**
* Utility method that extracts the core content in the request.
* This is what the publisher will publish.
*
* The attributes are validated.
*
* @throws AppException
* @return NotificationRequest Object
*/
private NotificationRequest extractNotificationRequestFromHttpRequest() {
NotificationRequest notificationRequest = null;
if (this.notificationRequest == null) {
try {
BufferedReader reader = httpServletRequest.getReader();
Stream<String> lines = reader.lines();
String requestBody = lines.collect(Collectors.joining("\n"));
NotificationRequest[] notificationRequestArray = GSON.fromJson(requestBody, NotificationRequest[].class);
notificationRequest = notificationRequestArray[0];
this.isHandshakeRequest = notificationRequest.getEventType().equals(EVENTGRID_VALIDATION_EVENT);
if (isHandshakeRequest) {
extractHandshakeData(notificationRequest);
} else {
extractNotificationData(notificationRequest);
}
} catch (Exception e) {
throw new AppException(HttpStatus.BAD_REQUEST.value(), "Request payload parsing error",
"Unable to parse request payload.", "Request contents are null or empty");
}
}
return notificationRequest;
}
private void extractHandshakeData(NotificationRequest notificationRequest) {
this.handshakeRequestData = GSON.fromJson(notificationRequest.getData(), HandshakeRequestData.class);
Preconditions.checkNotNull(this.handshakeRequestData.getValidationCode(), "Request payload parsing error" );
}
private void extractNotificationData(NotificationRequest notificationRequest) {
NotificationData notificationData = GSON.fromJson(notificationRequest.getData(), NotificationData.class);
verifyNotificationData(notificationData);
this.notificationData = notificationData;
}
// TODO: Clean up for using @NonNull Lombok.
// The Gson deserialisation happens through constructor made out of reflection,
// The unsafe initialisation inspires us to use Jackson
//
// Making Gson use a custom constructor, involves boilerplating, hence adding precondition checks. https://github.com/google/gson/blob/master/UserGuide.md#TOC-Writing-a-Deserializer
// The required condition is not operation as of now.https://github.com/google/gson/issues/61
private void verifyNotificationData(NotificationData notificationData) {
Preconditions.checkNotNull(notificationData, "Request payload parsing error" );
Preconditions.checkNotNull(notificationData.getData(), "Request payload parsing error" );
Preconditions.checkNotNull(notificationData.getMessageId(), "Request payload parsing error" );
Preconditions.checkNotNull(notificationData.getAttributes().get("correlation-id") , "Request payload parsing error" );
Preconditions.checkNotNull(notificationData.getAttributes().get("data-partition-id") , "Request payload parsing error" );
}
}
\ No newline at end of file
......@@ -15,10 +15,8 @@
package org.opengroup.osdu.notification.provider.azure.util;
import com.azure.cosmos.CosmosClient;
import com.azure.security.keyvault.secrets.SecretClient;
import com.azure.security.keyvault.secrets.models.KeyVaultSecret;
import org.opengroup.osdu.azure.KeyVaultFacade;
import org.opengroup.osdu.notification.provider.interfaces.IAppProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
......
// Copyright © Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package org.opengroup.osdu.notification.pubsub;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;
import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.notification.provider.azure.pubsub.EventGridHandshakeHandler;
import org.opengroup.osdu.notification.provider.azure.pubsub.EventGridRequestBodyExtractor;
import org.springframework.http.HttpStatus;
import java.io.IOException;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class EventGridHandshakeHandlerTest {
@Mock
EventGridRequestBodyExtractor eventGridRequestBodyExtractor;
@InjectMocks
@Spy
private EventGridHandshakeHandler sut;
@Test
public void should_returnValidResponse_getHandshakeResponse() throws IOException {
// Set up
when(sut.getHandshakeResponse()).thenReturn("testValidationCode");
String expectedResponse = "{\"ValidationResponse\":\"testValidationCode\"}";
// Act
String observedResponse = this.sut.getHandshakeResponse();
// Assert
Assert.assertEquals(observedResponse, expectedResponse);
}
@Test
public void should_throw_getHandshakeResponse() throws IOException {
// Set up
when(sut.getHandshakeResponse())
.thenThrow(new AppException(HttpStatus.BAD_REQUEST.value(), "Request payload parsing error", "" ));
try{
// Act
String observedResponse = this.sut.getHandshakeResponse();
// Assert
fail("Should Throw Exception");
} catch (AppException appException){
Assert.assertEquals(HttpStatus.BAD_REQUEST.value(), appException.getError().getCode());
Assert.assertEquals("Unable to parse request payload.", appException.getError().getMessage());
} catch (Exception exception) {
fail("Should Throw AppException");
}
}
}
// Copyright © Microsoft Corporation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package org.opengroup.osdu.notification.pubsub;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.opengroup.osdu.core.common.logging.JaxRsDpsLog;
import org.opengroup.osdu.core.common.model.http.AppException;
import org.opengroup.osdu.notification.provider.azure.pubsub.EventGridRequestBodyExtractor;
import org.springframework.http.HttpStatus;
import javax.servlet.http.HttpServletRequest;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.StringReader;
import java.util.Map;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class EventGridRequestBodyExtractorTest {
private EventGridRequestBodyExtractor sut;
@Mock
private HttpServletRequest httpServletRequest;
@Mock
private JaxRsDpsLog log;
@Test
public void should_returnTrue_isHandshakeRequest() throws IOException {
// Set up
String validHandshakeRequestRoot =
" [{\n" +
" \"id\": \"testId\",\n" +
" \"topic\": \"testTopic\",\n" +
" \"subject\": \"\",\n" +
" \"data\": {\n" +
" \"validationCode\": \"testValidationCode\",\n" +
" \"validationUrl\": \"testURL\"\n" +
" },\n" +
" \"eventType\": \"Microsoft.EventGrid.SubscriptionValidationEvent\",\n" +
" \"eventTime\": \"2020-08-14T11:18:55.9278057Z\",\n" +
" \"metadataVersion\": \"1\",\n" +
" \"dataVersion\": \"2\"\n" +
" }]";
BufferedReader reader = new BufferedReader(new StringReader(validHandshakeRequestRoot));
when(this.httpServletRequest.getReader()).thenReturn(reader);
sut = new EventGridRequestBodyExtractor(httpServletRequest, log);
// Act
boolean response = this.sut.isHandshakeRequest();
// Assert
Assert.assertTrue(response);
}
@Test
public void shouldThrow_whenRequestTypeIsNotright_isHandshakeRequest() throws IOException {
//SetUp
String invalidHandshakeRequestRoot =
" [{\n" +
" \"id\": \"testId\",\n" +
" \"topic\": \"testTopic\",\n" +
" \"subject\": \"\",\n" +
" \"data\": {\n" +
" \"validationCode\": \"testValidationCode\",\n" +
" \"validationUrl\": \"testURL\"\n" +
" },\n" +
" \"eventType\": \"SubscriptionValidationEvent\",\n" +
" \"eventTime\": \"2020-08-14T11:18:55.9278057Z\",\n" +
" \"metadataVersion\": \"1\",\n" +
" \"dataVersion\": \"2\"\n" +
" }]";
BufferedReader reader = new BufferedReader(new StringReader(invalidHandshakeRequestRoot));
when(this.httpServletRequest.getReader()).thenReturn(reader);
try{
// Act
sut = new EventGridRequestBodyExtractor(httpServletRequest, log);
// Assert
fail("Should Throw Exception");
} catch (AppException appException){
Assert.assertEquals(HttpStatus.BAD_REQUEST.value(), appException.getError().getCode());
Assert.assertEquals