sestsconnectionimpl.java

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

/*
 * Created on 20 Jun 2006
 * Created by Paul Gardner
 * Copyright (C) 2006 Aelitis, All Rights Reserved.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 * 
 * AELITIS, SAS au capital de 46,603.30 euros
 * 8 Allee Lenotre, La Grille Royale, 78600 Le Mesnil le Roi, France.
 *
 */

package org.gudy.azureus2.pluginsimpl.local.utils.security;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.security.spec.AlgorithmParameterSpec;
import java.util.*;

import javax.crypto.Cipher;
import javax.crypto.spec.IvParameterSpec;
import javax.crypto.spec.SecretKeySpec;


import org.gudy.azureus2.core3.logging.LogEvent;
import org.gudy.azureus2.core3.logging.LogIDs;
import org.gudy.azureus2.core3.logging.Logger;
import org.gudy.azureus2.core3.util.AESemaphore;
import org.gudy.azureus2.core3.util.AEThread;
import org.gudy.azureus2.core3.util.Debug;
import org.gudy.azureus2.core3.util.DirectByteBuffer;
import org.gudy.azureus2.core3.util.SystemTime;
import org.gudy.azureus2.plugins.messaging.MessageException;
import org.gudy.azureus2.plugins.messaging.generic.GenericMessageConnection;
import org.gudy.azureus2.plugins.messaging.generic.GenericMessageConnectionListener;
import org.gudy.azureus2.plugins.messaging.generic.GenericMessageEndpoint;
import org.gudy.azureus2.plugins.utils.PooledByteBuffer;
import org.gudy.azureus2.plugins.utils.security.SEPublicKey;
import org.gudy.azureus2.plugins.utils.security.SEPublicKeyLocator;
import org.gudy.azureus2.plugins.utils.security.SESecurityManager;
import org.gudy.azureus2.pluginsimpl.local.messaging.GenericMessageConnectionImpl;
import org.gudy.azureus2.pluginsimpl.local.utils.PooledByteBufferImpl;

import com.aelitis.azureus.core.AzureusCore;
import com.aelitis.azureus.core.security.CryptoManagerException;
import com.aelitis.azureus.core.security.CryptoSTSEngine;
import com.aelitis.azureus.core.util.bloom.BloomFilter;
import com.aelitis.azureus.core.util.bloom.BloomFilterFactory;

