ddp_utils.ws_intercom.windscribe_proxy

import ddp_utils.ws_intercom.windscribe_proxy

Provide browser-neutral Windscribe proxy transport services.

The browser layer owns WebDriver creation. This module only turns Windscribe’s authenticated HTTPS proxy endpoints into a local SOCKS5 endpoint that any browser backend can consume.

Examples

Decode extension credentials and acquire a local SOCKS5 endpoint:

from ddp_utils.ws_intercom.windscribe_proxy import (
    WindscribeCredentials,
    WindscribeProxyManager,
)

credentials = WindscribeCredentials.from_extension_encoded(
    username="dXNlcg==",
    password="cGFzcw==",
)
manager = WindscribeProxyManager(credentials)
with manager.acquire(location) as lease:
    print(lease.proxy_url)

exception ddp_utils.ws_intercom.windscribe_proxy.ProxyAuthenticationError

Bases: WindscribeProxyError

Indicate that an upstream proxy rejected its credentials.

Examples

Stop host fallback when shared credentials are invalid:

raise ProxyAuthenticationError("proxy authentication failed")

exception ddp_utils.ws_intercom.windscribe_proxy.ProxyConnectionError

Bases: WindscribeProxyError

Indicate that a local or upstream proxy tunnel could not be established.

Examples

Surface exhaustion of all candidate hosts:

raise ProxyConnectionError("all proxy hosts failed")

class ddp_utils.ws_intercom.windscribe_proxy.ProxyVerification(host: str | None, ip: str | None, country_code: str | None, raw: str)

Bases: object

Describe a verified proxy egress identity.

Variables

Name	Type	Description
host	str | None	Upstream proxy host used for the check, when known.
ip	str | None	Normalized externally visible IP address.
country_code	str | None	Uppercase country code when returned and validated.
raw	str	Original verification response body, excluded from repr.

Examples

Inspect normalized egress data:

print(verification.ip, verification.country_code)

exception ddp_utils.ws_intercom.windscribe_proxy.ProxyVerificationError

Bases: WindscribeProxyError

Indicate invalid or unexpected proxy egress verification data.

Examples

Reject a response exposing the direct IP:

raise ProxyVerificationError("proxy returned the direct IP")

exception ddp_utils.ws_intercom.windscribe_proxy.WindscribeAdapterError(message: str, *, code: str = 'ADAPTER_ERROR', retryable: bool = False, details: Any = None)

Bases: WindscribeCatalogError

Describe a failed canonical DDP extension-adapter request.

Variables

Name	Description
code	Stable machine-readable failure code.
retryable	Whether repeating the operation may succeed.
details	Optional structured provider diagnostics.

Examples

Preserve an adapter timeout contract:

error = WindscribeAdapterError(
    "adapter timed out",
    code="ADAPTER_RESPONSE_TIMEOUT",
    retryable=True,
)

Initialize an adapter failure without depending on Selenium types.

Parameters

Name	Type	Description
message	str	Human-readable failure description.
code	str	Stable machine-readable code.
retryable	bool	Whether the caller may safely retry.
details	Any	Optional structured diagnostic payload.

Examples

Mark a malformed response as terminal:

error = WindscribeAdapterError(
    "missing result",
    code="INVALID_ADAPTER_RESPONSE",
)

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeAdapterLocationSource(driver: Any, *, request_timeout: float = 20.0, poll_interval: float = 0.25)

Bases: object

Fetch locations through the canonical DDP adapter on an open dashboard page.

The supplied object only needs Selenium’s execute_async_script method. This class never imports Selenium and never creates, navigates, or closes a browser. The current page must be one of the origins allowed by the extension content bridge, such as https://wsi.ddp.am.

Examples

Bind an already-open dashboard WebDriver:

source = WindscribeAdapterLocationSource(driver)

Bind the source to an existing WebDriver-compatible page.

Parameters

Name	Type	Description
driver	Any	Object providing execute_async_script.
request_timeout	float	Per-adapter-request timeout cap in seconds.
poll_interval	float	Delay between refresh-operation polls.

