Masstransit: Quartz integration does not work with bus encryption

[Ashtonian]

Hello!

Working on a project with MT 3 and the quartz scheduler, we ran into an issue using the quartz integration piece with encryption.

MassTransit.Pipeline.Filters.RescueReceiveContextFilter<MassTransit.ReceiveContext> Error: 0 : Rescuing exception, System.InvalidOperationException: Only JSON and XML messages can be scheduled 
   at MassTransit.QuartzIntegration.ScheduleMessageConsumer.<CreateJobDetail>d__6.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at System.Runtime.CompilerServices.ConfiguredTaskAwaitable`1.ConfiguredTaskAwaiter.GetResult() 
   at MassTransit.QuartzIntegration.ScheduleMessageConsumer.<Consume>d__3.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at MassTransit.Pipeline.ConsumerFactories.DelegateConsumerFactory`1.<Send>d__2`1.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at MassTransit.Pipeline.Filters.ConsumerMessageFilter`2.<MassTransit-Pipeline-IFilter<MassTransit-ConsumeContext<TMessage>>-Send>d__4.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw() 
   at MassTransit.Pipeline.Filters.ConsumerMessageFilter`2.<MassTransit-Pipeline-IFilter<MassTransit-ConsumeContext<TMessage>>-Send>d__4.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at MassTransit.Pipeline.Filters.TeeConsumeFilter`1.<>c__DisplayClass7_0.<<Send>b__0>d.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at 

System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at MassTransit.Pipeline.Filters.TeeConsumeFilter`1.<Send>d__7.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at MassTransit.Pipeline.Filters.MessageConsumeFilter`1.<MassTransit-Pipeline-IFilter<MassTransit-ConsumeContext>-Send>d__7.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw() 
   at MassTransit.Pipeline.Filters.MessageConsumeFilter`1.<MassTransit-Pipeline-IFilter<MassTransit-ConsumeContext>-Send>d__7.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at MassTransit.Pipeline.Filters.DeserializeFilter.<Send>d__4.MoveNext() 
--- End of stack trace from previous location where exception was thrown --- 
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task) 
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) 
   at MassTransit.Pipeline.Filters.RescueReceiveContextFilter`1.<MassTransit-Pipeline-IFilter<MassTransit-ReceiveContext>-Send>d__5.MoveNext()

after looking through the repoI think it's being thrown here line 55? after looking at the content-type before publish is "application/vnd.masstransit+aes"

[phatboyg]

Are you configuring the quartz service with support for the AES encrypted messages? It would need to be configured the same. If you are, this seems surprising that it's not working since it's a transparent layer underneath it.

[Ashtonian]

Thanks for the quick response!!

Are you configuring the quartz service with support for the AES encrypted messages?

Do you mean configuring it with a bus that has UseEncryptedSerializer() set?

Here is our configuration

return Bus.Factory.CreateUsingInMemory( x =>
            {
                x.SetTransportProvider( transportCache );
                x.UseEncryptedSerializer( container.GetInstance< ICryptoStreamProvider >() );
                x.ReceiveEndpoint( "SomeEndpoint",
                                   e =>
                                   {
                                       e.StateMachineSaga( container.GetInstance< SomeStateMachine >(), container.GetInstance< ISagaRepository< SomeCycle > >() );

                                       //publish consumers
                                       e.LoadFrom( container );
                                   } );

                x.ReceiveEndpoint( "Quartz",
                                   e =>
                                   {
                                       e.Consumer( () => new ScheduleMessageConsumer( container.GetInstance< IScheduler >() ) );
                                       e.Consumer( () => new CancelScheduledMessageConsumer( container.GetInstance< IScheduler >() ) );
                                   } );
            } );

Later:

            var busControl = someContainer.GetInstance< IBusControl >();


            var quartzScheduler = CreateQuartzScheduler( databaseConnectionString, busControl );
            quartzScheduler.Start();

We are using an in memory bus for integration testing with the folowing packages from MT for your reference.

  <package id="MassTransit" version="3.2.0" targetFramework="net451" />
  <package id="MassTransit.Automatonymous" version="3.2.0" targetFramework="net451" />
  <package id="MassTransit.Quartz" version="3.2.0" targetFramework="net451" />
  <package id="MassTransit.SimpleInjector" version="3.2.0" targetFramework="net451" />

[phatboyg]

There is a nice UseInMemoryScheduler() extension method that wires this up for you.

That said it seems like something may not be working with encryption so I'll see what's happening. FWIW I'm not sure the message will be encrypted in the Quartz database unless you enable data encryption in the database itself.

[Ashtonian]

will have to check out the UseInMemoryScheduler()!

You are correct on the DB encryption, this issue came out because of the encryption requirement leaking via the same bus, we don't even need the quarts messages (currently) to be encrypted. We are planning bypassing this via either a separate bus or by using quarts independently temporarily.

Thanks for looking at this!

[phatboyg]

So yeah, getting the scheduler to work with encrypted messages is going to be complicated, since encrypted messages are binary and we are using a string column in SQL to store the message. There is also some JSON token manipulation happening under the hood to include the original data in the message.

I'm not saying it's impossible, but I spent an hour or two trying to get it to work and realized that the binary encrypted body is a hard nut to crack.

[kljuco]

Any new info on this? Is there a proper workaround? We are sending sensitive information over the bus and this actually is stoping us from using encryption.
Thank you