pyoutlineapi

 82class AsyncOutlineClient:
 83    """
 84    Asynchronous client for the Outline VPN Server API.
 85
 86    Args:
 87        api_url: Base URL for the Outline server API
 88        cert_sha256: SHA-256 fingerprint of the server's TLS certificate
 89        json_format: Return raw JSON instead of Pydantic models
 90        timeout: Request timeout in seconds
 91
 92    Examples:
 93        >>> async def doo_something():
 94        ...     async with AsyncOutlineClient(
 95        ...         "https://example.com:1234/secret",
 96        ...         "ab12cd34..."
 97        ...     ) as client:
 98        ...         server_info = await client.get_server_info()
 99    """
100
101    def __init__(
102        self,
103        api_url: str,
104        cert_sha256: str,
105        *,
106        json_format: bool = True,
107        timeout: float = 30.0,
108    ) -> None:
109        self._api_url = api_url.rstrip("/")
110        self._cert_sha256 = cert_sha256
111        self._json_format = json_format
112        self._timeout = aiohttp.ClientTimeout(total=timeout)
113        self._ssl_context: Optional[Fingerprint] = None
114        self._session: Optional[aiohttp.ClientSession] = None
115
116    async def __aenter__(self) -> AsyncOutlineClient:
117        """Set up client session for context manager."""
118        self._session = aiohttp.ClientSession(
119            timeout=self._timeout,
120            raise_for_status=False,
121            connector=aiohttp.TCPConnector(ssl=self._get_ssl_context()),
122        )
123        return self
124
125    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
126        """Clean up client session."""
127        if self._session:
128            await self._session.close()
129            self._session = None
130
131    @overload
132    async def _parse_response(
133        self,
134        response: ClientResponse,
135        model: type[BaseModel],
136        json_format: Literal[True],
137    ) -> JsonDict: ...
138
139    @overload
140    async def _parse_response(
141        self,
142        response: ClientResponse,
143        model: type[BaseModel],
144        json_format: Literal[False],
145    ) -> BaseModel: ...
146
147    @overload
148    async def _parse_response(
149        self, response: ClientResponse, model: type[BaseModel], json_format: bool
150    ) -> Union[JsonDict, BaseModel]: ...
151
152    @ensure_context
153    async def _parse_response(
154        self, response: ClientResponse, model: type[BaseModel], json_format: bool = True
155    ) -> ResponseType:
156        """
157        Parse and validate API response data.
158
159        Args:
160            response: API response to parse
161            model: Pydantic model for validation
162            json_format: Whether to return raw JSON
163
164        Returns:
165            Validated response data
166
167        Raises:
168            ValueError: If response validation fails
169        """
170        try:
171            data = await response.json()
172            validated = model.model_validate(data)
173            return validated.model_dump() if json_format else validated
174        except aiohttp.ContentTypeError as e:
175            raise ValueError("Invalid response format") from e
176        except Exception as e:
177            raise ValueError(f"Validation error: {e}") from e
178
179    @staticmethod
180    async def _handle_error_response(response: ClientResponse) -> None:
181        """Handle error responses from the API."""
182        try:
183            error_data = await response.json()
184            error = ErrorResponse.model_validate(error_data)
185            raise APIError(f"{error.code}: {error.message}", response.status)
186        except ValueError:
187            raise APIError(
188                f"HTTP {response.status}: {response.reason}", response.status
189            )
190
191    @ensure_context
192    async def _request(
193        self,
194        method: str,
195        endpoint: str,
196        *,
197        json: Any = None,
198        params: Optional[dict[str, Any]] = None,
199    ) -> Any:
200        """Make an API request."""
201        url = self._build_url(endpoint)
202
203        async with self._session.request(
204            method,
205            url,
206            json=json,
207            params=params,
208            raise_for_status=False,
209        ) as response:
210            if response.status >= 400:
211                await self._handle_error_response(response)
212
213            if response.status == 204:
214                return True
215
216            try:
217                await response.json()
218                return response
219            except aiohttp.ContentTypeError:
220                return await response.text()
221            except Exception as e:
222                raise APIError(f"Failed to parse response: {e}", response.status)
223
224    def _build_url(self, endpoint: str) -> str:
225        """Build and validate the full URL for the API request."""
226        if not isinstance(endpoint, str):
227            raise ValueError("Endpoint must be a string")
228
229        url = f"{self._api_url}/{endpoint.lstrip('/')}"
230        parsed_url = urlparse(url)
231
232        if not all([parsed_url.scheme, parsed_url.netloc]):
233            raise ValueError(f"Invalid URL: {url}")
234
235        return url
236
237    def _get_ssl_context(self) -> Optional[Fingerprint]:
238        """Create an SSL context if a certificate fingerprint is provided."""
239        if not self._cert_sha256:
240            return None
241
242        try:
243            return Fingerprint(binascii.unhexlify(self._cert_sha256))
244        except binascii.Error as e:
245            raise ValueError(f"Invalid certificate SHA256: {self._cert_sha256}") from e
246        except Exception as e:
247            raise OutlineError("Failed to create SSL context") from e
248
249    async def get_server_info(self) -> Union[JsonDict, Server]:
250        """
251        Get server information.
252
253        Returns:
254            Server information including name, ID, and configuration.
255
256        Examples:
257            >>> async def doo_something():
258            ...     async with AsyncOutlineClient(
259            ...         "https://example.com:1234/secret",
260            ...         "ab12cd34..."
261            ...     ) as client:
262            ...         server = await client.get_server_info()
263            ...         print(f"Server {server.name} running version {server.version}")
264        """
265        response = await self._request("GET", "server")
266        return await self._parse_response(
267            response, Server, json_format=self._json_format
268        )
269
270    async def rename_server(self, name: str) -> bool:
271        """
272        Rename the server.
273
274        Args:
275            name: New server name
276
277        Returns:
278            True if successful
279
280        Examples:
281            >>> async def doo_something():
282            ...     async with AsyncOutlineClient(
283            ...         "https://example.com:1234/secret",
284            ...         "ab12cd34..."
285            ...     ) as client:
286            ...     success = await client.rename_server("My VPN Server")
287            ...     if success:
288            ...         print("Server renamed successfully")
289        """
290        return await self._request("PUT", "name", json={"name": name})
291
292    async def set_hostname(self, hostname: str) -> bool:
293        """
294        Set server hostname for access keys.
295
296        Args:
297            hostname: New hostname or IP address
298
299        Returns:
300            True if successful
301
302        Raises:
303            APIError: If hostname is invalid
304
305        Examples:
306            >>> async def doo_something():
307            ...     async with AsyncOutlineClient(
308            ...         "https://example.com:1234/secret",
309            ...         "ab12cd34..."
310            ...     ) as client:
311            ...         await client.set_hostname("vpn.example.com")
312            ...         # Or use IP address
313            ...         await client.set_hostname("203.0.113.1")
314        """
315        return await self._request(
316            "PUT", "server/hostname-for-access-keys", json={"hostname": hostname}
317        )
318
319    async def set_default_port(self, port: int) -> bool:
320        """
321        Set default port for new access keys.
322
323        Args:
324            port: Port number (1025-65535)
325
326        Returns:
327            True if successful
328
329        Raises:
330            APIError: If port is invalid or in use
331
332        Examples:
333            >>> async def doo_something():
334            ...     async with AsyncOutlineClient(
335            ...         "https://example.com:1234/secret",
336            ...         "ab12cd34..."
337            ...     ) as client:
338            ...         await client.set_default_port(8388)
339
340        """
341        if port < 1025 or port > 65535:
342            raise ValueError("Privileged ports are not allowed. Use range: 1025-65535")
343
344        return await self._request(
345            "PUT", "server/port-for-new-access-keys", json={"port": port}
346        )
347
348    async def get_metrics_status(self) -> dict[str, Any] | BaseModel:
349        """
350        Get whether metrics collection is enabled.
351
352        Returns:
353            Current metrics collection status
354
355        Examples:
356            >>> async def doo_something():
357            ...     async with AsyncOutlineClient(
358            ...         "https://example.com:1234/secret",
359            ...         "ab12cd34..."
360            ...     ) as client:
361            ...         if await client.get_metrics_status():
362            ...             print("Metrics collection is enabled")
363        """
364        response = await self._request("GET", "metrics/enabled")
365        data = await self._parse_response(
366            response, MetricsStatusResponse, json_format=self._json_format
367        )
368        return data
369
370    async def set_metrics_status(self, enabled: bool) -> bool:
371        """
372        Enable or disable metrics collection.
373
374        Args:
375            enabled: Whether to enable metrics
376
377        Returns:
378            True if successful
379
380        Examples:
381            >>> async def doo_something():
382            ...     async with AsyncOutlineClient(
383            ...         "https://example.com:1234/secret",
384            ...         "ab12cd34..."
385            ...     ) as client:
386            ...         # Enable metrics
387            ...         await client.set_metrics_status(True)
388            ...         # Check new status
389            ...         is_enabled = await client.get_metrics_status()
390        """
391        return await self._request(
392            "PUT", "metrics/enabled", json={"metricsEnabled": enabled}
393        )
394
395    async def get_transfer_metrics(
396        self, period: MetricsPeriod = MetricsPeriod.MONTHLY
397    ) -> Union[JsonDict, ServerMetrics]:
398        """
399        Get transfer metrics for specified period.
400
401        Args:
402            period: Time period for metrics (DAILY, WEEKLY, or MONTHLY)
403
404        Returns:
405            Transfer metrics data for each access key
406
407        Examples:
408            >>> async def doo_something():
409            ...     async with AsyncOutlineClient(
410            ...         "https://example.com:1234/secret",
411            ...         "ab12cd34..."
412            ...     ) as client:
413            ...         # Get monthly metrics
414            ...         metrics = await client.get_transfer_metrics()
415            ...         # Or get daily metrics
416            ...         daily = await client.get_transfer_metrics(MetricsPeriod.DAILY)
417            ...         for user_id, bytes_transferred in daily.bytes_transferred_by_user_id.items():
418            ...             print(f"User {user_id}: {bytes_transferred / 1024**3:.2f} GB")
419        """
420        response = await self._request(
421            "GET", "metrics/transfer", params={"period": period.value}
422        )
423        return await self._parse_response(
424            response, ServerMetrics, json_format=self._json_format
425        )
426
427    async def create_access_key(
428        self,
429        *,
430        name: Optional[str] = None,
431        password: Optional[str] = None,
432        port: Optional[int] = None,
433        method: Optional[str] = None,
434        limit: Optional[DataLimit] = None,
435    ) -> Union[JsonDict, AccessKey]:
436        """
437        Create a new access key.
438
439        Args:
440            name: Optional key name
441            password: Optional password
442            port: Optional port number (1-65535)
443            method: Optional encryption method
444            limit: Optional data transfer limit
445
446        Returns:
447            New access key details
448
449        Examples:
450            >>> async def doo_something():
451            ...     async with AsyncOutlineClient(
452            ...         "https://example.com:1234/secret",
453            ...         "ab12cd34..."
454            ...     ) as client:
455            ...         # Create basic key
456            ...         key = await client.create_access_key(name="User 1")
457            ...
458            ...         # Create key with data limit
459            ...         _limit = DataLimit(bytes=5 * 1024**3)  # 5 GB
460            ...         key = await client.create_access_key(
461            ...             name="Limited User",
462            ...             port=8388,
463            ...             limit=_limit
464            ...         )
465            ...         print(f"Created key: {key.access_url}")
466        """
467        request = AccessKeyCreateRequest(
468            name=name, password=password, port=port, method=method, limit=limit
469        )
470        response = await self._request(
471            "POST", "access-keys", json=request.model_dump(exclude_none=True)
472        )
473        return await self._parse_response(
474            response, AccessKey, json_format=self._json_format
475        )
476
477    async def get_access_keys(self) -> Union[JsonDict, AccessKeyList]:
478        """
479        Get all access keys.
480
481        Returns:
482            List of all access keys
483
484        Examples:
485            >>> async def doo_something():
486            ...     async with AsyncOutlineClient(
487            ...         "https://example.com:1234/secret",
488            ...         "ab12cd34..."
489            ...     ) as client:
490            ...         keys = await client.get_access_keys()
491            ...         for key in keys.access_keys:
492            ...             print(f"Key {key.id}: {key.name or 'unnamed'}")
493            ...             if key.data_limit:
494            ...                 print(f"  Limit: {key.data_limit.bytes / 1024**3:.1f} GB")
495        """
496        response = await self._request("GET", "access-keys")
497        return await self._parse_response(
498            response, AccessKeyList, json_format=self._json_format
499        )
500
501    async def get_access_key(self, key_id: int) -> Union[JsonDict, AccessKey]:
502        """
503        Get specific access key.
504
505        Args:
506            key_id: Access key ID
507
508        Returns:
509            Access key details
510
511        Raises:
512            APIError: If key doesn't exist
513
514        Examples:
515            >>> async def doo_something():
516            ...     async with AsyncOutlineClient(
517            ...         "https://example.com:1234/secret",
518            ...         "ab12cd34..."
519            ...     ) as client:
520            ...         key = await client.get_access_key(1)
521            ...         print(f"Port: {key.port}")
522            ...         print(f"URL: {key.access_url}")
523        """
524        response = await self._request("GET", f"access-keys/{key_id}")
525        return await self._parse_response(
526            response, AccessKey, json_format=self._json_format
527        )
528
529    async def rename_access_key(self, key_id: int, name: str) -> bool:
530        """
531        Rename access key.
532
533        Args:
534            key_id: Access key ID
535            name: New name
536
537        Returns:
538            True if successful
539
540        Raises:
541            APIError: If key doesn't exist
542
543        Examples:
544            >>> async def doo_something():
545            ...     async with AsyncOutlineClient(
546            ...         "https://example.com:1234/secret",
547            ...         "ab12cd34..."
548            ...     ) as client:
549            ...         # Rename key
550            ...         await client.rename_access_key(1, "Alice")
551            ...
552            ...         # Verify new name
553            ...         key = await client.get_access_key(1)
554            ...         assert key.name == "Alice"
555        """
556        return await self._request(
557            "PUT", f"access-keys/{key_id}/name", json={"name": name}
558        )
559
560    async def delete_access_key(self, key_id: int) -> bool:
561        """
562        Delete access key.
563
564        Args:
565            key_id: Access key ID
566
567        Returns:
568            True if successful
569
570        Raises:
571            APIError: If key doesn't exist
572
573        Examples:
574            >>> async def doo_something():
575            ...     async with AsyncOutlineClient(
576            ...         "https://example.com:1234/secret",
577            ...         "ab12cd34..."
578            ...     ) as client:
579            ...         if await client.delete_access_key(1):
580            ...             print("Key deleted")
581
582        """
583        return await self._request("DELETE", f"access-keys/{key_id}")
584
585    async def set_access_key_data_limit(self, key_id: int, bytes_limit: int) -> bool:
586        """
587        Set data transfer limit for access key.
588
589        Args:
590            key_id: Access key ID
591            bytes_limit: Limit in bytes (must be positive)
592
593        Returns:
594            True if successful
595
596        Raises:
597            APIError: If key doesn't exist or limit is invalid
598
599        Examples:
600            >>> async def doo_something():
601            ...     async with AsyncOutlineClient(
602            ...         "https://example.com:1234/secret",
603            ...         "ab12cd34..."
604            ...     ) as client:
605            ...         # Set 5 GB limit
606            ...         limit = 5 * 1024**3  # 5 GB in bytes
607            ...         await client.set_access_key_data_limit(1, limit)
608            ...
609            ...         # Verify limit
610            ...         key = await client.get_access_key(1)
611            ...         assert key.data_limit and key.data_limit.bytes == limit
612        """
613        return await self._request(
614            "PUT",
615            f"access-keys/{key_id}/data-limit",
616            json={"limit": {"bytes": bytes_limit}},
617        )
618
619    async def remove_access_key_data_limit(self, key_id: int) -> bool:
620        """
621        Remove data transfer limit from access key.
622
623        Args:
624            key_id: Access key ID
625
626        Returns:
627            True if successful
628
629        Raises:
630            APIError: If key doesn't exist
631        """
632        return await self._request("DELETE", f"access-keys/{key_id}/data-limit")
633
634    @property
635    def session(self):
636        return self._session