Raises

Exception	Description
TypeError	driver lacks a callable execute_async_script.
ValueError	Either timing value is not positive.

Examples

Use shorter polling in an integration test:

source = WindscribeAdapterLocationSource(
    driver,
    request_timeout=10,
    poll_interval=0.1,
)

fetch_server_list(*, refresh: bool = False, timeout: float = 90.0) → Sequence[Mapping[str, Any]]

Read locations.list or execute and await locations.refresh.

Parameters

Name	Type	Description
refresh	bool	Run locations.refresh and poll its operation when true.
timeout	float	Total operation budget in seconds.

Returns

Type	Description
Sequence[Mapping[str, Any]]	Detached native server-list mappings.

Raises

Exception	Description
ValueError	timeout is not positive.
WindscribeAdapterError	Transport, envelope, operation, or result validation fails.

Examples

Force the extension to refresh its native catalog:

server_list = source.fetch_server_list(refresh=True, timeout=90)

exception ddp_utils.ws_intercom.windscribe_proxy.WindscribeCatalogError

Bases: WindscribeProxyError

Indicate an unavailable, invalid, empty, or ambiguous server catalog.

Examples

Reject a catalog without usable locations:

raise WindscribeCatalogError("catalog contains no locations")

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeCredentials(username: str, password: str)

Bases: object

Store decoded credentials shared by Windscribe proxy hosts.

Variables

Name	Type	Description
username	str	Upstream proxy username, excluded from repr.
password	str	Upstream proxy password, excluded from repr.

Examples

Construct already-decoded credentials:

credentials = WindscribeCredentials("proxy-user", "proxy-pass")

classmethod from_extension_encoded(username: str, password: str) → WindscribeCredentials

Decode Base64 values stored in extension serverCredentials.

Parameters

Name	Type	Description
username	str	Base64-encoded UTF-8 username.
password	str	Base64-encoded UTF-8 password.

Returns

Type	Description
WindscribeCredentials	Validated decoded credentials.

Raises

Exception	Description
ValueError	A value is empty, invalid Base64/UTF-8, or decodes to an invalid credential.

Examples

Decode values returned by the extension:

credentials = WindscribeCredentials.from_extension_encoded(
    "dXNlcg==",
    "cGFzc3dvcmQ=",
)

classmethod from_extension_state(state: Mapping[str, Any]) → WindscribeCredentials

Extract and decode proxy credentials from extension state.

Parameters

Name	Type	Description
state	Mapping[str, Any]	Extension state containing a serverCredentials mapping with encoded username and password values.

Returns

Type	Description
WindscribeCredentials	Validated decoded credentials.

Raises

Exception	Description
ValueError	The credentials object is missing, malformed, or contains invalid encoded values.

Examples

Decode a captured state object:

credentials = WindscribeCredentials.from_extension_state(state)

basic_authorization() → str

Build the HTTP Basic authorization value for these credentials.

Returns

Type	Description
str	Basic <base64(username:password)> header value.

Examples

Add credentials to an upstream CONNECT request:

header = credentials.basic_authorization()

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeLocation(hosts: Tuple[str, ...], port: int = 443, data_center_id: int | None = None, location_id: int | None = None, country_code: str | None = None, city: str | None = None, nick: str | None = None, location_name: str | None = None, short_name: str | None = None)

Bases: object

Describe one Windscribe data center and its proxy hosts.

Variables

Name	Type	Description
hosts	Tuple[str, ...]	Unique normalized host candidates in provider order.
port	int	Upstream HTTPS proxy port.
data_center_id	int | None	Provider data-center identifier.
location_id	int | None	Parent location identifier.
country_code	str | None	Uppercase country code when available.
city	str | None	Provider city label.
nick	str | None	Provider data-center nickname.
location_name	str | None	Parent location display name.
short_name	str | None	Parent location short name.

Examples

Define a single-host location:

location = WindscribeLocation(("proxy.example",), country_code="US")