public class 
SESTSConnectionImpl
	implements GenericMessageConnection
{	
	private static final LogIDs LOGID = LogIDs.NWMAN;

	private static final byte[]		AES_IV1				= 
   	{ 	(byte)0x15, (byte)0xE0, (byte)0x6B, (byte)0x7E, (byte)0x98, (byte)0x59, (byte)0xE4, (byte)0xA7, 
		(byte)0x34, (byte)0x66, (byte)0xAD, (byte)0x48, (byte)0x35, (byte)0xE2, (byte)0xD0, (byte)0x24 };

	private static final byte[]		AES_IV2				= { 
		(byte)0xC4, (byte)0xEF, (byte)0x06, (byte)0x3C, (byte)0x98, (byte)0x23, (byte)0xE8, (byte)0xB4, 
		(byte)0x26, (byte)0x58, (byte)0xAE, (byte)0xB9, (byte)0x2C, (byte)0x24, (byte)0xB6, (byte)0x11 };

	private final int	AES_KEY_SIZE_BYTES = AES_IV1.length;
	
	
	private static long					last_incoming_sts_create;
	
	private static final int			BLOOM_RECREATE				= 30*1000;
	private static final int			BLOOM_INCREASE				= 500;
	private static BloomFilter			generate_bloom				= BloomFilterFactory.createAddRemove4Bit(BLOOM_INCREASE);
	private static long					generate_bloom_create_time	= SystemTime.getCurrentTime();


	private AzureusCore						core;
	private GenericMessageConnectionImpl	connection;
	private SEPublicKey						my_public_key;
	private SEPublicKeyLocator				key_locator;
	private String							reason;
	private int								block_crypto;
	
	private CryptoSTSEngine	sts_engine;
	
	private List	listeners = new ArrayList();
	
	private boolean		sent_keys;
	private boolean		sent_auth;
	
	private PooledByteBuffer	pending_message;

	private AESemaphore	crypto_complete	= new AESemaphore( "SESTSConnection:send" );
	
	private Cipher	outgoing_cipher;
	private Cipher	incoming_cipher;
	
	
	private volatile boolean	failed;
	
	
	protected
	SESTSConnectionImpl(
		AzureusCore						_core,
		GenericMessageConnectionImpl	_connection,
		SEPublicKey						_my_public_key,
		SEPublicKeyLocator				_key_locator,
		String							_reason,
		int								_block_crypto )
	
		throws Exception
	{
		core			= _core;
		connection		= _connection;
		my_public_key	= _my_public_key;
		key_locator		= _key_locator;
		reason			= _reason;
		block_crypto	= _block_crypto;

		if ( connection.isIncoming()){
			
			rateLimit( connection.getEndpoint().getNotionalAddress());
		}
		
		sts_engine	= core.getCryptoManager().getECCHandler().getSTSEngine( reason );
		
		connection.addListener(
			new GenericMessageConnectionListener()
			{
				public void
				connected(
					GenericMessageConnection	connection )
				{
					reportConnected();
				}
				
				public void
				receive(
					GenericMessageConnection	connection,
					PooledByteBuffer			message )
				
					throws MessageException
				{
					SESTSConnectionImpl.this.receive( message );
				}
				
				public void
				failed(
					GenericMessageConnection	connection,
					Throwable 					error )
				
					throws MessageException
				{
					reportFailed( error );
				}
			});
	}
	
	
	protected static void
	rateLimit(
		InetSocketAddress	originator )
	
		throws Exception
	{
		synchronized( SESTSConnectionImpl.class ){
							
			int	hit_count = generate_bloom.add( originator.getAddress().getAddress());
			
			long	now = SystemTime.getCurrentTime();

				// allow up to 10% bloom filter utilisation
			
			if ( generate_bloom.getSize() / generate_bloom.getEntryCount() < 10 ){
				
				generate_bloom = BloomFilterFactory.createAddRemove4Bit(generate_bloom.getSize() + BLOOM_INCREASE );
				
				generate_bloom_create_time	= now;
				
	     		Logger.log(	new LogEvent(LOGID, "STS bloom: size increased to " + generate_bloom.getSize()));

			}else if ( now < generate_bloom_create_time || now - generate_bloom_create_time > BLOOM_RECREATE ){
				
				generate_bloom = BloomFilterFactory.createAddRemove4Bit(generate_bloom.getSize());
				
				generate_bloom_create_time	= now;
			}
				
			if ( hit_count >= 15 ){
				
	     		Logger.log(	new LogEvent(LOGID, "STS bloom: too many recent connection attempts from " + originator ));
	     		
	     		Debug.out( "STS: too many recent connection attempts from " + originator );
	     		
				throw( new IOException( "Too many recent connection attempts (sts)"));
			}
			
			long	since_last = now - last_incoming_sts_create;
			
			long	delay = 100 - since_last;
			
				// limit key gen operations to 10 a second
			
			if ( delay > 0 && delay < 100 ){
				
				try{
		    		Logger.log(	new LogEvent(LOGID, "STS: too many recent connection attempts, delaying " + delay ));
		    		 
					Thread.sleep( delay );
					
				}catch( Throwable e ){
				}
			}
			
			last_incoming_sts_create = now;
		}
	}
	
	public GenericMessageEndpoint
	getEndpoint()
	{
		return( connection.getEndpoint());
	}
	
	public int
	getMaximumMessageSize()
	{
		int	max = connection.getMaximumMessageSize();
		
		if ( outgoing_cipher != null ){
			
			max -= outgoing_cipher.getBlockSize();
		}
		
		return( max );
	}
	
	public void
	connect()
	
		throws MessageException
	{
		if ( connection.isIncoming()){

			connection.connect();

		}else{
			
			try{
				ByteBuffer	buffer = ByteBuffer.allocate( 32*1024 );
						
				sts_engine.getKeys( buffer );
						
				buffer.flip();
				
				sent_keys = true;
				
				connection.connect( buffer );
				
			}catch( CryptoManagerException	e ){
				
				throw( new MessageException( "Failed to get initial keys", e ));
			}
		}
	}
	
	protected void
	setFailed()
	{
		failed	= true;
		
		try{
			cryptoComplete();
			
		}catch( Throwable e ){
			
			Debug.printStackTrace( e );
		}
	}
	
	public void
	receive(
		PooledByteBuffer			message )
	
		throws MessageException
	{
		try{
			boolean	forward				= false;
			boolean	crypto_completed	= false;
			
			ByteBuffer	out_buffer = null;

			synchronized( this ){
		
				if ( crypto_complete.isReleasedForever()){
					
					forward	= true;
					
				}else{
					
						// basic sts flow:
						//   a -> puba -> b
						//   a <- pubb <- b
						//   a -> auta -> b
						//	 a <- autb <- b
						//   a -> data -> b
					
						// optimised
					
						//  a -> puba 		 -> b
						//  a <- pubb + auta <- b
						//  a -> autb + data -> b
					
						// therefore can be one or two messages in the payload
						// 	  1 crypto
						//    2 crypto (pub + auth)
						//	  crypto + data
					
						// initial a ->puba -> is done on first data send so data is ready for phase 3
					
					ByteBuffer	in_buffer = ByteBuffer.wrap( message.toByteArray());
					
					message.returnToPool();
						
						// piggyback pub key send
					
					if ( !sent_keys ){
						
							// we've received 
							//		a -> puba -> b
							// reply with
							//		a <- puba + auta <- b
						
						out_buffer = ByteBuffer.allocate( 64*1024 );

							// write our keys
						
						sts_engine.getKeys( out_buffer );
					
						sent_keys	= true;
						
							// read their keys
						
						sts_engine.putKeys( in_buffer );
					
							// write our auth
						
						sts_engine.getAuth( out_buffer );
						
						sent_auth 	= true;
						
					}else if ( !sent_auth ){
						
						out_buffer = ByteBuffer.allocate( 64*1024 );

						// we've received 
						//		a <- puba + auta <- b
						// reply with
						//		a -> autb + data -> b

							// read their keys
						
						sts_engine.putKeys( in_buffer );

							// write our auth
						
						sts_engine.getAuth( out_buffer );

						sent_auth = true;
					
							// 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();
						
						if ( pending_message != null ){
							
							byte[]	pending_bytes = pending_message.toByteArray();
							
							int	pending_size = pending_bytes.length;
							
							if ( outgoing_cipher != null ){