@Test public void receiveMoneyWithSequenceTooHighForSafeEncryption() throws Exception { streamReceiver = new StatelessStreamReceiver( serverSecretSupplier, streamConnectionGenerator, streamEncryptionService, StreamCodecContextFactory.oer() ); final InterledgerAddress receiverAddress = InterledgerAddress.of("example.receiver"); final StreamConnectionDetails connectionDetails = streamConnectionGenerator .generateConnectionDetails(serverSecretSupplier, receiverAddress); final SharedSecret sharedSecret = SharedSecret.of(connectionDetails.sharedSecret().key()); final StreamPacket testStreamPacket = StreamPacket.builder() .interledgerPacketType(InterledgerPacketType.PREPARE) .prepareAmount(UnsignedLong.ZERO) .sequence(StreamConnection.MAX_FRAMES_PER_CONNECTION) .addFrames(StreamMoneyFrame.builder() .streamId(UnsignedLong.ONE) .shares(UnsignedLong.ONE) .build()) .build(); final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(testStreamPacket, baos); final byte[] encryptedStreamPacketBytes = streamEncryptionService.encrypt(sharedSecret, baos.toByteArray()); final InterledgerCondition executionCondition = StreamUtils .generatedFulfillableFulfillment(sharedSecret, encryptedStreamPacketBytes) .getCondition(); final InterledgerPreparePacket preparePacket = InterledgerPreparePacket.builder() .destination(connectionDetails.destinationAddress()) .amount(UnsignedLong.valueOf(100L)) .expiresAt(Instant.EPOCH) .data(encryptedStreamPacketBytes) .typedData(testStreamPacket) .executionCondition(executionCondition) .build(); this.streamReceiver.receiveMoney(preparePacket, receiverAddress, DENOMINATION).handle( fulfillPacket -> { assertThat(fulfillPacket.getFulfillment().getCondition()).isEqualTo(executionCondition); final byte[] streamPacketBytes = streamEncryptionService.decrypt(sharedSecret, fulfillPacket.getData()); try { final StreamPacket streamPacket = streamCodecContext .read(StreamPacket.class, new ByteArrayInputStream(streamPacketBytes)); assertThat(streamPacket.frames().get(0).streamFrameType()).isEqualTo(StreamFrameType.ConnectionClose); } catch (Exception e) { throw new RuntimeException(e.getMessage(), e); } }, rejectPacket -> { throw new RuntimeException("Should have fulfilled"); } ); }
@Override public InterledgerResponsePacket receiveMoney( final InterledgerPreparePacket preparePacket, final InterledgerAddress receiverAddress, final Denomination denomination ) { Objects.requireNonNull(preparePacket); Objects.requireNonNull(receiverAddress); final SharedSecret spspSharedSecret = this.streamConnectionGenerator .deriveSecretFromAddress(serverSecretSupplier, preparePacket.getDestination()); final SharedSecret streamSharedSecret = SharedSecret.of(spspSharedSecret.key()); final StreamPacket streamPacket; try { if (preparePacket.getData().length == 0) { return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("No STREAM packet bytes available to decrypt") .triggeredBy(receiverAddress) .build(); } final byte[] streamPacketBytes = streamEncryptionService.decrypt(streamSharedSecret, preparePacket.getData()); streamPacket = streamCodecContext.read(StreamPacket.class, new ByteArrayInputStream(streamPacketBytes)); } catch (Exception e) { logger.error( "Unable to decrypt packet. preparePacket={} receiverAddress={} error={}", preparePacket, receiverAddress, e ); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("Could not decrypt data") .triggeredBy(receiverAddress) .build(); } final Builder<StreamFrame> responseFrames = ImmutableList.builder(); if (streamPacket.sequenceIsSafeForSingleSharedSecret()) { streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.StreamMoney) .map($ -> (StreamMoneyFrame) $) .forEach(streamMoneyFrame -> responseFrames.add(StreamMoneyMaxFrame.builder() .streamId(streamMoneyFrame.streamId()) .totalReceived(UnsignedLong.ZERO) .receiveMax(UnsignedLong.MAX_VALUE) .build())); streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.ConnectionNewAddress) .findFirst() .map(streamFrame -> responseFrames.add(ConnectionAssetDetailsFrame.builder() .sourceDenomination(denomination) .build())); } else { logger.warn("This STREAM Connection's sequence {} was too high for safe encryption. CLOSING the stream!", streamPacket.sequence()); responseFrames.add(ConnectionCloseFrame.builder() .errorCode(ErrorCodes.ProtocolViolation) .errorMessage("Sequence number was to too high for safe encryption") .build()); } final InterledgerFulfillment fulfillment = StreamUtils .generatedFulfillableFulfillment(streamSharedSecret, preparePacket.getData()); final boolean isFulfillable = fulfillment.getCondition().equals(preparePacket.getExecutionCondition()); if (isFulfillable && is(preparePacket.getAmount()).greaterThanEqualTo(streamPacket.prepareAmount())) { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.FULFILL) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .sharedSecret(streamSharedSecret) .build(); logger.debug( "Fulfilling prepare packet. preparePacket={} fulfillment={} returnableStreamPacketResponse={}", preparePacket, fulfillment, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerFulfillPacket.builder() .fulfillment(fulfillment) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } else { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.REJECT) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .build(); if (isFulfillable) { logger.debug("Packet is unfulfillable. preparePacket={}", preparePacket); } else if (is(preparePacket.getAmount()).lessThan(streamPacket.prepareAmount())) { logger.debug( "Received only: {} when we should have received at least: {}", preparePacket.getAmount(), streamPacket.prepareAmount() ); } logger.debug( "Rejecting Prepare and including encrypted stream packet. preparePacket={} returnableStreamPacketResponse={}", preparePacket, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F99_APPLICATION_ERROR) .message("STREAM packet not fulfillable (prepare amount < stream packet amount)") .triggeredBy(receiverAddress) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } }
StatelessStreamReceiver implements StreamReceiver { @Override public InterledgerResponsePacket receiveMoney( final InterledgerPreparePacket preparePacket, final InterledgerAddress receiverAddress, final Denomination denomination ) { Objects.requireNonNull(preparePacket); Objects.requireNonNull(receiverAddress); final SharedSecret spspSharedSecret = this.streamConnectionGenerator .deriveSecretFromAddress(serverSecretSupplier, preparePacket.getDestination()); final SharedSecret streamSharedSecret = SharedSecret.of(spspSharedSecret.key()); final StreamPacket streamPacket; try { if (preparePacket.getData().length == 0) { return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("No STREAM packet bytes available to decrypt") .triggeredBy(receiverAddress) .build(); } final byte[] streamPacketBytes = streamEncryptionService.decrypt(streamSharedSecret, preparePacket.getData()); streamPacket = streamCodecContext.read(StreamPacket.class, new ByteArrayInputStream(streamPacketBytes)); } catch (Exception e) { logger.error( "Unable to decrypt packet. preparePacket={} receiverAddress={} error={}", preparePacket, receiverAddress, e ); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("Could not decrypt data") .triggeredBy(receiverAddress) .build(); } final Builder<StreamFrame> responseFrames = ImmutableList.builder(); if (streamPacket.sequenceIsSafeForSingleSharedSecret()) { streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.StreamMoney) .map($ -> (StreamMoneyFrame) $) .forEach(streamMoneyFrame -> responseFrames.add(StreamMoneyMaxFrame.builder() .streamId(streamMoneyFrame.streamId()) .totalReceived(UnsignedLong.ZERO) .receiveMax(UnsignedLong.MAX_VALUE) .build())); streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.ConnectionNewAddress) .findFirst() .map(streamFrame -> responseFrames.add(ConnectionAssetDetailsFrame.builder() .sourceDenomination(denomination) .build())); } else { logger.warn("This STREAM Connection's sequence {} was too high for safe encryption. CLOSING the stream!", streamPacket.sequence()); responseFrames.add(ConnectionCloseFrame.builder() .errorCode(ErrorCodes.ProtocolViolation) .errorMessage("Sequence number was to too high for safe encryption") .build()); } final InterledgerFulfillment fulfillment = StreamUtils .generatedFulfillableFulfillment(streamSharedSecret, preparePacket.getData()); final boolean isFulfillable = fulfillment.getCondition().equals(preparePacket.getExecutionCondition()); if (isFulfillable && is(preparePacket.getAmount()).greaterThanEqualTo(streamPacket.prepareAmount())) { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.FULFILL) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .sharedSecret(streamSharedSecret) .build(); logger.debug( "Fulfilling prepare packet. preparePacket={} fulfillment={} returnableStreamPacketResponse={}", preparePacket, fulfillment, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerFulfillPacket.builder() .fulfillment(fulfillment) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } else { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.REJECT) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .build(); if (isFulfillable) { logger.debug("Packet is unfulfillable. preparePacket={}", preparePacket); } else if (is(preparePacket.getAmount()).lessThan(streamPacket.prepareAmount())) { logger.debug( "Received only: {} when we should have received at least: {}", preparePacket.getAmount(), streamPacket.prepareAmount() ); } logger.debug( "Rejecting Prepare and including encrypted stream packet. preparePacket={} returnableStreamPacketResponse={}", preparePacket, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F99_APPLICATION_ERROR) .message("STREAM packet not fulfillable (prepare amount < stream packet amount)") .triggeredBy(receiverAddress) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } } }
StatelessStreamReceiver implements StreamReceiver { @Override public InterledgerResponsePacket receiveMoney( final InterledgerPreparePacket preparePacket, final InterledgerAddress receiverAddress, final Denomination denomination ) { Objects.requireNonNull(preparePacket); Objects.requireNonNull(receiverAddress); final SharedSecret spspSharedSecret = this.streamConnectionGenerator .deriveSecretFromAddress(serverSecretSupplier, preparePacket.getDestination()); final SharedSecret streamSharedSecret = SharedSecret.of(spspSharedSecret.key()); final StreamPacket streamPacket; try { if (preparePacket.getData().length == 0) { return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("No STREAM packet bytes available to decrypt") .triggeredBy(receiverAddress) .build(); } final byte[] streamPacketBytes = streamEncryptionService.decrypt(streamSharedSecret, preparePacket.getData()); streamPacket = streamCodecContext.read(StreamPacket.class, new ByteArrayInputStream(streamPacketBytes)); } catch (Exception e) { logger.error( "Unable to decrypt packet. preparePacket={} receiverAddress={} error={}", preparePacket, receiverAddress, e ); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("Could not decrypt data") .triggeredBy(receiverAddress) .build(); } final Builder<StreamFrame> responseFrames = ImmutableList.builder(); if (streamPacket.sequenceIsSafeForSingleSharedSecret()) { streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.StreamMoney) .map($ -> (StreamMoneyFrame) $) .forEach(streamMoneyFrame -> responseFrames.add(StreamMoneyMaxFrame.builder() .streamId(streamMoneyFrame.streamId()) .totalReceived(UnsignedLong.ZERO) .receiveMax(UnsignedLong.MAX_VALUE) .build())); streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.ConnectionNewAddress) .findFirst() .map(streamFrame -> responseFrames.add(ConnectionAssetDetailsFrame.builder() .sourceDenomination(denomination) .build())); } else { logger.warn("This STREAM Connection's sequence {} was too high for safe encryption. CLOSING the stream!", streamPacket.sequence()); responseFrames.add(ConnectionCloseFrame.builder() .errorCode(ErrorCodes.ProtocolViolation) .errorMessage("Sequence number was to too high for safe encryption") .build()); } final InterledgerFulfillment fulfillment = StreamUtils .generatedFulfillableFulfillment(streamSharedSecret, preparePacket.getData()); final boolean isFulfillable = fulfillment.getCondition().equals(preparePacket.getExecutionCondition()); if (isFulfillable && is(preparePacket.getAmount()).greaterThanEqualTo(streamPacket.prepareAmount())) { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.FULFILL) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .sharedSecret(streamSharedSecret) .build(); logger.debug( "Fulfilling prepare packet. preparePacket={} fulfillment={} returnableStreamPacketResponse={}", preparePacket, fulfillment, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerFulfillPacket.builder() .fulfillment(fulfillment) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } else { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.REJECT) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .build(); if (isFulfillable) { logger.debug("Packet is unfulfillable. preparePacket={}", preparePacket); } else if (is(preparePacket.getAmount()).lessThan(streamPacket.prepareAmount())) { logger.debug( "Received only: {} when we should have received at least: {}", preparePacket.getAmount(), streamPacket.prepareAmount() ); } logger.debug( "Rejecting Prepare and including encrypted stream packet. preparePacket={} returnableStreamPacketResponse={}", preparePacket, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F99_APPLICATION_ERROR) .message("STREAM packet not fulfillable (prepare amount < stream packet amount)") .triggeredBy(receiverAddress) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } } StatelessStreamReceiver(
      final ServerSecretSupplier serverSecretSupplier, final StreamConnectionGenerator streamConnectionGenerator,
      final StreamEncryptionService streamEncryptionService, final CodecContext streamCodecContext
  ); }