58class OutlineError(Exception):
59    """Base exception for Outline client errors."""

62class APIError(OutlineError):
63    """Raised when API requests fail."""
64
65    def __init__(self, message: str, status_code: Optional[int] = None) -> None:
66        super().__init__(message)
67        self.status_code = status_code

42class AccessKey(BaseModel):
43    """Access key details."""
44
45    id: int
46    name: Optional[str] = None
47    password: str
48    port: int = Field(gt=0, lt=65536)
49    method: str
50    access_url: str = Field(alias="accessUrl")
51    data_limit: Optional[DataLimit] = Field(None, alias="dataLimit")

119class AccessKeyCreateRequest(BaseModel):
120    """
121    Request parameters for creating an access key.
122    Per OpenAPI: /access-keys POST request body
123    """
124
125    name: Optional[str] = None
126    method: Optional[str] = None
127    password: Optional[str] = None
128    port: Optional[int] = Field(None, gt=0, lt=65536)
129    limit: Optional[DataLimit] = None

54class AccessKeyList(BaseModel):
55    """List of access keys."""
56
57    access_keys: list[AccessKey] = Field(alias="accessKeys")

30class DataLimit(BaseModel):
31    """Data transfer limit configuration."""
32
33    bytes: int = Field(gt=0)
34
35    @field_validator("bytes")
36    def validate_bytes(cls, v: int) -> int:
37        if v < 0:
38            raise ValueError("bytes must be positive")
39        return v