classmethod from_data_center(data_center: Mapping[str, Any], *, location_id: int | None = None, country_code: str | None = None, location_name: str | None = None, short_name: str | None = None, port: int = 443) → WindscribeLocation

Build a location from one Windscribe data-center mapping.

Parameters

Name	Type	Description
data_center	Mapping[str, Any]	Provider mapping containing a hosts sequence.
location_id	int | None	Optional parent location identifier.
country_code	str | None	Explicit country code or the data-center value.
location_name	str | None	Optional parent location display name.
short_name	str | None	Optional parent location short name.
port	int	Upstream proxy port.

Returns

Type	Description
WindscribeLocation	Normalized location.

Raises

Exception	Description
ValueError	hosts is not a non-string sequence or contains no usable host after normalization.

Examples

Parse a provider data-center group:

location = WindscribeLocation.from_data_center(data_center)

classmethod from_server_list(server_list: Sequence[Mapping[str, Any]], *, port: int = 443) → Tuple[WindscribeLocation, ...]

Flatten the extension’s location/group hierarchy into proxy locations.

Invalid or hostless groups are ignored so one unavailable data center does not make the complete server catalog unusable. At least one valid data center is required.

Parameters

Name	Type	Description
server_list	Sequence[Mapping[str, Any]]	Native extension location objects.
port	int	Default port when a data center omits one.

Returns

Type	Description
Tuple[WindscribeLocation, …]	Usable normalized locations in provider order.

Raises

Exception	Description
WindscribeCatalogError	Input is not an array or contains no usable data center.

Examples

Normalize a manager-provided server list:

locations = WindscribeLocation.from_server_list(server_list)

ordered_hosts(*, shuffle: bool = False, rng: Random | None = None, excluded_hosts: Iterable[str] = ()) → Tuple[str, ...]

Return eligible hosts in connection-attempt order.

Parameters

Name	Type	Description
shuffle	bool	Randomize remaining candidates when True.
rng	Random | None	Optional deterministic random source; otherwise random.SystemRandom is used.
excluded_hosts	Iterable[str]	Hosts to normalize and remove before ordering.

Returns

Type	Description
Tuple[str, …]	Tuple of non-excluded hosts in selected order.

Examples

Skip a host that already failed:

hosts = location.ordered_hosts(excluded_hosts=[failed_host])

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeLocationSource(*args, **kwargs)

Bases: Protocol

Define a transport that supplies native Windscribe server lists.

Examples

Accept any compatible source in application code:

def refresh(source: WindscribeLocationSource):
    return source.fetch_server_list(refresh=True)

fetch_server_list(*, refresh: bool = False, timeout: float = 90.0) → Sequence[Mapping[str, Any]]

Read the current list or force Windscribe’s native server refresh.

Parameters

Name	Type	Description
refresh	bool	Request a native refresh before returning data.
timeout	float	Total operation budget in seconds.

Returns

Type	Description
Sequence[Mapping[str, Any]]	Native location mappings.

Examples

Read the currently available list:

server_list = source.fetch_server_list()

exception ddp_utils.ws_intercom.windscribe_proxy.WindscribeProxyError

Bases: RuntimeError

Base exception for Windscribe proxy transport failures.

Examples

Catch every transport-specific failure at one boundary:

try:
    lease = manager.acquire(location)
except WindscribeProxyError as error:
    report(error)

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeProxyLease(bridge: WindscribeSocksBridge, location: WindscribeLocation, selected_host: str, attempted_hosts: Tuple[str, ...] = ())

Bases: object

Own one acquired local endpoint and its deterministic cleanup.

Variables

Name	Type	Description
bridge	WindscribeSocksBridge	Running local SOCKS bridge.
location	WindscribeLocation	Selected Windscribe location.
selected_host	str	Upstream host that succeeded.
attempted_hosts	Tuple[str, ...]	Hosts attempted through successful acquisition.

Examples

Release the bridge at block exit:

with manager.acquire(location) as lease:
    create_browser(proxy=lease.proxy_url)

property proxy_url: str

Return this lease’s local SOCKS5 URL.

Returns

URL delegated to the running bridge.

Examples

Pass the acquired endpoint to browser configuration:

configure_browser_proxy(lease.proxy_url)

property address: Tuple[str, int]

Return this lease’s local SOCKS5 address.

Returns

(host, port) delegated to the running bridge.

Examples

Inspect the allocated port:

host, port = lease.address

close() → None

Stop the underlying bridge and release this lease idempotently.

Examples

Release an unused acquisition:

lease.close()

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeProxyManager(credentials: WindscribeCredentials, *, listen_host: str = '127.0.0.1', connect_timeout: float = 15.0, idle_timeout: float = 120.0, ssl_context: ssl.SSLContext | None = None, probe_host: str | None = 'checkip.windscribe.com', probe_port: int = 443, location_source: WindscribeLocationSource | None = None, location_selector: 'WindscribeProjectLocationSelector' | None = None, catalog_cache_path: str | os.PathLike[str] | None = None, catalog_ttl: float = 3600.0)

Bases: object

Acquire local Windscribe proxy leases without creating a browser.

Examples

Resolve and acquire one configured location:

manager = WindscribeProxyManager(credentials, location_source=source)
with manager.acquire(manager.get_location(city="Chicago")) as lease:
    create_browser(proxy=lease.proxy_url)

Configure acquisition, catalog, cache, and selection services.

Parameters

Name	Type	Description
credentials	WindscribeCredentials	Upstream credentials shared by created bridges.
listen_host	str	Local interface used by each SOCKS5 endpoint.
connect_timeout	float	Upstream connection timeout in seconds.
idle_timeout	float	Tunnel idle timeout in seconds.
ssl_context	Optional[ssl.SSLContext]	Optional shared TLS context.
probe_host	Optional[str]	Destination used to validate CONNECT, or None to disable the default probe target.
probe_port	int	Probe destination port.
location_source	Optional[WindscribeLocationSource]	Live native server-list provider.
location_selector	Optional['WindscribeProjectLocationSelector']	Optional project history and selection policy.
catalog_cache_path	Optional[Union[str, os.PathLike[str]]]	Optional persistent JSON catalog path.
catalog_ttl	float	Default maximum catalog age in seconds.

Raises

Exception	Description
ValueError	catalog_ttl is negative.

Examples

Enable a persistent one-hour catalog cache:

manager = WindscribeProxyManager(
    credentials,
    location_source=source,
    catalog_cache_path=".runtime/windscribe-catalog.json",
    catalog_ttl=3_600,
)

property last_cache_error: WindscribeCatalogError | None

Return the latest non-fatal persistent-cache error.

Returns

Most recent cache error, or None after successful cache work or explicit cache clearing.

Examples

Surface degraded caching without failing live acquisition:

if manager.last_cache_error:
    warn(manager.last_cache_error)

set_location_source(source: WindscribeLocationSource | None) → None

Attach or detach the live source used for catalog refreshes.

Parameters

Name	Type	Description
source	WindscribeLocationSource | None	Compatible source, or None for cache-only operation.

Examples

Replace a browser adapter source at runtime:

manager.set_location_source(new_source)

clear_location_cache(*, persistent: bool = False) → None

Clear memory state and optionally remove the persistent catalog.

Parameters

Name	Type	Description
persistent	bool	Remove the configured cache file when true.

Raises

Exception	Description
WindscribeCatalogError	Persistent cache removal fails.

Examples

Force the next lookup to use the live source:

manager.clear_location_cache(persistent=True)

fetch_catalog(*, refresh: bool = False, timeout: float = 90.0, max_age: float | None = None, allow_stale: bool = False) → WindscribeServerCatalog

Return a cached catalog or fetch it through the configured location source.

refresh=True invokes Windscribe’s native locations.refresh flow. Without refresh, a fresh memory/disk snapshot is preferred and a missing snapshot is read using locations.list. allow_stale is an explicit offline fallback when the source is unavailable.