StatelessStreamReceiver implements StreamReceiver { @Override public InterledgerResponsePacket receiveMoney( final InterledgerPreparePacket preparePacket, final InterledgerAddress receiverAddress, final Denomination denomination ) { Objects.requireNonNull(preparePacket); Objects.requireNonNull(receiverAddress); final SharedSecret spspSharedSecret = this.streamConnectionGenerator .deriveSecretFromAddress(serverSecretSupplier, preparePacket.getDestination()); final SharedSecret streamSharedSecret = SharedSecret.of(spspSharedSecret.key()); final StreamPacket streamPacket; try { if (preparePacket.getData().length == 0) { return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("No STREAM packet bytes available to decrypt") .triggeredBy(receiverAddress) .build(); } final byte[] streamPacketBytes = streamEncryptionService.decrypt(streamSharedSecret, preparePacket.getData()); streamPacket = streamCodecContext.read(StreamPacket.class, new ByteArrayInputStream(streamPacketBytes)); } catch (Exception e) { logger.error( "Unable to decrypt packet. preparePacket={} receiverAddress={} error={}", preparePacket, receiverAddress, e ); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("Could not decrypt data") .triggeredBy(receiverAddress) .build(); } final Builder<StreamFrame> responseFrames = ImmutableList.builder(); if (streamPacket.sequenceIsSafeForSingleSharedSecret()) { streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.StreamMoney) .map($ -> (StreamMoneyFrame) $) .forEach(streamMoneyFrame -> responseFrames.add(StreamMoneyMaxFrame.builder() .streamId(streamMoneyFrame.streamId()) .totalReceived(UnsignedLong.ZERO) .receiveMax(UnsignedLong.MAX_VALUE) .build())); streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.ConnectionNewAddress) .findFirst() .map(streamFrame -> responseFrames.add(ConnectionAssetDetailsFrame.builder() .sourceDenomination(denomination) .build())); } else { logger.warn("This STREAM Connection's sequence {} was too high for safe encryption. CLOSING the stream!", streamPacket.sequence()); responseFrames.add(ConnectionCloseFrame.builder() .errorCode(ErrorCodes.ProtocolViolation) .errorMessage("Sequence number was to too high for safe encryption") .build()); } final InterledgerFulfillment fulfillment = StreamUtils .generatedFulfillableFulfillment(streamSharedSecret, preparePacket.getData()); final boolean isFulfillable = fulfillment.getCondition().equals(preparePacket.getExecutionCondition()); if (isFulfillable && is(preparePacket.getAmount()).greaterThanEqualTo(streamPacket.prepareAmount())) { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.FULFILL) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .sharedSecret(streamSharedSecret) .build(); logger.debug( "Fulfilling prepare packet. preparePacket={} fulfillment={} returnableStreamPacketResponse={}", preparePacket, fulfillment, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerFulfillPacket.builder() .fulfillment(fulfillment) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } else { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.REJECT) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .build(); if (isFulfillable) { logger.debug("Packet is unfulfillable. preparePacket={}", preparePacket); } else if (is(preparePacket.getAmount()).lessThan(streamPacket.prepareAmount())) { logger.debug( "Received only: {} when we should have received at least: {}", preparePacket.getAmount(), streamPacket.prepareAmount() ); } logger.debug( "Rejecting Prepare and including encrypted stream packet. preparePacket={} returnableStreamPacketResponse={}", preparePacket, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F99_APPLICATION_ERROR) .message("STREAM packet not fulfillable (prepare amount < stream packet amount)") .triggeredBy(receiverAddress) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } } StatelessStreamReceiver(
      final ServerSecretSupplier serverSecretSupplier, final StreamConnectionGenerator streamConnectionGenerator,
      final StreamEncryptionService streamEncryptionService, final CodecContext streamCodecContext
  ); @Override StreamConnectionDetails setupStream(final InterledgerAddress receiverAddress); @Override InterledgerResponsePacket receiveMoney(
      final InterledgerPreparePacket preparePacket, final InterledgerAddress receiverAddress,
      final Denomination denomination
  ); }
