96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, urlsplit, urlunsplit
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClientVersion:
|
|
major: int = 0
|
|
minor: int = 1
|
|
|
|
def user_agent(self, device_id: str) -> str:
|
|
return f"vpnapp/v{self.major}.{self.minor}/{device_id}"
|
|
|
|
|
|
class SubscriptionError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class SubscriptionClient:
|
|
"""Small device-side client for the backend contract.
|
|
|
|
Storage is injected deliberately: Android should provide a Keystore-backed
|
|
implementation through the native bridge, while desktop can use an OS
|
|
credential store. The client never logs the token or decrypted payload.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
load_secret: Callable[[str], str | None],
|
|
save_secret: Callable[[str, str], None],
|
|
version: ClientVersion = ClientVersion(),
|
|
timeout: float = 15.0,
|
|
) -> None:
|
|
self._load_secret = load_secret
|
|
self._save_secret = save_secret
|
|
self.version = version
|
|
self.timeout = timeout
|
|
|
|
def enroll_url(self, url: str) -> dict[str, str]:
|
|
parsed = urlsplit(url)
|
|
query = parse_qs(parsed.query)
|
|
token = query.get("enroll", [""])[0]
|
|
parts = [part for part in parsed.path.split("/") if part]
|
|
if len(parts) != 2 or parts[0] != "d" or not parts[1] or not token:
|
|
raise SubscriptionError("enrollment URL must look like /d/{device_id}?enroll=...")
|
|
device_id = parts[1]
|
|
subscription_url = urlunsplit((parsed.scheme, parsed.netloc, f"/d/{device_id}", "", ""))
|
|
self._save_secret("device_id", device_id)
|
|
self._save_secret("device_token", token)
|
|
self._save_secret("subscription_url", subscription_url)
|
|
return {"device_id": device_id, "subscription_url": subscription_url}
|
|
|
|
async def refresh(self) -> dict[str, Any]:
|
|
url = self._load_secret("subscription_url")
|
|
device_id = self._load_secret("device_id")
|
|
token = self._load_secret("device_token")
|
|
if not url or not device_id or not token:
|
|
raise SubscriptionError("device is not enrolled")
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"User-Agent": self.version.user_agent(device_id),
|
|
"X-Device-Token": token,
|
|
}
|
|
try:
|
|
return await asyncio.to_thread(self._request, url, headers)
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode("utf-8", errors="replace")[:200]
|
|
raise SubscriptionError(f"subscription request failed ({exc.code}): {detail}") from exc
|
|
except (urllib.error.URLError, TimeoutError) as exc:
|
|
raise SubscriptionError("subscription server is unavailable") from exc
|
|
|
|
def apply(self, payload: dict[str, Any], controller: Any) -> None:
|
|
"""Map the server payload into the split local profile models."""
|
|
controller.signbox.config["vless_subscription_url"] = payload.get(
|
|
"vless_subscription_url", ""
|
|
)
|
|
controller.enhanced.config["amnezia_account_name"] = payload.get(
|
|
"amnezia_account_name", ""
|
|
)
|
|
controller.enhanced.config["amnezia_servers"] = payload.get("amnezia", {})
|
|
controller.freeturn.key = str(payload.get("freeturn_key", ""))
|
|
|
|
def _request(self, url: str, headers: dict[str, str]) -> dict[str, Any]:
|
|
request = urllib.request.Request(url, headers=headers, method="GET")
|
|
with urllib.request.urlopen(request, timeout=self.timeout) as response:
|
|
value = json.loads(response.read().decode("utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise SubscriptionError("subscription response is not an object")
|
|
return value
|