use std::{
hash::{DefaultHasher, Hash, Hasher},
io::Read,
net::IpAddr,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use bitcoin::{
consensus,
p2p::{address::AddrV2, ServiceFlags},
};
pub mod io;
const ONE_MINUTE: Duration = Duration::from_secs(60);
const ONE_WEEK: Duration = Duration::from_secs(604800);
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Record {
addr: AddrV2,
port: u16,
source: SourceId,
services: ServiceFlags,
failed_attempts: u8,
last_connection: Option<Duration>,
last_attempt: Option<Duration>,
}
impl Record {
fn compute_size(&self) -> u8 {
let mut size = 0;
match &self.addr {
AddrV2::I2p(_) => size += 34,
AddrV2::Ipv4(_) => size += 6,
AddrV2::Ipv6(_) => size += 18,
AddrV2::TorV2(_) => size += 12,
AddrV2::TorV3(_) => size += 34,
AddrV2::Cjdns(_) => size += 18,
AddrV2::Unknown(len, _) => size += *len,
}
size += 2;
size += self.source.0.len() as u8;
size += 8;
size += 1;
size += 9;
size += 9;
size
}
pub fn new(addr: AddrV2, port: u16, services: ServiceFlags, source: &IpAddr) -> Self {
let source = source.source_id();
Self {
addr,
port,
source,
services,
failed_attempts: 0,
last_connection: None,
last_attempt: None,
}
}
pub fn new_from_addrv2_source(
addr: AddrV2,
port: u16,
services: ServiceFlags,
source: &AddrV2,
) -> Self {
let source = match source {
AddrV2::Ipv4(ip) => IpAddr::V4(*ip).source_id(),
AddrV2::Ipv6(ip) => IpAddr::V6(*ip).source_id(),
_ => SourceId([1u8; 4]),
};
Self {
addr,
port,
source,
services,
failed_attempts: 0,
last_connection: None,
last_attempt: None,
}
}
pub fn deserialize<R: Read>(reader: &mut R) -> Result<Self, std::io::Error> {
let mut size_buf = [0u8; 1];
reader.read_exact(&mut size_buf)?;
let size = u8::from_le_bytes(size_buf);
let mut content_buf = vec![0u8; size as usize];
reader.read_exact(&mut content_buf)?;
let (addr, len) =
consensus::deserialize_partial::<AddrV2>(&content_buf).expect("must have 33 bytes");
let mut content_slice = &content_buf[len..];
let mut port_buf = [0u8; 2];
content_slice.read_exact(&mut port_buf)?;
let port = u16::from_le_bytes(port_buf);
let mut source_buf = [0u8; 4];
content_slice.read_exact(&mut source_buf)?;
let source = SourceId(source_buf);
let mut service_buf = [0u8; 8];
content_slice.read_exact(&mut service_buf)?;
let services = ServiceFlags::from(u64::from_le_bytes(service_buf));
let mut failed_buf = [0u8; 1];
content_slice.read_exact(&mut failed_buf)?;
let failed_attempts = u8::from_le_bytes(failed_buf);
let mut record = Record {
addr,
port,
source,
services,
failed_attempts,
last_connection: None,
last_attempt: None,
};
let mut last_attempt_buf = [0u8; 1];
content_slice.read_exact(&mut last_attempt_buf)?;
let should_read = u8::from_le_bytes(last_attempt_buf);
match should_read {
0u8 => {
content_slice.read_exact(&mut [0u8; 8])?;
}
1u8 => {
let mut time_buf = [0u8; 8];
content_slice.read_exact(&mut time_buf)?;
let secs = u64::from_le_bytes(time_buf);
record.last_attempt = Some(Duration::from_secs(secs));
}
_ => panic!("invalid time encoding"),
}
let mut last_conn_buf = [0u8; 1];
content_slice.read_exact(&mut last_conn_buf)?;
let should_read = u8::from_le_bytes(last_conn_buf);
match should_read {
0u8 => {
content_slice.read_exact(&mut [0u8; 8])?;
}
1u8 => {
let mut time_buf = [0u8; 8];
content_slice.read_exact(&mut time_buf)?;
let secs = u64::from_le_bytes(time_buf);
record.last_connection = Some(Duration::from_secs(secs));
}
_ => panic!("invalid time encoding"),
}
Ok(record)
}
fn destination_id(&self) -> DestinationId {
self.addr.destination_id()
}
pub fn network_addr(&self) -> (AddrV2, u16) {
(self.addr.clone(), self.port)
}
pub fn service_flags(&self) -> ServiceFlags {
self.services
}
pub fn update_service_flags(&mut self, flags: ServiceFlags) {
self.services = flags;
}
pub fn serialize(self) -> Vec<u8> {
let len = self.compute_size();
let mut buf = Vec::with_capacity(len.into());
buf.push(len);
let addr = consensus::serialize(&self.addr);
buf.extend(addr);
buf.extend(self.port.to_le_bytes());
buf.extend(self.source.0);
buf.extend(self.services.to_u64().to_le_bytes());
buf.push(self.failed_attempts);
if let Some(last_attempt) = self.last_attempt {
buf.push(0x01);
let secs = last_attempt.as_secs().to_le_bytes();
buf.extend(secs);
} else {
buf.extend([0u8; 9]);
}
if let Some(last_conn) = self.last_connection {
buf.push(0x01);
let secs = last_conn.as_secs().to_le_bytes();
buf.extend(secs);
} else {
buf.extend([0u8; 9]);
}
buf
}
pub fn is_terrible(&self, maximum_tries: u8, maximum_weekly_tries: u8) -> bool {
if let Some(attempt) = self.last_attempt {
if attempt < ONE_MINUTE {
return false;
}
if self.failed_attempts > maximum_weekly_tries && attempt < ONE_WEEK {
return true;
}
}
if self.failed_attempts > maximum_tries {
return true;
}
false
}
}
#[derive(Debug)]
pub struct Table<const B: usize, const S: usize, const W: usize> {
buckets: [Bucket<S>; B],
}
impl<const B: usize, const S: usize, const W: usize> Table<B, S, W> {
const RUN: usize = B / W;
fn derive_bucket(record: &Record) -> usize {
let salt = u32::from_le_bytes(record.source.0) as usize % W;
let range = (salt * Self::RUN) % B;
let index = u32::from_le_bytes(record.destination_id().0) as usize % W;
(range + index) % B
}
fn random_bucket() -> usize {
let mut hasher = DefaultHasher::new();
SystemTime::now().hash(&mut hasher);
u32::from_le_bytes(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
) as usize
% B
}
fn random_slot() -> usize {
let mut hasher = DefaultHasher::new();
SystemTime::now().hash(&mut hasher);
u32::from_le_bytes(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
) as usize
% S
}
fn random_from_bucket(bucket: &Bucket<S>) -> Option<Record> {
if bucket.is_empty() {
return None;
}
let slot_index = Self::random_slot();
let mut tmp = (slot_index + 1) % S;
while tmp.ne(&slot_index) {
let record = bucket.get(tmp);
if let Some(record) = record {
let Some(last_attempt) = record.last_attempt else {
return Some(record);
};
if last_attempt > ONE_MINUTE {
return Some(record);
}
}
tmp = (tmp + 1) % S;
}
bucket.get(slot_index)
}
pub fn new() -> Self {
let buckets: [Bucket<S>; B] = [const { Bucket::new() }; B];
Self { buckets }
}
pub fn add(&mut self, record: &Record) -> Option<Record> {
let bucket_index = Self::derive_bucket(record);
self.buckets[bucket_index].add(record.clone())
}
pub fn remove(&mut self, record: &Record) {
let bucket_index = Self::derive_bucket(record);
self.buckets[bucket_index].remove(record);
}
pub fn count(&self, record: &Record) -> usize {
self.buckets
.iter()
.filter(|bucket| bucket.has_record(record))
.count()
}
pub fn is_empty(&self) -> bool {
self.buckets.iter().all(|bucket| bucket.is_empty())
}
pub fn select(&self) -> Option<Record> {
if self.is_empty() {
return None;
};
let bucket_index = Self::random_bucket();
let bucket = &self.buckets[bucket_index];
if bucket.is_empty() {
let mut tmp = (bucket_index + 1) % B;
while tmp.ne(&bucket_index) {
let bucket = &self.buckets[tmp];
let random_record = Self::random_from_bucket(bucket);
if random_record.is_some() {
return random_record;
}
tmp = (tmp + 1) % B;
}
None
} else {
Self::random_from_bucket(bucket)
}
}
pub fn successful_connection(&mut self, record: &Record) {
let bucket_index = Self::derive_bucket(record);
let bucket = &mut self.buckets[bucket_index];
bucket.successful_connection(record);
}
pub fn failed_connection(&mut self, record: &Record) {
let bucket_index = Self::derive_bucket(record);
let bucket = &mut self.buckets[bucket_index];
bucket.failed_connection(record);
}
}
impl<const B: usize, const S: usize, const W: usize> Default for Table<B, S, W> {
fn default() -> Self {
Table::<B, S, W>::new()
}
}
#[derive(Debug)]
struct Bucket<const S: usize> {
records: [Option<Record>; S],
}
impl<const S: usize> Bucket<S> {
fn derive_slot(record: &Record) -> usize {
let index = u32::from_le_bytes(record.destination_id().0) as usize;
index % S
}
const fn new() -> Self {
let records: [Option<Record>; S] = [const { None }; S];
Self { records }
}
fn add(&mut self, record: Record) -> Option<Record> {
let slot = Self::derive_slot(&record);
match &self.records[slot] {
Some(occupied) => Some(occupied.clone()),
None => {
self.records[slot] = Some(record);
None
}
}
}
fn remove(&mut self, record: &Record) {
let slot = Self::derive_slot(record);
self.records[slot] = None;
}
fn has_record(&self, record: &Record) -> bool {
let slot = Self::derive_slot(record);
match &self.records[slot] {
Some(cmp) => cmp.eq(record),
None => false,
}
}
fn is_empty(&self) -> bool {
self.records.iter().all(|record| record.is_none())
}
fn successful_connection(&mut self, record: &Record) {
let slot = Self::derive_slot(record);
let new_flags = record.services;
if let Some(record) = &mut self.records[slot] {
record.last_attempt = Some(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("time went backwards"),
);
record.last_connection = Some(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("time went backwards"),
);
record.failed_attempts = 0;
record.services = new_flags;
}
}
fn failed_connection(&mut self, record: &Record) {
let slot = Self::derive_slot(record);
if let Some(record) = &mut self.records[slot] {
record.last_attempt = Some(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("time went backwards"),
);
record.failed_attempts += 1;
}
}
fn get(&self, index: usize) -> Option<Record> {
let index = index % S;
self.records[index].clone()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, std::hash::Hash)]
struct SourceId([u8; 4]);
trait SourceIdExt {
fn source_id(&self) -> SourceId;
}
impl SourceIdExt for IpAddr {
fn source_id(&self) -> SourceId {
let mut hasher = DefaultHasher::new();
match self {
Self::V4(ipv4) => {
let octets = ipv4.octets();
let first_two_octets = [octets[0], octets[1]];
first_two_octets.hash(&mut hasher);
let hash = hasher.finish();
let bytes: [u8; 4] = hash.to_le_bytes()[..4].try_into().expect("hash is u64");
SourceId(bytes)
}
Self::V6(ipv6) => {
let octets = ipv6.octets();
let first_four_octets = [octets[0], octets[1], octets[2], octets[3]];
first_four_octets.hash(&mut hasher);
let hash = hasher.finish();
let bytes = hash.to_le_bytes()[..4].try_into().expect("hash is u64");
SourceId(bytes)
}
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, std::hash::Hash)]
struct DestinationId([u8; 4]);
trait DestinationIdExt {
fn destination_id(&self) -> DestinationId;
}
impl DestinationIdExt for AddrV2 {
fn destination_id(&self) -> DestinationId {
let mut hasher = DefaultHasher::new();
match self {
Self::Ipv4(ipv4) => {
ipv4.octets().hash(&mut hasher);
DestinationId(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
)
}
Self::Ipv6(ipv6) => {
ipv6.octets().hash(&mut hasher);
DestinationId(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
)
}
Self::I2p(i2p) => {
i2p.hash(&mut hasher);
DestinationId(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
)
}
Self::TorV3(tv3) => {
tv3.hash(&mut hasher);
DestinationId(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
)
}
Self::Cjdns(ipv6) => {
ipv6.octets().hash(&mut hasher);
DestinationId(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
)
}
_ => {
"unknown network address".hash(&mut hasher);
DestinationId(
hasher.finish().to_le_bytes()[..4]
.try_into()
.expect("hash is u64"),
)
}
}
}
}
#[cfg(test)]
mod tests {
use std::{
hash::{DefaultHasher, Hash, Hasher},
net::{IpAddr, Ipv4Addr},
time::SystemTime,
};
use bitcoin::p2p::{address::AddrV2, ServiceFlags};
use crate::{Record, Table};
const LOCAL_HOST: Ipv4Addr = Ipv4Addr::new(127, 0, 0, 1);
const DUMB: AddrV2 = AddrV2::Ipv4(LOCAL_HOST);
const BUCKETS: usize = 256;
const SLOTS: usize = 16;
const RANGE: usize = 16;
pub fn random_record() -> Record {
let mut hasher = DefaultHasher::new();
let now = SystemTime::now();
now.hash(&mut hasher);
let bytes = hasher.finish();
let ip = bytes.to_le_bytes();
let dest = Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]);
let source = Ipv4Addr::new(ip[4], ip[5], ip[6], ip[7]);
let now = SystemTime::now();
now.hash(&mut hasher);
let bytes = hasher.finish();
let addr_v2 = AddrV2::Ipv4(dest);
let mut record = Record::new(addr_v2, 8333, ServiceFlags::NETWORK, &IpAddr::V4(source));
record.failed_attempts += bytes.to_le_bytes()[0];
record
}
#[test]
fn test_simple_table_situations() {
let mut table = Table::<BUCKETS, SLOTS, RANGE>::new();
assert!(table.is_empty());
assert!(table.select().is_none());
let record = Record::new(DUMB, 8333, ServiceFlags::NONE, &IpAddr::V4(LOCAL_HOST));
table.add(&record);
assert!(!table.is_empty());
for _ in 0..BUCKETS * SLOTS {
assert!(table.select().is_some());
}
for _ in 0..BUCKETS * SLOTS {
assert!(table.add(&record).is_some());
}
assert_eq!(table.count(&record), 1);
}
#[test]
fn test_encoding_roundtrip() {
for _ in 0..BUCKETS * SLOTS {
let want = random_record();
let bytes = want.clone().serialize();
let got = Record::deserialize(&mut bytes.as_slice()).unwrap();
assert_eq!(want, got);
}
}
}