diff --git a/docs/NETWORK_CONFIG.md b/docs/NETWORK_CONFIG.md index ffb01651..dd42ff9a 100644 --- a/docs/NETWORK_CONFIG.md +++ b/docs/NETWORK_CONFIG.md @@ -55,6 +55,43 @@ You can even set a specific server number this way, e.g., `--proxy nordvpn:gb236 Note that `gb` is used instead of `uk` to be more consistent across regional systems. +### protonvpn (dict) + +Enable Proton VPN as a proxy provider. Proton mints short-lived HTTPS proxy credentials per session, so +authentication reuses an exported `account.proton.me` browser session rather than static credentials. + +Export the cookies for `account.proton.me` (Netscape `cookies.txt` or a JSON cookie list) — they must include +the `AUTH-` and `REFRESH-` cookies — and save them at `/vpn/protonvpn.txt`. **No yaml is +required**: the provider auto-loads when that file exists. Exporting straight from the logged-in web page is +fine; cookies for other Proton apps (e.g. `mail.proton.me`) are ignored automatically. The session refreshes on +its own when the access token expires, so you only re-export if it is fully revoked. **Free Proton accounts +work**, limited to the free-tier exit countries. + +Configuration is optional — only add a block to override defaults: + +```yaml +proxy_providers: + protonvpn: + cookie_path: /path/to/protonvpn.txt # optional; defaults to /vpn/protonvpn.txt + server_map: + stream-us: us:ny # optional alias -> country / country:city / countryNN +``` + +Query formats (after the provider prefix). City uses the NordVPN-style colon, like `nordvpn:us:seattle`: + +- `--proxy protonvpn:us` — random server in the country (the chosen server name is logged) +- `--proxy protonvpn:us:ny` / `protonvpn:ca:vancouver` — by city (name or initials) +- `--proxy protonvpn:de203` — pin Proton server **#203** (the number shown in the log, e.g. `DE#203`) + +Use `uk` (Proton's code for the United Kingdom); `gb` is accepted and mapped to `uk` automatically. Proton's +server numbers are not sequential (a country may have `#203`, `#813`, …), so pick the number from a previous +run's log line rather than guessing. + +Selection only offers servers your plan can use (paid accounts are detected automatically) and **prefers paid +servers over the free pool**, falling back to free only when no paid server is available — a pinned `#NN` always +wins. **Tor and Secure Core servers are excluded** (both are slow for a plain proxy). Secure Core servers use +port 443, all others 4443 — handled automatically. + ### surfsharkvpn (dict) Enable Surfshark VPN proxy service using Surfshark Service credentials (not your login password). diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index 66fe718e..06648afb 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -50,7 +50,7 @@ from unshackle.core.drm import DRM_T, ClearKeyCENC, MonaLisa, PlayReady, Widevin from unshackle.core.events import events from unshackle.core.music import (MusicAudioIntegrityError, MusicMetadataResult, MusicPlanner, MusicRenderer, file_md5, verify_music_audio, write_music_manifest, write_music_metadata) -from unshackle.core.proxies import Basic, ExpressVPN, Gluetun, Hola, NordVPN, SurfsharkVPN, WindscribeVPN +from unshackle.core.proxies import Basic, ExpressVPN, Gluetun, Hola, NordVPN, ProtonVPN, SurfsharkVPN, WindscribeVPN from unshackle.core.service import Service from unshackle.core.services import Services from unshackle.core.title_cacher import get_account_hash @@ -1039,10 +1039,15 @@ class dl: with console.status("Loading Proxy Providers...", spinner="dots"): if config.proxy_providers.get("basic"): self.proxy_providers.append(Basic(**config.proxy_providers["basic"])) - if config.proxy_providers.get("expressvpn"): - self.proxy_providers.append(ExpressVPN(**config.proxy_providers["expressvpn"])) + # ExpressVPN/ProtonVPN auto-load when their default cookie file exists (no yaml needed) + expressvpn = ExpressVPN(**(config.proxy_providers.get("expressvpn") or {})) + if config.proxy_providers.get("expressvpn") or expressvpn.cookie_path.is_file(): + self.proxy_providers.append(expressvpn) if config.proxy_providers.get("nordvpn"): self.proxy_providers.append(NordVPN(**config.proxy_providers["nordvpn"])) + proton = ProtonVPN(**(config.proxy_providers.get("protonvpn") or {})) + if config.proxy_providers.get("protonvpn") or proton.cookie_path.is_file(): + self.proxy_providers.append(proton) if config.proxy_providers.get("surfsharkvpn"): self.proxy_providers.append(SurfsharkVPN(**config.proxy_providers["surfsharkvpn"])) if config.proxy_providers.get("windscribevpn"): @@ -1059,9 +1064,10 @@ class dl: if re.match(r"^[a-z]+:.+$", proxy, re.IGNORECASE): # requesting proxy from a specific proxy provider requested_provider, proxy = proxy.split(":", maxsplit=1) - # Match simple region codes (us, ca, uk1) or provider:region format (nordvpn:ca, windscribe:us) - if re.match(r"^[a-z]{2}(?:[-][a-z0-9]+)*(?:\d+)?$", proxy, re.IGNORECASE) or re.match( - r"^[a-z]+:[a-z]{2}(?:[-][a-z0-9]+)*(?:\d+)?$", proxy, re.IGNORECASE + # Match simple region codes (us, ca, uk1, us:ny) or provider:region (nordvpn:ca, protonvpn:us:ny). + # ':' is allowed as a city separator (e.g. nordvpn:us:seattle, protonvpn:de:berlin). + if re.match(r"^[a-z]{2}(?:[-:][a-z0-9]+)*(?:\d+)?$", proxy, re.IGNORECASE) or re.match( + r"^[a-z]+:[a-z]{2}(?:[-:][a-z0-9]+)*(?:\d+)?$", proxy, re.IGNORECASE ): proxy = proxy.lower() # Preserve the original user query (region code) for service-specific proxy_map overrides. diff --git a/unshackle/commands/search.py b/unshackle/commands/search.py index f4a34250..76fe5938 100644 --- a/unshackle/commands/search.py +++ b/unshackle/commands/search.py @@ -16,7 +16,7 @@ from unshackle.core import binaries from unshackle.core.config import config from unshackle.core.console import console from unshackle.core.constants import context_settings -from unshackle.core.proxies import Basic, ExpressVPN, Gluetun, Hola, NordVPN, SurfsharkVPN, WindscribeVPN +from unshackle.core.proxies import Basic, ExpressVPN, Gluetun, Hola, NordVPN, ProtonVPN, SurfsharkVPN, WindscribeVPN from unshackle.core.service import Service from unshackle.core.services import Services from unshackle.core.utils.click_types import ContextData @@ -67,10 +67,14 @@ def search(ctx: click.Context, no_proxy: bool, profile: Optional[str] = None, pr with console.status("Loading Proxy Providers...", spinner="dots"): if config.proxy_providers.get("basic"): proxy_providers.append(Basic(**config.proxy_providers["basic"])) - if config.proxy_providers.get("expressvpn"): - proxy_providers.append(ExpressVPN(**config.proxy_providers["expressvpn"])) + expressvpn = ExpressVPN(**(config.proxy_providers.get("expressvpn") or {})) + if config.proxy_providers.get("expressvpn") or expressvpn.cookie_path.is_file(): + proxy_providers.append(expressvpn) if config.proxy_providers.get("nordvpn"): proxy_providers.append(NordVPN(**config.proxy_providers["nordvpn"])) + proton = ProtonVPN(**(config.proxy_providers.get("protonvpn") or {})) + if config.proxy_providers.get("protonvpn") or proton.cookie_path.is_file(): + proxy_providers.append(proton) if config.proxy_providers.get("surfsharkvpn"): proxy_providers.append(SurfsharkVPN(**config.proxy_providers["surfsharkvpn"])) if config.proxy_providers.get("windscribevpn"): diff --git a/unshackle/core/proxies/__init__.py b/unshackle/core/proxies/__init__.py index 2615b807..2b313043 100644 --- a/unshackle/core/proxies/__init__.py +++ b/unshackle/core/proxies/__init__.py @@ -3,7 +3,8 @@ from .expressvpn import ExpressVPN from .gluetun import Gluetun from .hola import Hola from .nordvpn import NordVPN +from .proton import ProtonVPN from .surfsharkvpn import SurfsharkVPN from .windscribevpn import WindscribeVPN -__all__ = ("Basic", "ExpressVPN", "Gluetun", "Hola", "NordVPN", "SurfsharkVPN", "WindscribeVPN") +__all__ = ("Basic", "ExpressVPN", "Gluetun", "Hola", "NordVPN", "ProtonVPN", "SurfsharkVPN", "WindscribeVPN") diff --git a/unshackle/core/proxies/proton.py b/unshackle/core/proxies/proton.py new file mode 100644 index 00000000..1475050b --- /dev/null +++ b/unshackle/core/proxies/proton.py @@ -0,0 +1,361 @@ +from __future__ import annotations + +import json +import logging +import os +import random +import re +from pathlib import Path +from typing import Optional + +import requests + +from unshackle.core.config import config +from unshackle.core.proxies.proxy import Proxy + +log = logging.getLogger("proxies.proton") + + +class ProtonVPN(Proxy): + """ + Proton VPN HTTPS proxy provider. + + Reuses an exported account.proton.me session (the AUTH-/REFRESH- cookies) to call + the VPN API, resolves a server by country/city, mints short-lived proxy credentials, and + returns an authenticated HTTPS proxy URL (https://user:pass@server:4443). + + Query format (after the provider prefix, e.g. "protonvpn:us"): + us random server in the country (paid preferred over free) + us12 Proton server #12 (the number shown in the connection log) + us:ny city, NordVPN-style + + Free Proton accounts work, limited to the free-tier exit countries. + """ + + API_BASE = "https://account.proton.me/api" + APP_VERSION = "browser-vpn@1.3.5" + PROXY_PORT = 4443 + SECURE_CORE_PORT = 443 + SECURE_CORE_FEATURE = 1 # Features bitmask: 1=SecureCore, 2=Tor, 4=P2P, 8=Streaming, 16=IPv6 + TOR_FEATURE = 2 + EXCLUDE_FEATURES = SECURE_CORE_FEATURE | TOR_FEATURE # skip Secure Core + Tor; both are slow for a plain proxy + TOKEN_DURATION = 1200 # free tier caps the granted lifetime lower + COUNTRY_ALIASES = {"gb": "uk"} # Proton labels the United Kingdom "UK", not the ISO "GB" + + def __init__( + self, + server_map: Optional[dict[str, str]] = None, + cookie_path: Optional[str] = None, + cache_path: Optional[str] = None, + timeout: float = 10.0, + ): + self.server_map = {str(k).lower().strip(): str(v).lower().strip() for k, v in (server_map or {}).items()} + self.timeout = timeout + + self.uid: Optional[str] = None + self.access_token: Optional[str] = None + self.refresh_token: Optional[str] = None + self.logicals: Optional[list[dict]] = None + self.max_tier: Optional[int] = None + self.last_name: Optional[str] = None + self.last_city: Optional[str] = None + self.last_host: Optional[str] = None + + self.cookie_path = Path(cookie_path).expanduser() if cookie_path else self.default_cookie_path() + self.cache_path = Path(cache_path).expanduser() if cache_path else self.default_cache_path() + + def __repr__(self) -> str: + try: + servers = self.get_logicals() # eager so the "Loaded ProtonVPN: …" line shows the catalog size + except Exception: + servers = self.logicals or [] + if not servers: + return "Proton VPN HTTPS Proxy" + countries = len({str(s.get("ExitCountry") or "").upper() for s in servers if s.get("ExitCountry")}) + return ( + f"{countries} Countr{'ies' if countries != 1 else 'y'} ({len(servers)} Server{'s' if servers != 1 else ''})" + ) + + def get_proxy(self, query: str) -> Optional[str]: + server = self.resolve_server(query.strip().lower()) + if not server: + return None + creds = self.get_browser_token() + if not creds: + log.error("Proton: could not obtain proxy credentials") + return None + + self.last_host = host = str(server.get("Domain") or "") + port = self.SECURE_CORE_PORT if int(server.get("Features") or 0) & self.SECURE_CORE_FEATURE else self.PROXY_PORT + # Username (a JWT) and Password are URL-safe, but quote so requests rebuilds Basic auth cleanly + user, pwd = (requests.utils.quote(c, safe="") for c in creds) + return f"https://{user}:{pwd}@{host}:{port}" + + def last_connection_display(self) -> Optional[str]: + if not self.last_name or not self.last_host: + return None + label = f"{self.last_name} - {self.last_city}" if self.last_city else self.last_name + return f"({label}): {self.last_host.replace('.protonvpn.net', '')}" + + def resolve_server(self, query: str) -> Optional[dict]: + self.last_name = self.last_city = self.last_host = None + query = self.server_map.get(query, query) + country, city, server_num = self.parse_query(query) + + pool = [] + if country: + max_tier = self.get_max_tier() # only servers the account can actually connect to + pool = [ + s + for s in self.get_logicals() + if str(s.get("ExitCountry") or "").lower() == country + and s.get("Status") == 1 + and int(s.get("Tier") or 0) <= max_tier + and not int(s.get("Features") or 0) & self.EXCLUDE_FEATURES + ] + if city: + pool = self.match_city(pool, city) + + if not pool: + log.warning("Proton: no server matched query '%s'", query) + return None + return self.select(pool, server_num) + + def parse_query(self, query: str) -> tuple[Optional[str], Optional[str], Optional[int]]: + """Parse "us", "us12"/"us-12" (server #12), and "us:ny" (city).""" + city = None + if ":" in query: + query, city = query.split(":", 1) + city = city.strip() or None + + match = re.match(r"^([a-z]{2})-?(\d+)?$", query.strip()) + if not match: + return None, None, None + country = self.COUNTRY_ALIASES.get(match.group(1), match.group(1)) + server_num = int(match.group(2)) if match.group(2) else None + + valid = {str(s.get("ExitCountry") or "").lower() for s in self.get_logicals()} + return (country, city, server_num) if country in valid else (None, None, None) + + def match_city(self, servers: list[dict], city_query: str) -> list[dict]: + city_query = city_query.strip().lower() + exact, prefix, substring = [], [], [] + for s in servers: + city = str(s.get("City") or "") + slug = slugify(city) + abbreviation = "".join(w[0] for w in re.findall(r"[a-z]+", city.lower())) + if city_query in (slug, abbreviation): + exact.append(s) + elif slug.startswith(city_query): + prefix.append(s) + elif city_query in slug: + substring.append(s) + return exact or prefix or substring + + def select(self, pool: list[dict], server_num: Optional[int]) -> dict: + if server_num is not None: + matches = [s for s in pool if server_no(s) == server_num] + if matches: + pool = matches + else: + log.warning("Proton: server #%d not found (%d available), picking random", server_num, len(pool)) + else: + pool = [s for s in pool if int(s.get("Tier") or 0) > 0] or pool + chosen = random.choice(pool) + self.last_name = chosen.get("Name") + self.last_city = chosen.get("City") + return chosen + + def get_logicals(self) -> list[dict]: + if self.logicals is None: + response = self.api("GET", "/vpn/v1/logicals") + self.logicals = (response.json().get("LogicalServers") or []) if response is not None else [] + return self.logicals + + def get_max_tier(self) -> int: + """The account's plan tier (0=free, 2=plus). Caps which servers it can connect to.""" + if self.max_tier is None: + response = self.api("GET", "/vpn") + vpn = (response.json().get("VPN") or {}) if response is not None else {} + self.max_tier = int(vpn.get("MaxTier") or 0) + return self.max_tier + + def get_browser_token(self) -> Optional[tuple[str, str]]: + response = self.api("GET", f"/vpn/v1/browser/token?Duration={self.TOKEN_DURATION}") + if response is None: + return None + body = response.json() + if body.get("Username") and body.get("Password"): + return body["Username"], body["Password"] + log.error("Proton: browser/token response missing credentials") + return None + + def api(self, method: str, path: str) -> Optional[requests.Response]: + """Call the Proton API, refreshing the session once on a 401.""" + if not self.access_token: + self.load_session() + if not self.access_token: + log.error("Proton: no session available; export account.proton.me cookies to %s", self.cookie_path) + return None + + for attempt in range(2): + response = requests.request(method, f"{self.API_BASE}{path}", headers=self.headers(), timeout=self.timeout) + if response.status_code == 401 and attempt == 0 and self.refresh_session(): + continue + if response.status_code == 401: + log.error("Proton: session expired; re-export account.proton.me cookies to %s", self.cookie_path) + return None + if not response.ok: + raise ValueError(f"Proton request failed with HTTP {response.status_code}: {path}") + return response + return None + + def headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self.access_token}", + "x-pm-uid": self.uid or "", + "x-pm-appversion": self.APP_VERSION, + "Accept": "application/json", + } + + def refresh_session(self) -> bool: + if not self.refresh_token or not self.uid: + return False + response = requests.post( + f"{self.API_BASE}/auth/refresh", + json={ + "UID": self.uid, + "ResponseType": "token", + "GrantType": "refresh_token", + "RefreshToken": self.refresh_token, + "RedirectURI": "https://protonvpn.com", + }, + headers={"x-pm-uid": self.uid, "x-pm-appversion": self.APP_VERSION, "Accept": "application/json"}, + timeout=self.timeout, + ) + if not response.ok: + log.warning("Proton: session refresh failed with HTTP %s", response.status_code) + return False + body = response.json() + self.access_token = body.get("AccessToken") or self.access_token + self.refresh_token = body.get("RefreshToken") or self.refresh_token + self.save_session() + return True + + def load_session(self) -> None: + cached = self.load_cache() + + api_host = self.API_BASE.split("://", 1)[-1].split("/", 1)[0] + cookie_uid = cookie_access = cookie_refresh = None + for name, value, domain in self.load_cookies(): + if domain and domain.lstrip(".") != api_host: + continue + if name.startswith("AUTH-"): + cookie_uid, cookie_access = name[len("AUTH-") :], value + elif name.startswith("REFRESH-"): + cookie_refresh = value + + # Cached tokens win, but only for the same account and not older than a re-exported cookie file + same_account = cookie_uid is None or cached.get("uid") == cookie_uid + cookie_newer = ( + self.cookie_path.is_file() + and self.cache_path.is_file() + and self.cookie_path.stat().st_mtime > self.cache_path.stat().st_mtime + ) + if cached.get("access_token") and same_account and not cookie_newer: + self.uid = cached["uid"] + self.access_token = cached["access_token"] + self.refresh_token = cached.get("refresh_token") + return + + self.uid, self.access_token = cookie_uid, cookie_access + # A cookie-only re-export may lack REFRESH-; reuse the cache's refresh token if it's the same account + self.refresh_token = cookie_refresh or (cached.get("refresh_token") if same_account else None) + if self.uid and self.access_token: + self.save_session() + + def load_cookies(self) -> list[tuple[str, str, str]]: + """Return [(name, value, domain), ...] from a Netscape cookies.txt or JSON cookie list.""" + if not self.cookie_path.is_file(): + return [] + try: + content = self.cookie_path.read_text(encoding="utf-8").strip() + except OSError as error: + log.error("Proton: failed to read cookies file %s: %s", self.cookie_path, error) + return [] + + try: + data = json.loads(content) + except json.JSONDecodeError: + # Netscape: domain \t flag \t path \t secure \t expiry \t name \t value + # Proton's AUTH-/REFRESH- cookies are HttpOnly, exported as "#HttpOnly_\t..." + out = [] + for line in content.splitlines(): + line = line.removeprefix("#HttpOnly_") + parts = line.split("\t") + if not line.startswith("#") and len(parts) >= 7: + out.append((parts[5], parts[6], parts[0])) + return out + + items = data.get("cookies") if isinstance(data, dict) else data + return [ + (str(c["name"]), str(c.get("value", "")), str(c.get("domain", ""))) + for c in (items or []) + if isinstance(c, dict) and c.get("name") + ] + + def load_cache(self) -> dict: + if not self.cache_path.is_file(): + return {} + try: + data = json.loads(self.cache_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + log.warning("Proton: failed to read token cache %s: %s", self.cache_path, error) + return {} + return data if isinstance(data, dict) else {} + + def save_session(self) -> None: + tokens = {"uid": self.uid, "access_token": self.access_token, "refresh_token": self.refresh_token} + try: + self.cache_path.parent.mkdir(parents=True, exist_ok=True) + write_private(self.cache_path, json.dumps({k: v for k, v in tokens.items() if v}, indent=2)) + except OSError as error: + log.error("Proton: failed to save token cache %s: %s", self.cache_path, error) + + def default_cookie_path(self) -> Path: + cookies_dir = Path(config.directories.cookies) + for folder in ("vpn", "vpns"): + for fname in ("protonvpn.txt", "proton.txt"): + if (candidate := cookies_dir / folder / fname).is_file(): + return candidate + return cookies_dir / "vpn" / "protonvpn.txt" + + def default_cache_path(self) -> Path: + return Path(config.directories.cache) / "global" / "proton_tokens.json" + + +def slugify(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") + + +def server_no(server: dict) -> Optional[int]: + """Proton's server number from the logical name, e.g. "US#231" -> 231, "US-CO#21" -> 21.""" + match = re.search(r"#(\d+)", str(server.get("Name") or "")) + return int(match.group(1)) if match else None + + +def write_private(path: Path, content: str) -> None: + """Write content to path created with owner-only (0600) permissions.""" + if hasattr(os, "fchmod"): + fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + os.fchmod(fd, 0o600) + os.write(fd, content.encode("utf-8")) + finally: + os.close(fd) + else: + path.write_text(content, encoding="utf-8") + try: + path.chmod(0o600) + except OSError: + pass diff --git a/unshackle/core/proxies/resolve.py b/unshackle/core/proxies/resolve.py index 7e851950..1c1df39f 100644 --- a/unshackle/core/proxies/resolve.py +++ b/unshackle/core/proxies/resolve.py @@ -22,16 +22,22 @@ def initialize_proxy_providers() -> List[Any]: from unshackle.core.proxies.expressvpn import ExpressVPN from unshackle.core.proxies.hola import Hola from unshackle.core.proxies.nordvpn import NordVPN + from unshackle.core.proxies.proton import ProtonVPN from unshackle.core.proxies.surfsharkvpn import SurfsharkVPN proxy_config = getattr(main_config, "proxy_providers", {}) if proxy_config.get("basic"): proxy_providers.append(Basic(**proxy_config["basic"])) - if proxy_config.get("expressvpn"): - proxy_providers.append(ExpressVPN(**proxy_config["expressvpn"])) + # ExpressVPN/ProtonVPN auto-load when their default cookie file exists (no yaml needed) + expressvpn = ExpressVPN(**(proxy_config.get("expressvpn") or {})) + if proxy_config.get("expressvpn") or expressvpn.cookie_path.is_file(): + proxy_providers.append(expressvpn) if proxy_config.get("nordvpn"): proxy_providers.append(NordVPN(**proxy_config["nordvpn"])) + proton = ProtonVPN(**(proxy_config.get("protonvpn") or {})) + if proxy_config.get("protonvpn") or proton.cookie_path.is_file(): + proxy_providers.append(proton) if proxy_config.get("surfsharkvpn"): proxy_providers.append(SurfsharkVPN(**proxy_config["surfsharkvpn"])) if hasattr(binaries, "HolaProxy") and binaries.HolaProxy: