Best Checklist NordVPN in 2026: Tested & Reviewed

#!/usr/bin/env python3
"""
nordvpn_benchmark.py — Automated throughput benchmarking for NordVPN protocols.
Requires: pip install speedtest-cli nordvpn-py
Run: sudo python3 nordvpn_benchmark.py
"""

import csv
import json
import logging
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

try:
    import speedtest
except ImportError:
    print("ERROR: speedtest-cli not installed. Run: pip install speedtest-cli", file=sys.stderr)
    sys.exit(1)

# Configuration
PROTOCOLS = ["wireguard", "openvpn_udp", "openvpn_tcp", "nordlynx"]
COUNTRIES = ["de", "us", "se", "jp", "sg"]
OUTPUT_FILE = Path("nordvpn_benchmark_results.csv")
RETRIES = 3
STABILIZE_SECONDS = 10  # Wait after connect before testing
DISCONNECT_TIMEOUT = 30

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("benchmark.log"),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)


def run_shell(args: list[str], timeout: int = 60) -> tuple[int, str, str]:
    """Execute a shell command and return (returncode, stdout, stderr)."""
    try:
        result = subprocess.run(
            args,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return result.returncode, result.stdout.strip(), result.stderr.strip()
    except subprocess.TimeoutExpired:
        logger.error(f"Command timed out after {timeout}s: {' '.join(args)}")
        return -1, "", "timeout"
    except FileNotFoundError as e:
        logger.error(f"Command not found: {e}")
        return -2, "", str(e)


def disconnect_vpn() -> bool:
    """Disconnect any active NordVPN session."""
    rc, out, err = run_shell(["nordvpn", "disconnect"], timeout=DISCONNECT_TIMEOUT)
    if rc != 0:
        logger.warning(f"Disconnect returned {rc}: {err}")
        # Force kill as fallback
        run_shell(["pkill", "-9", "nordvpnd"])
        time.sleep(2)
    return True


def connect_vpn(protocol: str, country: str) -> bool:
    """Connect to NordVPN with specified protocol and country."""
    logger.info(f"Connecting: protocol={protocol}, country={country}")
    rc, out, err = run_shell(
        ["nordvpn", "connect", "--protocol", protocol, "--country", country],
        timeout=60
    )
    if rc != 0:
        logger.error(f"Connect failed: {err}")
        return False
    # Verify connection is actually established
    time.sleep(3)
    rc2, out2, _ = run_shell(["nordvpn", "status"])
    if "Connected" not in out2:
        logger.error(f"Status check failed after connect: {out2}")
        return False
    # Stabilization period for tunnel throughput
    logger.info(f"Connected. Stabilizing for {STABILIZE_SECONDS}s...")
    time.sleep(STABILIZE_SECONDS)
    return True


def run_speedtest() -> dict | None:
    """Run speedtest-cli and return parsed results."""
    try:
        st = speedtest.Speedtest()
        st.get_best_server()
        st.download()
        st.upload()
        results = st.results.dict()
        return {
            "download_mbps": round(results["download"] / 1_000_000, 2),
            "upload_mbps": round(results["upload"] / 1_000_000, 2),
            "ping_ms": results["ping"],
            "server_name": results["server"]["name"],
            "server_sponsor": results["server"]["sponsor"],
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
    except speedtest.ConfigRetrievalError as e:
        logger.error(f"Speedtest config retrieval failed: {e}")
        return None
    except speedtest.SpeedtestBestServerFailure as e:
        logger.error(f"Speedtest best server failure: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected speedtest error: {e}")
        return None


def main():
    logger.info("=== NordVPN Protocol Benchmark Starting ===")
    disconnect_vpn()
    time.sleep(2)

    fieldnames = [
        "timestamp", "protocol", "country",
        "download_mbps", "upload_mbps", "ping_ms",
        "server_name", "success", "notes"
    ]
    rows = []

    for protocol in PROTOCOLS:
        for country in COUNTRIES:
            for attempt in range(1, RETRIES + 1):
                logger.info(f"Attempt {attempt}/{RETRIES}: {protocol}/{country}")

                if not connect_vpn(protocol, country):
                    rows.append({
                        "timestamp": datetime.now(timezone.utc).isoformat(),
                        "protocol": protocol,
                        "country": country,
                        "download_mbps": "",
                        "upload_mbps": "",
                        "ping_ms": "",
                        "server_name": "",
                        "success": False,
                        "notes": f"connect_failed_attempt_{attempt}"
                    })
                    disconnect_vpn()
                    time.sleep(5)
                    continue

                result = run_speedtest()
                disconnect_vpn()
                time.sleep(5)  # Cool-down between tests

                if result:
                    row = {
                        **result,
                        "protocol": protocol,
                        "country": country,
                        "success": True,
                        "notes": f"attempt_{attempt}"
                    }
                    rows.append(row)
                    logger.info(f"  -> DL: {row['download_mbps']} Mbps, UL: {row['upload_mbps']} Mbps, Ping: {row['ping_ms']} ms")
                    break  # Success, move to next combination
                else:
                    rows.append({
                        "timestamp": datetime.now(timezone.utc).isoformat(),
                        "protocol": protocol,
                        "country": country,
                        "download_mbps": "",
                        "upload_mbps": "",
                        "ping_ms": "",
                        "server_name": "",
                        "success": False,
                        "notes": f"speedtest_failed_attempt_{attempt}"
                    })

    # Write results to CSV
    with open(OUTPUT_FILE, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

    successful = sum(1 for r in rows if r["success"])
    logger.info(f"=== Benchmark Complete: {successful}/{len(rows)} tests succeeded ===")
    logger.info(f"Results written to {OUTPUT_FILE}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
nordvpn_leak_test.py — Automated DNS, WebRTC, and IP leak detection for NordVPN.
Requires: pip install requests
Run: python3 nordvpn_leak_test.py
"""

import json
import logging
import subprocess
import sys
import time
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# External services that report your IP and DNS configuration
IP_CHECK_SERVICES = [
    "https://api.ipify.org?format=json",
    "https://ipinfo.io/json",
    "https://ifconfig.me/json",
    "https://check.torproject.org/api/ip",
]

DNS_LEAK_SERVICES = [
    "https://dnsleaktest.com/json",
    "https://www.dnsleak.com/json",
]

WEB_RTC_CHECK = "https://diafygi.webtn.org/webrtc-test/"


@dataclass
class LeakResult:
    service: str
    vpn_ip: Optional[str] = None
    real_ip: Optional[str] = None
    dns_servers: list = field(default_factory=list)
    webrtc_leak: bool = False
    isp_exposed: bool = False
    timestamp: str = ""
    error: Optional[str] = None


def fetch_json(url: str, timeout: int = 10) -> dict | None:
    """Fetch JSON from a URL with basic error handling."""
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.URLError as e:
        logger.warning(f"URL fetch failed for {url}: {e}")
        return None
    except json.JSONDecodeError as e:
        logger.warning(f"JSON decode failed for {url}: {e}")
        return None


def get_baseline_ip() -> str | None:
    """Get real IP without VPN active."""
    logger.info("Capturing baseline IP (VPN must be disconnected)...")
    for svc in IP_CHECK_SERVICES:
        data = fetch_json(svc)
        if data:
            ip = data.get("ip") or data.get("origin")
            if ip:
                logger.info(f"  Baseline IP: {ip}")
                return ip
    logger.warning("Could not determine baseline IP")
    return None


def check_ip_leak(baseline_ip: str) -> list[LeakResult]:
    """Check if VPN is leaking the real IP address."""
    results = []
    for svc in IP_CHECK_SERVICES:
        result = LeakResult(service=svc, timestamp=datetime.now(timezone.utc).isoformat())
        data = fetch_json(svc)
        if data:
            current_ip = data.get("ip") or data.get("origin")
            result.vpn_ip = current_ip
            result.real_ip = baseline_ip
            if current_ip == baseline_ip:
                result.isp_exposed = True
                logger.critical(f"IP LEAK DETECTED on {svc}: {current_ip}")
            else:
                logger.info(f"  IP safe on {svc}: {current_ip}")
        else:
            result.error = "fetch_failed"
        results.append(result)
    return results


def check_dns_leak() -> list[LeakResult]:
    """Check if DNS queries are leaking outside the VPN tunnel."""
    results = []
    for svc in DNS_LEAK_SERVICES:
        result = LeakResult(service=svc, timestamp=datetime.now(timezone.utc).isoformat())
        data = fetch_json(svc)
        if data:
            # Extract DNS server IPs from the response
            dns_entries = []
            if isinstance(data, dict):
                for key, val in data.items():
                    if "ip" in key.lower() or "dns" in key.lower():
                        if isinstance(val, str) and val:
                            dns_entries.append(val)
                        elif isinstance(val, list):
                            dns_entries.extend(val)
            result.dns_servers = dns_entries
            logger.info(f"  DNS servers on {svc}: {dns_entries}")
        else:
            result.error = "fetch_failed"
        results.append(result)
    return results


def get_nordvpn_ip() -> str | None:
    """Get current NordVPN-assigned IP."""
    rc, out, err = subprocess.run(
        ["nordvpn", "status"],
        capture_output=True, text=True, timeout=10
    )
    if rc != 0:
        return None
    for line in out.split("\n"):
        if "IP" in line:
            ip = line.split(":")[-1].strip()
            if ip:
                return ip
    return None


def run_full_leak_test():
    """Execute the complete leak test suite."""
    logger.info("=== NordVPN Leak Test Suite Starting ===")

    # Step 1: Get baseline (disconnected) IP
    baseline = get_baseline_ip()
    if not baseline:
        logger.error("Cannot proceed without baseline IP. Check internet connectivity.")
        sys.exit(1)

    # Step 2: Get VPN-assigned IP
    vpn_ip = get_nordvpn_ip()
    if not vpn_ip:
        logger.error("NordVPN does not appear to be connected. Run: nordvpn connect")
        sys.exit(1)
    logger.info(f"VPN-assigned IP: {vpn_ip}")

    # Step 3: Run IP leak check
    logger.info("\n--- IP Leak Check ---")
    ip_results = check_ip_leak(baseline)

    # Step 4: Run DNS leak check
    logger.info("\n--- DNS Leak Check ---")
    dns_results = check_dns_leak()

    # Step 5: Summarize
    leaks = [r for r in ip_results if r.isp_exposed]
    dns_exposed = [r for r in dns_results if r.error]

    print("\n" + "=" * 60)
    print("LEAK TEST SUMMARY")
    print("=" * 60)
    print(f"Baseline IP:  {baseline}")
    print(f"VPN IP:       {vpn_ip}")
    print(f"IP Leaks:     {len(leaks)}")
    print(f"DNS Issues:   {len(dns_exposed)}")
    print(f"Overall:      {'FAIL — leaks detected' if leaks else 'PASS — no IP leaks'}")
    print("=" * 60)

    # Output structured results for CI integration
    output = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "baseline_ip": baseline,
        "vpn_ip": vpn_ip,
        "ip_leaks": len(leaks),
        "ip_results": [{"service": r.service, "exposed": r.isp_exposed} for r in ip_results],
        "dns_results": [{"service": r.service, "servers": r.dns_servers} for r in dns_results],
    }
    with open("leak_test_results.json", "w") as f:
        json.dump(output, f, indent=2)
    logger.info("Results written to leak_test_results.json")

    return len(leaks) == 0


if __name__ == "__main__":
    success = run_full_leak_test()
    sys.exit(0 if success else 1)

#!/usr/bin/env python3
"""
nordvpn_healthcheck.py — Automated health check and kill-switch validation for NordVPN.
Requires: pip install requests
Run: sudo python3 nordvpn_healthcheck.py
"""

import logging
import subprocess
import sys
import time
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

CHECK_URL = "https://api.ipify.org"
CHECK_INTERVAL = 1  # seconds between connectivity checks
MAX_WAIT = 30  # maximum seconds to wait for kill-switch activation


@dataclass
class HealthReport:
    timestamp: str
    vpn_ip: str
    baseline_ip: str
    kill_switch_active: bool
    downtime_ms: int
    reconnected: bool
    reconnection_time_s: float
    notes: str = ""


def check_connectivity() -> Optional[str]:
    """Check if we can reach the internet. Returns IP if reachable, None otherwise."""
    try:
        req = urllib.request.Request(CHECK_URL, headers={"User-Agent": "HealthCheck/1.0"})
        with urllib.request.urlopen(req, timeout=5) as resp:
            return resp.read().decode().strip()
    except Exception:
        return None


def get_vpn_status() -> dict:
    """Get current NordVPN status as a dict."""
    try:
        result = subprocess.run(
            ["nordvpn", "status"],
            capture_output=True, text=True, timeout=10
        )
        status = {"connected": False, "ip": None, "country": None}
        for line in result.stdout.strip().split("\n"):
            if ":" in line:
                key, val = line.split(":", 1)
                status[key.strip().lower()] = val.strip()
        status["connected"] = status.get("status", "").lower() == "connected"
        return status
    except subprocess.TimeoutExpired:
        return {"connected": False, "ip": None, "country": None, "error": "timeout"}
    except FileNotFoundError:
        logger.error("NordVPN CLI not found. Is nordvpn installed?")
        return {"connected": False, "ip": None, "country": None, "error": "binary_not_found"}


def validate_kill_switch(baseline_ip: str) -> HealthReport:
    """
    Validate kill switch by disconnecting VPN and measuring:
    1. Time until internet goes dark (kill-switch activation)
    2. Time until reconnection
    """
    report = HealthReport(
        timestamp=datetime.now(timezone.utc).isoformat(),
        vpn_ip="",
        baseline_ip=baseline_ip,
        kill_switch_active=False,
        downtime_ms=0,
        reconnected=False,
        reconnection_time_s=0.0
    )

    status = get_vpn_status()
    if not status.get("connected"):
        report.notes = "VPN not connected at test start"
        logger.error("VPN not connected. Cannot test kill switch.")
        return report

    report.vpn_ip = status.get("ip", "")
    logger.info(f"VPN connected. IP: {report.vpn_ip}")

    # Step 1: Verify connectivity through VPN
    ip_with_vpn = check_connectivity()
    if ip_with_vpn != report.vpn_ip:
        logger.warning(f"Connectivity check IP mismatch: {ip_with_vpn} vs VPN IP {report.vpn_ip}")

    # Step 2: Disconnect VPN
    logger.info("Disconnecting VPN to test kill-switch...")
    subprocess.run(["nordvpn", "disconnect"], capture_output=True, timeout=15)

    # Step 3: Check how long network stays dark
    start = time.monotonic()
    first_failure = None
    while time.monotonic() - start < MAX_WAIT:
        ip = check_connectivity()
        if ip is None:
            if first_failure is None:
                first_failure = time.monotonic()
                logger.info(f"  Network went dark at {first_failure - start:.2f}s")
        else:
            if first_failure is not None:
                # Network came back — kill switch worked, then reconnected
                report.reconnected = True
                report.reconnection_time_s = round(time.monotonic() - start, 2)
                report.downtime_ms = int((first_failure - start) * 1000)
                report.kill_switch_active = True
                logger.info(f"  Network restored at {report.reconnection_time_s}s, downtime: {report.downtime_ms}ms")
                break
        time.sleep(CHECK_INTERVAL)

    if not report.reconnected and first_failure:
        report.kill_switch_active = True
        report.downtime_ms = int((time.monotonic() - start) * 1000)
        logger.info(f"  Kill switch active. Network still dark after {report.downtime_ms}ms")
    elif not report.kill_switch_active:
        logger.warning("Kill switch did NOT activate — traffic leaked during disconnect")

    # Step 4: Reconnect for continued use
    logger.info("Reconnecting VPN...")
    subprocess.run(
        ["nordvpn", "connect", "--protocol", "wireguard"],
        capture_output=True, timeout=30
    )
    time.sleep(5)

    return report


def run_health_check_suite():
    """Run the complete health check suite."""
    logger.info("=== NordVPN Health Check Suite ===")

    baseline_ip = check_connectivity()
    logger.info(f"Baseline connectivity IP: {baseline_ip}")

    report = validate_kill_switch(baseline_ip)

    print("\n" + "=" * 50)
    print("HEALTH CHECK REPORT")
    print("=" * 50)
    print(f"Timestamp:            {report.timestamp}")
    print(f"VPN IP:               {report.vpn_ip}")
    print(f"Kill Switch Active:   {report.kill_switch_active}")
    print(f"Downtime:             {report.downtime_ms} ms")
    print(f"Reconnected:          {report.reconnected}")
    print(f"Reconnection Time:    {report.reconnection_time_s}s")
    print(f"Notes:                {report.notes}")
    print("=" * 50)

    # Store results
    with open("healthcheck_report.json", "w") as f:
        f.write(report.__class__.__annotations__.__class__.__dict__)
        json_content = {
            "timestamp": report.timestamp,
            "vpn_ip": report.vpn_ip,
            "baseline_ip": report.baseline_ip,
            "kill_switch_active": report.kill_switch_active,
            "downtime_ms": report.downtime_ms,
            "reconnected": report.reconnected,
            "reconnection_time_s": report.reconnection_time_s,
            "notes": report.notes
        }
        import json
        f.write(json.dumps(json_content, indent=2))
    logger.info("Report saved to healthcheck_report.json")

    return report


if __name__ == "__main__":
    run_health_check_suite()

# Example: CI health gate
#!/bin/bash
set -euo pipefail

nordvpn connect --protocol wireguard --country de
sleep 10

SPEED=$(python3 -c "import speedtest; st = speedtest.Speedtest(); st.download(); print(round(st.results.download / 1_000_000, 1))")
echo "Current throughput: ${SPEED} Mbps"

if (( $(echo "$SPEED < 100" | bc -l) )); then
    echo "ERROR: VPN throughput below 100 Mbps threshold"
    nordvpn disconnect
    exit 1
fi

echo "VPN healthy. Proceeding with deployment..."
nordvpn disconnect

# Split tunnel: route only 0.0.0.0/1 and 128.0.0.0/1 through VPN
sudo ip rule add from all lookup 200
sudo ip route add default via $(nordvpn status | grep IP | awk '{print $3}') dev nordlynx table 200
sudo ip route add 0.0.0.0/1 via $(nordvpn status | grep IP | awk '{print $3}') dev nordlynx table 200
sudo ip route add 128.0.0.0/1 via $(nordvpn status | grep IP | awk '{print $3}') dev nordlynx table 200

from prometheus_client import start_http_server, Gauge
import subprocess, time, re

VPN_CONNECTED = Gauge('nordvpn_connected', 'VPN connection status (1=connected)')
VPN_LATENCY = Gauge('nordvpn_latency_ms', 'VPN tunnel latency in ms')
VPN_UPTIME = Gauge('nordvpn_uptime_seconds', 'VPN connection uptime')

def parse_status():
    result = subprocess.run(['nordvpn', 'status'], capture_output=True, text=True)
    output = result.stdout
    connected = 1 if 'Connected' in output else 0
    ip_match = re.search(r'IP:\s+([\d.]+)', output)
    VPN_CONNECTED.set(connected)
    if connected:
        # Measure latency to VPN endpoint
        ping_result = subprocess.run(
            ['ping', '-c', '3', '-q', ip_match.group(1) if ip_match else '10.5.0.1'],
            capture_output=True, text=True
        )
        avg_match = re.search(r'avg = ([\d.]+)', ping_result.stdout)
        VPN_LATENCY.set(float(avg_match.group(1)) if avg_match else 0)

if __name__ == '__main__':
    start_http_server(8000)
    while True:
        parse_status()
        time.sleep(30)