MessageProcessor.cc

/*
 * Copyright 2003 Michael A. Marsh, Cornell University. All rights reserved.
 * This software is released under the modified BSD license.
 * See the file LICENSE in the top-level directory for details.
 */
//
// $Id: MessageProcessor.cc,v 1.3 2004/05/19 15:56:24 mmarsh Exp $
//
// $Log: MessageProcessor.cc,v $
// Revision 1.3  2004/05/19 15:56:24  mmarsh
// Added copyright and license statements.
//
// Revision 1.2  2003/11/04 22:31:45  mmarsh
// *** empty log message ***
//
//

#include "MessageProcessor.h"
#include "InitActivity.h"
#include "InitEvent.h"
#include "StateInfo.h"
#include "CODEX_Server/ServerResponseEvent.h"
#include "FailureEvent.h"

#include "timing.h"

using namespace CODEX_APSS;

MessageProcessor::MessageProcessor(
   CODEX_Events::DeadPileType& deadPile,
   CODEX_Events::QType& eventQueue,
   CODEX_Server::ServerResponseHandler* destination,
   CODEX_Server::UnicastRequestHandler* requestHandler ) :
   CODEX_Events::Activity( deadPile, eventQueue ),
   m_destination( destination ),
   m_requestHandler( requestHandler )
{
}

MessageProcessor::~MessageProcessor()
{
}

bool
MessageProcessor::handler( RoutedMessageEvent< SignedInitMsg >& event )
{
#ifdef TIMING
   ActiveTimer.start();
#endif
   StateInfo* stateInfo = StateInfo::instance();
   CODEX_Server::ServerState* serverState =
      CODEX_Server::ServerState::instance();
   const SignedInitMsg& signedMessage = event.message();
   const InitMsg& message = signedMessage.message();
   unsigned int coordinator = message.coordinator().value();

   // Get the shares for this label.
   ShareSetType shares;
   try
   {
      const LSType& labeledShares =
         stateInfo->sharing( message.label(), coordinator );
      shares.addShare( labeledShares.share() );
   }
   catch ( ... )
   {
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }

   // Get the list of subsharings for this label.
   SubshareList& sList = stateInfo->splittings( message.label() );

   const unsigned int NumShares = ShareType::NumShares;

   // If there is not a subsharing for each share, we must generate
   // new subsharings.
   typedef CODEX_VSS::ShareSplitting< ShareType >  SplitType;
   unsigned int num = message.label().num().value();
   if ( sList.size() != NumShares )
   {
      sList.clear();
      sList.resize( NumShares );
      const CODEX_VSS::Range& range = stateInfo->subshareRange(num);
      OneWay oneWay( stateInfo->witness(num).args() );
      for ( unsigned int i = 0 ; i < NumShares ; ++i )
      {
         // We only care about shares that we have _now_.  No recovery
         // is performed.
         if ( shares(i).initialized() )
         {
            ShareSetType sSet;
            SplitType::split( shares(i).value(), sSet, range );
            SublabelType sLabel( message.label(), i, sSet, oneWay );
            sList[i] = LabeledSet( sSet, sLabel );
         }
      }
   }

   // Create the Activity that will coordinate subshare distribution.
   // We don't need to keep the pointer anywhere, since it's self-closing.
   InitActivity* ia = new InitActivity( m_deadPile,
                                        m_queue,
                                        message.version().value(),
                                        coordinator,
                                        serverState->hostNum(),
                                        num,
                                        event.seqNum(),
                                        sList,
                                        m_destination,
                                        m_requestHandler );

   // Send the event.
   sendEvent( new InitEvent( this, ia ), event.source() );
#ifdef TIMING
   ActiveTimer.stop();
#endif
   return true;
}