Parameters

Name	Type	Description
refresh	bool	Force the native provider refresh workflow.
timeout	float	Live-source operation budget in seconds.
max_age	float | None	Accepted cache age, or configured TTL when omitted.
allow_stale	bool	Return an available stale snapshot if live fetch fails.

Returns

Type	Description
WindscribeServerCatalog	Fresh, refreshed, or explicitly accepted stale catalog.

Raises

Exception	Description
ValueError	timeout is not positive or effective age is negative.
WindscribeCatalogError	No usable cache/source exists or fetching and parsing fail without stale fallback.

Examples

Allow offline use of the most recent snapshot:

catalog = manager.fetch_catalog(allow_stale=True)

fetch_server_list(*, refresh: bool = False, timeout: float = 90.0, max_age: float | None = None, allow_stale: bool = False) → List[Dict[str, Any]]

Return a detached native server list from the selected catalog.

Parameters

Name	Type	Description
refresh	bool	Force native refresh.
timeout	float	Live-source operation budget in seconds.
max_age	float | None	Accepted cache age override.
allow_stale	bool	Permit stale fallback.

Returns

Type	Description
List[Dict[str, Any]]	Mutable JSON-compatible location objects.

Examples

Persist native manager data:

server_list = manager.fetch_server_list(allow_stale=True)

fetch_locations(*, refresh: bool = False, timeout: float = 90.0, max_age: float | None = None, allow_stale: bool = False) → Tuple[WindscribeLocation, ...]

Return normalized locations ready for selection or acquisition.

Parameters

Name	Type	Description
refresh	bool	Force native refresh.
timeout	float	Live-source operation budget in seconds.
max_age	float | None	Accepted cache age override.
allow_stale	bool	Permit stale fallback.

Returns

Type	Description
Tuple[WindscribeLocation, …]	Catalog location tuple.

Examples

Feed locations to a project selector:

locations = manager.fetch_locations()

find_locations(*, refresh: bool = False, timeout: float = 90.0, max_age: float | None = None, allow_stale: bool = False, **filters: Any) → Tuple[WindscribeLocation, ...]

Fetch a catalog and return locations matching exact filters.

Parameters

Name	Type	Description
refresh	bool	Force native refresh.
timeout	float	Live-source operation budget in seconds.
max_age	float | None	Accepted cache age override.
allow_stale	bool	Permit stale fallback.
**filters	Any	Filters accepted by WindscribeServerCatalog.find_locations().

Returns

Type	Description
Tuple[WindscribeLocation, …]	Matching normalized locations.

Examples

Find eligible Atlanta data centers:

locations = manager.find_locations(city="Atlanta")

get_location(*, refresh: bool = False, timeout: float = 90.0, max_age: float | None = None, allow_stale: bool = False, **filters: Any) → WindscribeLocation

Fetch a catalog and return one unambiguous matching location.

Parameters

Name	Type	Description
refresh	bool	Force native refresh.
timeout	float	Live-source operation budget in seconds.
max_age	float | None	Accepted cache age override.
allow_stale	bool	Permit stale fallback.
**filters	Any	Filters accepted by WindscribeServerCatalog.get_location().

Returns

Type	Description
WindscribeLocation	Sole matching normalized location.

Raises

Exception	Description
WindscribeCatalogError	No match exists or filters are ambiguous.

Examples

Resolve one catalog entry:

location = manager.get_location(data_center_id=42)

select_random_location(*, pool: str | int | WindscribeLocation | Iterable[str | int | WindscribeLocation] | None = None, blacklist: str | int | WindscribeLocation | Iterable[str | int | WindscribeLocation] | None = None, exclude_recent: int | None = None, rng: Random | None = None, refresh: bool = False, timeout: float = 90.0, max_age: float | None = None, allow_stale: bool = False) → WindscribeLocation

Select a random eligible catalog location using the project policy.

