use std::fs;
use std::hash::{Hash, Hasher};
use std::io;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use flate2::read::MultiGzDecoder;
use reqwest::{Client, RequestBuilder, StatusCode};
use tokio::task::JoinHandle;
use tracing::{error, info, warn};
use super::verify::Guard;
use super::{DatabaseFormat, GeoIpDownloadError};
use crate::Secret;
const PART_EXT: &str = "part";
const STAGE_EXT: &str = "staged";
const LOCK_EXT: &str = "lock";
const PROGRESS_INTERVAL: Duration = Duration::from_secs(30);
const RESUME_WINDOW_SECS: u64 = 24 * 60 * 60;
const MMDB_MARKER: &[u8] = b"\xab\xcd\xefMaxMind.com";
const METADATA_TAIL_BYTES: u64 = 128 * 1024;
const SHA256_HEX_LEN: usize = 64;
const CHECKSUM_BODY_LIMIT: usize = 4 * 1024;
const WRITE_BATCH_BYTES: usize = 1024 * 1024;
const REDACTED: &str = "<redacted>";
pub(crate) fn loggable(url: &str) -> String {
let (mut safe, query) = match reqwest::Url::parse(url) {
Ok(mut parsed) => {
let query = parsed.query().map(str::to_owned);
parsed.set_query(None);
parsed.set_fragment(None);
let _ = parsed.set_username("");
let _ = parsed.set_password(None);
(parsed.to_string(), query)
}
Err(_) => unparsed(url),
};
let Some(query) = query else {
return safe;
};
for (position, pair) in query.split('&').enumerate() {
safe.push(if position == 0 { '?' } else { '&' });
safe.push_str(pair.split_once('=').map_or(pair, |(name, _)| name));
safe.push('=');
safe.push_str(REDACTED);
}
safe
}
fn unparsed(url: &str) -> (String, Option<String>) {
let (base, query) = match url.split_once('?') {
Some((base, query)) => (base, Some(query.to_string())),
None => (url, None),
};
let safe = match base.split_once("://") {
Some((scheme, rest)) => {
let (authority, path) = rest.split_at(rest.find('/').unwrap_or(rest.len()));
let host = authority
.rsplit_once('@')
.map_or(authority, |(_, host)| host);
format!("{scheme}://{host}{path}")
}
None => base
.rsplit_once('@')
.map_or(base, |(_, tail)| tail)
.to_string(),
};
(safe, query)
}
const DIGEST_BUFFER_BYTES: usize = 64 * 1024;
const MAX_EXPANDED_BYTES: u64 = 4 * 1024 * 1024 * 1024;
const MAX_BODY_BYTES: u64 = MAX_EXPANDED_BYTES;
const MARKUP_OPENERS: [&[u8]; 3] = [b"<!doctype", b"<html", b"<?xml"];
const MARKUP_HEAD_BYTES: usize = 64;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Archive {
Raw,
Gzip,
TarGz { member: &'static str },
}
#[derive(Clone)]
pub(crate) enum Credential {
None,
Basic {
username: Secret,
password: Secret,
fields: &'static str,
},
QueryToken {
name: &'static str,
value: Secret,
fields: &'static str,
},
}
impl std::fmt::Debug for Credential {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let kind = match self {
Self::None => "None",
Self::Basic { .. } => "Basic(***REDACTED***)",
Self::QueryToken { .. } => "QueryToken(***REDACTED***)",
};
f.write_str(kind)
}
}
impl Credential {
fn apply(&self, request: RequestBuilder) -> RequestBuilder {
match self {
Self::None => request,
Self::Basic {
username, password, ..
} => request.basic_auth(username.expose(), Some(password.expose())),
Self::QueryToken { name, value, .. } => request.query(&[(*name, value.expose())]),
}
}
fn identify(&self, state: &mut impl Hasher) {
match self {
Self::None => state.write_u8(0),
Self::Basic {
username, password, ..
} => {
state.write_u8(1);
username.expose().hash(state);
password.expose().hash(state);
}
Self::QueryToken { name, value, .. } => {
state.write_u8(2);
name.hash(state);
value.expose().hash(state);
}
}
}
const fn fields(&self) -> Option<&'static str> {
match self {
Self::None => None,
Self::Basic { fields, .. } | Self::QueryToken { fields, .. } => Some(fields),
}
}
}
pub(crate) fn client(
injected: Option<&Client>,
connect_timeout: Duration,
read_timeout: Duration,
) -> Result<Client, GeoIpDownloadError> {
if let Some(client) = injected {
return Ok(client.clone());
}
Ok(Client::builder()
.tls_backend_rustls()
.user_agent(format!("factbook/{}", crate::VERSION))
.referer(false)
.connect_timeout(connect_timeout)
.read_timeout(read_timeout)
.build()?)
}
#[derive(Debug)]
pub(crate) struct Transfer {
pub(crate) url: String,
pub(crate) fallback_url: Option<String>,
pub(crate) checksum_url: Option<String>,
pub(crate) dest: PathBuf,
pub(crate) archive: Archive,
pub(crate) format: DatabaseFormat,
pub(crate) credential: Credential,
}
impl Transfer {
pub(crate) fn identify(&self, state: &mut impl Hasher) {
self.url.hash(state);
self.fallback_url.hash(state);
self.checksum_url.hash(state);
self.dest.hash(state);
self.credential.identify(state);
}
#[cfg(test)]
pub(crate) async fn run(self, client: &Client) -> Result<PathBuf, GeoIpDownloadError> {
self.run_guarded(client, Guard::OFF).await
}
#[cfg(test)]
pub(crate) async fn run_within(
self,
client: &Client,
limit: u64,
) -> Result<PathBuf, GeoIpDownloadError> {
self.fetch(client, Guard::OFF, limit).await
}
pub(crate) async fn run_guarded(
self,
client: &Client,
guard: Guard,
) -> Result<PathBuf, GeoIpDownloadError> {
self.fetch(client, guard, MAX_BODY_BYTES).await
}
async fn fetch(
self,
client: &Client,
guard: Guard,
limit: u64,
) -> Result<PathBuf, GeoIpDownloadError> {
if let Some(parent) = self.dest.parent() {
fs::create_dir_all(parent)?;
}
let part = with_extension(&self.dest, PART_EXT);
let lock_path = with_extension(&self.dest, LOCK_EXT);
let lock = fs::OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&lock_path)?;
match lock.try_lock() {
Ok(()) => {}
Err(fs::TryLockError::WouldBlock) => {
return Err(GeoIpDownloadError::Busy {
path: self.dest.display().to_string(),
});
}
Err(fs::TryLockError::Error(e)) => {
warn!(
dest = %self.dest.display(),
error = %e,
"file locking unavailable, so a concurrent writer cannot be detected"
);
}
}
info!(
url = %loggable(&self.url),
dest = %self.dest.display(),
archive = ?self.archive,
"downloading GeoIP database"
);
let lock = Arc::new(lock);
let bytes = match self.stream_with_fallback(client, &part, &lock, limit).await {
Ok(bytes) => bytes,
Err(e) => {
if fs::metadata(&part).is_ok_and(|meta| meta.len() == 0) {
let _ = fs::remove_file(&part);
}
return Err(e);
}
};
if let Some(checksum_url) = self.checksum_url.clone() {
let expected = fetch_checksum(client, &checksum_url, &self.credential).await?;
let checked = part.clone();
let actual = tokio::task::spawn_blocking(move || sha256_of(&checked)).await??;
if actual != expected {
let _ = fs::remove_file(&part);
return Err(self.rejected(GeoIpDownloadError::ChecksumMismatch {
url: loggable(&checksum_url),
expected,
actual,
}));
}
}
let dest = self.dest.clone();
let archive = self.archive;
let format = self.format;
let staged = with_extension(&dest, STAGE_EXT);
let final_size = tokio::task::spawn_blocking(move || {
#[cfg(test)]
super::testkit::checkpoint(&dest);
let result = materialise_guarded(&part, &staged, &dest, archive, format, &guard);
let _ = fs::remove_file(&part);
if result.is_err() {
let _ = fs::remove_file(&staged);
}
drop(lock);
result
})
.await?
.map_err(|e| self.rejected(e))?;
info!(
dest = %self.dest.display(),
downloaded_bytes = bytes,
database_bytes = final_size,
"GeoIP database ready"
);
Ok(self.dest)
}
fn rejected(&self, error: GeoIpDownloadError) -> GeoIpDownloadError {
error!(
url = %loggable(&self.url),
dest = %self.dest.display(),
reason = %error,
"rejected a GeoIP download, keeping the file already on disk"
);
error
}
async fn stream_with_fallback(
&self,
client: &Client,
part: &Path,
lock: &Arc<fs::File>,
limit: u64,
) -> Result<u64, GeoIpDownloadError> {
let primary = self.stream_to(client, &self.url, part, lock, limit).await;
let Some(fallback) = self.fallback_url.as_deref() else {
return primary;
};
if !matches!(
primary,
Err(GeoIpDownloadError::UnexpectedStatus { status: 404, .. })
) {
return primary;
}
warn!(
url = %loggable(&self.url),
fallback = %loggable(fallback),
"GeoIP database not published at that URL, trying the previous one"
);
let _ = fs::remove_file(part);
self.stream_to(client, fallback, part, lock, limit).await
}
async fn stream_to(
&self,
client: &Client,
url: &str,
part: &Path,
lock: &Arc<fs::File>,
limit: u64,
) -> Result<u64, GeoIpDownloadError> {
let offset = resumable_offset(part);
let resume_from = if resume_is_checkable(self.archive, self.checksum_url.is_some()) {
offset
} else {
0
};
match self
.stream_from(client, url, part, lock, resume_from, limit)
.await
{
Err(GeoIpDownloadError::UnexpectedStatus { status: 416, .. }) if resume_from > 0 => {
let _ = fs::remove_file(part);
self.stream_from(client, url, part, lock, 0, limit).await
}
other => other,
}
}
async fn stream_from(
&self,
client: &Client,
url: &str,
part: &Path,
lock: &Arc<fs::File>,
resume_from: u64,
limit: u64,
) -> Result<u64, GeoIpDownloadError> {
let mut request = self.credential.apply(client.get(url));
if resume_from > 0 {
request = request.header(reqwest::header::RANGE, format!("bytes={resume_from}-"));
}
let mut response = request.send().await.map_err(reqwest::Error::without_url)?;
let status = response.status();
if !status.is_success() {
return Err(refused(url, status, response.headers(), &self.credential));
}
let resuming = resume_from > 0 && status == StatusCode::PARTIAL_CONTENT;
let mut written = if resuming { resume_from } else { 0 };
let carried = written;
let expected = response
.content_length()
.map(|length| length.saturating_add(written));
if expected.is_some_and(|expected| expected > limit) {
let _ = fs::remove_file(part);
return Err(GeoIpDownloadError::TooLarge {
url: loggable(url),
limit,
});
}
let file = if resuming {
fs::OpenOptions::new().append(true).open(part)?
} else {
fs::File::create(part)?
};
let mut writer = PartWriter::new(file, part, lock);
if resume_from > 0 {
info!(
url = %loggable(url),
resumed_from = resume_from,
honoured = resuming,
"continuing an interrupted GeoIP download"
);
}
let started = Instant::now();
let mut reported = Instant::now();
loop {
let chunk = match response.chunk().await {
Ok(Some(chunk)) => chunk,
Ok(None) => break,
Err(e) => {
if writer.finish().await.is_err() {
let _ = fs::remove_file(part);
}
return Err(e.without_url().into());
}
};
if written.saturating_add(chunk.len() as u64) > limit {
drop(writer);
let _ = fs::remove_file(part);
return Err(GeoIpDownloadError::TooLarge {
url: loggable(url),
limit,
});
}
if let Err(e) = writer.write(&chunk).await {
drop(writer);
let _ = fs::remove_file(part);
return Err(e.into());
}
written += chunk.len() as u64;
if reported.elapsed() >= PROGRESS_INTERVAL {
info!(
url = %loggable(url),
downloaded_bytes = written,
expected_bytes = expected,
rate_kib_per_sec = rate_kib_per_sec(written - carried, started.elapsed()),
"downloading GeoIP database"
);
reported = Instant::now();
}
}
if let Err(e) = writer.finish().await {
let _ = fs::remove_file(part);
return Err(e.into());
}
if let Some(expected) = expected
&& written != expected
{
return Err(GeoIpDownloadError::Truncated {
url: loggable(url),
expected,
actual: written,
});
}
Ok(written)
}
}
struct PartWriter {
file: Option<fs::File>,
pending: Option<JoinHandle<(fs::File, Vec<u8>, io::Result<()>)>>,
gathered: Vec<u8>,
spare: Vec<u8>,
lock: Arc<fs::File>,
#[cfg(test)]
path: PathBuf,
}
impl PartWriter {
fn new(file: fs::File, path: &Path, lock: &Arc<fs::File>) -> Self {
#[cfg(not(test))]
let _ = path;
Self {
file: Some(file),
pending: None,
gathered: Vec::with_capacity(WRITE_BATCH_BYTES),
spare: Vec::new(),
lock: Arc::clone(lock),
#[cfg(test)]
path: path.to_path_buf(),
}
}
async fn write(&mut self, chunk: &[u8]) -> io::Result<()> {
self.gathered.extend_from_slice(chunk);
if self.gathered.len() >= WRITE_BATCH_BYTES {
self.send().await?;
}
Ok(())
}
async fn finish(mut self) -> io::Result<()> {
if !self.gathered.is_empty() {
self.send().await?;
}
self.settle().await?;
match self.file.take() {
Some(mut file) => io::Write::flush(&mut file),
None => Err(lost()),
}
}
async fn send(&mut self) -> io::Result<()> {
self.settle().await?;
let mut file = self.file.take().ok_or_else(lost)?;
let mut batch = std::mem::replace(&mut self.gathered, std::mem::take(&mut self.spare));
let lock = Arc::clone(&self.lock);
#[cfg(test)]
let path = self.path.clone();
self.pending = Some(tokio::task::spawn_blocking(move || {
#[cfg(test)]
super::testkit::checkpoint(&path);
let written = io::Write::write_all(&mut file, &batch);
drop(lock);
batch.clear();
(file, batch, written)
}));
Ok(())
}
async fn settle(&mut self) -> io::Result<()> {
let Some(write) = self.pending.take() else {
return Ok(());
};
let (file, spare, written) = write.await.map_err(io::Error::from)?;
self.file = Some(file);
self.spare = spare;
written
}
}
fn lost() -> io::Error {
io::Error::other("the part file was lost to a write that did not complete")
}
fn refused(
url: &str,
status: StatusCode,
headers: &reqwest::header::HeaderMap,
credential: &Credential,
) -> GeoIpDownloadError {
if status == StatusCode::TOO_MANY_REQUESTS {
return GeoIpDownloadError::RateLimited {
url: loggable(url),
retry_after_secs: headers
.get(reqwest::header::RETRY_AFTER)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.trim().parse().ok()),
};
}
if let Some(fields) = credential.fields() {
if status == StatusCode::UNAUTHORIZED {
return GeoIpDownloadError::CredentialRejected {
url: loggable(url),
status: status.as_u16(),
fields,
};
}
if status == StatusCode::FORBIDDEN {
return GeoIpDownloadError::NotEntitled { url: loggable(url) };
}
}
GeoIpDownloadError::UnexpectedStatus {
url: loggable(url),
status: status.as_u16(),
}
}
const fn resume_is_checkable(archive: Archive, has_digest: bool) -> bool {
has_digest || !matches!(archive, Archive::Raw)
}
fn resumable_offset(part: &Path) -> u64 {
let Ok(metadata) = fs::metadata(part) else {
return 0;
};
let age = metadata
.modified()
.ok()
.and_then(|modified| SystemTime::now().duration_since(modified).ok());
if age.is_some_and(|age| age.as_secs() < RESUME_WINDOW_SECS) {
metadata.len()
} else {
let _ = fs::remove_file(part);
0
}
}
const fn rate_kib_per_sec(written: u64, elapsed: Duration) -> u64 {
let seconds = elapsed.as_secs();
written / 1024 / if seconds == 0 { 1 } else { seconds }
}
async fn fetch_checksum(
client: &Client,
url: &str,
credential: &Credential,
) -> Result<String, GeoIpDownloadError> {
let mut response = credential
.apply(client.get(url))
.send()
.await
.map_err(reqwest::Error::without_url)?;
let status = response.status();
if !status.is_success() {
return Err(GeoIpDownloadError::UnexpectedStatus {
url: loggable(url),
status: status.as_u16(),
});
}
let mut bytes = Vec::new();
while let Some(chunk) = response
.chunk()
.await
.map_err(reqwest::Error::without_url)?
{
bytes.extend_from_slice(&chunk);
if bytes.len() > CHECKSUM_BODY_LIMIT {
return Err(GeoIpDownloadError::MalformedChecksum { url: loggable(url) });
}
}
let body = String::from_utf8_lossy(&bytes);
let digest = body.split_whitespace().next().unwrap_or_default();
if digest.len() != SHA256_HEX_LEN || !digest.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(GeoIpDownloadError::MalformedChecksum { url: loggable(url) });
}
Ok(digest.to_ascii_lowercase())
}
fn sha256_of(path: &Path) -> Result<String, GeoIpDownloadError> {
use sha2::Digest;
let mut file = fs::File::open(path)?;
let mut hasher = sha2::Sha256::new();
let mut buffer = vec![0u8; DIGEST_BUFFER_BYTES];
loop {
let read = io::Read::read(&mut file, &mut buffer)?;
if read == 0 {
break;
}
hasher.update(&buffer[..read]);
}
let mut hex = String::with_capacity(SHA256_HEX_LEN);
for byte in hasher.finalize() {
use std::fmt::Write;
let _ = write!(hex, "{byte:02x}");
}
Ok(hex)
}
fn holds_a_database(path: &Path, format: DatabaseFormat) -> Result<bool, io::Error> {
match format {
DatabaseFormat::Mmdb => {
let mut file = fs::File::open(path)?;
let length = file.metadata()?.len();
io::Seek::seek(
&mut file,
io::SeekFrom::Start(length.saturating_sub(METADATA_TAIL_BYTES)),
)?;
let mut tail = Vec::new();
io::Read::read_to_end(&mut file, &mut tail)?;
Ok(tail
.windows(MMDB_MARKER.len())
.any(|window| window == MMDB_MARKER))
}
DatabaseFormat::Csv | DatabaseFormat::Json => {
let mut head = vec![0u8; MARKUP_HEAD_BYTES];
let read = io::Read::read(&mut fs::File::open(path)?, &mut head)?;
let head = head[..read].to_ascii_lowercase();
Ok(!MARKUP_OPENERS.iter().any(|opener| head.starts_with(opener)))
}
}
}
#[cfg(test)]
fn materialise(
part: &Path,
staged: &Path,
dest: &Path,
archive: Archive,
format: DatabaseFormat,
) -> Result<u64, GeoIpDownloadError> {
materialise_guarded(part, staged, dest, archive, format, &Guard::OFF)
}
fn materialise_guarded(
part: &Path,
staged: &Path,
dest: &Path,
archive: Archive,
format: DatabaseFormat,
guard: &Guard,
) -> Result<u64, GeoIpDownloadError> {
let source = fs::File::open(part)?;
match archive {
Archive::Raw => {
fs::rename(part, staged)?;
}
Archive::Gzip => {
let decoder = MultiGzDecoder::new(io::BufReader::new(source));
let mut out = io::BufWriter::new(fs::File::create(staged)?);
copy_bounded(decoder, &mut out, dest, MAX_EXPANDED_BYTES)?;
}
Archive::TarGz { member } => {
extract_member(source, staged, member, dest)?;
}
}
if !holds_a_database(staged, format)? {
return Err(GeoIpDownloadError::NotADatabase {
url: dest.display().to_string(),
});
}
guard.admit(staged, dest, format)?;
let size = fs::metadata(staged)?.len();
fs::OpenOptions::new()
.write(true)
.open(staged)?
.sync_all()?;
fs::rename(staged, dest)?;
if let Some(parent) = dest.parent() {
let _ = fs::File::open(parent).and_then(|dir| dir.sync_all());
}
Ok(size)
}
fn copy_bounded(
source: impl io::Read,
out: &mut impl io::Write,
dest: &Path,
limit: u64,
) -> Result<(), GeoIpDownloadError> {
let written = io::copy(&mut source.take(limit + 1), out)?;
io::Write::flush(out)?;
if written > limit {
return Err(GeoIpDownloadError::TooLarge {
url: dest.display().to_string(),
limit,
});
}
Ok(())
}
fn extract_member(
source: fs::File,
staged: &Path,
member: &'static str,
dest: &Path,
) -> Result<(), GeoIpDownloadError> {
let decoder = MultiGzDecoder::new(io::BufReader::new(source));
let mut archive = tar::Archive::new(decoder);
for entry in archive.entries()? {
let mut entry = entry?;
let is_file = entry.header().entry_type().is_file();
let is_match = is_file && entry.path()?.file_name().is_some_and(|name| name == member);
if is_match {
let mut out = io::BufWriter::new(fs::File::create(staged)?);
copy_bounded(&mut entry, &mut out, dest, MAX_EXPANDED_BYTES)?;
return Ok(());
}
}
Err(GeoIpDownloadError::ArchiveMemberMissing { member })
}
fn with_extension(path: &Path, extension: &str) -> PathBuf {
let mut name = path.as_os_str().to_os_string();
name.push(".");
name.push(extension);
PathBuf::from(name)
}
#[cfg(test)]
mod tests {
use tar::EntryType;
use super::*;
use crate::geoip::download::testkit::{gzip, mmdb_body, tar_gz, tar_gz_of};
#[test]
fn temp_names_append_rather_than_replace() {
let dest = Path::new("/var/lib/geoip/dbip-city-lite.mmdb");
assert_eq!(
with_extension(dest, PART_EXT),
PathBuf::from("/var/lib/geoip/dbip-city-lite.mmdb.part")
);
assert_eq!(
with_extension(dest, STAGE_EXT),
PathBuf::from("/var/lib/geoip/dbip-city-lite.mmdb.staged")
);
}
#[test]
fn credential_debug_never_shows_the_secret() {
let basic = Credential::Basic {
username: "account-1234".into(),
password: "licence-abcd".into(),
fields: "auto_download.maxmind_account_id",
};
let token = Credential::QueryToken {
name: "token",
value: "token-wxyz".into(),
fields: "auto_download.ipinfo_token",
};
assert_eq!(format!("{basic:?}"), "Basic(***REDACTED***)");
assert_eq!(format!("{token:?}"), "QueryToken(***REDACTED***)");
assert_eq!(format!("{:?}", Credential::None), "None");
assert_eq!(basic.fields(), Some("auto_download.maxmind_account_id"));
assert_eq!(Credential::None.fields(), None);
}
#[test]
fn transfer_debug_never_shows_the_secret() {
let transfer = Transfer {
url: "https://example.invalid/db.mmdb".into(),
fallback_url: None,
checksum_url: None,
dest: PathBuf::from("/tmp/db.mmdb"),
archive: Archive::Raw,
format: DatabaseFormat::Mmdb,
credential: Credential::QueryToken {
name: "token",
value: "token-wxyz".into(),
fields: "auto_download.ipinfo_token",
},
};
let rendered = format!("{transfer:?}");
assert!(!rendered.contains("token-wxyz"), "{rendered}");
assert!(rendered.contains("REDACTED"), "{rendered}");
}
#[test]
fn a_body_that_expands_past_the_ceiling_is_refused() {
let dest = Path::new("/tmp/db.mmdb");
let mut out = Vec::new();
copy_bounded(&b"12345678"[..], &mut out, dest, 8).unwrap();
assert_eq!(out, b"12345678");
let mut out = Vec::new();
let err = copy_bounded(&b"123456789"[..], &mut out, dest, 8).unwrap_err();
assert!(
matches!(err, GeoIpDownloadError::TooLarge { limit: 8, .. }),
"{err:?}"
);
}
#[test]
fn a_logged_url_keeps_its_parameter_names_and_drops_their_values() {
assert_eq!(
loggable("https://internal.example/feed.csv?api_key=AKIAsecret"),
"https://internal.example/feed.csv?api_key=<redacted>"
);
assert_eq!(
loggable("https://example.invalid/db?suffix=tar.gz&token=abc"),
"https://example.invalid/db?suffix=<redacted>&token=<redacted>"
);
assert_eq!(
loggable("https://example.invalid/db?bare"),
"https://example.invalid/db?bare=<redacted>"
);
assert_eq!(
loggable("https://example.invalid/db.mmdb"),
"https://example.invalid/db.mmdb"
);
}
#[test]
fn a_logged_url_drops_the_userinfo_basic_auth_is_sent_from() {
for (url, shown) in [
(
"https://user:pw@internal.example/feed.csv",
"https://internal.example/feed.csv",
),
(
"https://token@internal.example:8443/feed.csv?key=abc",
"https://internal.example:8443/feed.csv?key=<redacted>",
),
(
"https://example.invalid/feed@2.csv",
"https://example.invalid/feed@2.csv",
),
(
"https:user:sekrit@example.net/feed.csv",
"https://example.net/feed.csv",
),
(
"https:///user:sekrit@example.net/feed.csv",
"https://example.net/feed.csv",
),
(
"https:\\\\user:sekrit@example.net\\feed.csv#sekrit",
"https://example.net/feed.csv",
),
(
"https://user:sekrit@exa mple.net/x.csv",
"https://exa mple.net/x.csv",
),
("https:user:sekrit@exa mple.net/x.csv", "exa mple.net/x.csv"),
("not a url", "not a url"),
] {
assert_eq!(loggable(url), shown, "{url}");
}
}
#[test]
fn the_default_client_builds() {
let built = client(None, Duration::from_secs(30), Duration::from_secs(60));
assert!(built.is_ok());
}
#[test]
fn an_injected_client_is_used_as_it_stands() {
let injected = Client::new();
let built = client(
Some(&injected),
Duration::from_secs(30),
Duration::from_secs(60),
);
assert!(built.is_ok());
}
#[test]
fn the_progress_rate_survives_a_sub_second_interval() {
assert_eq!(rate_kib_per_sec(4096, Duration::from_millis(10)), 4);
assert_eq!(rate_kib_per_sec(0, Duration::ZERO), 0);
assert_eq!(
rate_kib_per_sec(10 * 1024 * 30, Duration::from_secs(30)),
10
);
}
#[test]
fn an_absent_part_file_resumes_from_zero() {
let dir = tempfile::tempdir().unwrap();
assert_eq!(resumable_offset(&dir.path().join("absent.part")), 0);
}
#[test]
fn a_recent_part_file_offers_its_length() {
let dir = tempfile::tempdir().unwrap();
let part = dir.path().join("db.mmdb.part");
fs::write(&part, b"twelve bytes").unwrap();
assert_eq!(resumable_offset(&part), 12);
assert!(part.exists(), "a resumable part file is kept");
}
#[test]
fn materialise_gzip_writes_the_decompressed_file() {
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("db.mmdb");
let part = with_extension(&dest, PART_EXT);
let staged = with_extension(&dest, STAGE_EXT);
let payload = mmdb_body(b"a body that round-trips");
fs::write(&part, gzip(&payload)).unwrap();
let size = materialise(&part, &staged, &dest, Archive::Gzip, DatabaseFormat::Mmdb).unwrap();
assert_eq!(usize::try_from(size).unwrap(), payload.len());
assert_eq!(fs::read(&dest).unwrap(), payload);
assert!(!staged.exists(), "staged file must be renamed away");
}
#[test]
fn materialise_raw_renames_the_body_into_place() {
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("db.mmdb");
let part = with_extension(&dest, PART_EXT);
let staged = with_extension(&dest, STAGE_EXT);
let payload = mmdb_body(b"raw body");
fs::write(&part, &payload).unwrap();
let size = materialise(&part, &staged, &dest, Archive::Raw, DatabaseFormat::Mmdb).unwrap();
assert_eq!(usize::try_from(size).unwrap(), payload.len());
assert_eq!(fs::read(&dest).unwrap(), payload);
}
#[test]
fn materialise_rejects_a_body_that_is_not_a_database() {
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("db.mmdb");
let part = with_extension(&dest, PART_EXT);
let staged = with_extension(&dest, STAGE_EXT);
fs::write(&part, b"<html><title>Log in</title></html>").unwrap();
let err =
materialise(&part, &staged, &dest, Archive::Raw, DatabaseFormat::Mmdb).unwrap_err();
assert!(
matches!(err, GeoIpDownloadError::NotADatabase { .. }),
"{err:?}"
);
assert!(!dest.exists());
}
#[test]
fn materialise_tar_gz_extracts_the_named_member() {
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("GeoLite2-City.mmdb");
let part = with_extension(&dest, PART_EXT);
let staged = with_extension(&dest, STAGE_EXT);
let payload = mmdb_body(b"city database bytes");
fs::write(&part, tar_gz("GeoLite2-City.mmdb", &payload)).unwrap();
let size = materialise(
&part,
&staged,
&dest,
Archive::TarGz {
member: "GeoLite2-City.mmdb",
},
DatabaseFormat::Mmdb,
)
.unwrap();
assert_eq!(usize::try_from(size).unwrap(), payload.len());
assert_eq!(fs::read(&dest).unwrap(), payload);
}
#[test]
fn materialise_tar_gz_reports_a_missing_member() {
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("GeoLite2-ASN.mmdb");
let part = with_extension(&dest, PART_EXT);
let staged = with_extension(&dest, STAGE_EXT);
fs::write(&part, tar_gz_of(&[])).unwrap();
let err = materialise(
&part,
&staged,
&dest,
Archive::TarGz {
member: "GeoLite2-ASN.mmdb",
},
DatabaseFormat::Mmdb,
)
.unwrap_err();
assert!(
matches!(err, GeoIpDownloadError::ArchiveMemberMissing { member } if member == "GeoLite2-ASN.mmdb"),
"{err:?}"
);
assert!(!dest.exists());
}
fn extract(
archive: &[u8],
member: &'static str,
) -> (tempfile::TempDir, PathBuf, Result<u64, GeoIpDownloadError>) {
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join(member);
let part = with_extension(&dest, PART_EXT);
let staged = with_extension(&dest, STAGE_EXT);
fs::write(&part, archive).unwrap();
let result = materialise(
&part,
&staged,
&dest,
Archive::TarGz { member },
DatabaseFormat::Mmdb,
);
(dir, dest, result)
}
#[test]
fn materialise_tar_gz_takes_the_first_member_carrying_the_name() {
let first = mmdb_body(b"the copy the extraction stops at");
let second = mmdb_body(b"a second copy filed under the same name");
let archive = tar_gz_of(&[
(
"GeoLite2-City_20241231/GeoLite2-City.mmdb",
EntryType::Regular,
&first,
),
(
"GeoLite2-City_20250131/GeoLite2-City.mmdb",
EntryType::Regular,
&second,
),
]);
let (_dir, dest, result) = extract(&archive, "GeoLite2-City.mmdb");
assert_eq!(usize::try_from(result.unwrap()).unwrap(), first.len());
assert_eq!(fs::read(&dest).unwrap(), first);
}
#[test]
fn materialise_tar_gz_does_not_take_a_directory_for_the_member() {
let archive = tar_gz_of(&[("GeoLite2-City.mmdb", EntryType::Directory, &[][..])]);
let (dir, dest, result) = extract(&archive, "GeoLite2-City.mmdb");
let err = result.unwrap_err();
assert!(
matches!(err, GeoIpDownloadError::ArchiveMemberMissing { member } if member == "GeoLite2-City.mmdb"),
"{err:?}"
);
assert!(!dest.exists());
assert!(!with_extension(&dest, STAGE_EXT).exists());
assert_eq!(
fs::read_dir(dir.path()).unwrap().count(),
1,
"nothing was staged"
);
}
#[test]
fn materialise_tar_gz_reaches_a_member_behind_unrelated_entries() {
let payload = mmdb_body(b"city database bytes");
let archive = tar_gz_of(&[
("GeoLite2-City_20241231/", EntryType::Directory, &[][..]),
(
"GeoLite2-City_20241231/COPYRIGHT.txt",
EntryType::Regular,
b"(c) MaxMind",
),
(
"GeoLite2-City_20241231/LICENSE.txt",
EntryType::Regular,
b"terms",
),
(
"GeoLite2-City_20241231/GeoLite2-City-Blocks-IPv4.csv",
EntryType::Regular,
b"network,geoname_id\n",
),
(
"GeoLite2-City_20241231/GeoLite2-City.mmdb",
EntryType::Regular,
&payload,
),
]);
let (_dir, dest, result) = extract(&archive, "GeoLite2-City.mmdb");
assert_eq!(usize::try_from(result.unwrap()).unwrap(), payload.len());
assert_eq!(fs::read(&dest).unwrap(), payload);
}
#[test]
fn a_digest_is_lowercase_hex_of_the_whole_file() {
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join("body");
fs::write(&file, b"").unwrap();
assert_eq!(
sha256_of(&file).unwrap(),
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
);
}
}