bool
MessageProcessor::handler( RoutedMessageEvent< SignedEstablishMsg >& event )
{
#ifdef TIMING
   ActiveTimer.start();
#endif
   StateInfo* stateInfo = StateInfo::instance();
   CODEX_Server::ServerState* serverState =
      CODEX_Server::ServerState::instance();
   const SignedEstablishMsg& signedMessage = event.message();
   const EstablishMsg& message = signedMessage.message();
   unsigned int establisher = message.establisher().value();

   BIGNUM * digest = 0;
   try
   {
      // The subshares are stored when the message is stored in the
      // verifier.  We just need to compose the response, cache it,
      // and respond.
      EstablishedMsg response( message );

      const CODEX_Ciphers::HashFunction& hashFunc = serverState->hashFunc();
      digest = response.digest( hashFunc );
      const CODEX_Ciphers::RSAPrivateKey& key = serverState->privateKey();
      CODEX_Ciphers::RSASignature* signature = key.sign( digest );
      BN_free(digest);
      digest = 0;
      if ( 0 == signature )
      {
         sendEvent( 0, event.source(), true ); // NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }
      SignedEstablishedMsg signedResponse( response, *signature );
      delete signature;

      stateInfo->addEstablishResponse( signedResponse );

      // Send the response.
      CODEX_Quorum::Message m;
      unsigned char server[CODEX_Server::ServerState::nSID];
      memcpy( server, event.server(), CODEX_Server::ServerState::nSID );
      // Make sure the "outgoing" mask isn't set.
      server[0] &= ~(CODEX_Server::ServerState::OutgoingMask);
      int serverNum = server[0];

      // Fill message headers.
      m.fill( server, CODEX_Server::ServerState::nSID );
      m.fill( event.seqNum(), CODEX_Server::ServerState::nMID );
      m.fill( stateInfo->domain() );
      m.fill( kEstablishedMsg | SignatureMask );
      int length = signedResponse.marshal(0);
      unsigned char* buffer = new unsigned char[length];
      unsigned char* pBuffer = buffer;
      signedResponse.marshal(&pBuffer);
      m.fill( buffer, length );
      delete [] buffer;

      sendEvent( new CODEX_Server::ServerResponseEvent( this,
                                                        m_destination,
                                                        m,
                                                        serverNum ),
                 event.source() );
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

bool
MessageProcessor::handler( RoutedMessageEvent< SignedComputeMsg >& event )
{
#ifdef TIMING
   ActiveTimer.start();
#endif
   StateInfo* stateInfo = StateInfo::instance();
   CODEX_Server::ServerState* serverState =
      CODEX_Server::ServerState::instance();
   const SignedComputeMsg& signedMessage = event.message();
   const ComputeMsg& message = signedMessage.message();
   unsigned int coordinator = message.coordinator().value();
   unsigned int version =
      message.subshareLabel(0).label().version().value() + 1;
   unsigned int num = message.subshareLabel(0).label().num().value();
   unsigned int self = serverState->hostNum();

   // Sanity check -- have we moved past this version already?
   if ( version <= stateInfo->version(num) )
   {
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }

   // Check for missing subshares.  Initiate subshare recovery if necessary.
   //   This will return false, not true, since the request still needs
   //   to be handled.
   const ShareSetType* subsharesets[ ShareType::NumShares ];
   bool needRecovery = false;
   for ( unsigned int i = 0 ; i < ShareType::NumShares ; ++i )
   {
      try
      {
         subsharesets[i] =
            &(stateInfo->subsharing( message.subshareLabel(i), coordinator ));
         for ( unsigned int j = 0 ; j < ShareType::NumShares ; ++j )
         {
            if ( ( ! (*subsharesets[i])(j).initialized() ) &&
                 subsharesets[i]->shouldHave( j, self ) )
            {
               // At this point we know that recovery must be performed for
               // subshare label i.
               needRecovery = true;
               stateInfo->recover( message.subshareLabel(i) );
               break;
            }
         }
      }
      catch ( CODEX_Server::KeySharesNotFoundException& )
      {
         needRecovery = true;
         stateInfo->recover( message.subshareLabel(i) );
      }
   }
   if ( needRecovery )
   {
      // At least one subsharing is being recovered, so stop processing
      // for now.
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return false;
   }

   // Compute the new shares and label
   ShareSetType newShareSet;
   LabelType::VType::ValueType labelVals[ ShareType::NumShares ];
   OneWay oneWay( stateInfo->witness(num).args() );
   LabelType::VType vr( oneWay );
   for ( unsigned int i = 0 ; i < ShareType::NumShares ; ++i )
   {
      // Compute the share
      if ( newShareSet.shouldHave( i, self ) )
      {
         // We have already ensured that if a shareset *should* have i,
         // then it *does*.
         ShareSetType shareSet;
         for ( unsigned int j = 0 ; j < ShareType::NumShares ; ++j )
         {
            shareSet.setShare( j, (*subsharesets[j])(i) );
         }
         // Compute this share
         ShareType::ValueType shareVal;
         shareSet.recover( shareVal );
         newShareSet.setShare( i, shareVal );
      }

      // Compute the label
      LabelType::VType::ValueType sublabelVals[ ShareType::NumShares ];
      for ( unsigned int j = 0 ; j < ShareType::NumShares ; ++j )
      {
         sublabelVals[j] = message.subshareLabel(j).vc(i);
      }
      vr( sublabelVals, labelVals[i] );
   }
   ShareType newShare( newShareSet, self );
   LabelType label( num, version, coordinator, labelVals );

   // Make sure the label is valid.
   if ( ! label.verify( stateInfo->witness(num).witness(), oneWay ) )
   {
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
   if ( ! label.check( newShare, oneWay ) )
   {
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }

   LSType* newLabeledShare = new LSType( newShare, label );
   stateInfo->addSharing( newLabeledShare );

   // Compose the ComputedMsg response
   ComputedMsg response( label, CODEX_ASN1::Integer(serverState->hostNum()) );

   // Sign the response
   BIGNUM * digest = 0;
   try
   {
      const CODEX_Ciphers::HashFunction& hashFunc = serverState->hashFunc();
      digest = response.digest( hashFunc );
      const CODEX_Ciphers::RSAPrivateKey& key = serverState->privateKey();
      CODEX_Ciphers::RSASignature* signature = key.sign( digest );
      BN_free(digest);
      digest = 0;
      if ( 0 == signature )
      {
         sendEvent( 0, event.source(), true ); // NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }
      SignedComputedMsg signedResponse( response, *signature );
      delete signature;

      // Cache the signed response
      stateInfo->addComputeResponse( signedResponse );

      // Send the signed response
      CODEX_Quorum::Message m;
      unsigned char server[CODEX_Server::ServerState::nSID];
      memcpy( server, event.server(), CODEX_Server::ServerState::nSID );
      // Make sure the "outgoing" mask isn't set.
      server[0] &= ~(CODEX_Server::ServerState::OutgoingMask);
      int serverNum = server[0];

      // Fill message headers.
      m.fill( server, CODEX_Server::ServerState::nSID );
      m.fill( event.seqNum(), CODEX_Server::ServerState::nMID );
      m.fill( stateInfo->domain() );
      m.fill( kComputedMsg | SignatureMask );
      int length = signedResponse.marshal(0);
      unsigned char* buffer = new unsigned char[length];
      unsigned char* pBuffer = buffer;
      signedResponse.marshal(&pBuffer);
      m.fill( buffer, length );
      delete [] buffer;

      sendEvent( new CODEX_Server::ServerResponseEvent( this,
                                                        m_destination,
                                                        m,
                                                        serverNum ),
                 event.source() );
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

bool
MessageProcessor::handler( RoutedMessageEvent< SignedRecoverMsg >& event )
{
#ifdef TIMING
   ActiveTimer.start();
#endif
   StateInfo* stateInfo = StateInfo::instance();
   CODEX_Server::ServerState* serverState =
      CODEX_Server::ServerState::instance();
   const SignedRecoverMsg& signedMessage = event.message();
   const RecoverMsg& message = signedMessage.message();
   unsigned int requester = message.requester().value();

   BIGNUM * digest = 0;
   try
   {
      // Get the subsharing.
      const ShareSetType& shareset =
         stateInfo->subsharing( message.sublabel(), requester );

      // Get the subset for the requesting server.
      unsigned char server[CODEX_Server::ServerState::nSID];
      memcpy( server, event.server(), CODEX_Server::ServerState::nSID );
      // Make sure the "outgoing" mask isn't set.
      server[0] &= ~(CODEX_Server::ServerState::OutgoingMask);
      int serverNum = server[0];
      ShareType shares( shareset, serverNum );

      // Compose the response
      RecoveredMsg response( message, shares );

      // Sign the response
      const CODEX_Ciphers::HashFunction& hashFunc = serverState->hashFunc();
      digest = response.digest( hashFunc );
      const CODEX_Ciphers::RSAPrivateKey& key = serverState->privateKey();
      CODEX_Ciphers::RSASignature* signature = key.sign( digest );
      BN_free(digest);
      digest = 0;
      if ( 0 == signature )
      {
         sendEvent( 0, event.source(), true ); // NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }
      SignedRecoveredMsg signedResponse( response, *signature );
      delete signature;

      // Send the signed response.
      CODEX_Quorum::Message m;

      // Fill message headers.
      m.fill( server, CODEX_Server::ServerState::nSID );
      m.fill( event.seqNum(), CODEX_Server::ServerState::nMID );
      m.fill( stateInfo->domain() );
      m.fill( kRecoveredMsg | SignatureMask );
      int length = signedResponse.marshal(0);
      unsigned char* buffer = new unsigned char[length];
      unsigned char* pBuffer = buffer;
      signedResponse.marshal(&pBuffer);
      m.fill( buffer, length );
      delete [] buffer;

      sendEvent( new CODEX_Server::ServerResponseEvent( this,
                                                        m_destination,
                                                        m,
                                                        serverNum ),
                 event.source() );
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
   catch ( CODEX_Server::KeySharesNotFoundException& )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( new FailureEvent( this, m_destination, event ),
                 event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

bool
MessageProcessor::handler( RoutedMessageEvent< SignedFinishedMsg >& event )
{
   // Everything that needs to be done with a FinishedMsg should have
   // been done at verification time.  No response is needed.
   return true;
}

---