The location is not added to history until acquire succeeds. This prevents failed proxy attempts from consuming entries in the recent connection window.

Parameters

Name	Type	Description
pool	str | int | WindscribeLocation | Iterable[str | int | WindscribeLocation] | None	Optional selector allowlist.
blacklist	str | int | WindscribeLocation | Iterable[str | int | WindscribeLocation] | None	Optional selector denylist.
exclude_recent	int | None	Recent-history window override.
rng	Random | None	Optional deterministic random source.
refresh	bool	Force native refresh.
timeout	float	Live-source operation budget in seconds.
max_age	float | None	Accepted cache age override.
allow_stale	bool	Permit stale fallback.

Returns

Type	Description
WindscribeLocation	Random eligible location selected by project policy.

Raises

Exception	Description
WindscribeCatalogError	No project selector is configured or its policy cannot produce an eligible location.

Examples

Select only from an explicit city pool:

location = manager.select_random_location(pool=["Chicago", "Dallas"])

acquire(location: WindscribeLocation, *, shuffle_hosts: bool = False, rng: Random | None = None, excluded_hosts: Iterable[str] = (), probe: bool = True) → WindscribeProxyLease

Return a local proxy lease for the first working location host.

Each candidate receives a new bridge. Failed bridges are stopped before the next host. Selection history is updated only after successful probe and lease construction.

Parameters

Name	Type	Description
location	WindscribeLocation	Location whose host candidates are attempted.
shuffle_hosts	bool	Randomize eligible candidate order.
rng	Random | None	Optional deterministic random source.
excluded_hosts	Iterable[str]	Hosts to omit from this acquisition.
probe	bool	Validate CONNECT against configured probe target before accepting a host.

Returns

Type	Description
WindscribeProxyLease	Lease owning the first successful local bridge.

Raises

Exception	Description
ProxyAuthenticationError	Shared credentials are rejected; host fallback stops immediately.
ProxyConnectionError	No host remains or every host fails.

Examples

Acquire while skipping a previously blocked host:

lease = manager.acquire(location, excluded_hosts=blocked_hosts)

open(location: WindscribeLocation, *, shuffle_hosts: bool = False, rng: Random | None = None, excluded_hosts: Iterable[str] = (), probe: bool = True) → WindscribeProxyLease

Return a local proxy lease for the first working location host.

Each candidate receives a new bridge. Failed bridges are stopped before the next host. Selection history is updated only after successful probe and lease construction.

Parameters

Name	Type	Description
location	WindscribeLocation	Location whose host candidates are attempted.
shuffle_hosts	bool	Randomize eligible candidate order.
rng	Random | None	Optional deterministic random source.
excluded_hosts	Iterable[str]	Hosts to omit from this acquisition.
probe	bool	Validate CONNECT against configured probe target before accepting a host.

Returns

Type	Description
WindscribeProxyLease	Lease owning the first successful local bridge.

Raises

Exception	Description
ProxyAuthenticationError	Shared credentials are rejected; host fallback stops immediately.
ProxyConnectionError	No host remains or every host fails.

Examples

Acquire while skipping a previously blocked host:

lease = manager.acquire(location, excluded_hosts=blocked_hosts)

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeServerCatalog(server_list: Tuple[Mapping[str, Any], ...], locations: Tuple[WindscribeLocation, ...], fetched_at: float, source: str = 'adapter')

Bases: object

Hold a validated native snapshot and normalized proxy locations.

Variables

Name	Type	Description
server_list	Tuple[Mapping[str, Any], ...]	Detached native provider objects.
locations	Tuple[WindscribeLocation, ...]	Parsed locations ready for selection and acquisition.
fetched_at	float	Snapshot acquisition time as a Unix timestamp.
source	str	Diagnostic source label such as adapter or cache.

Examples

Build a catalog from manager data:

catalog = WindscribeServerCatalog.from_server_list(server_list)

classmethod from_server_list(server_list: Sequence[Mapping[str, Any]], *, fetched_at: float | None = None, source: str = 'adapter', port: int = 443) → WindscribeServerCatalog

