sestsconnectionimpl.java

这是一个基于java编写的torrent的P2P源码

								pending_size =  (( pending_size + AES_KEY_SIZE_BYTES -1 )/AES_KEY_SIZE_BYTES)*AES_KEY_SIZE_BYTES;
								
								if ( pending_size == 0 ){
									
									pending_size = AES_KEY_SIZE_BYTES;
								}
							}
							
							if ( out_buffer.remaining() >= pending_size ){
								
								if ( outgoing_cipher != null ){
									
									
									out_buffer.put( outgoing_cipher.doFinal( pending_bytes ));
									
								}else{
								
									out_buffer.put( pending_bytes );
								}
								
									// don't deallocate the pending message, the original caller does this
																
								pending_message	= null;
							}
						}
						
						crypto_completed	= true;
						
					}else{
							// we've received
							//		a -> autb + data -> b
						
							// read their auth
						
						sts_engine.putAuth( in_buffer );

							// check we wanna talk to this person
						
						byte[]	rem_key = sts_engine.getRemotePublicKey();
						
						if ( !key_locator.accept( new SEPublicKeyImpl( my_public_key.getType(), rem_key ))){
							
							throw( new MessageException( "remote public key not accepted" ));
						}
						
						setupBlockCrypto();

						crypto_completed	= true;
						
							// pick up any remaining data for delivery
						
						if ( in_buffer.hasRemaining()){
							
							message = new PooledByteBufferImpl( new DirectByteBuffer( in_buffer.slice()));
							
							forward	= true;
						}
					}
				}
			}
				
			if ( out_buffer != null ){
				
				out_buffer.flip();
				
				connection.send( new PooledByteBufferImpl( new DirectByteBuffer( out_buffer )));
			}
			
			if ( crypto_completed ){
				
				cryptoComplete();
			}
			if ( forward ){
				
				receiveContent( message );
			}
		}catch( Throwable e ){
			
			reportFailed( e );
			
			if ( e instanceof MessageException ){
				
				throw((MessageException)e);
				
			}else{
				
				throw( new MessageException( "Receive failed", e ));
			}
		}
	}

	protected void
	setupBlockCrypto()
	
		throws MessageException
	{
		if ( !failed ){
			
			if ( block_crypto == SESecurityManager.BLOCK_ENCRYPTION_NONE ){
				
				return;
			}
			
			try{
				byte[]	shared_secret = sts_engine.getSharedSecret();
								
			    SecretKeySpec	secret_key_spec1 = new SecretKeySpec(shared_secret, 0, 16, "AES" );
			    SecretKeySpec	secret_key_spec2 = new SecretKeySpec(shared_secret, 8, 16, "AES" );
		        
			    AlgorithmParameterSpec	param_spec1 = 	new IvParameterSpec( AES_IV1);
			    AlgorithmParameterSpec	param_spec2 = 	new IvParameterSpec( AES_IV2);      
			        
			    Cipher cipher1 = Cipher.getInstance( "AES/CBC/PKCS5Padding" );
			    Cipher cipher2 = Cipher.getInstance( "AES/CBC/PKCS5Padding" );
		        
			    if ( connection.isIncoming()){
			    	
			        cipher1.init( Cipher.ENCRYPT_MODE, secret_key_spec1, param_spec1 );
			        cipher2.init( Cipher.DECRYPT_MODE, secret_key_spec2, param_spec2 );
			        
			        incoming_cipher	= cipher2;
			        outgoing_cipher	= cipher1;
			        
			    }else{
			    	
			        cipher1.init( Cipher.DECRYPT_MODE, secret_key_spec1, param_spec1 );
			        cipher2.init( Cipher.ENCRYPT_MODE, secret_key_spec2, param_spec2 );
			        
			        incoming_cipher	= cipher1;
			        outgoing_cipher	= cipher2;
			    }

			}catch( Throwable e ){
				
				throw( new MessageException( "Failed to setup block encryption", e ));
			}
		}
	}

	protected void
	cryptoComplete()
	
		throws MessageException
	{
		crypto_complete.releaseForever();
	}
	
	public void
	send(
		PooledByteBuffer			message )
	
		throws MessageException
	{
		if ( failed ){
			
			throw( new MessageException( "Connection failed" ));
		}
		
		try{
			if ( crypto_complete.isReleasedForever()){
				
				sendContent( message );
				
			}else{
				
					// not complete, stash the message so it has a chance of being piggybacked on
					// the crypto protocol exchange
				
				synchronized( this ){
					
					if ( pending_message == null ){
						
						pending_message = message;
					}
				}
			}
			
			crypto_complete.reserve();
						
				// if the pending message couldn't be piggy backed it'll still be allocated
				
			boolean	send_it = false;
				
			synchronized( this ){

				if ( pending_message == message ){
					
					pending_message	= null;
					
					send_it	= true;
				}
			}
			
			if ( send_it ){
				
				sendContent( message );
			}

		}catch( Throwable e ){
			
			setFailed();
			
			if ( e instanceof MessageException ){
				
				throw((MessageException)e);
				
			}else{
				
				throw( new MessageException( "Send failed", e ));
			}
		}
	}
	
	protected void
	sendContent(
		PooledByteBuffer			message )
	
		throws MessageException
	{
		if ( outgoing_cipher != null ){
			
			try{
				byte[]	plain	=  message.toByteArray();
				byte[]	enc		= outgoing_cipher.doFinal( plain );
			
				PooledByteBuffer	temp = new PooledByteBufferImpl( enc );
				
				try{
					connection.send( temp );
					
						// successfull send -> release caller's buffer
					
					message.returnToPool();
					
				}catch( Throwable e ){
					
						// failed semantics are to not release the caller's buffer
					
					temp.returnToPool();
					
					throw( e );
				}
				
			}catch( Throwable e ){
				
				throw( new MessageException( "Failed to encrypt data", e ));
			}
		}else{
				// sanity check - never allow unencrypted outbound if block enc selected
			
			if ( block_crypto != SESecurityManager.BLOCK_ENCRYPTION_NONE ){
				
				connection.close();
				
				throw( new MessageException( "Crypto isn't setup" ));
			}
		
			connection.send( message );
		}
	}
	
	protected void
	receiveContent(
		PooledByteBuffer			message )
	
		throws MessageException
	{
		boolean	buffer_handled = false;
		
		try{
			if ( incoming_cipher != null ){
			
				try{
					byte[]	enc 	= message.toByteArray();
					byte[]	plain 	= incoming_cipher.doFinal( enc );
	
					PooledByteBuffer	temp = new PooledByteBufferImpl( plain );
	
					message.returnToPool();
					
					buffer_handled	= true;
					
					message	= temp;
					
				}catch( Throwable e ){
					
					throw( new MessageException( "Failed to decrypt data", e ));
				}
				
			}else if ( block_crypto != SESecurityManager.BLOCK_ENCRYPTION_NONE ){
				
				throw( new MessageException( "Crypto isn't setup" ));
			}
			
			for (int i=0;i<listeners.size();i++){
				
				PooledByteBuffer	message_to_deliver;
				
				if ( i == 0 ){
					
					message_to_deliver	= message;
					
				}else{
				
						// unlikely we'll ever have > 1 receiver....
					
					message_to_deliver = new PooledByteBufferImpl( message.toByteArray());
				}
				
				try{
					((GenericMessageConnectionListener)listeners.get(i)).receive( this, message_to_deliver );
					
					if ( message_to_deliver == message ){
						
						buffer_handled	= true;
					}
				}catch( Throwable e ){
					
					message_to_deliver.returnToPool();
					
					buffer_handled	= true;
					
					Debug.printStackTrace( e );
				}
			}
		}finally{
			
			if ( !buffer_handled ){
				
				message.returnToPool();
			}
		}
	}
	
	public void
	close()
	
		throws MessageException
	{
		connection.close();
	}
	
	protected void
	reportConnected()
	{
			// we've got to take this off the current thread to avoid the connection even causing immediate
			// submission of a message which then block this thread awaiting crypto completion. "this" thread
			// is currently the selector thread which then screws the crypto protocol...
		
		new AEThread( "SESTSConnection:connected", true )
		{
			public void
			runSupport()
			{
				for (int i=0;i<listeners.size();i++){
					
					try{
						((GenericMessageConnectionListener)listeners.get(i)).connected( SESTSConnectionImpl.this );
						
					}catch( Throwable e ){
						
						Debug.printStackTrace( e );
					}
				}
			}
		}.start();
		
	}
	
	protected void
	reportFailed(
		Throwable	error )
	{
		setFailed();
		
		for (int i=0;i<listeners.size();i++){
			
			try{
				((GenericMessageConnectionListener)listeners.get(i)).failed( this, error );
				
			}catch( Throwable e ){
				
				Debug.printStackTrace( e );
			}
		}
	}
	
	public void
	addListener(
		GenericMessageConnectionListener		listener )
	{
		listeners.add( listener );
	}
	
	public void
	removeListener(
		GenericMessageConnectionListener		listener )
	{
		listeners.remove( listener );
	}
}