From bc9d79de026a066540ef050fa598d45a3493ed89 Mon Sep 17 00:00:00 2001 From: Bram Date: Fri, 14 Aug 2026 20:03:05 +0200 Subject: [PATCH] optimize --- .../gluetun-pia-wireguard-rotator/Dockerfile | 17 +- .../gluetun-pia-wireguard-rotator/README.md | 70 ++-- .../entrypoint.py | 2 +- Dockers/gluetun-pia-wireguard-rotator/pia.py | 360 ++++++++++++++++ .../gluetun-pia-wireguard-rotator/rotate.py | 391 +++++++----------- 5 files changed, 552 insertions(+), 288 deletions(-) create mode 100644 Dockers/gluetun-pia-wireguard-rotator/pia.py diff --git a/Dockers/gluetun-pia-wireguard-rotator/Dockerfile b/Dockers/gluetun-pia-wireguard-rotator/Dockerfile index 0cc472f..18dbc70 100644 --- a/Dockers/gluetun-pia-wireguard-rotator/Dockerfile +++ b/Dockers/gluetun-pia-wireguard-rotator/Dockerfile @@ -1,21 +1,13 @@ -FROM golang:1.25-alpine AS builder - -# ccarpinteri fork: uses PIA's central token API instead of regional -# /authv3/generateToken (CN-only certs fail on modern Go / new Server-* hosts). -RUN apk add --no-cache git ca-certificates \ - && go install github.com/ccarpinteri/pia-wg-config@v1.4.0 - FROM docker:cli -RUN apk add --no-cache bash ca-certificates tzdata python3 py3-pip \ +RUN apk add --no-cache ca-certificates tzdata python3 py3-pip \ && python3 -m venv /opt/venv \ - && /opt/venv/bin/pip install --no-cache-dir 'croniter==6.2.4' \ + && /opt/venv/bin/pip install --no-cache-dir 'croniter==6.2.4' 'cryptography>=42,<47' \ && apk del py3-pip ENV PATH="/opt/venv/bin:$PATH" -COPY --from=builder /go/bin/pia-wg-config /usr/local/bin/pia-wg-config -COPY rotate.py entrypoint.py entrypoint.sh /usr/local/bin/ +COPY pia.py rotate.py entrypoint.py entrypoint.sh /usr/local/bin/ RUN chmod +x /usr/local/bin/rotate.py /usr/local/bin/entrypoint.py /usr/local/bin/entrypoint.sh @@ -28,6 +20,9 @@ ENV REGION_SELECT=fastest ENV SERVERLIST_CACHE_PATH=/config/cache/pia-serverlist.json ENV SERVERLIST_CACHE_TTL=24h ENV SERVERLIST_CACHE_MAX_AGE=168h +ENV TOKEN_CACHE_PATH=/config/cache/pia-token.json +ENV TOKEN_CACHE_TTL=20h +ENV PIA_CA_PATH=/config/cache/ca.rsa.4096.crt ENV WG_CONFIG_MAX_AGE=7d ENV LATENCY_SWITCH_MARGIN_MS=15 diff --git a/Dockers/gluetun-pia-wireguard-rotator/README.md b/Dockers/gluetun-pia-wireguard-rotator/README.md index 752274d..fa33987 100644 --- a/Dockers/gluetun-pia-wireguard-rotator/README.md +++ b/Dockers/gluetun-pia-wireguard-rotator/README.md @@ -1,6 +1,6 @@ # gluetun-pia-wireguard-rotator -Sidecar die op een cron-schema (en bij container-start) de **snelste** PIA WireGuard-region kiest (TCP-latency) via [pia-wg-config](https://github.com/ccarpinteri/pia-wg-config), `wg0.conf` op het gedeelde gluetun-volume schrijft, en een configureerbare lijst containers herstart. +Sidecar die op een cron-schema (en bij container-start) de **snelste** PIA WireGuard-server kiest (TCP-latency), lokaal een keypair maakt, via PIA `addKey` registreert, `wg0.conf` op het gedeelde gluetun-volume schrijft, en afhankelijke containers herstart. ## Vereisten @@ -19,12 +19,14 @@ Sidecar die op een cron-schema (en bij container-start) de **snelste** PIA WireG | `PIA_PASS` | PIA-wachtwoord | | `PIA_REGIONS` | CSV (`nl_amsterdam,france,belgium`) of JSON-array (`["nl_amsterdam","france"]`) | -Region-codes moeten overeenkomen met `pia-wg-config` (niet de OpenVPN-namen uit Gluetun's ingebouwde PIA-provider). +Region-IDs komen uit de PIA serverlist (niet de OpenVPN-namen uit Gluetun's ingebouwde PIA-provider). Lijst opvragen: ```bash -docker run --rm --entrypoint pia-wg-config bramkel/gluetun-pia-wireguard-rotator:latest --list-regions +docker run --rm --entrypoint /opt/venv/bin/python \ + bramkel/gluetun-pia-wireguard-rotator:latest \ + /usr/local/bin/rotate.py --list-regions ``` ### Optional @@ -35,33 +37,34 @@ docker run --rm --entrypoint pia-wg-config bramkel/gluetun-pia-wireguard-rotator | `GLUETUN_CONTAINER` | `m3u-filter-vpn` | Gluetun-container; wordt **altijd als eerste** herstart | | `WG_CONFIG_PATH` | `/config/wireguard/wg0.conf` | Pad waar `wg0.conf` wordt geschreven | | `ROTATOR_STATE_PATH` | `/config/rotator-state.json` | Laatste rotatie-metadata | -| `ROTATE_CRON` | `0 3 * * *` | 5-veld cron-expressie (minuut uur dag-van-maand maand dag-van-week), in `TZ`. Macros: `@hourly`, `@daily`, `@weekly`, `@monthly`, `@yearly` | -| `REGION_SELECT` | `fastest` | `fastest` = laagste TCP-latency naar WG-servers in `PIA_REGIONS`; `random` = willekeurig (slaat vorige region over indien mogelijk) | +| `ROTATE_CRON` | `0 3 * * *` | 5-veld cron-expressie, in `TZ`. Macros: `@hourly`, `@daily`, `@weekly`, `@monthly`, `@yearly` | +| `REGION_SELECT` | `fastest` | `fastest` = laagste TCP-latency; `random` = willekeurige regio/server | | `LATENCY_PORT` | `1337` | TCP-poort voor latency-probes | | `LATENCY_TIMEOUT_SECONDS` | `2` | Timeout per probe | | `LATENCY_SAMPLES` | `2` | Aantal samples per server-IP (gemiddelde) | -| `LATENCY_SWITCH_MARGIN_MS` | `15` | Alleen switchen van regio als de winst ≥ deze marge is (minder churn / token-calls) | -| `SERVERLIST_CACHE_PATH` | `/config/cache/pia-serverlist.json` | Disk-cache voor PIA serverlist (gedeeld met `pia-wg-config`) | -| `SERVERLIST_CACHE_TTL` | `24h` | Gebruik cache zonder refresh (`Ns`/`Nm`/`Nh`/`Nd` of seconden) | -| `SERVERLIST_CACHE_MAX_AGE` | `168h` | Maximale leeftijd; daarna verplicht vernieuwen (stale fallback bij fetch-fout) | -| `WG_CONFIG_MAX_AGE` | `7d` | Geen nieuwe token/config zolang regio gelijk blijft en `wg0.conf` jonger is | -| `FORCE_ROTATE` | `false` | `true` = altijd nieuwe config + container-restarts, cache-skip negeren | -| `RATE_LIMIT_WAIT_SECONDS` | `3600` | Wachttijd bij PIA rate-limit (`429` / `too_many_attempts`) vóór retry | +| `LATENCY_SWITCH_MARGIN_MS` | `15` | Alleen switchen als de winst ≥ deze marge is | +| `SERVERLIST_CACHE_PATH` | `/config/cache/pia-serverlist.json` | Disk-cache voor PIA serverlist | +| `SERVERLIST_CACHE_TTL` | `24h` | Cache zonder refresh (`Ns`/`Nm`/`Nh`/`Nd` of seconden) | +| `SERVERLIST_CACHE_MAX_AGE` | `168h` | Max leeftijd; stale fallback bij fetch-fout | +| `TOKEN_CACHE_PATH` | `/config/cache/pia-token.json` | Disk-cache voor PIA auth-token | +| `TOKEN_CACHE_TTL` | `20h` | Hergebruik token i.p.v. opnieuw inloggen | +| `PIA_CA_PATH` | `/config/cache/ca.rsa.4096.crt` | Gecachete PIA CA voor `addKey` TLS | +| `WG_CONFIG_MAX_AGE` | `7d` | Geen nieuwe token/`addKey` zolang endpoint gelijk blijft en config jonger is | +| `FORCE_ROTATE` | `false` | `true` = altijd nieuwe config + restarts | +| `FORCE_TOKEN_REFRESH` | `false` | `true` = token-cache negeren | +| `RATE_LIMIT_WAIT_SECONDS` | `3600` | Wachttijd bij PIA rate-limit vóór retry | | `TZ` | `Europe/Brussels` | Tijdzone voor scheduling | -`ROTATE_CRON` voorbeelden: `0 */6 * * *` (elke 6 uur), `0 3 * * 1-5` (weekdagen 03:00), `@hourly`. Quote de waarde in Compose (`'ROTATE_CRON=0 3 * * *'`) zodat YAML `*` niet speciaal interpreteert. Oude `ROTATE_AT=HH:MM` werkt nog als `ROTATE_CRON` leeg is. +`ROTATE_CRON` voorbeelden: `0 */6 * * *`, `0 3 * * 1-5`, `@hourly`. Quote in Compose: `'ROTATE_CRON=0 3 * * *'`. ## Output -- `wireguard/wg0.conf` op het gedeelde volume — Gluetun leest dit als `/gluetun/wireguard/wg0.conf` en dit **overschrijft** `WIREGUARD_*` environment variables -- `rotator-state.json` — laatste gekozen region, latency-resultaten, timestamp en herstartte containers +- `wireguard/wg0.conf` — Gluetun leest dit als `/gluetun/wireguard/wg0.conf` (overschrijft `WIREGUARD_*` env-vars) +- `rotator-state.json` — gekozen region/server, latency, timestamps +- `cache/` — serverlist, token, PIA CA ## Compose-integratie -Zie [`docker-compose.example.yml`](docker-compose.example.yml) voor een volledig voorbeeld met `m3u-filter-vpn`, m3u-editor en xtream-proxy. - -### Nieuwe service toevoegen - ```yaml gluetun-pia-wireguard-rotator: image: bramkel/gluetun-pia-wireguard-rotator:latest @@ -83,11 +86,9 @@ Zie [`docker-compose.example.yml`](docker-compose.example.yml) voor een volledig - m3u-filter-vpn ``` -Zet `PIA_USER` en `PIA_PASSWORD` in een host-`.env` (niet inline in compose). - ### Gluetun opschonen (aanbevolen na eerste succesvolle rotatie) -Zodra `wg0.conf` bestaat, heeft het bestand voorrang op env-vars. Verwijder uit `m3u-filter-vpn` om verwarring te voorkomen: +Verwijder uit gluetun zodra `wg0.conf` bestaat: - `WIREGUARD_ENDPOINT_IP` - `WIREGUARD_PUBLIC_KEY` @@ -102,27 +103,18 @@ Behoud minimaal: - VPN_TYPE=wireguard ``` -Optioneel host-`.env`-keys (`WIREGUARD_*`) opruimen als die niet meer gebruikt worden. - ## Deploy -1. Push/build image (`Dockers/gluetun-pia-wireguard-rotator/**` triggert Gitea CI → `bramkel/gluetun-pia-wireguard-rotator:latest`) +1. Push/build image (`Dockers/gluetun-pia-wireguard-rotator/**` → `bramkel/gluetun-pia-wireguard-rotator:latest`) 2. `docker compose up -d gluetun-pia-wireguard-rotator` -3. Controleer logs: `docker logs gluetun-pia-wireguard-rotator` en `docker logs m3u-filter-vpn` +3. Logs: `docker logs gluetun-pia-wireguard-rotator` ## Gedrag -1. Bij start / cron: latency meten (of random kiezen) -2. Serverlist komt uit disk-cache (`SERVERLIST_CACHE_*`); token/API alleen bij echte config-refresh -3. Geen `pia-wg-config` + geen restarts als regio gelijk blijft én `wg0.conf` jonger is dan `WG_CONFIG_MAX_AGE` -4. Anders: nieuwe config schrijven, `GLUETUN_CONTAINER` eerst herstarten, daarna `RESTART_CONTAINERS` +1. Latency meten over WG-servers in `PIA_REGIONS` (of random) +2. **Pin** de snelste server-IP (niet alleen regio) +3. Keypair lokaal genereren; token + `addKey` alleen bij echte refresh +4. Caches: serverlist, token (~20u), CA-cert +5. Skip token/`addKey`/restarts als endpoint gelijk blijft én config jonger dan `WG_CONFIG_MAX_AGE` -**Let op:** zet gluetun **niet** in `RESTART_CONTAINERS`; gebruik `GLUETUN_CONTAINER` daarvoor. Sidecars met `network_mode: service:...` horen in `RESTART_CONTAINERS`. - -**Let op:** elke rotatie veroorzaakt kort downtime voor alle VPN-afhankelijke services. - -## Security - -- De Docker socket geeft de rotator rechten om containers te herstarten; mount read-only waar mogelijk -- `wg0.conf` bevat private keys (`chmod 600`) -- Bewaar PIA-credentials in `.env`, niet in version control +**Let op:** zet gluetun **niet** in `RESTART_CONTAINERS`; gebruik `GLUETUN_CONTAINER`. Elke echte rotatie geeft korte downtime. diff --git a/Dockers/gluetun-pia-wireguard-rotator/entrypoint.py b/Dockers/gluetun-pia-wireguard-rotator/entrypoint.py index fcef0bd..682f6ee 100644 --- a/Dockers/gluetun-pia-wireguard-rotator/entrypoint.py +++ b/Dockers/gluetun-pia-wireguard-rotator/entrypoint.py @@ -13,7 +13,7 @@ from zoneinfo import ZoneInfo from croniter import croniter -# Matches rotate.sh EXIT_RATE_LIMITED (EX_TEMPFAIL) +# Matches rotate.py / pia.py EXIT_RATE_LIMITED (EX_TEMPFAIL) EXIT_RATE_LIMITED = 75 CRON_MACROS = { diff --git a/Dockers/gluetun-pia-wireguard-rotator/pia.py b/Dockers/gluetun-pia-wireguard-rotator/pia.py new file mode 100644 index 0000000..7325396 --- /dev/null +++ b/Dockers/gluetun-pia-wireguard-rotator/pia.py @@ -0,0 +1,360 @@ +#!/usr/bin/env python3 +"""Minimal PIA WireGuard client: serverlist, token, keygen, addKey.""" + +from __future__ import annotations + +import base64 +import json +import os +import re +import socket +import ssl +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from datetime import datetime +from http.client import HTTPSConnection +from pathlib import Path +from typing import Any + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + +EXIT_RATE_LIMITED = 75 +SERVERLIST_URL = "https://serverlist.piaservers.net/vpninfo/servers/v6" +TOKEN_URL = "https://www.privateinternetaccess.com/api/client/v2/token" +PIA_CA_URL = ( + "https://raw.githubusercontent.com/pia-foss/manual-connections/master/ca.rsa.4096.crt" +) + +_DURATION_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400} + + +def log(msg: str) -> None: + print(f"[{datetime.now().astimezone().isoformat(timespec='seconds')}] {msg}", file=sys.stderr) + + +def parse_duration_seconds(raw: str, default_seconds: int) -> int: + value = (raw or "").strip().lower() + if not value: + return default_seconds + if value.isdigit(): + return int(value) + match = re.fullmatch(r"(\d+)([smhd])", value) + if not match: + raise SystemExit(f"Invalid duration '{raw}' (use seconds or Ns/Nm/Nh/Nd)") + return int(match.group(1)) * _DURATION_UNITS[match.group(2)] + + +def env_bool(name: str, default: bool = False) -> bool: + raw = os.environ.get(name) + if raw is None or raw.strip() == "": + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +def serverlist_cache_path() -> Path: + return Path(os.environ.get("SERVERLIST_CACHE_PATH", "/config/cache/pia-serverlist.json")) + + +def token_cache_path() -> Path: + return Path(os.environ.get("TOKEN_CACHE_PATH", "/config/cache/pia-token.json")) + + +def ca_cache_path() -> Path: + return Path(os.environ.get("PIA_CA_PATH", "/config/cache/ca.rsa.4096.crt")) + + +@dataclass(frozen=True) +class WgServer: + region: str + ip: str + cn: str + + +@dataclass(frozen=True) +class WgKeys: + private_key: str + public_key: str + + +def generate_wg_keys() -> WgKeys: + private = X25519PrivateKey.generate() + priv_bytes = private.private_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PrivateFormat.Raw, + encryption_algorithm=serialization.NoEncryption(), + ) + pub_bytes = private.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + return WgKeys( + private_key=base64.b64encode(priv_bytes).decode("ascii"), + public_key=base64.b64encode(pub_bytes).decode("ascii"), + ) + + +def parse_serverlist_bytes(raw: bytes) -> dict[str, Any]: + try: + data, _ = json.JSONDecoder().raw_decode(raw.decode("utf-8", errors="replace")) + except json.JSONDecodeError as exc: + raise SystemExit(f"Failed to parse PIA server list: {exc}") from exc + if not isinstance(data, dict) or "regions" not in data: + raise SystemExit("PIA server list missing regions") + return data + + +def fetch_serverlist() -> dict[str, Any]: + cache_path = serverlist_cache_path() + ttl = parse_duration_seconds(os.environ.get("SERVERLIST_CACHE_TTL", "24h"), 86400) + max_age = parse_duration_seconds(os.environ.get("SERVERLIST_CACHE_MAX_AGE", "168h"), 604800) + force = env_bool("SERVERLIST_FORCE_REFRESH", False) + now = time.time() + + cached_raw: bytes | None = None + cache_age: float | None = None + if cache_path.is_file() and not force: + try: + cached_raw = cache_path.read_bytes() + cache_age = now - cache_path.stat().st_mtime + except OSError as exc: + log(f"Server list cache unreadable ({exc}); fetching fresh copy") + + if cached_raw is not None and cache_age is not None and cache_age <= ttl: + log(f"Using cached PIA server list (age {int(cache_age)}s <= TTL {ttl}s)") + return parse_serverlist_bytes(cached_raw) + + try: + with urllib.request.urlopen(SERVERLIST_URL, timeout=30) as resp: + text = resp.read().decode("utf-8", errors="replace") + data, end = json.JSONDecoder().raw_decode(text) + stripped = text[:end].encode("utf-8") + if not isinstance(data, dict) or "regions" not in data: + raise SystemExit("PIA server list missing regions") + cache_path.parent.mkdir(parents=True, exist_ok=True) + tmp = cache_path.with_suffix(cache_path.suffix + ".tmp") + tmp.write_bytes(stripped) + os.replace(tmp, cache_path) + log(f"Fetched and cached PIA server list -> {cache_path}") + return data + except (urllib.error.URLError, json.JSONDecodeError, OSError) as exc: + if cached_raw is not None and cache_age is not None and cache_age <= max_age: + log(f"Server list fetch failed ({exc}); using stale cache (age {int(cache_age)}s)") + return parse_serverlist_bytes(cached_raw) + raise SystemExit(f"Failed to fetch PIA server list: {exc}") from exc + + +def list_regions(serverlist: dict[str, Any] | None = None) -> list[dict[str, Any]]: + data = serverlist or fetch_serverlist() + regions = [] + for region in data.get("regions", []): + regions.append( + { + "id": region.get("id"), + "name": region.get("name"), + "country": region.get("country"), + "port_forward": bool(region.get("port_forward")), + "offline": bool(region.get("offline")), + "wg_servers": len((region.get("servers") or {}).get("wg") or []), + } + ) + regions.sort(key=lambda item: (item.get("country") or "", item.get("id") or "")) + return regions + + +def region_wg_servers(serverlist: dict[str, Any], region_id: str) -> list[WgServer]: + for region in serverlist.get("regions", []): + if region.get("id") != region_id: + continue + if region.get("offline"): + log(f"Region {region_id} is marked offline; skipping") + return [] + servers = (region.get("servers") or {}).get("wg") or [] + out: list[WgServer] = [] + for server in servers: + if not isinstance(server, dict) or not server.get("ip") or not server.get("cn"): + continue + out.append(WgServer(region=region_id, ip=server["ip"], cn=server["cn"])) + return out + log(f"Region {region_id} not found in PIA server list; skipping") + return [] + + +def ensure_pia_ca() -> Path: + path = ca_cache_path() + ttl = parse_duration_seconds(os.environ.get("PIA_CA_CACHE_TTL", "30d"), 30 * 86400) + if path.is_file() and (time.time() - path.stat().st_mtime) <= ttl: + return path + try: + with urllib.request.urlopen(PIA_CA_URL, timeout=30) as resp: + data = resp.read() + if b"BEGIN CERTIFICATE" not in data: + raise SystemExit("Downloaded PIA CA does not look like a certificate") + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_bytes(data) + os.replace(tmp, path) + log(f"Cached PIA CA certificate -> {path}") + except (urllib.error.URLError, OSError) as exc: + if path.is_file(): + log(f"PIA CA download failed ({exc}); using existing {path}") + return path + raise SystemExit(f"Failed to download PIA CA certificate: {exc}") from exc + return path + + +def _rate_limited(body: str, status_code: int) -> bool: + return status_code == 429 or "too_many_attempts" in body + + +def get_token(username: str, password: str) -> str: + cache_path = token_cache_path() + ttl = parse_duration_seconds(os.environ.get("TOKEN_CACHE_TTL", "20h"), 20 * 3600) + force = env_bool("FORCE_TOKEN_REFRESH", False) + + if cache_path.is_file() and not force: + try: + cached = json.loads(cache_path.read_text(encoding="utf-8")) + token = cached.get("token") + obtained = float(cached.get("obtained_at", 0)) + if isinstance(token, str) and token and (time.time() - obtained) <= ttl: + log(f"Using cached PIA token (age {int(time.time() - obtained)}s <= TTL {ttl}s)") + return token + except (OSError, json.JSONDecodeError, TypeError, ValueError): + pass + + form = urllib.parse.urlencode({"username": username, "password": password}).encode() + req = urllib.request.Request( + TOKEN_URL, + data=form, + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + body = resp.read().decode("utf-8", errors="replace") + status = resp.status + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + if _rate_limited(body, exc.code): + raise SystemExit(EXIT_RATE_LIMITED) from exc + raise SystemExit(f"PIA token request failed with status {exc.code}: {body}") from exc + except urllib.error.URLError as exc: + raise SystemExit(f"PIA token request failed: {exc}") from exc + + if _rate_limited(body, status): + raise SystemExit(EXIT_RATE_LIMITED) + + try: + payload = json.loads(body) + except json.JSONDecodeError as exc: + raise SystemExit(f"Invalid PIA token response: {body}") from exc + + token = payload.get("token") + if not isinstance(token, str) or not token: + raise SystemExit(f"PIA token response missing token: {body}") + + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text( + json.dumps({"token": token, "obtained_at": time.time()}, indent=2) + "\n", + encoding="utf-8", + ) + cache_path.chmod(0o600) + log("Fetched and cached new PIA token") + return token + + +class _HTTPSConnectionToIP(HTTPSConnection): + """HTTPS connection to a fixed IP while presenting server_hostname for SNI/verify.""" + + def __init__(self, ip: str, port: int, server_hostname: str, context: ssl.SSLContext): + super().__init__(server_hostname, port=port, context=context, timeout=30) + self._connect_ip = ip + self._server_hostname = server_hostname + + def connect(self) -> None: + sock = socket.create_connection((self._connect_ip, self.port), self.timeout) + self.sock = self._context.wrap_socket(sock, server_hostname=self._server_hostname) + + +def add_key(server: WgServer, token: str, public_key: str) -> dict[str, Any]: + ca_path = ensure_pia_ca() + context = ssl.create_default_context(cafile=str(ca_path)) + query = urllib.parse.urlencode({"pt": token, "pubkey": public_key}) + path = f"/addKey?{query}" + + conn = _HTTPSConnectionToIP(server.ip, 1337, server.cn, context) + try: + conn.request("GET", path, headers={"Content-Type": "application/json"}) + resp = conn.getresponse() + body = resp.read().decode("utf-8", errors="replace") + status = resp.status + finally: + conn.close() + + if _rate_limited(body, status): + raise SystemExit(EXIT_RATE_LIMITED) + if status != 200: + raise SystemExit(f"addKey failed for {server.cn}/{server.ip}: status {status}: {body}") + + try: + payload = json.loads(body) + except json.JSONDecodeError as exc: + raise SystemExit(f"Invalid addKey response: {body}") from exc + + if payload.get("status") not in (None, "OK"): + raise SystemExit(f"addKey rejected: {body}") + + required = ("server_key", "server_ip", "peer_ip", "dns_servers") + missing = [key for key in required if key not in payload] + if missing: + raise SystemExit(f"addKey response missing {missing}: {body}") + if not payload["dns_servers"]: + raise SystemExit(f"addKey response missing dns_servers: {body}") + + return payload + + +def render_wg_config(keys: WgKeys, addkey: dict[str, Any], server: WgServer) -> str: + endpoint_ip = addkey.get("server_ip") or server.ip + endpoint_port = addkey.get("server_port") or 1337 + dns = addkey["dns_servers"][0] + return ( + "[Interface]\n" + f"PrivateKey = {keys.private_key}\n" + f"Address = {addkey['peer_ip']}\n" + f"DNS = {dns}\n" + "[Peer]\n" + f"PublicKey = {addkey['server_key']}\n" + "AllowedIPs = 0.0.0.0/0\n" + f"Endpoint = {endpoint_ip}:{endpoint_port}\n" + "PersistentKeepalive = 25\n" + ) + + +def generate_wg_config( + username: str, + password: str, + server: WgServer, + outfile: Path, +) -> dict[str, Any]: + log(f"Requesting PIA token (cached when possible)") + token = get_token(username, password) + log("Generating local WireGuard keypair") + keys = generate_wg_keys() + log(f"Registering pubkey via addKey on {server.cn} ({server.ip})") + addkey = add_key(server, token, keys.public_key) + config = render_wg_config(keys, addkey, server) + outfile.write_text(config, encoding="utf-8") + outfile.chmod(0o600) + return { + "region": server.region, + "server_ip": server.ip, + "server_cn": server.cn, + "endpoint_ip": addkey.get("server_ip") or server.ip, + "peer_ip": addkey.get("peer_ip"), + } diff --git a/Dockers/gluetun-pia-wireguard-rotator/rotate.py b/Dockers/gluetun-pia-wireguard-rotator/rotate.py index c8d8a9c..bacf4e0 100644 --- a/Dockers/gluetun-pia-wireguard-rotator/rotate.py +++ b/Dockers/gluetun-pia-wireguard-rotator/rotate.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Generate a PIA WireGuard config and restart dependent containers.""" +"""Pick a PIA region/server and write a WireGuard config for Gluetun.""" from __future__ import annotations @@ -12,23 +12,12 @@ import subprocess import sys import tempfile import time -import urllib.error -import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from pathlib import Path from typing import Any -# EX_TEMPFAIL — entrypoint retries after RATE_LIMIT_WAIT_SECONDS -EXIT_RATE_LIMITED = 75 -SERVERLIST_URL = "https://serverlist.piaservers.net/vpninfo/servers/v6" - -_DURATION_UNITS = { - "s": 1, - "m": 60, - "h": 3600, - "d": 86400, -} +import pia def log(msg: str) -> None: @@ -42,34 +31,6 @@ def require_env(name: str) -> str: return value -def env_bool(name: str, default: bool = False) -> bool: - raw = os.environ.get(name) - if raw is None or raw.strip() == "": - return default - return raw.strip().lower() in {"1", "true", "yes", "on"} - - -def parse_duration_seconds(raw: str, default_seconds: int) -> int: - value = (raw or "").strip().lower() - if not value: - return default_seconds - if value.isdigit(): - return int(value) - match = re.fullmatch(r"(\d+)([smhd])", value) - if not match: - raise SystemExit(f"Invalid duration '{raw}' (use seconds or Ns/Nm/Nh/Nd)") - return int(match.group(1)) * _DURATION_UNITS[match.group(2)] - - -def duration_go(seconds: int) -> str: - """Format seconds as a Go duration string (pia-wg-config uses time.ParseDuration).""" - if seconds % 3600 == 0: - return f"{seconds // 3600}h" - if seconds % 60 == 0: - return f"{seconds // 60}m" - return f"{seconds}s" - - def parse_list(raw: str) -> list[str]: raw = raw.strip() if not raw: @@ -99,86 +60,6 @@ def read_state(state_path: Path) -> dict[str, Any]: return data if isinstance(data, dict) else {} -def read_previous_region(state_path: Path) -> str | None: - region = read_state(state_path).get("region") - return region if isinstance(region, str) and region else None - - -def serverlist_cache_path() -> Path: - return Path(os.environ.get("SERVERLIST_CACHE_PATH", "/config/cache/pia-serverlist.json")) - - -def fetch_serverlist() -> dict[str, Any]: - """Fetch PIA server list, with on-disk cache shared with pia-wg-config.""" - cache_path = serverlist_cache_path() - ttl = parse_duration_seconds(os.environ.get("SERVERLIST_CACHE_TTL", "24h"), 86400) - max_age = parse_duration_seconds(os.environ.get("SERVERLIST_CACHE_MAX_AGE", "168h"), 604800) - force = env_bool("SERVERLIST_FORCE_REFRESH", False) - now = time.time() - - cached_raw: bytes | None = None - cache_age: float | None = None - if cache_path.is_file() and not force: - try: - cached_raw = cache_path.read_bytes() - cache_age = now - cache_path.stat().st_mtime - except OSError as exc: - log(f"Server list cache unreadable ({exc}); fetching fresh copy") - cached_raw = None - cache_age = None - - if cached_raw is not None and cache_age is not None and cache_age <= ttl: - log(f"Using cached PIA server list (age {int(cache_age)}s <= TTL {ttl}s)") - return parse_serverlist_bytes(cached_raw) - - try: - with urllib.request.urlopen(SERVERLIST_URL, timeout=30) as resp: - raw = resp.read() - text = raw.decode("utf-8", errors="replace") - data, end = json.JSONDecoder().raw_decode(text) - stripped = text[:end].encode("utf-8") - if not isinstance(data, dict) or "regions" not in data: - raise SystemExit("PIA server list missing regions") - cache_path.parent.mkdir(parents=True, exist_ok=True) - tmp = cache_path.with_suffix(cache_path.suffix + ".tmp") - tmp.write_bytes(stripped) - os.replace(tmp, cache_path) - log(f"Fetched and cached PIA server list -> {cache_path}") - return data - except (urllib.error.URLError, json.JSONDecodeError, OSError) as exc: - if cached_raw is not None and cache_age is not None and cache_age <= max_age: - log(f"Server list fetch failed ({exc}); using stale cache (age {int(cache_age)}s)") - return parse_serverlist_bytes(cached_raw) - raise SystemExit(f"Failed to fetch PIA server list: {exc}") from exc - - -def parse_serverlist_bytes(raw: bytes) -> dict[str, Any]: - try: - data, _ = json.JSONDecoder().raw_decode(raw.decode("utf-8", errors="replace")) - except json.JSONDecodeError as exc: - raise SystemExit(f"Failed to parse PIA server list: {exc}") from exc - if not isinstance(data, dict) or "regions" not in data: - raise SystemExit("PIA server list missing regions") - return data - - -def region_wg_servers(serverlist: dict[str, Any], region_id: str) -> list[dict[str, str]]: - for region in serverlist.get("regions", []): - if region.get("id") != region_id: - continue - if region.get("offline"): - log(f"Region {region_id} is marked offline; skipping") - return [] - servers = region.get("servers", {}).get("wg") or [] - return [ - {"ip": s["ip"], "cn": s.get("cn", "")} - for s in servers - if isinstance(s, dict) and s.get("ip") - ] - log(f"Region {region_id} not found in PIA server list; skipping") - return [] - - def tcp_latency_ms(ip: str, port: int, timeout: float) -> float | None: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) @@ -202,55 +83,51 @@ def average_tcp_latency_ms(ip: str, port: int, timeout: float, samples: int) -> return sum(readings) / len(readings) -def pick_fastest_region( +def pick_fastest( candidates: list[str], - previous: str | None, -) -> tuple[str, list[dict[str, Any]]]: + previous_region: str | None, + previous_server_ip: str | None, +) -> tuple[pia.WgServer, list[dict[str, Any]]]: port = int(os.environ.get("LATENCY_PORT", "1337")) timeout = float(os.environ.get("LATENCY_TIMEOUT_SECONDS", "2")) samples = max(1, int(os.environ.get("LATENCY_SAMPLES", "2"))) margin = float(os.environ.get("LATENCY_SWITCH_MARGIN_MS", "15")) - serverlist = fetch_serverlist() - probes: list[tuple[str, str]] = [] - server_counts: dict[str, int] = {} + serverlist = pia.fetch_serverlist() + servers: list[pia.WgServer] = [] for region_id in candidates: - servers = region_wg_servers(serverlist, region_id) - if not servers: - continue - server_counts[region_id] = len(servers) - for server in servers: - probes.append((region_id, server["ip"])) - - if not probes: + servers.extend(pia.region_wg_servers(serverlist, region_id)) + if not servers: raise SystemExit("No reachable WireGuard servers found for configured regions") - best_by_region: dict[str, dict[str, Any]] = { - region_id: { - "region": region_id, - "latency_ms": None, - "server_ip": None, - "servers": server_counts[region_id], - "failures": 0, - } - for region_id in server_counts - } - - with ThreadPoolExecutor(max_workers=min(32, len(probes))) as pool: + best_by_region: dict[str, dict[str, Any]] = {} + with ThreadPoolExecutor(max_workers=min(32, len(servers))) as pool: futures = { - pool.submit(average_tcp_latency_ms, ip, port, timeout, samples): (region_id, ip) - for region_id, ip in probes + pool.submit(average_tcp_latency_ms, server.ip, port, timeout, samples): server + for server in servers } for future in as_completed(futures): - region_id, ip = futures[future] + server = futures[future] value = future.result() - current = best_by_region[region_id] + current = best_by_region.setdefault( + server.region, + { + "region": server.region, + "latency_ms": None, + "server_ip": None, + "server_cn": None, + "servers": 0, + "failures": 0, + }, + ) + current["servers"] += 1 if value is None: current["failures"] += 1 continue if current["latency_ms"] is None or value < current["latency_ms"]: current["latency_ms"] = round(value, 2) - current["server_ip"] = ip + current["server_ip"] = server.ip + current["server_cn"] = server.cn results = sorted( best_by_region.values(), @@ -260,12 +137,12 @@ def pick_fastest_region( item["region"], ), ) - for item in results: latency = "timeout" if item["latency_ms"] is None else f"{item['latency_ms']:.2f}ms" log( f"Latency {item['region']}: {latency} " - f"(best_ip={item['server_ip']}, servers={item['servers']}, failures={item['failures']})" + f"(best={item['server_cn']}/{item['server_ip']}, " + f"servers={item['servers']}, failures={item['failures']})" ) reachable = [item for item in results if item["latency_ms"] is not None] @@ -273,46 +150,68 @@ def pick_fastest_region( raise SystemExit("All latency probes failed; cannot select fastest region") winner = reachable[0] - if previous: - previous_result = next((item for item in reachable if item["region"] == previous), None) + if previous_region: + previous_result = next((item for item in reachable if item["region"] == previous_region), None) if previous_result is not None: improvement = previous_result["latency_ms"] - winner["latency_ms"] - if winner["region"] != previous and improvement < margin: + same_server = ( + previous_server_ip + and previous_result["server_ip"] == previous_server_ip + ) + if winner["region"] != previous_region and improvement < margin: log( - f"Keeping current region {previous} " + f"Keeping current region {previous_region} " f"({previous_result['latency_ms']:.2f}ms); " f"best {winner['region']} only {improvement:.2f}ms faster " f"(margin {margin:g}ms)" ) winner = previous_result - elif winner["region"] == previous: - log(f"Current region {previous} is still fastest") + elif winner["region"] == previous_region and same_server: + log(f"Current endpoint {previous_server_ip} is still fastest in {previous_region}") + elif winner["region"] == previous_region: + log(f"Current region {previous_region} is still fastest") - log(f"Fastest region: {winner['region']} ({winner['latency_ms']:.2f}ms via {winner['server_ip']})") - return winner["region"], results + server = pia.WgServer( + region=winner["region"], + ip=winner["server_ip"], + cn=winner["server_cn"], + ) + log(f"Selected {server.region} via {server.cn} ({server.ip}) at {winner['latency_ms']:.2f}ms") + return server, results -def pick_random_region(candidates: list[str], previous: str | None) -> str: - pool = [region for region in candidates if region != previous] or list(candidates) - if previous and previous not in pool and len(candidates) == 1: - log(f"Only one region configured; reusing previous: {previous}") - elif previous and previous not in pool: - log(f"Excluding previous region: {previous}") - return random.choice(pool) +def pick_random(candidates: list[str], previous_region: str | None) -> pia.WgServer: + serverlist = pia.fetch_serverlist() + pool = [region for region in candidates if region != previous_region] or list(candidates) + if previous_region and previous_region not in pool and len(candidates) == 1: + log(f"Only one region configured; reusing previous: {previous_region}") + elif previous_region and previous_region not in pool: + log(f"Excluding previous region: {previous_region}") + + random.shuffle(pool) + for region_id in pool: + servers = pia.region_wg_servers(serverlist, region_id) + if servers: + server = random.choice(servers) + log(f"Randomly selected {server.region} via {server.cn} ({server.ip})") + return server + raise SystemExit("No WireGuard servers found for configured regions") -def pick_region(state_path: Path) -> tuple[str, str, list[dict[str, Any]]]: +def pick_server(state_path: Path) -> tuple[pia.WgServer, str, list[dict[str, Any]]]: regions = parse_regions() - previous = read_previous_region(state_path) + state = read_state(state_path) + previous_region = state.get("region") if isinstance(state.get("region"), str) else None + previous_server_ip = state.get("server_ip") if isinstance(state.get("server_ip"), str) else None mode = os.environ.get("REGION_SELECT", "fastest").strip().lower() or "fastest" if mode == "random": - return pick_random_region(regions, previous), mode, [] + return pick_random(regions, previous_region), mode, [] if mode != "fastest": raise SystemExit(f"Invalid REGION_SELECT '{mode}' (expected fastest|random)") - region, results = pick_fastest_region(regions, previous) - return region, mode, results + server, results = pick_fastest(regions, previous_region, previous_server_ip) + return server, mode, results def parse_restart_containers() -> list[str]: @@ -337,48 +236,6 @@ def restart_containers(containers: list[str]) -> None: raise SystemExit(f"docker restart failed for container={container}") -def generate_wg_config(region: str, outfile: Path) -> None: - user = require_env("PIA_USER") - password = require_env("PIA_PASS") - cache_path = serverlist_cache_path() - ttl = parse_duration_seconds(os.environ.get("SERVERLIST_CACHE_TTL", "24h"), 86400) - max_age = parse_duration_seconds(os.environ.get("SERVERLIST_CACHE_MAX_AGE", "168h"), 604800) - - # Ensure cache exists so pia-wg-config can reuse it. - fetch_serverlist() - - cmd = [ - "pia-wg-config", - "-v", - "-r", - region, - "-o", - str(outfile), - "--serverlist-cache", - str(cache_path), - "--serverlist-cache-ttl", - duration_go(ttl), - "--serverlist-cache-max-age", - duration_go(max_age), - user, - password, - ] - result = subprocess.run(cmd, check=False, capture_output=True, text=True) - if result.stdout: - print(result.stdout, end="", file=sys.stderr) - if result.stderr: - print(result.stderr, end="", file=sys.stderr) - - output = f"{result.stdout}{result.stderr}" - if result.returncode == 0: - return - if "too_many_attempts" in output or "status 429" in output: - print(f"pia-wg-config rate-limited for region={region}", file=sys.stderr) - raise SystemExit(EXIT_RATE_LIMITED) - print(f"pia-wg-config failed for region={region}", file=sys.stderr) - raise SystemExit(1) - - def config_age_seconds(wg_path: Path) -> float | None: if not wg_path.is_file(): return None @@ -388,19 +245,26 @@ def config_age_seconds(wg_path: Path) -> float | None: return None -def should_skip_regeneration(region: str, previous: str | None, wg_path: Path) -> bool: - if env_bool("FORCE_ROTATE", False): +def should_skip_regeneration( + server: pia.WgServer, + state: dict[str, Any], + wg_path: Path, + latency_results: list[dict[str, Any]], +) -> bool: + if pia.env_bool("FORCE_ROTATE", False): log("FORCE_ROTATE=true; regenerating WireGuard config") return False - if previous != region: + previous_region = state.get("region") + previous_server_ip = state.get("server_ip") + if previous_region != server.region: return False age = config_age_seconds(wg_path) if age is None: return False - max_age = parse_duration_seconds(os.environ.get("WG_CONFIG_MAX_AGE", "7d"), 604800) + max_age = pia.parse_duration_seconds(os.environ.get("WG_CONFIG_MAX_AGE", "7d"), 604800) if age > max_age: log(f"Existing WireGuard config is stale (age {int(age)}s > max {max_age}s); regenerating") return False @@ -412,16 +276,36 @@ def should_skip_regeneration(region: str, previous: str | None, wg_path: Path) - ): return False + # Same region, different server: only regenerate if clearly faster. + if previous_server_ip and server.ip != previous_server_ip: + margin = float(os.environ.get("LATENCY_SWITCH_MARGIN_MS", "15")) + previous_latency = state.get("latency_ms") + winner = next((item for item in latency_results if item.get("server_ip") == server.ip), None) + new_latency = winner.get("latency_ms") if winner else None + if ( + isinstance(previous_latency, (int, float)) + and isinstance(new_latency, (int, float)) + and (previous_latency - new_latency) < margin + ): + log( + f"Keeping current server {previous_server_ip}; " + f"{server.ip} only {previous_latency - new_latency:.2f}ms faster " + f"(margin {margin:g}ms)" + ) + return True + log(f"Switching server within {server.region}: {previous_server_ip} -> {server.ip}") + return False + log( - f"Skipping PIA token/config request; region unchanged ({region}) " - f"and config age {int(age)}s <= {max_age}s" + f"Skipping PIA token/addKey; endpoint unchanged " + f"({server.region}/{server.ip}) and config age {int(age)}s <= {max_age}s" ) return True def write_state( state_path: Path, - region: str, + server: pia.WgServer, mode: str, restarted: list[str], latency_results: list[dict[str, Any]], @@ -429,10 +313,19 @@ def write_state( skipped: bool = False, ) -> None: previous = read_state(state_path) - winner = next((item for item in latency_results if item.get("region") == region), None) + winner = next( + ( + item + for item in latency_results + if item.get("region") == server.region and item.get("server_ip") == server.ip + ), + next((item for item in latency_results if item.get("region") == server.region), None), + ) now = datetime.now().astimezone().isoformat(timespec="seconds") payload: dict[str, Any] = { - "region": region, + "region": server.region, + "server_ip": server.ip, + "server_cn": server.cn, "selection": mode, "wg_config": os.environ.get("WG_CONFIG_PATH", "/config/wireguard/wg0.conf"), "restarted_containers": restarted, @@ -445,7 +338,8 @@ def write_state( payload["rotated_at"] = now if winner and winner.get("latency_ms") is not None: payload["latency_ms"] = winner["latency_ms"] - payload["server_ip"] = winner.get("server_ip") + elif skipped and previous.get("latency_ms") is not None: + payload["latency_ms"] = previous["latency_ms"] if latency_results: payload["latency_results"] = latency_results @@ -461,20 +355,31 @@ def rotate_once() -> None: state_path = Path(os.environ.get("ROTATOR_STATE_PATH", "/config/rotator-state.json")) wg_path = Path(os.environ.get("WG_CONFIG_PATH", "/config/wireguard/wg0.conf")) - previous = read_previous_region(state_path) + state = read_state(state_path) - region, mode, latency_results = pick_region(state_path) - log(f"Selected region: {region} (mode={mode})") + server, mode, latency_results = pick_server(state_path) + log(f"Selected endpoint: {server.region} / {server.cn} / {server.ip} (mode={mode})") - if should_skip_regeneration(region, previous, wg_path): - write_state(state_path, region, mode, [], latency_results, skipped=True) - log(f"Rotation skipped for region={region}") + if should_skip_regeneration(server, state, wg_path, latency_results): + # If we decided to keep the previous server IP, persist that identity. + keep_ip = state.get("server_ip") if isinstance(state.get("server_ip"), str) else server.ip + keep_cn = state.get("server_cn") if isinstance(state.get("server_cn"), str) else server.cn + if keep_ip != server.ip: + server = pia.WgServer(region=server.region, ip=keep_ip, cn=str(keep_cn or server.cn)) + write_state(state_path, server, mode, [], latency_results, skipped=True) + log(f"Rotation skipped for {server.region}/{server.ip}") return with tempfile.TemporaryDirectory(prefix="pia-rotate-") as tmp: tmp_conf = Path(tmp) / "wg0.conf" - log("Generating WireGuard config with pia-wg-config") - generate_wg_config(region, tmp_conf) + log("Generating WireGuard config via native PIA client") + pia.generate_wg_config( + require_env("PIA_USER"), + require_env("PIA_PASS"), + server, + tmp_conf, + ) + text = tmp_conf.read_text(encoding="utf-8", errors="replace") if not re.search(r"^\[Interface\]", text, flags=re.MULTILINE): raise SystemExit("Generated config missing [Interface] section") @@ -486,9 +391,21 @@ def rotate_once() -> None: restarted = parse_restart_containers() restart_containers(restarted) - write_state(state_path, region, mode, restarted, latency_results, skipped=False) - log(f"Rotation complete for region={region}") + write_state(state_path, server, mode, restarted, latency_results, skipped=False) + log(f"Rotation complete for {server.region}/{server.ip}") + + +def list_regions_main() -> None: + for region in pia.list_regions(): + print( + f"{region['id']:<35} {region['name']:<35} " + f"country={region['country']:<4} port-forward={region['port_forward']!s:<5} " + f"offline={region['offline']} wg={region['wg_servers']}" + ) if __name__ == "__main__": - rotate_once() + if len(sys.argv) > 1 and sys.argv[1] in {"--list-regions", "list-regions"}: + list_regions_main() + else: + rotate_once()