Validate native server data and build a detached catalog.

Parameters

Name	Type	Description
server_list	Sequence[Mapping[str, Any]]	JSON-compatible native location mappings.
fetched_at	float | None	Explicit snapshot time or current time when omitted.
source	str	Diagnostic source label.
port	int	Default proxy port for parsed data centers.

Returns

Type	Description
WindscribeServerCatalog	Validated catalog with normalized locations.

Raises

Exception	Description
WindscribeCatalogError	Input cannot form a usable catalog.

Examples

Preserve an externally supplied fetch time:

catalog = WindscribeServerCatalog.from_server_list(
    server_list,
    fetched_at=received_at,
    source="manager",
)

is_fresh(max_age: float, *, now: float | None = None) → bool

Return whether snapshot age is within a maximum number of seconds.

Parameters

Name	Type	Description
max_age	float	Non-negative accepted age.
now	float | None	Comparison timestamp, or current time when omitted.

Returns

Type	Description
bool	True when effective age does not exceed max_age.

Raises

Exception	Description
ValueError	max_age is negative.

Examples

Accept a catalog cached for one hour:

usable = catalog.is_fresh(3_600)

to_server_list() → List[Dict[str, Any]]

Return a detached JSON-compatible native server-list copy.

Returns

Type	Description
List[Dict[str, Any]]	Mutable list whose changes cannot mutate this catalog.

Examples

Hand native data to a cache writer:

payload["server_list"] = catalog.to_server_list()

find_locations(*, location_id: int | None = None, data_center_id: int | None = None, country_code: str | None = None, city: str | None = None, nick: str | None = None, location_name: str | None = None, short_name: str | None = None) → Tuple[WindscribeLocation, ...]

Return locations matching every supplied exact filter.

Parameters

Name	Type	Description
location_id	int | None	Parent location identifier.
data_center_id	int | None	Data-center identifier.
country_code	str | None	Country code normalized to uppercase.
city	str | None	Case-insensitive exact city.
nick	str | None	Case-insensitive exact data-center nickname.
location_name	str | None	Case-insensitive exact parent display name.
short_name	str | None	Case-insensitive exact parent short name.

Returns

Type	Description
Tuple[WindscribeLocation, …]	Matching locations in catalog order.

Examples

Select every Chicago data center:

matches = catalog.find_locations(city="Chicago")

get_location(**filters: Any) → WindscribeLocation

Return exactly one location matching the supplied filters.

Parameters

Name	Type	Description
**filters	Any	Keyword filters accepted by find_locations().

Returns

Type	Description
WindscribeLocation	The sole matching location.

Raises

Exception	Description
WindscribeCatalogError	No location matches or filters are ambiguous.

Examples

Resolve one data center by identifier:

location = catalog.get_location(data_center_id=42)

class ddp_utils.ws_intercom.windscribe_proxy.WindscribeSocksBridge(upstream_host: str, upstream_port: int, credentials: WindscribeCredentials, *, listen_host: str = '127.0.0.1', listen_port: int = 0, connect_timeout: float = 15.0, idle_timeout: float = 120.0, ssl_context: SSLContext | None = None)

Bases: object

Expose local SOCKS5 through one authenticated Windscribe HTTPS proxy.

Examples

Start and deterministically stop a local endpoint:

with WindscribeSocksBridge(host, 443, credentials) as bridge:
    configure_browser_proxy(bridge.proxy_url)

Configure a stopped local bridge.

Parameters

Name	Type	Description
upstream_host	str	Windscribe HTTPS proxy hostname.
upstream_port	int	Upstream proxy port.
credentials	WindscribeCredentials	Shared upstream authentication credentials.
listen_host	str	Local interface for the SOCKS5 server.
listen_port	int	Local port, or zero to allocate a free port.
connect_timeout	float	Socket and upstream setup timeout in seconds.
idle_timeout	float	Bidirectional tunnel idle timeout in seconds.
ssl_context	Optional[ssl.SSLContext]	TLS context, or a default verified context when omitted.

