dhtpluginstoragemanager.java

JAVA多torrent下载程序P2P源码 This is a stripped down version of Manfred Duchrows Programmer s Freind class

		boolean			exhaustive )
	{
		//System.out.println( "DHT get existing diversification: put = " + put_operation  );
		
		HashWrapper	wrapper = new HashWrapper( key );
		
			// must always return a value - original if no diversification exists
		
		try{
			storage_mon.enter();
		
			byte[][]	res = followDivChain( wrapper, put_operation, exhaustive );
			
			if ( !Arrays.equals( res[0], key )){
				
				String	trace = "";
				
				for (int i=0;i<res.length;i++){
					trace += (i==0?"":",") + DHTLog.getString2( res[i] );
				}
				
				log.log( "SM: get div: " + DHTLog.getString2(key) + ", put = " + put_operation + ", exh = " + exhaustive + " -> " + trace );
			}
			
			return( res );
			
		}finally{
			
			storage_mon.exit();
		}
	}
	
	public byte[][]
	createNewDiversification(
		DHTTransportContact	cause,
		byte[]				key,
		boolean				put_operation,
		byte				diversification_type,
		boolean				exhaustive )
	{
		//System.out.println( "DHT create new diversification: put = " + put_operation +", type = " + diversification_type );
		
		HashWrapper	wrapper = new HashWrapper( key );
		
		try{
			storage_mon.enter();
		
			diversification	div = lookupDiversification( wrapper );
		
			boolean	created = false;
			
			if ( div == null ){
				
				div = createDiversification( wrapper, diversification_type );
				
				created	= true;			
			}
		
			byte[][] res = followDivChain( wrapper, put_operation, exhaustive );
		
			String	trace = "";
			
			for (int i=0;i<res.length;i++){
				
				trace += (i==0?"":",") + DHTLog.getString2( res[i] );
			}
			
			log.log( "SM: create div: " + DHTLog.getString2(key) + 
						", new = " + created + ", put = " + put_operation + 
						", exh = " + exhaustive + 
						", type = " + DHT.DT_STRINGS[diversification_type] + " -> " + trace +
						", cause = " + (cause==null?"<unknown>":cause.getString()));
			

			return( res );
			
		}finally{
			
			storage_mon.exit();
		}
	} 
	
	protected byte[][]
	followDivChain(
		HashWrapper	wrapper,
		boolean		put_operation,
		boolean		exhaustive )
	{
		List	list = new ArrayList();
		
		list.add( wrapper );
		
		list	= followDivChain( list, put_operation, 0, exhaustive, new ArrayList());
		
		byte[][]	res = new byte[list.size()][];
		
		for (int i=0;i<list.size();i++){
			
			res[i] = ((HashWrapper)list.get(i)).getBytes();
		}
		
		return( res );
	}
	
	protected List
	followDivChain(
		List		list_in,
		boolean		put_operation,
		int			depth,
		boolean		exhaustive,
		List		keys_done )
	{
		List	list_out = new ArrayList();
	
		/*
		String	indent = "";
		for(int i=0;i<depth;i++){
			indent+= "  ";
		}
		System.out.println( indent + "->" );
		*/
		
			// for each entry, if there are no diversifications then we just return the value
			// for those with divs we replace their entry with the diversified set (which can
			// include the entry itself under some circumstances )
		
		for (int i=0;i<list_in.size();i++){
			
			HashWrapper	wrapper = (HashWrapper)list_in.get(i);
		
			diversification	div = lookupDiversification( wrapper );

			if ( div == null ){
				
				if ( !list_out.contains( wrapper )){
					
					list_out.add(wrapper);
				}
				
			}else{
				
				if ( keys_done.contains( wrapper )){
					
						// we've recursed on the key, this means that a prior diversification wanted
						// the key included, so include it now
					
					if ( !list_out.contains( wrapper )){
						
						list_out.add(wrapper);
					}
					
					continue;
				}
				
				keys_done.add( wrapper );
				
					// replace this entry with the diversified keys 
				
				List	new_list = followDivChain( div.getKeys( put_operation, exhaustive ), put_operation, depth+1, exhaustive, keys_done );
				
				for (int j=0;j<new_list.size();j++){
					
					Object	entry =  new_list.get(j);
					
					if ( !list_out.contains( entry )){
						
						list_out.add(entry);
					}
				}
			}
		}
		// System.out.println( indent + "<-" );

		return( list_out );
	}
	
	protected storageKey
	getStorageKey(
		HashWrapper		key )
	{
		storageKey	res = (storageKey)local_storage_keys.get( key );
		
		if ( res == null ){
			
				// someout could be spamming us with crap, prevent things from getting
				// out of control
			
			if ( local_storage_keys.size() >= MAX_STORAGE_KEYS ){
				
				res = new storageKey( this, DHT.DT_SIZE, key ); 

				Debug.out( "DHTStorageManager: max key limit exceeded" );
				
				log.log( "SM: max storage key limit exceeded - " + DHTLog.getString2( key.getBytes()));
				
			}else{

				res = new storageKey( this, DHT.DT_NONE, key ); 
			
				local_storage_keys.put( key, res );
			}
		}
		
		return( res );
	}
	
	protected void
	deleteStorageKey(
		storageKey		key )
	{
		if ( local_storage_keys.remove( key ) != null ){
		
			if ( key.getDiversificationType() != DHT.DT_NONE ){
				
				writeDiversifications();
			}
		}
	}
	
	protected void
	readDiversifications()
	{
		try{
			storage_mon.enter();
			
			Map	map = readMapFromFile( "diverse" );
	
			List	keys = (List)map.get("local");
			
			if ( keys != null ){
				
				long	now = SystemTime.getCurrentTime();
				
				for (int i=0;i<keys.size();i++){
					
					storageKey d = storageKey.deserialise(this, (Map)keys.get(i));
						
					long	time_left = d.getExpiry() - now;
					
					if ( time_left > 0 ){
					
						local_storage_keys.put( d.getKey(), d );
						
					}else{
						
						log.log( "SM: serialised sk: " + DHTLog.getString2( d.getKey().getBytes()) + " expired" );
					}
				}
			}
			List	divs = (List)map.get("remote");
			
			if ( divs != null ){
				
				long	now = SystemTime.getCurrentTime();
				
				for (int i=0;i<divs.size();i++){
					
					diversification d = diversification.deserialise( this, (Map)divs.get(i));
						
					long	time_left = d.getExpiry() - now;

					if ( time_left > 0 ){
					
						remote_diversifications.put( d.getKey(), d );
						
					}else{
						
						log.log( "SM: serialised div: " + DHTLog.getString2( d.getKey().getBytes()) + " expired" );
					}
				}
			}
			
		}finally{
			
			storage_mon.exit();
		}
	}
	
	protected void
	writeDiversifications()
	{
		try{
			storage_mon.enter();
	
			Map	map = new HashMap();
			
			List	keys = new ArrayList();
			
			map.put( "local", keys );
			
			Iterator	it = local_storage_keys.values().iterator();
			
			while( it.hasNext()){
			
				storageKey	key = (storageKey)it.next();
				
				if ( key.getDiversificationType() != DHT.DT_NONE ){
					
					keys.add(key.serialise());
				}
			}
			
			List	divs = new ArrayList();
			
			map.put( "remote", divs );
			
			it = remote_diversifications.values().iterator();
			
			while( it.hasNext()){
			
				divs.add(((diversification)it.next()).serialise());
			}
			
			writeMapToFile( map, "diverse" );
			
		}catch( Throwable e ){
			
			Debug.printStackTrace(e);
	
		}finally{
			
			storage_mon.exit();
		}
	}
	
	protected diversification
	lookupDiversification(
		HashWrapper	wrapper )
	{
		diversification	div = (diversification)remote_diversifications.get(wrapper);
		
		if ( div != null ){
			
			if ( div.getExpiry() < SystemTime.getCurrentTime()){
				
				log.log( "SM: div: " + DHTLog.getString2( div.getKey().getBytes()) + " expired" );

				remote_diversifications.remove( wrapper );
				
				div = null;
			}
		}
		
		return( div );
	}
	
	protected diversification
	createDiversification(
		HashWrapper			wrapper,
		byte				type )
	{
		diversification	div = new diversification( this, wrapper, type );
			
		remote_diversifications.put( wrapper, div );
		
		writeDiversifications();
		
		return( div );
	}
	
	protected static String
	formatExpiry(
		long	l )
	{
		long	diff = l - SystemTime.getCurrentTime();
		
		return( (diff<0?"-":"") + DisplayFormatters.formatTime(Math.abs(diff)));
	}
	
	
		// key blocks
	
	protected void
	readKeyBlocks()
	{
		try{
			key_block_mon.enter();
			
			Map	map = readMapFromFile( "block" );
	
			List	entries = (List)map.get( "entries" );
			
			int	now_secs = (int)(SystemTime.getCurrentTime()/1000);
			
			ByteArrayHashMap	new_map = new ByteArrayHashMap();
			
			if ( entries != null ){
			
				for (int i=0;i<entries.size();i++){
					
					try{
						Map	m = (Map)entries.get(i);
						
						byte[]	request = (byte[])m.get( "req" );
						byte[]	cert	= (byte[])m.get( "cert" );
						int		recv	= ((Long)m.get( "received" )).intValue();
						boolean	direct	= ((Long)m.get( "direct" )).longValue()==1;
						
						if ( recv > now_secs ){
							
							recv	= now_secs;
						}

						keyBlock	kb = new keyBlock( request, cert, recv, direct );

							// direct "add" values never timeout, however direct "removals" do, as do 
							// indirect values
						
						if ( ( direct && kb.isAdd()) || now_secs - recv < KEY_BLOCK_TIMEOUT_SECS ){
						
							if ( verifyKeyBlock( request, cert )){
								
								log.log( "KB: deserialised " + DHTLog.getString2( kb.getKey()) + ",add=" + kb.isAdd() + ",dir=" + kb.isDirect());
						
								new_map.put( kb.getKey(), kb );
							}
						}
						
					}catch( Throwable e ){
						
						Debug.printStackTrace(e);
					}
				}
			}
			
			key_block_map_cow		= new_map;
			key_blocks_direct_cow	= buildKeyBlockDetails( new_map );
			
		}finally{
			
			key_block_mon.exit();
		}
	}
	
	protected DHTStorageBlock[]
	buildKeyBlockDetails(
		ByteArrayHashMap		map )
	{
		List	kbs = map.values();
		
		Iterator	it = kbs.iterator();
		
		while( it.hasNext()){
			
			keyBlock	kb = (keyBlock)it.next();
			
			if ( !kb.isDirect()){
				
				it.remove();
			}
		}
		
		DHTStorageBlock[]	new_blocks = new DHTStorageBlock[kbs.size()];
		
		kbs.toArray( new_blocks );
		
		return( new_blocks );
	}
	
	protected void
	writeKeyBlocks()
	{
		try{
			key_block_mon.enter();
	
			Map	map = new HashMap();
			
			List	entries = new ArrayList();
			
			map.put( "entries", entries );
			
			List	kbs = key_block_map_cow.values();
			
			for (int i=0;i<kbs.size();i++){
								
				keyBlock	kb = (keyBlock)kbs.get(i);
									
				Map	m = new HashMap();
				
				m.put( "req", kb.getRequest());
				m.put( "cert", kb.getCertificate());
				m.put( "received", new Long(kb.getReceived()));
				m.put( "direct", new Long(kb.isDirect()?1:0));
				
				entries.add( m );
			}
			
			writeMapToFile( map, "block" );
			
		}catch( Throwable e ){
			
			Debug.printStackTrace(e);
	
		}finally{
			
			key_block_mon.exit();
		}
	}
	
	
	public DHTStorageBlock
	keyBlockRequest(
		DHTTransportContact		originating_contact,
		byte[]					request,
		byte[]					signature )
	{
			// request is 4 bytes flags, 4 byte time, K byte key
			// flag: MSB 00 -> unblock, 01 ->block
		
		if ( request.length <= 8 ){
			
			return( null );
		}
				
		keyBlock	kb = 
			new keyBlock(request, signature, (int)(SystemTime.getCurrentTime()/1000), originating_contact != null );
		
		try{
			key_block_mon.enter();

			boolean	add_it	= false;
			
			try{
				keyBlock	old = (keyBlock)key_block_map_cow.get( kb.getKey());
								
				if ( old != null ){
				
						// never override a direct value with an indirect one as direct = first hand knowledge
						// whereas indirect is hearsay

					if ( old.isDirect() && !kb.isDirect()){
					
						return( null );
					}
					 
						// don't let older instructions override newer ones
					
					if ( old.getCreated() > kb.getCreated()){
						
						return( null );
					}
				}
				
				if ( kb.isAdd()){
					
					if ( old == null || !old.isAdd()){
						
						if ( !verifyKeyBlock( kb, originating_contact )){
							
							return( null );
						}
					
						add_it	= true;
					}
					
					return( kb );
					
				}else{
					
						// only direct operations can "remove" blocks
					
					if ( kb.isDirect() && ( old == null || old.isAdd())){
					
						if ( !verifyKeyBlock( kb, originating_contact )){
							
							return( null );
						}
						
						add_it	= true;
					}
					
					return( null );
				}
			}finally{
				
				if ( add_it ){
					
					ByteArrayHashMap new_map = key_block_map_cow.duplicate();
					
					new_map.put( kb.getKey(), kb );
					
						// seeing as we've received this from someone there's no point in replicating it
						// back to them later - mark them to prevent this
					
					if ( originating_contact != null ){
						
						kb.sentTo( originating_contact );
					}
					
					key_block_map_cow		= new_map;
					key_blocks_direct_cow	= buildKeyBlockDetails( key_block_map_cow );
					
					writeKeyBlocks();
				}
			}
		}finally{
			
			key_block_mon.exit();
		}
	}
	
	protected boolean
	verifyKeyBlock(
		keyBlock				kb,
		DHTTransportContact		originator )
	{
		byte[]	id = originator==null?new byte[20]:originator.getID();
		
		BloomFilter	filter = kb_verify_fail_bloom;
		
		long	now = SystemTime.getCurrentTime();
		
		if ( 	filter == null || 
				kb_verify_fail_bloom_create_time > now ||
				now - kb_verify_fail_bloom_create_time > 30*60*1000 ){
			
			kb_verify_fail_bloom_create_time	= now;
			
			filter = BloomFilterFactory.createAddOnly(4000);
			
			kb_verify_fail_bloom	= filter;
		}
		
		if ( filter.contains( id )){
			
			log.log( "KB: request verify denied" );
			
			return( false );
		}
		
		try{
			Signature	verifier = Signature.getInstance("MD5withRSA" );
			
			verifier.initVerify( key_block_public_key );
			
			verifier.update( kb.getRequest() );

			if ( !verifier.verify( kb.getCertificate())){
			
				log.log( "KB: request verify failed for " + DHTLog.getString2( kb.getKey()));

				filter.add( id );
				
				return( false );
			}
			
			log.log( "KB: request verify ok " + DHTLog.getString2( kb.getKey()) + ", add = " + kb.isAdd() + ", direct = " + kb.isDirect());

			return( true );
			
		}catch( Throwable e ){
			
			return( false );
		}
	}
	
	public static boolean
	verifyKeyBlock(
		byte[]		request,
		byte[]		signature )
	{
		try{
			Signature	verifier = Signature.getInstance("MD5withRSA" );
			
			verifier.initVerify( key_block_public_key );
			
			verifier.update( request );

			if ( !verifier.verify( signature )){
			
				return( false );
			}
			
			return( true );
			
		}catch( Throwable e ){
			
			return( false );
		}
	}
	
	public DHTStorageBlock
	getKeyBlockDetails(
		byte[]		key )
	{
		keyBlock	kb = (keyBlock)key_block_map_cow.get( key );
		
		if ( kb == null || !kb.isAdd()){
			
			return( null );
		}
		
		if ( !kb.getLogged()){
			
			kb.setLogged();
			
			log.log( "KB: Access to key '" + DHTLog.getFullString( kb.getKey()) + "' denied as it is blocked" );
		}