diff --git a/Dockers/gluetun-pia-wireguard-rotator/pia.py b/Dockers/gluetun-pia-wireguard-rotator/pia.py index 7325396..d70b87f 100644 --- a/Dockers/gluetun-pia-wireguard-rotator/pia.py +++ b/Dockers/gluetun-pia-wireguard-rotator/pia.py @@ -26,10 +26,21 @@ 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" +GTOKEN_URL = "https://www.privateinternetaccess.com/gtoken/generateToken" PIA_CA_URL = ( "https://raw.githubusercontent.com/pia-foss/manual-connections/master/ca.rsa.4096.crt" ) +# Cloudflare error 1010 blocks Python's default User-Agent. +HTTP_HEADERS = { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36" + ), + "Accept": "application/json, text/plain, */*", + "Accept-Language": "en-US,en;q=0.9", +} + _DURATION_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400} @@ -167,13 +178,21 @@ def list_regions(serverlist: dict[str, Any] | None = None) -> list[dict[str, Any def region_wg_servers(serverlist: dict[str, Any], region_id: str) -> list[WgServer]: + return _region_servers(serverlist, region_id, "wg") + + +def region_meta_servers(serverlist: dict[str, Any], region_id: str) -> list[WgServer]: + return _region_servers(serverlist, region_id, "meta") + + +def _region_servers(serverlist: dict[str, Any], region_id: str, kind: 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 [] + servers = (region.get("servers") or {}).get(kind) or [] out: list[WgServer] = [] for server in servers: if not isinstance(server, dict) or not server.get("ip") or not server.get("cn"): @@ -211,7 +230,122 @@ 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: +def _cloudflare_blocked(status_code: int, body: str) -> bool: + return status_code == 403 and ("error code: 1010" in body or "Attention Required" in body) + + +def _basic_auth_header(username: str, password: str) -> str: + token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") + return f"Basic {token}" + + +def _parse_token_body(body: str, source: str) -> str: + try: + payload = json.loads(body) + except json.JSONDecodeError as exc: + raise RuntimeError(f"{source}: invalid JSON: {body[:200]}") from exc + token = payload.get("token") + if not isinstance(token, str) or not token: + raise RuntimeError(f"{source}: missing token in response: {body[:200]}") + return token + + +def _store_token(token: str) -> str: + cache_path = token_cache_path() + 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) + return token + + +def _token_via_central(username: str, password: str) -> str: + form = urllib.parse.urlencode({"username": username, "password": password}).encode() + req = urllib.request.Request( + TOKEN_URL, + data=form, + method="POST", + headers={ + **HTTP_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 + if _cloudflare_blocked(exc.code, body): + raise RuntimeError(f"central token API blocked by Cloudflare (1010)") from exc + raise RuntimeError(f"central token API status {exc.code}: {body[:200]}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"central token API network error: {exc}") from exc + + if _rate_limited(body, status): + raise SystemExit(EXIT_RATE_LIMITED) + return _parse_token_body(body, "central token API") + + +def _token_via_gtoken(username: str, password: str) -> str: + req = urllib.request.Request( + GTOKEN_URL, + method="GET", + headers={ + **HTTP_HEADERS, + "Authorization": _basic_auth_header(username, password), + }, + ) + 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 + if _cloudflare_blocked(exc.code, body): + raise RuntimeError("gtoken API blocked by Cloudflare (1010)") from exc + raise RuntimeError(f"gtoken API status {exc.code}: {body[:200]}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"gtoken API network error: {exc}") from exc + + if _rate_limited(body, status): + raise SystemExit(EXIT_RATE_LIMITED) + return _parse_token_body(body, "gtoken API") + + +def _token_via_meta(username: str, password: str, meta: WgServer) -> str: + ca_path = ensure_pia_ca() + context = ssl.create_default_context(cafile=str(ca_path)) + conn = _HTTPSConnectionToIP(meta.ip, 443, meta.cn, context, timeout=10) + try: + conn.request( + "GET", + "/authv3/generateToken", + headers={ + **HTTP_HEADERS, + "Authorization": _basic_auth_header(username, password), + }, + ) + 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 RuntimeError(f"meta {meta.cn}/{meta.ip} status {status}: {body[:200]}") + return _parse_token_body(body, f"meta {meta.cn}") + + +def get_token(username: str, password: str, preferred_region: str | None = None) -> 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) @@ -227,52 +361,73 @@ def get_token(username: str, password: str) -> str: 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"}, - ) + errors: list[str] = [] + + for label, getter in ( + ("central v2 token API", lambda: _token_via_central(username, password)), + ("gtoken API", lambda: _token_via_gtoken(username, password)), + ): + try: + log(f"Requesting PIA token via {label}") + token = getter() + log(f"Fetched and cached new PIA token via {label}") + return _store_token(token) + except SystemExit: + raise + except Exception as exc: # noqa: BLE001 - try next auth method + errors.append(f"{label}: {exc}") + log(f"Token via {label} failed: {exc}") + + # Meta servers bypass Cloudflare; prefer the selected region, then any region. 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 + serverlist = fetch_serverlist() + except SystemExit as exc: + errors.append(f"serverlist for meta auth: {exc}") + serverlist = None - if _rate_limited(body, status): - raise SystemExit(EXIT_RATE_LIMITED) + meta_candidates: list[WgServer] = [] + if serverlist is not None: + if preferred_region: + meta_candidates.extend(region_meta_servers(serverlist, preferred_region)) + if not meta_candidates: + for region in serverlist.get("regions", []): + rid = region.get("id") + if not isinstance(rid, str): + continue + meta_candidates.extend(region_meta_servers(serverlist, rid)) + if len(meta_candidates) >= 8: + break + # Prefer legacy CN hostnames; new Server-* meta hosts often hang on /authv3. + meta_candidates.sort(key=lambda item: item.cn.startswith("Server-")) - try: - payload = json.loads(body) - except json.JSONDecodeError as exc: - raise SystemExit(f"Invalid PIA token response: {body}") from exc + for meta in meta_candidates[:5]: + label = f"meta {meta.cn} ({meta.ip})" + try: + log(f"Requesting PIA token via {label}") + token = _token_via_meta(username, password, meta) + log(f"Fetched and cached new PIA token via {label}") + return _store_token(token) + except SystemExit: + raise + except Exception as exc: # noqa: BLE001 + errors.append(f"{label}: {exc}") + log(f"Token via {label} failed: {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 + raise SystemExit("PIA token request failed:\n- " + "\n- ".join(errors)) 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) + def __init__( + self, + ip: str, + port: int, + server_hostname: str, + context: ssl.SSLContext, + timeout: float = 30, + ): + super().__init__(server_hostname, port=port, context=context, timeout=timeout) self._connect_ip = ip self._server_hostname = server_hostname @@ -342,8 +497,8 @@ def generate_wg_config( server: WgServer, outfile: Path, ) -> dict[str, Any]: - log(f"Requesting PIA token (cached when possible)") - token = get_token(username, password) + log("Requesting PIA token (cached when possible)") + token = get_token(username, password, preferred_region=server.region) log("Generating local WireGuard keypair") keys = generate_wg_keys() log(f"Registering pubkey via addKey on {server.cn} ({server.ip})")