Examples

Reserve an automatically selected loopback port:

bridge = WindscribeSocksBridge(host, 443, credentials)

property address: Tuple[str, int]

Return the active local SOCKS5 listening address.

Returns

(host, port) pair assigned to the running server.

Raises

Exception	Description
RuntimeError	The bridge has not been started or has been stopped.

Examples

Pass a structured endpoint to a client:

host, port = bridge.address

property proxy_url: str

Return the active endpoint as a socks5://host:port URL.

Returns

Browser-compatible local proxy URL.

Raises

Exception	Description
RuntimeError	The bridge is not running.

Examples

Configure a browser proxy:

browser_proxy = bridge.proxy_url

property last_error: Exception | None

Return the latest client-handler exception, if one was recorded.

Returns

Last exception under the bridge error lock, or None.

Examples

Inspect a failed background connection:

if bridge.last_error is not None:
    report(bridge.last_error)

start() → WindscribeSocksBridge

Start the local SOCKS5 server in a daemon thread idempotently.

Returns

Type	Description
WindscribeSocksBridge	This bridge with an available address.

Examples

Start and read the allocated port:

bridge.start()
print(bridge.address)

stop() → None

Stop the server and close all tracked sockets idempotently.

Examples

Release the endpoint explicitly:

bridge.stop()

close() → None

Stop the server and close all tracked sockets idempotently.

Examples

Release the endpoint explicitly:

bridge.stop()

probe(target_host: str, target_port: int = 443) → None

Validate TLS, authentication, and CONNECT without a browser.

Parameters

Name	Type	Description
target_host	str	Destination used for the CONNECT probe.
target_port	int	Destination port.

Raises

Exception	Description
ProxyAuthenticationError	Upstream credentials are rejected.
ProxyConnectionError	The tunnel cannot be established.
OSError	Socket or TLS setup fails before protocol normalization.

Examples

Verify a bridge before creating a browser:

bridge.probe("checkip.windscribe.com", 443)

ddp_utils.ws_intercom.windscribe_proxy.normalize_verification(body: str, *, host: str | None = None, expected_country_code: str | None = None, direct_ip: str | None = None) → ProxyVerification

Validate and normalize a proxy egress response.

Parameters

Name	Type	Description
body	str	Raw body from an IP or geolocation endpoint.
host	str | None	Optional Windscribe host associated with the check.
expected_country_code	str | None	Optional two-letter country code that the response must report.
direct_ip	str | None	Optional non-proxied IP address. Matching it means that the proxy did not change the visible egress address.

Returns

Type	Description
ProxyVerification	A normalized ProxyVerification record containing the parsed IP, country code, source host, and original response.

Raises

Exception	Description
ProxyVerificationError	The response has no valid IP, reports the direct IP, or does not match the expected country.
ValueError	direct_ip is supplied but is not a valid IP address.

Examples

Require a US proxy address different from the direct address:

result = normalize_verification(
    '{"ip": "203.0.113.8", "country_code": "US"}',
    expected_country_code="US",
    direct_ip="198.51.100.4",
)

ddp_utils.ws_intercom.windscribe_proxy.parse_verification_response(body: str) → Tuple[str, str | None]

Extract an IP address and optional country code from an endpoint response.

The response may be a plain-text IP address or a JSON object using common IP and country-code field names. Extra whitespace in a plain-text response is ignored after its first token.

Parameters

Name	Type	Description
body	str	Raw response body returned by an IP or geolocation endpoint.

Returns

Type	Description
Tuple[str, str | None]	A pair containing the reported IP string and an uppercase-independent country-code value when the endpoint supplied one.

Raises

Exception	Description
ProxyVerificationError	The response is empty, is a non-object JSON value, or contains no recognized IP field.

Examples

Parse a JSON geolocation response:

ip, country = parse_verification_response(
    '{"ip": "203.0.113.8", "country_code": "US"}'
)