Implemented rotating DNS upstreams from config. Fixed warnings.
This commit is contained in:
@@ -4,7 +4,7 @@
|
||||
"version": 0,
|
||||
"key_file": "default.key",
|
||||
"listen": "[::]:4244",
|
||||
"public": true,
|
||||
"public": false,
|
||||
"peers": [
|
||||
"test-ip4.alfis.name:4244",
|
||||
"test-ip6.alfis.name:4244"
|
||||
@@ -12,8 +12,8 @@
|
||||
"dns": {
|
||||
"port": 53,
|
||||
"forwarders": [
|
||||
"1.1.1.1",
|
||||
"8.8.8.8"
|
||||
"94.140.14.14:53",
|
||||
"94.140.15.15:53"
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ use sqlite::{Connection, State, Statement};
|
||||
|
||||
use crate::{Block, Bytes, Keystore, Transaction};
|
||||
use crate::settings::Settings;
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error};
|
||||
|
||||
const DB_NAME: &str = "blockchain.db";
|
||||
|
||||
@@ -2,6 +2,7 @@ use crate::Context;
|
||||
use std::sync::{Mutex, Arc};
|
||||
use crate::dns::filter::DnsFilter;
|
||||
use crate::dns::protocol::{DnsPacket, QueryType, DnsRecord, DnsQuestion};
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error};
|
||||
|
||||
pub struct BlockchainFilter {
|
||||
@@ -16,8 +17,8 @@ impl BlockchainFilter {
|
||||
|
||||
impl DnsFilter for BlockchainFilter {
|
||||
fn lookup(&self, qname: &str, qtype: QueryType) -> Option<DnsPacket> {
|
||||
let mut search;
|
||||
let mut subdomain;
|
||||
let search;
|
||||
let subdomain;
|
||||
let parts: Vec<&str> = qname.rsplitn(3, ".").collect();
|
||||
match parts.len() {
|
||||
1 => { return None; }
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
use crate::{Blockchain, Bus, Keystore};
|
||||
use crate::event::Event;
|
||||
use crate::settings::Settings;
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error};
|
||||
|
||||
pub struct Context {
|
||||
|
||||
+7
-7
@@ -2,7 +2,7 @@
|
||||
|
||||
use std::io::Write;
|
||||
use std::marker::{Send, Sync};
|
||||
use std::net::{TcpStream, UdpSocket};
|
||||
use std::net::{TcpStream, UdpSocket, ToSocketAddrs};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::mpsc::{channel, Sender};
|
||||
use std::sync::{Arc, Mutex};
|
||||
@@ -32,7 +32,7 @@ pub trait DnsClient {
|
||||
fn get_failed_count(&self) -> usize;
|
||||
|
||||
fn run(&self) -> Result<()>;
|
||||
fn send_query(&self, qname: &str, qtype: QueryType, server: (&str, u16), recursive: bool) -> Result<DnsPacket>;
|
||||
fn send_query(&self, qname: &str, qtype: QueryType, server: &str, recursive: bool) -> Result<DnsPacket>;
|
||||
}
|
||||
|
||||
/// The UDP client
|
||||
@@ -84,7 +84,7 @@ impl DnsNetworkClient {
|
||||
///
|
||||
/// This is much simpler than using UDP, since the kernel will take care of
|
||||
/// packet ordering, connection state, timeouts etc.
|
||||
pub fn send_tcp_query(&self, qname: &str, qtype: QueryType, server: (&str, u16), recursive: bool) -> Result<DnsPacket> {
|
||||
pub fn send_tcp_query<A: ToSocketAddrs>(&self, qname: &str, qtype: QueryType, server: A, recursive: bool) -> Result<DnsPacket> {
|
||||
let _ = self.total_sent.fetch_add(1, Ordering::Release);
|
||||
|
||||
// Prepare request
|
||||
@@ -125,7 +125,7 @@ impl DnsNetworkClient {
|
||||
/// worker thread, and returned to this thread through a channel. Thus this
|
||||
/// method is thread safe, and can be used from any number of threads in
|
||||
/// parallel.
|
||||
pub fn send_udp_query(&self, qname: &str, qtype: QueryType, server: (&str, u16), recursive: bool) -> Result<DnsPacket> {
|
||||
pub fn send_udp_query<A: ToSocketAddrs>(&self, qname: &str, qtype: QueryType, server: A, recursive: bool) -> Result<DnsPacket> {
|
||||
let _ = self.total_sent.fetch_add(1, Ordering::Release);
|
||||
|
||||
// Prepare request
|
||||
@@ -268,7 +268,7 @@ impl DnsClient for DnsNetworkClient {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn send_query(&self,qname: &str, qtype: QueryType, server: (&str, u16), recursive: bool) -> Result<DnsPacket> {
|
||||
fn send_query(&self,qname: &str, qtype: QueryType, server: &str, recursive: bool) -> Result<DnsPacket> {
|
||||
let packet = self.send_udp_query(qname, qtype, server, recursive)?;
|
||||
if !packet.header.truncated_message {
|
||||
return Ok(packet);
|
||||
@@ -284,7 +284,7 @@ pub mod tests {
|
||||
use super::*;
|
||||
use crate::dns::protocol::{DnsPacket, DnsRecord, QueryType};
|
||||
|
||||
pub type StubCallback = dyn Fn(&str, QueryType, (&str, u16), bool) -> Result<DnsPacket>;
|
||||
pub type StubCallback = dyn Fn(&str, QueryType, &str, bool) -> Result<DnsPacket>;
|
||||
|
||||
pub struct DnsStubClient {
|
||||
callback: Box<StubCallback>,
|
||||
@@ -313,7 +313,7 @@ pub mod tests {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn send_query(&self,qname: &str, qtype: QueryType, server: (&str, u16), recursive: bool) -> Result<DnsPacket> {
|
||||
fn send_query(&self,qname: &str, qtype: QueryType, server: &str, recursive: bool) -> Result<DnsPacket> {
|
||||
(self.callback)(qname, qtype, server, recursive)
|
||||
}
|
||||
}
|
||||
|
||||
+3
-3
@@ -38,7 +38,7 @@ impl ServerStatistics {
|
||||
|
||||
pub enum ResolveStrategy {
|
||||
Recursive,
|
||||
Forward { host: String, port: u16 },
|
||||
Forward { upstreams: Vec<String> },
|
||||
}
|
||||
|
||||
pub struct ServerContext {
|
||||
@@ -101,8 +101,8 @@ impl ServerContext {
|
||||
pub fn create_resolver(&self, ptr: Arc<ServerContext>) -> Box<dyn DnsResolver> {
|
||||
match self.resolve_strategy {
|
||||
ResolveStrategy::Recursive => Box::new(RecursiveDnsResolver::new(ptr)),
|
||||
ResolveStrategy::Forward { ref host, port } => {
|
||||
Box::new(ForwardingDnsResolver::new(ptr, (host.clone(), port)))
|
||||
ResolveStrategy::Forward { ref upstreams } => {
|
||||
Box::new(ForwardingDnsResolver::new(ptr, upstreams.clone()))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+13
-13
@@ -68,12 +68,12 @@ pub trait DnsResolver {
|
||||
/// This resolver uses an external DNS server to service a query
|
||||
pub struct ForwardingDnsResolver {
|
||||
context: Arc<ServerContext>,
|
||||
server: (String, u16),
|
||||
upstreams: Vec<String>,
|
||||
}
|
||||
|
||||
impl ForwardingDnsResolver {
|
||||
pub fn new(context: Arc<ServerContext>, server: (String, u16)) -> ForwardingDnsResolver {
|
||||
ForwardingDnsResolver { context, server }
|
||||
pub fn new(context: Arc<ServerContext>, upstreams: Vec<String>) -> ForwardingDnsResolver {
|
||||
ForwardingDnsResolver { context, upstreams }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -83,10 +83,11 @@ impl DnsResolver for ForwardingDnsResolver {
|
||||
}
|
||||
|
||||
fn perform(&mut self, qname: &str, qtype: QueryType) -> Result<DnsPacket> {
|
||||
let &(ref host, port) = &self.server;
|
||||
let index: usize = rand::random::<usize>() % self.upstreams.len();
|
||||
let upstream = self.upstreams[index].as_ref();
|
||||
let result = match self.context.cache.lookup(qname, qtype) {
|
||||
None => {
|
||||
self.context.client.send_query(qname, qtype, (host.as_str(), port), true)?
|
||||
self.context.client.send_query(qname, qtype, upstream, true)?
|
||||
}
|
||||
Some(packet) => packet
|
||||
};
|
||||
@@ -149,11 +150,11 @@ impl DnsResolver for RecursiveDnsResolver {
|
||||
|
||||
let ns_copy = ns.clone();
|
||||
|
||||
let server = (ns_copy.as_str(), 53);
|
||||
let server = format!("{}:{}", ns_copy.as_str(), 53);
|
||||
let response = self
|
||||
.context
|
||||
.client
|
||||
.send_query(qname, qtype.clone(), server, false)?;
|
||||
.send_query(qname, qtype.clone(), &server, false)?;
|
||||
|
||||
// If we've got an actual answer, we're done!
|
||||
if !response.answers.is_empty() && response.header.rescode == ResultCode::NOERROR {
|
||||
@@ -234,8 +235,7 @@ mod tests {
|
||||
match Arc::get_mut(&mut context) {
|
||||
Some(mut ctx) => {
|
||||
ctx.resolve_strategy = ResolveStrategy::Forward {
|
||||
host: "127.0.0.1".to_string(),
|
||||
port: 53,
|
||||
upstreams: vec![String::from("127.0.0.1:53")]
|
||||
};
|
||||
}
|
||||
None => panic!(),
|
||||
@@ -342,10 +342,10 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_recursive_resolver_match_order() {
|
||||
let context = create_test_context(Box::new(|_, _, (server, _), _| {
|
||||
let context = create_test_context(Box::new(|_, _, server, _| {
|
||||
let mut packet = DnsPacket::new();
|
||||
|
||||
if server == "127.0.0.1" {
|
||||
if server.starts_with("127.0.0.1") {
|
||||
packet.header.id = 1;
|
||||
|
||||
packet.answers.push(DnsRecord::A {
|
||||
@@ -355,7 +355,7 @@ mod tests {
|
||||
});
|
||||
|
||||
return Ok(packet);
|
||||
} else if server == "127.0.0.2" {
|
||||
} else if server.starts_with("127.0.0.2") {
|
||||
packet.header.id = 2;
|
||||
|
||||
packet.answers.push(DnsRecord::A {
|
||||
@@ -365,7 +365,7 @@ mod tests {
|
||||
});
|
||||
|
||||
return Ok(packet);
|
||||
} else if server == "127.0.0.3" {
|
||||
} else if server.starts_with("127.0.0.3") {
|
||||
packet.header.id = 3;
|
||||
|
||||
packet.answers.push(DnsRecord::A {
|
||||
|
||||
+2
-4
@@ -437,8 +437,7 @@ mod tests {
|
||||
match Arc::get_mut(&mut context) {
|
||||
Some(mut ctx) => {
|
||||
ctx.resolve_strategy = ResolveStrategy::Forward {
|
||||
host: "127.0.0.1".to_string(),
|
||||
port: 53,
|
||||
upstreams: vec![String::from("127.0.0.1:53")]
|
||||
};
|
||||
}
|
||||
None => panic!(),
|
||||
@@ -545,8 +544,7 @@ mod tests {
|
||||
match Arc::get_mut(&mut context2) {
|
||||
Some(mut ctx) => {
|
||||
ctx.resolve_strategy = ResolveStrategy::Forward {
|
||||
host: "127.0.0.1".to_string(),
|
||||
port: 53,
|
||||
upstreams: vec![String::from("127.0.0.1:53")]
|
||||
};
|
||||
}
|
||||
None => panic!(),
|
||||
|
||||
@@ -13,6 +13,7 @@ use std::path::Path;
|
||||
use serde::{Serialize, Deserialize, Serializer, Deserializer};
|
||||
// For deserialization
|
||||
use serde::de::{Error as DeError, Visitor};
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error};
|
||||
use crate::hash_is_good;
|
||||
|
||||
@@ -206,6 +207,8 @@ impl<'dd> Deserialize<'dd> for Bytes {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::Keystore;
|
||||
|
||||
#[test]
|
||||
pub fn test_signature() {
|
||||
let keystore: Keystore = Keystore::new();
|
||||
|
||||
+2
-1
@@ -19,6 +19,7 @@ use serde::Deserialize;
|
||||
use web_view::*;
|
||||
use getopts::Options;
|
||||
use simple_logger::{SimpleLogger};
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error, LevelFilter};
|
||||
|
||||
use alfis::{Blockchain, Bytes, Context, Keystore, Transaction};
|
||||
@@ -368,7 +369,7 @@ fn create_server_context(context: Arc<Mutex<Context>>, settings: &Settings) -> A
|
||||
server_context.dns_port = settings.dns.port;
|
||||
server_context.resolve_strategy = match settings.dns.forwarders.is_empty() {
|
||||
true => { ResolveStrategy::Recursive }
|
||||
false => { ResolveStrategy::Forward { host: settings.dns.forwarders[0].clone(), port: 53 }} // TODO refactor to use more resolvers
|
||||
false => { ResolveStrategy::Forward { upstreams: settings.dns.forwarders.clone() }}
|
||||
};
|
||||
server_context.filters.push(Box::new(BlockchainFilter::new(context)));
|
||||
match server_context.initialize() {
|
||||
|
||||
@@ -7,6 +7,7 @@ use chrono::Utc;
|
||||
use crypto::digest::Digest;
|
||||
use crypto::sha2::Sha256;
|
||||
use num_cpus;
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error};
|
||||
|
||||
use crate::{Block, Bytes, Context, hash_is_good, Transaction};
|
||||
|
||||
+3
-2
@@ -2,7 +2,7 @@ extern crate serde;
|
||||
extern crate serde_json;
|
||||
|
||||
use std::{io, thread};
|
||||
use std::io::{Read, Write, Error};
|
||||
use std::io::{Read, Write};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
@@ -10,10 +10,11 @@ use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
|
||||
use mio::{Events, Interest, Poll, Registry, Token};
|
||||
use mio::event::Event;
|
||||
use mio::net::{TcpListener, TcpStream};
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error};
|
||||
|
||||
use crate::{Context, Block, p2p::Message, p2p::State, p2p::Peer, p2p::Peers};
|
||||
use std::net::{SocketAddr, IpAddr, SocketAddrV4, Shutdown, ToSocketAddrs};
|
||||
use std::net::{SocketAddr, IpAddr, SocketAddrV4, ToSocketAddrs};
|
||||
|
||||
const SERVER: Token = Token(0);
|
||||
const POLL_TIMEOUT: Option<Duration> = Some(Duration::from_millis(3000));
|
||||
|
||||
+6
-5
@@ -6,6 +6,7 @@ use crate::p2p::{Peer, State, Message};
|
||||
use crate::p2p::network::LISTEN_PORT;
|
||||
use crate::p2p::network::next;
|
||||
use rand::random;
|
||||
#[allow(unused_imports)]
|
||||
use log::{trace, debug, info, warn, error};
|
||||
|
||||
pub struct Peers {
|
||||
@@ -33,12 +34,12 @@ impl Peers {
|
||||
}
|
||||
|
||||
pub fn close_peer(&mut self, registry: &Registry, token: &Token) {
|
||||
let mut peer = self.peers.get_mut(token);
|
||||
let peer = self.peers.get_mut(token);
|
||||
match peer {
|
||||
Some(mut peer) => {
|
||||
let mut stream = peer.get_stream();
|
||||
Some(peer) => {
|
||||
let stream = peer.get_stream();
|
||||
let _ = stream.shutdown(Shutdown::Both);
|
||||
registry.deregister(stream);
|
||||
let _ = registry.deregister(stream);
|
||||
info!("Peer connection {:?} has shut down", &peer.get_addr());
|
||||
|
||||
if !peer.disabled() && !peer.is_inbound() {
|
||||
@@ -169,7 +170,7 @@ fn skip_addr(addr: &SocketAddr) -> bool {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
SocketAddr::V6(addr) => {
|
||||
SocketAddr::V6(_addr) => {
|
||||
// TODO uncomment when stabilized
|
||||
// if addr.ip().is_unique_local() {
|
||||
// return true;
|
||||
|
||||
Reference in New Issue
Block a user