From ecf570a5b2c9174dab163142347371701375d990 Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sat, 8 Aug 2026 21:38:12 +0300 Subject: [PATCH] config: pydantic-settings loader with env > .env > yaml precedence --- src/netcheck/config.py | 183 +++++++++++++++++++++++++++++++++++++++++ tests/test_config.py | 83 +++++++++++++++++++ 2 files changed, 266 insertions(+) create mode 100644 src/netcheck/config.py create mode 100644 tests/test_config.py diff --git a/src/netcheck/config.py b/src/netcheck/config.py new file mode 100644 index 0000000..0c8d852 --- /dev/null +++ b/src/netcheck/config.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, Field, SecretStr +from pydantic_settings import ( + BaseSettings, + PydanticBaseSettingsSource, + SettingsConfigDict, + YamlConfigSettingsSource, +) + +DEFAULT_CONFIG_PATH = Path("config.yaml") +DEFAULT_ENV_PATH = Path(".env") + + +class Timeouts(BaseModel): + http_seconds: float = 8.0 + module_seconds: float = 30.0 + speedtest_seconds: float = 90.0 + dns_seconds: float = 4.0 + subprocess_seconds: float = 60.0 + + +class HostSpec(BaseModel): + label: str + host: str + + +class Probing(BaseModel): + reference_hosts: list[HostSpec] = Field( + default_factory=lambda: [ + HostSpec(label="cloudflare-dns", host="1.1.1.1"), + HostSpec(label="google-dns", host="8.8.8.8"), + HostSpec(label="quad9-dns", host="9.9.9.9"), + ] + ) + service_hosts: list[HostSpec] = Field( + default_factory=lambda: [ + HostSpec(label="cloudflare", host="cloudflare.com"), + HostSpec(label="google", host="google.com"), + HostSpec(label="github", host="github.com"), + ] + ) + ping_count: int = 20 + quick_ping_count: int = 5 + ping_interval_seconds: float = 0.25 + ping_timeout_seconds: float = 2.0 + mtr_cycles: int = 10 + quick_mtr_cycles: int = 1 + max_hops: int = 30 + trace_concurrency: int = 2 + + +class Speedtest(BaseModel): + enabled_tiers: list[str] = Field(default_factory=lambda: ["ookla_bin", "cloudflare", "fastcom"]) + download_sizes_bytes: list[int] = Field(default_factory=lambda: [1_000_000, 10_000_000, 25_000_000]) + upload_sizes_bytes: list[int] = Field(default_factory=lambda: [1_000_000, 5_000_000]) + cloudflare_base_url: str = "https://speed.cloudflare.com" + fastcom_api_url: str = "https://api.fast.com/netflix/speedtest/v2" + ndt7_locate_url: str = "https://locate.measurementlab.net/v2/nearest/ndt/ndt7" + bufferbloat_probe_interval_seconds: float = 0.2 + + +class Providers(BaseModel): + cf_trace_url: str = "https://www.cloudflare.com/cdn-cgi/trace" + ip_api_url: str = "http://ip-api.com/json/" + freeipapi_url: str = "https://freeipapi.com/api/json/" + ipinfo_url: str = "https://ipinfo.io/" + ipwhois_url: str = "https://ipwho.is/" + ripestat_base_url: str = "https://stat.ripe.net/data" + asrank_url: str = "https://api.asrank.caida.org/v2/graphql" + peeringdb_base_url: str = "https://www.peeringdb.com/api" + internetdb_url: str = "https://internetdb.shodan.io/" + abuseipdb_url: str = "https://api.abuseipdb.com/api/v2/check" + cymru_origin_zone: str = "origin.asn.cymru.com" + cymru_asn_zone: str = "asn.cymru.com" + ripestat_max_rows: int = 200 + ripestat_timeframe_days: int = 14 + peeringdb_cache_hours: int = 24 + firehol_refresh_hours: int = 24 + firehol_netsets: list[str] = Field(default_factory=list) + + +class Dnsbl(BaseModel): + zones: list[str] = Field( + default_factory=lambda: [ + "zen.spamhaus.org", + "bl.spamcop.net", + "b.barracudacentral.org", + "dnsbl.dronebl.org", + ] + ) + + +class Band(BaseModel): + good: float + warn: float + + +class BufferbloatBands(BaseModel): + a: float = 5.0 + b: float = 30.0 + c: float = 60.0 + d: float = 200.0 + e: float = 400.0 + + +class VpnBands(BaseModel): + likely: float = 0.40 + confirmed: float = 0.75 + + +class Thresholds(BaseModel): + latency_ms: Band = Band(good=40.0, warn=100.0) + jitter_ms: Band = Band(good=5.0, warn=20.0) + loss_pct: Band = Band(good=0.0, warn=2.0) + bufferbloat_ms: BufferbloatBands = BufferbloatBands() + vpn_confidence: VpnBands = VpnBands() + + +class Output(BaseModel): + logs_dir: str = "./logs" + cache_dir: str = "./.cache" + emoji: bool = True + + +class Watch(BaseModel): + interval_seconds: int = 60 + speedtest_every_n_cycles: int = 10 + dashboard_refresh_hz: int = 4 + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_prefix="NETCHECK_", + env_nested_delimiter="__", + env_file_encoding="utf-8", + extra="ignore", + yaml_file=None, + ) + + timeouts: Timeouts = Timeouts() + probing: Probing = Probing() + speedtest: Speedtest = Speedtest() + providers: Providers = Providers() + dnsbl: Dnsbl = Dnsbl() + thresholds: Thresholds = Thresholds() + output: Output = Output() + watch: Watch = Watch() + + ipinfo_token: SecretStr | None = Field(default=None, alias="IPINFO_TOKEN") + peeringdb_api_key: SecretStr | None = Field(default=None, alias="PEERINGDB_API_KEY") + abuseipdb_api_key: SecretStr | None = Field(default=None, alias="ABUSEIPDB_API_KEY") + + @classmethod + def settings_customise_sources( + cls, + settings_cls: type[BaseSettings], + init_settings: PydanticBaseSettingsSource, + env_settings: PydanticBaseSettingsSource, + dotenv_settings: PydanticBaseSettingsSource, + file_secret_settings: PydanticBaseSettingsSource, + ) -> tuple[PydanticBaseSettingsSource, ...]: + return ( + init_settings, + env_settings, + dotenv_settings, + YamlConfigSettingsSource(settings_cls), + file_secret_settings, + ) + + +def load_settings( + config_path: Path | None = None, + env_file: Path | None = None, +) -> Settings: + yaml_path = Path(config_path) if config_path is not None else DEFAULT_CONFIG_PATH + dotenv_path = Path(env_file) if env_file is not None else DEFAULT_ENV_PATH + Settings.model_config["yaml_file"] = yaml_path if yaml_path.exists() else None + kwargs: dict[str, Any] = {"_env_file": str(dotenv_path) if dotenv_path.exists() else None} + return Settings(**kwargs) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..38f7ab7 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import textwrap + +import pytest + +from netcheck.config import load_settings + + +@pytest.fixture() +def yaml_file(tmp_path): + path = tmp_path / "config.yaml" + path.write_text( + textwrap.dedent( + """ + timeouts: + http_seconds: 3.5 + probing: + ping_count: 7 + reference_hosts: + - { label: "yaml-host", host: "9.9.9.9" } + output: + logs_dir: "./yaml-logs" + emoji: false + thresholds: + latency_ms: { good: 11.0, warn: 22.0 } + """ + ).strip(), + encoding="utf-8", + ) + return path + + +def test_defaults_apply_when_yaml_omits_a_field(yaml_file, tmp_path): + s = load_settings(config_path=yaml_file, env_file=tmp_path / "missing.env") + assert s.probing.max_hops == 30 + assert s.timeouts.module_seconds == 30.0 + + +def test_yaml_overrides_defaults(yaml_file, tmp_path): + s = load_settings(config_path=yaml_file, env_file=tmp_path / "missing.env") + assert s.timeouts.http_seconds == 3.5 + assert s.probing.ping_count == 7 + assert s.output.emoji is False + assert s.thresholds.latency_ms.warn == 22.0 + assert s.probing.reference_hosts[0].host == "9.9.9.9" + + +def test_dotenv_overrides_yaml(yaml_file, tmp_path): + env_file = tmp_path / ".env" + env_file.write_text("NETCHECK_PROBING__PING_COUNT=13\n", encoding="utf-8") + s = load_settings(config_path=yaml_file, env_file=env_file) + assert s.probing.ping_count == 13 + + +def test_environment_overrides_dotenv(yaml_file, tmp_path, monkeypatch): + env_file = tmp_path / ".env" + env_file.write_text("NETCHECK_PROBING__PING_COUNT=13\n", encoding="utf-8") + monkeypatch.setenv("NETCHECK_PROBING__PING_COUNT", "99") + s = load_settings(config_path=yaml_file, env_file=env_file) + assert s.probing.ping_count == 99 + + +def test_secrets_default_to_none_so_the_tool_runs_unconfigured(tmp_path): + s = load_settings(config_path=tmp_path / "absent.yaml", env_file=tmp_path / "absent.env") + assert s.ipinfo_token is None + assert s.peeringdb_api_key is None + assert s.abuseipdb_api_key is None + + +def test_secrets_load_from_dotenv_without_the_nested_prefix(tmp_path, monkeypatch): + monkeypatch.delenv("IPINFO_TOKEN", raising=False) + env_file = tmp_path / ".env" + env_file.write_text("IPINFO_TOKEN=abc123\n", encoding="utf-8") + s = load_settings(config_path=tmp_path / "absent.yaml", env_file=env_file) + assert s.ipinfo_token is not None + assert s.ipinfo_token.get_secret_value() == "abc123" + assert "abc123" not in repr(s) + + +def test_missing_yaml_is_not_an_error(tmp_path): + s = load_settings(config_path=tmp_path / "nope.yaml", env_file=tmp_path / "nope.env") + assert s.probing.ping_count == 20