Reject messages with payload size > 1MiB

- Update validation in WireFrameDecoder class
- Write unit and component tests for that case

Closes ONAP-340

Change-Id: I68cb608fd76118719b12a83de1ef930160f8a162
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
Issue-ID: DCAEGEN2-601

2 files changed, 47 insertions, 7 deletions

diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt
index 49eeddaa..d917c71a 100644
--- a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/VesHvSpecification.kt
@@ -57,9 +57,11 @@ object VesHvSpecification : Spek({
             val validMessage = vesMessage(Domain.HVRANMEAS)
             val msgWithInvalidDomain = vesMessage(Domain.OTHER)
             val msgWithInvalidFrame = invalidWireFrame()
+            val msgWithTooBigPayload = vesMessageWithTooBigPayload(Domain.HVRANMEAS)
             val expectedRefCnt = 0
 
-            val handledEvents = sut.handleConnection(sink, validMessage, msgWithInvalidDomain, msgWithInvalidFrame)
+            val handledEvents = sut.handleConnection(
+                    sink, validMessage, msgWithInvalidDomain, msgWithInvalidFrame, msgWithTooBigPayload)
 
             assertThat(handledEvents).hasSize(1)
 
@@ -72,6 +74,9 @@ object VesHvSpecification : Spek({
             assertThat(msgWithInvalidFrame.refCnt())
                     .describedAs("message with invalid frame should be released")
                     .isEqualTo(expectedRefCnt)
+            assertThat(msgWithTooBigPayload.refCnt())
+                    .describedAs("message with payload exceeding 1MiB should be released")
+                    .isEqualTo(expectedRefCnt)
 
         }
 
@@ -148,4 +153,20 @@ object VesHvSpecification : Spek({
             assertThat(msg.message.header.eventId).describedAs("routed message eventId").isEqualTo("second")
         }
     }
+
+    describe("request validation") {
+        it("should reject message with payload greater than 1 MiB and all subsequent messages") {
+            val sink = StoringSink()
+            val sut = Sut(sink)
+            sut.configurationProvider.updateConfiguration(basicConfiguration)
+
+            val handledMessages = sut.handleConnection(sink,
+                    vesMessage(Domain.HVRANMEAS, "first"),
+                    vesMessageWithTooBigPayload(Domain.HVRANMEAS, "second"),
+                    vesMessage(Domain.HVRANMEAS, "third"))
+
+            assertThat(handledMessages).hasSize(1)
+            assertThat(handledMessages.first().message.header.eventId).isEqualTo("first")
+        }
+    }
 })
diff --git a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/messages.kt b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/messages.kt
index 8895d642..e620e6b9 100644
--- a/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/messages.kt
+++ b/hv-collector-ct/src/test/kotlin/org/onap/dcae/collectors/veshv/tests/component/messages.kt
@@ -20,8 +20,12 @@
 package org.onap.dcae.collectors.veshv.tests.component
 
 import com.google.protobuf.ByteString
+import io.netty.buffer.ByteBuf
 import io.netty.buffer.ByteBufAllocator
 import io.netty.buffer.PooledByteBufAllocator
+import org.onap.dcae.collectors.veshv.domain.*
+import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder.Companion.MAX_PAYLOAD_SIZE
+import org.onap.ves.HVRanMeasFieldsV5
 import org.onap.ves.VesEventV5.VesEvent
 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader
 import org.onap.ves.VesEventV5.VesEvent.CommonEventHeader.Domain
@@ -29,7 +33,7 @@ import java.util.*
 
 val allocator: ByteBufAllocator = PooledByteBufAllocator.DEFAULT
 
-fun vesMessage(domain: Domain = Domain.OTHER, id: String = UUID.randomUUID().toString()) = allocator.buffer().run {
+fun vesMessage(domain: Domain = Domain.OTHER, id: String = UUID.randomUUID().toString()): ByteBuf = allocator.buffer().run {
     writeByte(0xFF) // always 0xFF
     writeByte(0x01)   // version
     writeByte(0x01)   // content type = GPB
@@ -40,7 +44,7 @@ fun vesMessage(domain: Domain = Domain.OTHER, id: String = UUID.randomUUID().toS
 }
 
 
-fun invalidVesMessage() = allocator.buffer().run {
+fun invalidVesMessage(): ByteBuf = allocator.buffer().run {
     writeByte(0xFF) // always 0xFF
     writeByte(0x01)   // version
     writeByte(0x01)   // content type = GPB
@@ -51,17 +55,32 @@ fun invalidVesMessage() = allocator.buffer().run {
 
 }
 
-fun garbageFrame() = allocator.buffer().run {
+fun garbageFrame(): ByteBuf = allocator.buffer().run {
     writeBytes("the meaning of life is &@)(*_!".toByteArray())
 }
 
-fun invalidWireFrame() = allocator.buffer().run {
+fun invalidWireFrame(): ByteBuf = allocator.buffer().run {
     writeByte(0xFF)
     writeByte(0x01)   // version
     writeByte(0x01)   // content type = GPB
 }
 
-fun vesEvent(domain: Domain = Domain.HVRANMEAS, id: String = UUID.randomUUID().toString()) =
+fun vesMessageWithTooBigPayload(domain: Domain = Domain.OTHER, id: String = UUID.randomUUID().toString()): ByteBuf = allocator.buffer().run {
+    writeByte(0xFF) // always 0xFF
+    writeByte(0x01)   // version
+    writeByte(0x01)   // content type = GPB
+
+    val gpb = vesEvent(
+            domain,
+            id,
+            ByteString.copyFrom(ByteArray(MAX_PAYLOAD_SIZE))
+    ).toByteString().asReadOnlyByteBuffer()
+
+    writeInt(gpb.limit())  // ves event size in bytes
+    writeBytes(gpb)  // ves event as GPB bytes
+}
+
+fun vesEvent(domain: Domain = Domain.HVRANMEAS, id: String = UUID.randomUUID().toString(), hvRanMeasFields: ByteString = ByteString.EMPTY) =
         VesEvent.newBuilder()
                 .setCommonEventHeader(
                         CommonEventHeader.getDefaultInstance().toBuilder()
@@ -76,5 +95,5 @@ fun vesEvent(domain: Domain = Domain.HVRANMEAS, id: String = UUID.randomUUID().t
                                 .setStartEpochMicrosec(120034455)
                                 .setLastEpochMicrosec(120034459)
                                 .setSequence(1))
-                .setHvRanMeasFields(ByteString.EMPTY)
+                .setHvRanMeasFields(hvRanMeasFields)
                 .build()