114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
import requests
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Final
|
|
|
|
|
|
@dataclass
|
|
class Host:
|
|
"""Proxy host information."""
|
|
|
|
ip: str
|
|
port: str
|
|
signature: str
|
|
|
|
|
|
DELAY_PER_REQUEST: Final[float] = 1
|
|
"""Delay in seconds between host authentication requests."""
|
|
|
|
GET_ACCESS_TOKEN_URL: Final[str] = (
|
|
"https://api-pro.ducunt.com/rest/v1/security/tokens/accs"
|
|
)
|
|
"""URL to get proxy service authentication token for further API calls."""
|
|
|
|
GET_COUNTRIES_URL: Final[str] = (
|
|
"https://stats.ducunt.com/api/rest/v2/entrypoints/countries"
|
|
)
|
|
"""URL API endpoint to get countries."""
|
|
|
|
GET_ACCESS_PROXY_URL: Final[str] = (
|
|
"https://api-pro.ducunt.com/rest/v1/security/tokens/accs-proxy"
|
|
)
|
|
"""URL API to get proxies."""
|
|
|
|
# Get authentication token from proxy service
|
|
print("Getting authentication token from proxy service")
|
|
|
|
get_access_token_payload: dict[str, str | dict[str, str]] = {
|
|
"clientApp": {"name": "URBAN_VPN_BROWSER_EXTENSION"},
|
|
"type": "accs",
|
|
}
|
|
|
|
with requests.post(
|
|
GET_ACCESS_TOKEN_URL,
|
|
json=get_access_token_payload,
|
|
headers={
|
|
"Authorization": "Bearer On9GLm0c8B5GFWI9FYtdfPWlXfjhpF0Q",
|
|
},
|
|
) as reponse:
|
|
authentication_response: dict[str, Any] = reponse.json()
|
|
|
|
authentication_token: str = authentication_response["value"]
|
|
print("Authentication token recieved")
|
|
|
|
# Get region list from proxy service
|
|
|
|
print("Getting available regions from proxy service")
|
|
with requests.get(
|
|
GET_COUNTRIES_URL,
|
|
headers={
|
|
"X-Client-App": "URBAN_VPN_BROWSER_EXTENSION",
|
|
"Authorization": f"Bearer {authentication_token}",
|
|
},
|
|
) as reponse:
|
|
regions_response: dict[str, Any] = reponse.json()
|
|
|
|
regions: list[Any] = [
|
|
country["servers"]["elements"]
|
|
for country in regions_response["countries"]["elements"]
|
|
if country["servers"]["count"] > 0
|
|
]
|
|
|
|
hosts: list[Host] = [
|
|
Host(
|
|
ip=host["address"]["primary"]["ip"],
|
|
port=host["address"]["primary"]["port"],
|
|
signature=host["signature"] if "signature" in host else "",
|
|
)
|
|
for region in regions
|
|
for host in region
|
|
if host["type"] == "PROXY" and host["accessType"] == "ACCESSIBLE"
|
|
]
|
|
|
|
print("Successfully acquired regions and available hosts information")
|
|
print(f"Preparing to query authentication information for {len(hosts)} hosts")
|
|
|
|
with Path("proxies").open("w", encoding="utf-8") as proxy_file:
|
|
for i, host in enumerate(hosts):
|
|
time.sleep(DELAY_PER_REQUEST)
|
|
|
|
get_access_proxy_token_payload: dict[str, str | dict[str, str]] = {
|
|
"clientApp": {"name": "URBAN_VPN_BROWSER_EXTENSION"},
|
|
"signature": host.signature,
|
|
"type": "accs",
|
|
}
|
|
|
|
with requests.post(
|
|
GET_ACCESS_PROXY_URL,
|
|
json=get_access_proxy_token_payload,
|
|
headers={
|
|
"Authorization": f"Bearer {authentication_token}",
|
|
},
|
|
) as reponse:
|
|
host_information_response: Any = reponse.json()
|
|
|
|
proxy_connection_string: str = (
|
|
f"http://{host_information_response['value']}:1@{host.ip}:{host.port}\n"
|
|
)
|
|
|
|
proxy_file.write(proxy_connection_string)
|
|
proxy_file.flush()
|
|
|
|
print(f"Acquired information for host {i}")
|