StatelessStreamReceiver implements StreamReceiver { @Override public InterledgerResponsePacket receiveMoney( final InterledgerPreparePacket preparePacket, final InterledgerAddress receiverAddress, final Denomination denomination ) { Objects.requireNonNull(preparePacket); Objects.requireNonNull(receiverAddress); final SharedSecret spspSharedSecret = this.streamConnectionGenerator .deriveSecretFromAddress(serverSecretSupplier, preparePacket.getDestination()); final SharedSecret streamSharedSecret = SharedSecret.of(spspSharedSecret.key()); final StreamPacket streamPacket; try { if (preparePacket.getData().length == 0) { return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("No STREAM packet bytes available to decrypt") .triggeredBy(receiverAddress) .build(); } final byte[] streamPacketBytes = streamEncryptionService.decrypt(streamSharedSecret, preparePacket.getData()); streamPacket = streamCodecContext.read(StreamPacket.class, new ByteArrayInputStream(streamPacketBytes)); } catch (Exception e) { logger.error( "Unable to decrypt packet. preparePacket={} receiverAddress={} error={}", preparePacket, receiverAddress, e ); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F06_UNEXPECTED_PAYMENT) .message("Could not decrypt data") .triggeredBy(receiverAddress) .build(); } final Builder<StreamFrame> responseFrames = ImmutableList.builder(); if (streamPacket.sequenceIsSafeForSingleSharedSecret()) { streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.StreamMoney) .map($ -> (StreamMoneyFrame) $) .forEach(streamMoneyFrame -> responseFrames.add(StreamMoneyMaxFrame.builder() .streamId(streamMoneyFrame.streamId()) .totalReceived(UnsignedLong.ZERO) .receiveMax(UnsignedLong.MAX_VALUE) .build())); streamPacket.frames().stream() .filter(streamFrame -> streamFrame.streamFrameType() == StreamFrameType.ConnectionNewAddress) .findFirst() .map(streamFrame -> responseFrames.add(ConnectionAssetDetailsFrame.builder() .sourceDenomination(denomination) .build())); } else { logger.warn("This STREAM Connection's sequence {} was too high for safe encryption. CLOSING the stream!", streamPacket.sequence()); responseFrames.add(ConnectionCloseFrame.builder() .errorCode(ErrorCodes.ProtocolViolation) .errorMessage("Sequence number was to too high for safe encryption") .build()); } final InterledgerFulfillment fulfillment = StreamUtils .generatedFulfillableFulfillment(streamSharedSecret, preparePacket.getData()); final boolean isFulfillable = fulfillment.getCondition().equals(preparePacket.getExecutionCondition()); if (isFulfillable && is(preparePacket.getAmount()).greaterThanEqualTo(streamPacket.prepareAmount())) { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.FULFILL) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .sharedSecret(streamSharedSecret) .build(); logger.debug( "Fulfilling prepare packet. preparePacket={} fulfillment={} returnableStreamPacketResponse={}", preparePacket, fulfillment, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerFulfillPacket.builder() .fulfillment(fulfillment) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } else { final StreamPacket returnableStreamPacketResponse = StreamPacket.builder() .sequence(streamPacket.sequence()) .interledgerPacketType(InterledgerPacketType.REJECT) .prepareAmount(preparePacket.getAmount()) .frames(responseFrames.build()) .build(); if (isFulfillable) { logger.debug("Packet is unfulfillable. preparePacket={}", preparePacket); } else if (is(preparePacket.getAmount()).lessThan(streamPacket.prepareAmount())) { logger.debug( "Received only: {} when we should have received at least: {}", preparePacket.getAmount(), streamPacket.prepareAmount() ); } logger.debug( "Rejecting Prepare and including encrypted stream packet. preparePacket={} returnableStreamPacketResponse={}", preparePacket, returnableStreamPacketResponse ); try { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); streamCodecContext.write(returnableStreamPacketResponse, baos); final byte[] returnableStreamPacketBytes = baos.toByteArray(); final byte[] encryptedReturnableStreamPacketBytes = streamEncryptionService .encrypt(streamSharedSecret, returnableStreamPacketBytes); return InterledgerRejectPacket.builder() .code(InterledgerErrorCode.F99_APPLICATION_ERROR) .message("STREAM packet not fulfillable (prepare amount < stream packet amount)") .triggeredBy(receiverAddress) .data(encryptedReturnableStreamPacketBytes) .typedData(returnableStreamPacketResponse) .build(); } catch (IOException e) { throw new StreamException(e.getMessage(), e); } } } StatelessStreamReceiver(
      final ServerSecretSupplier serverSecretSupplier, final StreamConnectionGenerator streamConnectionGenerator,
      final StreamEncryptionService streamEncryptionService, final CodecContext streamCodecContext
  ); @Override StreamConnectionDetails setupStream(final InterledgerAddress receiverAddress); @Override InterledgerResponsePacket receiveMoney(
      final InterledgerPreparePacket preparePacket, final InterledgerAddress receiverAddress,
      final Denomination denomination
  );  }