use unsigned integers in database

Author: mat-1

Cargo.toml

@@ -33,13 +33,7 @@ parking_lot = "0.12.4"
 enum-utils = "0.1.2"
 flate2 = "1.1.2"
 tracing-appender = "0.2.3"
-sqlx = { version = "0.8.6", features = [
-    "runtime-tokio",
-    "postgres",
-    "tls-rustls",
-    "chrono",
-    "uuid",
-] }
+sqlx = { version = "0.8.6", features = ["runtime-tokio", "postgres", "tls-rustls", "chrono", "uuid"] }
 rustc-hash = "2.1.1"
 mongodb = "3.3.0"
 eyre = "0.6.12"

README.md

@@ -29,13 +29,20 @@ It is assumed that you know the basics of server scanning. Otherwise, I recommen
 
 Rename `example-config.toml` to `config.toml` and fill in the fields.
 
-Assuming you already have Postgres installed, you can make the database with the following queries:
+Installing Postgres with the [pg-uint128](https://github.com/pg-uint/pg-uint128) extension is required.
+
+Then, can make the database with the following queries:
 ```sql
 CREATE DATABASE matscan;
 CREATE USER matscan WITH PASSWORD 'replace me!!!';
 GRANT ALL PRIVILEGES ON DATABASE matscan TO matscan;
 GRANT CREATE ON SCHEMA public TO matscan;
 GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO matscan;
+
+-- the user will need superuser permissions on the first run to enable postgres extensions
+ALTER ROLE matscan superuser;
+-- after the first run, you should do ALTER ROLE matscan nosuperuser;
+
 -- PostgreSQL URI is postgres://matscan:replace-me@localhost/matscan
 ```
 

migrations/20250913210747_uint.sql

@@ -0,0 +1,21 @@
+-- migrate to unsigned integers to simplify some logic (especially when comparing ips/ports in the database)
+
+-- requires pg-uint128 to be installed, see https://github.com/pg-uint/pg-uint128
+
+create extension uint128;
+
+alter table server_players drop constraint server_players_server_ip_server_port_fkey;
+alter table servers drop constraint servers_pkey;
+
+alter table servers alter column ip set data type uint4 using ((ip::bigint + 4294967296::bigint) % 4294967296::bigint)::uint4;
+alter table servers alter column port set data type uint2 using ((port::bigint + 65536::bigint) % 65536::bigint)::uint2;
+alter table server_players alter column server_ip set data type uint4 using ((server_ip::bigint + 4294967296::bigint) % 4294967296::bigint)::uint4;
+alter table server_players alter column server_port set data type uint2 using ((server_port::bigint + 65536::bigint) % 65536::bigint)::uint2;
+alter table ips_with_aliased_servers alter column ip set data type uint4 using ((ip::bigint + 4294967296::bigint) % 4294967296::bigint)::uint4;
+
+alter table servers add primary key (ip, port);
+alter table server_players
+    add constraint server_players_server_ip_server_port_fkey
+    foreign key (server_ip, server_port)
+    references servers (ip, port)
+    on delete cascade;

src/database/collect_servers.rs

@@ -9,7 +9,7 @@ use rustc_hash::FxHashMap;
 use sqlx::Row;
 use tracing::info;
 
-use crate::database::Database;
+use crate::database::{Database, PgU16, PgU32};
 
 #[derive(Default)]
 pub struct CollectServersCache {
@@ -118,8 +118,8 @@ impl Database {
 
         let mut servers = Vec::new();
         while let Some(row) = rows.try_next().await? {
-            let ip = Ipv4Addr::from_bits(row.get::<i32, _>(0) as u32);
-            let port = row.get::<i16, _>(1) as u16;
+            let ip = Ipv4Addr::from_bits(row.get::<PgU32, _>(0).0);
+            let port = row.get::<PgU16, _>(1).0;
 
             servers.push(SocketAddrV4::new(ip, port));
 

src/database/migrate_mongo_to_postgres.rs

@@ -8,6 +8,8 @@ use futures_util::TryStreamExt;
 use sqlx::QueryBuilder;
 use uuid::Uuid;
 
+use crate::database::{PgU16, PgU32};
+
 pub async fn do_migration(mongodb_uri: &str, postgres_uri: &str) {
     let client_options = mongodb::options::ClientOptions::parse(mongodb_uri)
         .await
@@ -294,8 +296,8 @@ impl Database {
                 last_time_no_players_online = $27
             "#,
         )
-        .bind(addr.ip().to_bits() as i32)
-        .bind(addr.port() as i16)
+        .bind(PgU32(addr.ip().to_bits()) )
+        .bind(PgU16(addr.port()))
         .bind(s.last_pinged)
         .bind(s.is_online_mode)
         .bind(s.favicon_hash)
@@ -330,8 +332,8 @@ impl Database {
                 "INSERT INTO server_players (server_ip, server_port, uuid, username, online_mode, last_seen, first_seen) ",
             );
             query_builder.push_values(s.player_sample, |mut b, player| {
-                b.push_bind(addr.ip().to_bits() as i32)
-                    .push_bind(addr.port() as i16)
+                b.push_bind(PgU32(addr.ip().to_bits()))
+                    .push_bind(PgU16(addr.port()))
                     .push_bind(player.uuid)
                     .push_bind(player.name.map(|n| n.replace('\0', "")))
                     .push_bind(player.uuid.map(|u| match u.get_version_num() {

src/database/mod.rs

@@ -1,13 +1,26 @@
 pub mod collect_servers;
 pub mod migrate_mongo_to_postgres;
 
-use std::{collections::HashSet, net::Ipv4Addr, sync::Arc, time::Duration};
+use std::{
+    collections::HashSet,
+    fmt::{self, Display},
+    net::Ipv4Addr,
+    ops::Deref,
+    str::FromStr,
+    sync::Arc,
+    time::Duration,
+};
 
 use futures_util::stream::StreamExt;
 use lru_cache::LruCache;
 use parking_lot::Mutex;
 use rustc_hash::FxHashMap;
-use sqlx::{PgPool, Row};
+use sqlx::{
+    PgPool, Postgres, Row,
+    encode::IsNull,
+    error::BoxDynError,
+    postgres::{PgArgumentBuffer, PgTypeInfo, PgValueFormat, PgValueRef},
+};
 use tracing::{error, info};
 
 use crate::database::collect_servers::CollectServersCache;
@@ -80,8 +93,8 @@ impl Database {
             .fetch_all(&self.pool)
             .await?;
         for row in rows {
-            let ip = Ipv4Addr::from_bits(row.get::<i32, _>(0) as u32);
-            let allowed_port = row.get::<i16, _>(1) as u16;
+            let ip = Ipv4Addr::from_bits(row.get::<PgU32, _>(0).0);
+            let allowed_port = row.get::<PgU16, _>(1).0;
             aliased_ips_to_allowed_port.insert(ip, allowed_port);
         }
         self.shared.lock().aliased_ips_to_allowed_port = aliased_ips_to_allowed_port;
@@ -102,14 +115,14 @@ impl Database {
         let mut txn = self.pool.begin().await?;
 
         sqlx::query("INSERT INTO ips_with_aliased_servers (ip, allowed_port) VALUES ($1, $2)")
-            .bind(ip.to_bits() as i32)
-            .bind(allowed_port as i16)
+            .bind(PgU32(ip.to_bits()))
+            .bind(PgU16(allowed_port))
             .execute(&mut *txn)
             .await?;
         // delete all servers with this ip that aren't on the allowed port
         let delete_res = sqlx::query("DELETE FROM servers WHERE ip = $1 AND port != $2")
-            .bind(ip.to_bits() as i32)
-            .bind(allowed_port as i16)
+            .bind(PgU32(ip.to_bits()))
+            .bind(PgU16(allowed_port))
             .execute(&mut *txn)
             .await?;
         let deleted_count = delete_res.rows_affected();
@@ -146,7 +159,7 @@ impl Database {
         .fetch(&self.pool);
 
         while let Some(Ok(row)) = rows.next().await {
-            let ip = row.get::<i32, _>(0);
+            let ip = row.get::<PgU32, _>(0).0;
             let player_count = row.get::<i64, _>(1);
 
             let delete_count = player_count - KEEP_PLAYER_COUNT;
@@ -160,7 +173,7 @@ impl Database {
                 )
                 ",
             )
-            .bind(ip)
+            .bind(PgU32(ip))
             .bind(delete_count)
             .execute(&self.pool)
             .await?;
@@ -174,3 +187,79 @@ impl Database {
 pub fn sanitize_text_for_postgres(s: &str) -> String {
     s.replace('\0', "")
 }
+
+pub struct PgU32(pub u32);
+impl Deref for PgU32 {
+    type Target = u32;
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+impl Display for PgU32 {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+impl FromStr for PgU32 {
+    type Err = std::num::ParseIntError;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(Self(s.parse()?))
+    }
+}
+impl sqlx::Type<Postgres> for PgU32 {
+    fn type_info() -> PgTypeInfo {
+        PgTypeInfo::with_name("uint4")
+    }
+}
+impl sqlx::Decode<'_, Postgres> for PgU32 {
+    fn decode(value: PgValueRef<'_>) -> Result<Self, BoxDynError> {
+        Ok(match value.format() {
+            PgValueFormat::Binary => Self(u32::from_be_bytes(value.as_bytes()?.try_into()?)),
+            PgValueFormat::Text => value.as_str()?.parse()?,
+        })
+    }
+}
+impl sqlx::Encode<'_, Postgres> for PgU32 {
+    fn encode_by_ref(&self, buf: &mut PgArgumentBuffer) -> Result<IsNull, BoxDynError> {
+        buf.extend(&self.to_be_bytes());
+        Ok(IsNull::No)
+    }
+}
+
+pub struct PgU16(pub u16);
+impl Deref for PgU16 {
+    type Target = u16;
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+impl Display for PgU16 {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+impl FromStr for PgU16 {
+    type Err = std::num::ParseIntError;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(Self(s.parse()?))
+    }
+}
+impl sqlx::Type<Postgres> for PgU16 {
+    fn type_info() -> PgTypeInfo {
+        PgTypeInfo::with_name("uint2")
+    }
+}
+impl sqlx::Decode<'_, Postgres> for PgU16 {
+    fn decode(value: PgValueRef<'_>) -> Result<Self, BoxDynError> {
+        Ok(match value.format() {
+            PgValueFormat::Binary => Self(u16::from_be_bytes(value.as_bytes()?.try_into()?)),
+            PgValueFormat::Text => value.as_str()?.parse()?,
+        })
+    }
+}
+impl sqlx::Encode<'_, Postgres> for PgU16 {
+    fn encode_by_ref(&self, buf: &mut PgArgumentBuffer) -> Result<IsNull, BoxDynError> {
+        buf.extend(&self.to_be_bytes());
+        Ok(IsNull::No)
+    }
+}

src/net/raw_sockets.rs

@@ -7,13 +7,13 @@ use std::{
 
 #[repr(C)]
 #[derive(Debug, Clone)]
-struct ifreq {
+struct Ifreq {
     ifr_name: [libc::c_char; libc::IF_NAMESIZE],
     ifr_data: libc::c_int, /* ifr_ifindex or ifr_mtu */
 }
 
-fn ifreq_for(name: &str) -> ifreq {
-    let mut ifreq = ifreq {
+fn ifreq_for(name: &str) -> Ifreq {
+    let mut ifreq = Ifreq {
         ifr_name: [0; libc::IF_NAMESIZE],
         ifr_data: 0,
     };
@@ -25,11 +25,11 @@ fn ifreq_for(name: &str) -> ifreq {
 
 fn ifreq_ioctl(
     lower: libc::c_int,
-    ifreq: &mut ifreq,
+    ifreq: &mut Ifreq,
     cmd: libc::c_ulong,
 ) -> io::Result<libc::c_int> {
     unsafe {
-        let res = libc::ioctl(lower, cmd as _, ifreq as *mut ifreq);
+        let res = libc::ioctl(lower, cmd as _, ifreq as *mut Ifreq);
         if res == -1 {
             return Err(io::Error::last_os_error());
         }
@@ -47,7 +47,7 @@ pub const ETH_P_IEEE802154: libc::c_short = 0x00F6;
 pub struct RawSocket {
     protocol: libc::c_short,
     lower: libc::c_int,
-    ifreq: ifreq,
+    ifreq: Ifreq,
 }
 
 impl AsRawFd for RawSocket {

src/processing.rs

@@ -16,7 +16,10 @@ use sqlx::Row;
 use tokio::time::sleep;
 
 use crate::{
-    config::Config, database::Database, processing::minecraft::SamplePlayer, terminal_colors::*,
+    config::Config,
+    database::{Database, PgU16, PgU32},
+    processing::minecraft::SamplePlayer,
+    terminal_colors::*,
 };
 
 pub struct SharedData {
@@ -132,8 +135,8 @@ async fn handle_response_futures(
         tasks.push(async move {
             let mut processed_server_status = if let Ok(row) =
                 sqlx::query("SELECT last_pinged FROM servers WHERE ip = $1 AND port = $2")
-                    .bind(addr.ip().to_bits() as i32)
-                    .bind(addr.port() as i16)
+                    .bind(PgU32(addr.ip().to_bits()))
+                    .bind(PgU16(addr.port()))
                     .fetch_one(&db.pool)
                     .await
             {

src/processing/minecraft/mod.rs

@@ -21,7 +21,7 @@ use uuid::Uuid;
 use super::{ProcessableProtocol, SharedData};
 use crate::{
     config::Config,
-    database::{CachedIpHash, Database, sanitize_text_for_postgres},
+    database::{CachedIpHash, Database, PgU16, PgU32, sanitize_text_for_postgres},
     processing::minecraft::{
         passive_fingerprint::{PassiveMinecraftFingerprint, generate_passive_fingerprint},
         snipe::maybe_log_sniped,
@@ -335,8 +335,8 @@ pub async fn insert_server_to_db(
 
     let mut qb = InsertServerQueryBuilder::new();
     let now = chrono::Utc::now();
-    qb.field("ip", target.ip().to_bits() as i32);
-    qb.field("port", target.port() as i16);
+    qb.field("ip", PgU32(target.ip().to_bits()));
+    qb.field("port", PgU16(target.port()));
     qb.field("last_pinged", now);
     qb.field("is_online_mode", r.is_online_mode);
     qb.field("favicon_hash", r.favicon_hash);
@@ -383,8 +383,8 @@ pub async fn insert_server_to_db(
             "INSERT INTO server_players (server_ip, server_port, uuid, username, online_mode, last_seen) ",
         );
         query_builder.push_values(&r.player_sample, |mut b, player| {
-            b.push_bind(target.ip().to_bits() as i32)
-                .push_bind(target.port() as i16)
+            b.push_bind(PgU32(target.ip().to_bits()))
+                .push_bind(PgU16(target.port()))
                 .push_bind(player.uuid)
                 .push_bind(player.name.clone())
                 .push_bind(match player.uuid.get_version_num() {

src/processing/minecraft/snipe.rs

@@ -2,7 +2,7 @@ use std::{collections::HashMap, net::SocketAddrV4, sync::Arc};
 
 use parking_lot::Mutex;
 
-use crate::{config::Config, database::Database, processing::{SharedData, minecraft::{ANONYMOUS_PLAYER_NAME, PingResponse, }}};
+use crate::{config::Config, database::{Database, PgU16, PgU32}, processing::{SharedData, minecraft::{ANONYMOUS_PLAYER_NAME, PingResponse, }}};
 
 pub fn maybe_log_sniped(
     shared: &Arc<Mutex<SharedData>>,
@@ -98,8 +98,8 @@ pub fn maybe_log_sniped(
                             "SELECT FROM server_players WHERE username = '$1' AND server_ip = $2 AND server_port = $3 LIMIT 1",
                         )
                             .bind(ANONYMOUS_PLAYER_NAME)
-                            .bind(target.ip().to_bits() as i32)
-                            .bind(target.port() as i16).fetch_optional(&database.pool).await {
+                            .bind(PgU32(target.ip().to_bits()))
+                            .bind(PgU16(target.port())).fetch_optional(&database.pool).await {
                                 let has_historical_anon = has_historical_anon_res.is_some();
                                 if !has_historical_anon {
                                     send_to_webhook(