Skip to content

HTTP client API

client

Call Gluetun and fetch PIA endpoints without exposing connection secrets.

Control requests carry the shared API key; health probes and catalog downloads do not. Every request has a total deadline and a response-size limit so a slow or malformed server cannot hold the recovery loop indefinitely.

APIUnavailable 

APIUnavailable(
    reason: str = "request failed",
    status: int | None = None,
)

Bases: Exception

Keep retry semantics while carrying a safe failure category and HTTP status.

Parameters:

Name Type Description Default
reason str

Internal category; unrecognized values are never included in diagnostics.

 'request failed'
status int | None

HTTP response status when the server rejected the request.

 None
Source code in docker/privateerr/client.py 
39
40
41
42
def __init__(self, reason: str = "request failed", status: int | None = None) -> None:
    super().__init__("API request failed")
    self.reason = reason
    self.status = status

diagnostic 

diagnostic() -> str

Report only approved categories, never URLs, headers, bodies, or raw exceptions.

Source code in docker/privateerr/client.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def diagnostic(self) -> str:
    """Report only approved categories, never URLs, headers, bodies, or raw exceptions."""

    allowed = {
        "request failed",
        "HTTP response rejected",
        "request timed out",
        "connection failed",
        "invalid response",
        "response too large",
    }
    reason = self.reason if self.reason in allowed else "request failed"
    suffix = f" (HTTP {self.status})" if self.status is not None else ""
    return reason + suffix

NoRedirects

Bases: HTTPRedirectHandler

Keep authentication headers on the configured server.

redirect_request 

redirect_request(
    req: Request,
    fp: IO[bytes],
    code: int,
    msg: str,
    headers: Message,
    newurl: str,
) -> None

Reject redirects rather than forwarding a control API key to another destination.

Source code in docker/privateerr/client.py 
63
64
65
66
67
def redirect_request(
    self, req: Request, fp: IO[bytes], code: int, msg: str, headers: Message, newurl: str
) -> None:
    """Reject redirects rather than forwarding a control API key to another destination."""
    return None

Client 

Client(config: Config)

Use Gluetun's existing API and health listener; expose no additional server.

Source code in docker/privateerr/client.py 
93
94
95
96
97
def __init__(self, config: Config):
    self.config = config

    # Contact configured servers directly, ignoring inherited proxy environment variables.
    self.opener = build_opener(ProxyHandler({}), NoRedirects())

request 

request(
    url: str,
    *,
    timeout: float,
    method: str = "GET",
    body: Mapping[str, object] | None = None,
    key: str = "",
) -> bytes

Send one bounded request and return its body without logging request contents.

Parameters:

Name Type Description Default
url str

Complete destination; redirects and inherited proxies are not followed.

 required
timeout float

Maximum total seconds for connection setup and response reads.

 required
method str

HTTP method used for this request.

 'GET'
body Mapping[str, object] | None

Optional object to encode as JSON for a settings update.

 None
key str

Optional API credential; omit it for health and catalog requests.

 ''

Returns:

Type Description
bytes

The response body, limited to four MiB.

Raises:

Type Description
APIUnavailable

If transport, status, size, or deadline checks fail.
Source code in docker/privateerr/client.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def request(
    self,
    url: str,
    *,
    timeout: float,
    method: str = "GET",
    body: Mapping[str, object] | None = None,
    key: str = "",
) -> bytes:
    """Send one bounded request and return its body without logging request contents.

    Args:
        url: Complete destination; redirects and inherited proxies are not followed.
        timeout: Maximum total seconds for connection setup and response reads.
        method: HTTP method used for this request.
        body: Optional object to encode as JSON for a settings update.
        key: Optional API credential; omit it for health and catalog requests.

    Returns:
        The response body, limited to four MiB.

    Raises:
        APIUnavailable: If transport, status, size, or deadline checks fail.
    """
    headers: dict[str, str] = {}

    # Callers must explicitly supply authentication; health and catalog calls omit it.
    if key:
        headers["X-API-Key"] = key

    data = None

    # Serialize settings only for calls that submit a body.
    if body is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(body).encode()

    request = Request(url, data=data, headers=headers, method=method)

    # The socket timeout bounds individual waits; the outer deadline bounds the whole call.
    try:
        with deadline(timeout), self.opener.open(request, timeout=5) as response:
            if response.status != 200:
                raise APIUnavailable("HTTP response rejected", response.status)

            # Bound retained data even if a server sends an unexpectedly large response.
            payload = response.read(4 * 1024 * 1024 + 1)

            if len(payload) > 4 * 1024 * 1024:
                raise APIUnavailable("response too large")

            return payload
    except HTTPError as error:
        # Release the error response without including its potentially sensitive body.
        error.close()
        raise APIUnavailable("HTTP response rejected", error.code) from None
    except TimeoutError:
        raise APIUnavailable("request timed out") from None
    except URLError as error:
        # urllib wraps connection-stage timeouts, unlike timeouts while reading a response.
        reason = (
            "request timed out"
            if isinstance(error.reason, TimeoutError)
            else "connection failed"
        )
        raise APIUnavailable(reason) from None
    except OSError:
        raise APIUnavailable("connection failed") from None
    except HTTPException, ValueError:
        raise APIUnavailable("invalid response") from None

