tcpprotocoldecoderphe.java

基于JXTA开发平台的下载软件开发源代码

						
					if ( read_buffer == null ){
												
						read_buffer = ByteBuffer.allocate( dh_public_key_bytes.length + 6 );
						
						protocol_substate	= 1;
					}					
						
					while( true ){
						
						read( read_buffer );
								
						if ( read_buffer.hasRemaining()){
							
							break;
						}
						
						if ( protocol_substate == 1 ){
							
							read_buffer.flip();
							
							byte[] other_dh_public_key_bytes_etc = read_buffer.array();
			
							completeDH( other_dh_public_key_bytes_etc );
			
							byte[]	etc = read_cipher.update( other_dh_public_key_bytes_etc, DH_SIZE_BYTES, 6 );
							
							byte	other_supported_protocols = etc[3];
							
							int	common_protocols = my_supported_protocols & other_supported_protocols;
							
							if (( common_protocols & CRYPTO_PLAIN )!= 0 ){
								
								selected_protocol = CRYPTO_PLAIN;
								
							}else if (( common_protocols & CRYPTO_XOR )!= 0 ){
								
								selected_protocol = CRYPTO_XOR;
								
							}else if (( common_protocols & CRYPTO_RC4 )!= 0 ){
								
								selected_protocol = CRYPTO_RC4;
								
							}else if (( common_protocols & CRYPTO_AES )!= 0 ){
								
								selected_protocol = CRYPTO_AES;
								
							}else{
								
								throw( new IOException( 
										"No crypto protocol in common: mine = " + 
											Integer.toHexString((byte)my_supported_protocols) + ", theirs = " +
											Integer.toHexString((byte)other_supported_protocols)));
				
							}
								
							int	padding	= (( etc[4] & 0xff ) << 8 ) + ( etc[5] & 0xff );
							
							if ( padding > PADDING_MAX ){
								
								throw( new IOException( "Invalid padding '" + padding + "'" ));
							}
							
							read_buffer = ByteBuffer.allocate( padding );
							
							protocol_substate	= 2;
							
						}else{
							
							read_buffer	= null;
				        						
							protocol_state	= PS_OUTBOUND_3;
							
							break;
						}
					}
				}else if ( protocol_state == PS_INBOUND_3 ){
					
					
					//	B skips Pa bytes until receives SHA1(S)
					//	B decrypts "selected method" + len(Pc) and skips len(Pc) bytes 
						
					if ( read_buffer == null ){
												
						read_buffer = ByteBuffer.allocate( 20 + PADDING_MAX );
						
						read_buffer.limit( 20 );
						
						protocol_substate	= 1;
					}					
					
					while( true ){
						
						read( read_buffer );
								
						if ( read_buffer.hasRemaining()){
						
							break;
						}
						
						if ( protocol_substate == 1 ){
							
							int	limit = read_buffer.limit();
							
							read_buffer.position( limit - 20 );
							
							boolean match	= true;
							
							for (int i=0;i<20;i++){
								
								if ( read_buffer.get() != sha1_secret_bytes[i] ){
									
									match	= false;
									
									break;
								}
							}
							
							if ( match ){
							
								read_buffer = ByteBuffer.allocate( 6 );
								
								protocol_substate	= 2;
								
								break;
							
							}else{
								
								if ( limit == read_buffer.capacity()){
									
									throw( new IOException( "PHE skip to SHA1 marker failed" ));
								}
								
								read_buffer.limit( limit + 1 );
								
								read_buffer.position( limit );
							}
						}else if ( protocol_substate == 2 ){
							
							read_buffer.flip();
								
							byte[]	etc = read_cipher.update( read_buffer.array());
							
							selected_protocol = etc[3];
							
							int	padding	= (( etc[4] & 0xff ) << 8 ) + ( etc[5] & 0xff );
							
							if ( padding > PADDING_MAX ){
								
								throw( new IOException( "Invalid padding '" + padding + "'" ));
							}

							read_buffer = ByteBuffer.allocate( padding );
							
							protocol_substate	= 3;
														
						}else{
							
							read_buffer	= null;
				        
							handshakeComplete();
							
							break;
						}
					}
				}
		
				if ( handshake_complete ){
					
					read_selector.cancel( channel );
					
					write_selector.cancel( channel );
					
					loop	= false;
					
					complete();
					
				}else{
				
					if ( read_buffer == null ){
						
						read_selector.pauseSelects( channel );
						
					}else{
						
						read_selector.resumeSelects ( channel );
						
						loop	= false;
						
					}
					
					if ( write_buffer == null ){
						
						write_selector.pauseSelects( channel );
						
					}else{
						
						write_selector.resumeSelects ( channel );
						
						loop	= false;
					}
				}
			}
		}catch( Throwable e ){
						
			failed( e );
			
			if ( e instanceof IOException ){
				
				throw((IOException)e);
				
			}else{
				
				throw( new IOException( Debug.getNestedExceptionMessage(e)));
			}
		}finally{
			
			process_mon.exit();
		}
	}
	*/
	
	protected void
	read(
		ByteBuffer		buffer )
	
		throws IOException
	{
		int	len = channel.read( buffer );
	
		// System.out.println( "read:" + this + "/" + protocol_state + "/" + protocol_substate + " -> " + len +"[" + buffer +"]");
		
		if ( len < 0 ){
			
			throw( new IOException( "end of stream on socket read - phe: " + getString()));
		}
		
		bytes_read += len;
	}
	
	protected void
	write(
		ByteBuffer		buffer )
	
		throws IOException
	{
		int	len = channel.write( buffer );
		
		// System.out.println( "write:" + this + "/" + protocol_state + "/" + protocol_substate + " -> " + len +"[" + buffer +"]");

		if ( len < 0 ){
			
			throw( new IOException( "bytes written < 0 " ));			
		}
		
		bytes_written += len;
	}
	
	public boolean 
	selectSuccess(
		VirtualChannelSelector 	selector, 
		SocketChannel 			sc, 
		Object 					attachment)
	{
		try{
			int	old_bytes_read	= bytes_read;
			
			process();
			
			if ( selector == write_selector ){
				
				return( true );
				
			}else{
				
				boolean	progress = bytes_read != old_bytes_read;
				
				if ( progress ){
					
					last_read_time = SystemTime.getCurrentTime();
				}
				
				return( progress );
			}
			
		}catch( Throwable  e ){
			
			failed( e );
			
			return( false );
		}
	}

	public void 
	selectFailure(
		VirtualChannelSelector 	selector, 
		SocketChannel 			sc, 
		Object 					attachment, 
		Throwable				msg )
	{
		failed( msg );
	}
	
	protected byte[]
	bigIntegerToBytes(
		BigInteger	bi,
		int			num_bytes )
	{
		String	str = bi.toString(16);
		
		while( str.length() < num_bytes*2 ){
			str = "0" + str;
		}
		
		return( ByteFormatter.decodeString(str));
	}
	
	protected BigInteger
	bytesToBigInteger(
		byte[]	bytes,
		int		offset,
		int		len )
	{		
		return( new BigInteger( ByteFormatter.encodeString( bytes, offset, len  ), 16 ));
	}
	
	protected static synchronized byte[]
	getRandomPadding(
		int		max_len )
	{
		byte[]	bytes = new byte[ random.nextInt(max_len)];
		
		random.nextBytes(bytes);
		
		return( bytes );
	}
	
	protected static synchronized byte[]
   	getZeroPadding()
   	{
   		byte[]	bytes = new byte[ random.nextInt(PADDING_MAX)];
   		
   		return( bytes );
   	}
	
	protected static KeyPair
	generateDHKeyPair(
		SocketChannel	channel,
		boolean			outbound )
	
		throws IOException
	{
		synchronized( dh_key_generator ){
			
			if ( !outbound ){
				
				int	hit_count = generate_bloom.add( channel.socket().getInetAddress().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, "PHE 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, "PHE bloom: too many recent connection attempts from " + channel.socket().getInetAddress()));
		     		
					throw( new IOException( "Too many recent connection attempts (phe)"));
				}
				
				long	since_last = now - last_dh_incoming_key_generate;
				
				long	delay = 100 - since_last;
				
					// limit key gen operations to 10 a second
				
				if ( delay > 0 && delay < 100 ){
					
					try{
						Thread.sleep( delay );
						
					}catch( Throwable e ){
					}
				}
				
				last_dh_incoming_key_generate = now;
			}
			
			KeyPair	res = dh_key_generator.generateKeyPair();
			
			return( res );
		}
	}
	
	protected void
	complete()
	{
		// System.out.println( (outbound?"out: ":"in :") + " complete, r " + bytes_read + ", w " + bytes_written + ", initial data = " + initial_data_in.length + "/" + initial_data_out.length );

		processing_complete	= true;
		
		adapter.decodeComplete( this );	
	}
	
	protected void
	failed(
		Throwable 	cause )
	{
		// System.out.println( (outbound?"out: ":"in :") + " failed, " + cause.getMessage());

		processing_complete	= true;
		
		read_selector.cancel( channel );
		
		write_selector.cancel( channel );

		adapter.decodeFailed( this, cause );
	}
	
	public boolean
	isComplete(
		long		now )
	{
		return( processing_complete );
	}
	
	public TCPTransportHelperFilter
	getFilter()
	{
		return( filter );
	}
	
	public long
	getLastReadTime()
	{
		long	now = SystemTime.getCurrentTime();
		
		if ( last_read_time > now ){
			
			last_read_time	= now;
		}
		
		return( last_read_time );
	}
	
	public String
	getString()
	{
		return( "state=" + protocol_state + ",sub=" + protocol_substate + ",in=" + bytes_read + ",out=" + bytes_written);
	}
	
	public SocketChannel
	getChannel()
	{
		return( channel );
	}
}