MessageVerifier.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: MessageVerifier.cc,v 1.3 2004/05/19 15:56:24 mmarsh Exp $
//
// $Log: MessageVerifier.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 "MessageVerifier.h"
#include "StateInfo.h"
#include "CODEX_Server/ServerResponseEvent.h"
#include "FailureEvent.h"

#include "timing.h"

using namespace CODEX_APSS;

MessageVerifier::MessageVerifier(
   CODEX_Events::DeadPileType& deadPile,
   CODEX_Events::QType& eventQueue,
   RoutedMessageHandler* destination,
   CODEX_Server::ServerResponseHandler* responder ) :
   CODEX_Events::Activity( deadPile, eventQueue ),
   m_destination( destination ),
   m_responder( responder )
{
}

MessageVerifier::~MessageVerifier()
{
}

bool
MessageVerifier::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();

   BIGNUM * digest = 0;
   try
   {
      const CODEX_Ciphers::RSAPublicKey& serverKey =
         serverState->publicKey( coordinator );

      // Check the signature on the message.
      digest = message.digest( serverState->hashFunc() );
      if ( ! serverKey.verifySignature( signedMessage.signature(), digest ) )
      {
         throw CODEX_Server::SignatureVerificationFailedException( __FILE__ ,
                                                                   __LINE__ );
      }
      BN_free( digest );  digest = 0;

      // Check the version.  Send a finished message if out-of-date.
      unsigned int num = message.label().num().value();
      unsigned int version = message.version().value();
      if ( version < stateInfo->version(num) )
      {
         stateInfo->sendFinished( num, coordinator );
         sendEvent( new FailureEvent( this, m_responder, event ),
                    event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // Check for a previous init message from this coordinator:
      const InitRecord& priorInit = stateInfo->initRecord( num, coordinator );
      if ( priorInit.request().initialized() )
      {
         unsigned int pVersion = priorInit.request().version().value();
         if ( pVersion == version )
         {
            if ( priorInit.request().label() != message.label() )
            {
               // Inconsistent -- ignore
               //add logging here?
               sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
            // otherwise, look for cached response
            if ( priorInit.response().initialized() )
            {
               // send the cached 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( kContributeMsg | SignatureMask );
               const SignedContributeMsg& cm = priorInit.response();
               int length = cm.marshal(0);
               unsigned char* buffer = new unsigned char[length];
               unsigned char* pBuffer = buffer;
               cm.marshal(&pBuffer);
               m.fill( buffer, length );
               delete [] buffer;

               sendEvent(
                  new CODEX_Server::ServerResponseEvent( this,
                                                         m_responder,
                                                         m,
                                                         serverNum ),
                  event.source() );
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
            else
            {
               // We're still processing the request
               sendEvent( 0, event.source() ); // ACK
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
         }
      }

      // Make sure the new version is consistent with the label
      if ( ( message.version().value() -
             message.label().version().value() ) != 1 )
      {
         sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // The message is good -- store and re-route it to our successor.
      stateInfo->addInitMsg( signedMessage );
      event.reRoute( m_destination );
#ifdef TIMING
      InitActTimer[num].start();
      ActiveTimer.stop();
#endif
      return false;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

bool
MessageVerifier::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
   {
      const CODEX_Ciphers::RSAPublicKey& serverKey =
         serverState->publicKey( establisher );

      const CODEX_Ciphers::HashFunction& hashFunc = serverState->hashFunc();

      // Check the signature on the message.
      digest = message.digest( hashFunc );
      if ( ! serverKey.verifySignature( signedMessage.signature(), digest ) )
      {
         throw CODEX_Server::SignatureVerificationFailedException( __FILE__ ,
                                                                   __LINE__ );
      }
      BN_free( digest );  digest = 0;

      // Check the version.  Send a finished message if out-of-date.
      unsigned int num = message.sublabel().label().num().value();
      unsigned int version = message.version().value();
      if ( version < stateInfo->version(num) )
      {
         stateInfo->sendFinished( num, establisher );
         sendEvent( new FailureEvent( this, m_responder, event ),
                    event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // Check for a previous establish message from this server:
      const EstablishRecord& priorEst =
         stateInfo->establishRecord( num,
                                     establisher,
                                     message.sublabel().id().value(),
                                     message.sublabel().label() );
      if ( priorEst.request().initialized() )
      {
         unsigned int pVersion = priorEst.request().version().value();
         if ( pVersion == version )
         {
            if ( priorEst.request().sublabel() != message.sublabel() )
            {
               // Inconsistent -- ignore
               //add logging here?
               sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
            // otherwise, look for cached response
            if ( priorEst.response().initialized() )
            {
               // send the cached 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 );
               const SignedEstablishedMsg& em = priorEst.response();
               int length = em.marshal(0);
               unsigned char* buffer = new unsigned char[length];
               unsigned char* pBuffer = buffer;
               em.marshal(&pBuffer);
               m.fill( buffer, length );
               delete [] buffer;

               sendEvent(
                  new CODEX_Server::ServerResponseEvent( this,
                                                         m_responder,
                                                         m,
                                                         serverNum ),
                  event.source() );
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
            else
            {
               // We're still processing the request
               sendEvent( 0, event.source() ); // ACK
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
         }
      }

      OneWay oneWay( stateInfo->witness(num).args() );

      // Check the validity of the label
      if ( ! message.sublabel().label().verify(
         stateInfo->witness(num).witness(), oneWay ) )
      {
         sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // Check the validity of the sublabel
      if ( ! message.sublabel().verify( oneWay ) )
      {
         sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // Check the validity of the subshares
      if ( ! message.sublabel().check(message.shares(),oneWay) )
      {
         sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // The message is good -- store and re-route it to our successor.
      stateInfo->addEstablishMsg( signedMessage );
      event.reRoute( m_destination );
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return false;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

bool
MessageVerifier::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();

   BIGNUM * digest = 0;
   try
   {
      const CODEX_Ciphers::RSAPublicKey& serverKey =
         serverState->publicKey( coordinator );

      // Check the signature on the message.
      digest = message.digest( serverState->hashFunc() );
      if ( ! serverKey.verifySignature( signedMessage.signature(), digest ) )
      {
         throw CODEX_Server::SignatureVerificationFailedException( __FILE__ ,
                                                                   __LINE__ );
      }
      BN_free( digest );  digest = 0;

      // Check the version.  Send a finished message if out-of-date.
      unsigned int num = message.subshareLabel(0).label().num().value();
      unsigned int version =
         message.subshareLabel(0).label().version().value();
      if ( version < stateInfo->version(num) )
      {
         stateInfo->sendFinished( num, coordinator );
         sendEvent( new FailureEvent( this, m_responder, event ),
                    event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // Check for a previous compute message from this coordinator:
      const ComputeRecord& priorComp = stateInfo->computeRecord( num,
                                                                 coordinator );
      if ( priorComp.request().initialized() )
      {
         unsigned int pVersion =
            priorComp.request().subshareLabel(0).label().version().value();
         if ( pVersion == version )
         {
            for ( unsigned int i = 0 ; i < ComputeMsg::NumShares ; ++i )
            {
               if ( priorComp.request().subshareLabel(i) !=
                    message.subshareLabel(i) )
               {
                  // Inconsistent -- ignore
                  //add logging here?
                  sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
                  ActiveTimer.stop();
#endif
                  return true;
               }
            }
            // If consistent, look for cached response
            if ( priorComp.response().initialized() )
            {
               // send the cached 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 );
               const SignedComputedMsg& cm = priorComp.response();
               int length = cm.marshal(0);
               unsigned char* buffer = new unsigned char[length];
               unsigned char* pBuffer = buffer;
               cm.marshal(&pBuffer);
               m.fill( buffer, length );
               delete [] buffer;

               sendEvent(
                  new CODEX_Server::ServerResponseEvent( this,
                                                         m_responder,
                                                         m,
                                                         serverNum ),
                  event.source() );
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
            else
            {
               // We're still processing the request
               sendEvent( 0, event.source() ); // ACK
#ifdef TIMING
               ActiveTimer.stop();
#endif
               return true;
            }
         }
      }

      // Verify that the form of the message is valid.
      for ( unsigned int i = 0 ; i < ShareType::NumShares ; ++i )
      {
         unsigned int id = message.subshareLabel(i).id().value();
         if ( id != i )
         {
            sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
            ActiveTimer.stop();
#endif
            return true;
         }
      }

      // The message is good -- store and re-route it to our successor.
      stateInfo->addComputeMsg( signedMessage );
      event.reRoute( m_destination );
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return false;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

bool
MessageVerifier::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
   {
      const CODEX_Ciphers::RSAPublicKey& serverKey =
         serverState->publicKey( requester );

      // Check the signature on the message.
      digest = message.digest( serverState->hashFunc() );
      if ( ! serverKey.verifySignature( signedMessage.signature(), digest ) )
      {
         throw CODEX_Server::SignatureVerificationFailedException( __FILE__ ,
                                                                   __LINE__ );
      }
      BN_free( digest );  digest = 0;

      // Check the version.  Send a finished message if out-of-date.
      unsigned int num = message.sublabel().label().num().value();
      unsigned int version = message.version().value();
      if ( version < stateInfo->version(num) )
      {
         stateInfo->sendFinished( num, requester );
         sendEvent( new FailureEvent( this, m_responder, event ),
                    event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // The message is good -- re-route it to our successor.
      event.reRoute( m_destination );
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return false;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

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

   BIGNUM * digest = 0;
   try
   {
      const CODEX_Ciphers::RSAPublicKey& serverKey =
         serverState->publicKey( coordinator );

      // Check the signature on the message.
      digest = message.digest( serverState->hashFunc() );
      if ( ! serverKey.verifySignature( signedMessage.signature(), digest ) )
      {
         throw CODEX_Server::SignatureVerificationFailedException( __FILE__ ,
                                                                   __LINE__ );
      }
      BN_free( digest );  digest = 0;

      // Make sure there is sufficient evidence attached.
      CODEX_Quorum::QuorumSystem* qs = serverState->quorumSystem();
      if ( 0 == qs )
      {
         throw CODEX_Server::NoQuorumSystemException( __FILE__ , __LINE__ );
      }
      unsigned int quorumSize = qs->quorumSize();
      if ( message.evidence().size() < quorumSize )
      {
         sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }

      // Check the version.  Store the finished message if it's in the current
      // or new epoch, unless it's a duplicate.
      unsigned int num =
         message.evidence().element(0)->message().shareLabel().num().value();
      unsigned int version = message.version().value();
      if ( version < stateInfo->version(num) )
      {
         sendEvent( new FailureEvent( this, m_responder, event ),
                    event.source(), true ); // send a NACK
#ifdef TIMING
         ActiveTimer.stop();
#endif
         return true;
      }
      unsigned int coord = message.coordinator().value();
      const SignedFinishedMsg& fm = stateInfo->finishedMsg( num, coord );
      if ( fm.message().initialized() )
      {
         unsigned int pVers = fm.message().version().value();
         if ( version <= pVers )
         {
            // ignore
            sendEvent( 0, event.source() );
#ifdef TIMING
            ActiveTimer.stop();
#endif
            return true;
         }
      }

      // Verify that the evidence supports the message.
      ComputedArray::ArrayItr itr = message.evidence().begin();
      ComputedArray::ArrayItr end = message.evidence().end();
      for ( ; itr != end ; ++itr )
      {
         const ComputedMsg& cm = (*itr)->message();
         // Check that the coordinator matches the coordinator for the
         // computed message.
         unsigned int eCoord = cm.shareLabel().id().value();
         if ( eCoord != coordinator )
         {
            sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
            ActiveTimer.stop();
#endif
            return true;
         }

         // Check that the version matches the version for the computed
         // messages.
         unsigned int eVers = cm.shareLabel().version().value();
         if ( eVers != version )
         {
            sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
            ActiveTimer.stop();
#endif
            return true;
         }

         // Check the signature on the computed message.
         unsigned int s = cm.computor().value();
         digest = cm.digest( serverState->hashFunc() );
         if ( ! serverState->publicKey(s).verifySignature( (*itr)->signature(),
                                                           digest ) )
         {
            throw
               CODEX_Server::SignatureVerificationFailedException( __FILE__ ,
                                                                   __LINE__ );
         }
         BN_free( digest );  digest = 0;
      }

#ifdef TIMING
      // We have to stop the timer slightly early, since addFinishedMsg()
      // might move us into the next epoch and print the timing statistics.
      ActiveTimer.stop();
#endif
      // The message is good -- store it.
      stateInfo->addFinishedMsg( signedMessage );
      return false;
   }
   catch ( ... )
   {
      if ( 0 != digest ) BN_free( digest );
      sendEvent( 0, event.source(), true ); // send a NACK
#ifdef TIMING
      ActiveTimer.stop();
#endif
      return true;
   }
}

---