get 

get(route: str) -> dict[str, object]

Read a control API object, rejecting responses the supervisor cannot interpret.

Source code in docker/privateerr/client.py 
170
171
172
173
174
175
176
177
178
179
def get(self, route: str) -> dict[str, object]:
    """Read a control API object, rejecting responses the supervisor cannot interpret."""
    try:
        result = json.loads(
            self.request(self.config.api_url + route, timeout=30, key=self.config.api_key)
        )

        return object_fields(result)
    except (ValueError, UnicodeError) as error:
        raise APIUnavailable("invalid response") from error

apply 

apply(settings: Mapping[str, object]) -> None

Submit connection fields; the supervisor separately verifies settings and health.

Parameters:

Name Type Description Default
settings Mapping[str, object]

Minimal connection update; unrelated Gluetun settings are omitted.

 required

Raises:

Type Description
APIUnavailable

If the response cannot confirm the request. The update may still have succeeded, so callers must reconcile it before retrying.
Source code in docker/privateerr/client.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def apply(self, settings: Mapping[str, object]) -> None:
    """Submit connection fields; the supervisor separately verifies settings and health.

    Args:
        settings: Minimal connection update; unrelated Gluetun settings are omitted.

    Raises:
        APIUnavailable: If the response cannot confirm the request. The update may
            still have succeeded, so callers must reconcile it before retrying.
    """
    self.request(
        self.config.api_url + "/v1/vpn/settings",
        timeout=30,
        method="PUT",
        body=settings,
        key=self.config.api_key,
    )

healthy 

healthy() -> bool

Probe the separate health listener without sending the control API key.

Source code in docker/privateerr/client.py 
199
200
201
202
203
204
205
def healthy(self) -> bool:
    """Probe the separate health listener without sending the control API key."""
    try:
        self.request(self.config.health_url, timeout=10)
        return True
    except APIUnavailable:
        return False

catalog 

catalog(*, timeout: float = 20) -> dict[str, object]

Read PIA's advertised regions and WireGuard servers without API credentials.

Parameters:

Name Type Description Default
timeout float

Maximum total request seconds; generation can supply its remaining budget instead of starting another full request window.

 20

Returns:

Type Description
dict[str, object]

Catalog object with a regions list. Nested fields remain untrusted and must be validated by the endpoint selector or metadata generator.

Raises:

Type Description
APIUnavailable

If retrieval fails or the first response line is not a JSON object containing a regions list.
Note

Only the first line is parsed; trailing catalog signature data is not verified by this method. Transport uses the configured catalog URL.

Source code in docker/privateerr/client.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def catalog(self, *, timeout: float = 20) -> dict[str, object]:
    """Read PIA's advertised regions and WireGuard servers without API credentials.

    Args:
        timeout: Maximum total request seconds; generation can supply its
            remaining budget instead of starting another full request window.

    Returns:
        Catalog object with a regions list. Nested fields remain untrusted
            and must be validated by the endpoint selector or metadata generator.

    Raises:
        APIUnavailable: If retrieval fails or the first response line is not
            a JSON object containing a regions list.

    Note:
        Only the first line is parsed; trailing catalog signature data is not
        verified by this method. Transport uses the configured catalog URL.
    """

    # PIA serves JSON on the first line followed by signature data outside that JSON document.
    try:
        result = json.loads(
            self.request(self.config.catalog_url, timeout=timeout).splitlines()[0]
        )

        result = object_fields(result)

        if not isinstance(result.get("regions"), list):
            raise ValueError

        return result
    except (ValueError, IndexError, UnicodeError) as error:
        raise APIUnavailable("invalid response") from error

deadline 

deadline(seconds: float) -> Generator[None]

Bound DNS, connection setup, and body reads in the single-threaded Linux supervisor.

Source code in docker/privateerr/client.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@contextmanager
def deadline(seconds: float) -> Generator[None]:
    """Bound DNS, connection setup, and body reads in the single-threaded Linux supervisor."""

    def expired(signum: int, frame: FrameType | None) -> None:
        """Interrupt blocking network operations when the total request time expires."""
        raise TimeoutError("HTTP deadline exceeded")

    # A socket timeout alone does not bound DNS lookup or a body arriving a few bytes at a time.
    previous = signal.signal(signal.SIGALRM, expired)
    signal.setitimer(signal.ITIMER_REAL, seconds)

    try:
        yield
    finally:
        # Cancel this request's alarm before later work can receive it.
        signal.setitimer(signal.ITIMER_REAL, 0)
        signal.signal(signal.SIGALRM, previous)