pyoutlineapi

171class APIError(OutlineError):
172    """HTTP API request failure.
173
174    Automatically determines retry eligibility based on HTTP status code.
175
176    Attributes:
177        status_code: HTTP status code (if available)
178        endpoint: API endpoint that failed
179        response_data: Raw response data (may contain sensitive info)
180
181    Example:
182        >>> error = APIError("Not found", status_code=404, endpoint="/server")
183        >>> error.is_client_error  # True
184        >>> error.is_retryable  # False
185    """
186
187    __slots__ = ("endpoint", "response_data", "status_code")
188
189    def __init__(
190        self,
191        message: str,
192        *,
193        status_code: int | None = None,
194        endpoint: str | None = None,
195        response_data: dict[str, Any] | None = None,
196    ) -> None:
197        """Initialize API error with sanitized endpoint.
198
199        Args:
200            message: Error message
201            status_code: HTTP status code
202            endpoint: API endpoint (will be sanitized)
203            response_data: Response data (may contain sensitive info)
204        """
205        from .common_types import Validators
206
207        # Sanitize endpoint for safe logging
208        safe_endpoint = (
209            Validators.sanitize_endpoint_for_logging(endpoint) if endpoint else None
210        )
211
212        # Build safe details (optimization: avoid dict creation if all None)
213        safe_details: dict[str, Any] | None = None
214        if status_code is not None or safe_endpoint is not None:
215            safe_details = {}
216            if status_code is not None:
217                safe_details["status_code"] = status_code
218            if safe_endpoint is not None:
219                safe_details["endpoint"] = safe_endpoint
220
221        # Build internal details (optimization: avoid dict creation if all None)
222        details: dict[str, Any] | None = None
223        if status_code is not None or endpoint is not None:
224            details = {}
225            if status_code is not None:
226                details["status_code"] = status_code
227            if endpoint is not None:
228                details["endpoint"] = endpoint
229
230        super().__init__(message, details=details, safe_details=safe_details)
231
232        # Store attributes directly (faster access than dict lookups)
233        self.status_code = status_code
234        self.endpoint = endpoint
235        self.response_data = response_data
236
237    @property
238    def is_retryable(self) -> bool:
239        """Check if error is retryable based on status code."""
240        return (
241            self.status_code in Constants.RETRY_STATUS_CODES
242            if self.status_code
243            else False
244        )
245
246    @property
247    def is_client_error(self) -> bool:
248        """Check if error is a client error (4xx status).
249
250        Returns:
251            True if status code is 400-499
252        """
253        return self.status_code is not None and 400 <= self.status_code < 500
254
255    @property
256    def is_server_error(self) -> bool:
257        """Check if error is a server error (5xx status).
258
259        Returns:
260            True if status code is 500-599
261        """
262        return self.status_code is not None and 500 <= self.status_code < 600
263
264    @property
265    def is_rate_limit_error(self) -> bool:
266        """Check if error is a rate limit error (429 status).
267
268        Returns:
269            True if status code is 429
270        """
271        return self.status_code == 429

137class AccessKey(BaseValidatedModel):
138    """Access key model matching API schema with optimized properties.
139
140    SCHEMA: Based on OpenAPI /access-keys endpoint
141    """
142
143    id: str
144    name: str | None = None
145    password: str
146    port: Port
147    method: str
148    access_url: str = Field(alias="accessUrl")
149    data_limit: DataLimit | None = Field(None, alias="dataLimit")
150
151    @field_validator("name", mode="before")
152    @classmethod
153    def validate_name(cls, v: str | None) -> str | None:
154        """Validate and normalize name from API."""
155        if v is None:
156            return None
157
158        if isinstance(v, str):
159            stripped = v.strip()
160            if not stripped:
161                return None
162
163            if len(stripped) > Constants.MAX_NAME_LENGTH:
164                raise ValueError(
165                    f"Name too long: {len(stripped)} (max {Constants.MAX_NAME_LENGTH})"
166                )
167            return stripped
168
169    @field_validator("id")
170    @classmethod
171    def validate_id(cls, v: str) -> str:
172        """Validate key ID.
173
174        :param v: Key ID
175        :return: Validated key ID
176        :raises ValueError: If ID is invalid
177        """
178        return Validators.validate_key_id(v)
179
180    @property
181    def has_data_limit(self) -> bool:
182        """Check if key has data limit (optimized None check).
183
184        :return: True if data limit exists
185        """
186        return self.data_limit is not None
187
188    @property
189    def display_name(self) -> str:
190        """Get display name with optimized conditional.
191
192        :return: Display name
193        """
194        return self.name if self.name else f"Key-{self.id}"

491class AccessKeyCreateRequest(BaseValidatedModel):
492    """Request model for creating access keys.
493
494    SCHEMA: Based on POST /access-keys request body
495    """
496
497    name: str | None = Field(default=None, min_length=1, max_length=255)
498    method: str | None = None
499    password: str | None = None
500    port: Port | None = None
501    limit: DataLimit | None = None

197class AccessKeyList(BaseValidatedModel):
198    """List of access keys with optimized utility methods.
199
200    SCHEMA: Based on GET /access-keys response
201    """
202
203    access_keys: list[AccessKey] = Field(alias="accessKeys")
204
205    @cached_property
206    def count(self) -> int:
207        """Get number of access keys (cached).
208
209        NOTE: Cached because list is immutable after creation
210
211        :return: Key count
212        """
213        return len(self.access_keys)
214
215    @property
216    def is_empty(self) -> bool:
217        """Check if list is empty (uses cached count).
218
219        :return: True if no keys
220        """
221        return self.count == 0
222
223    def get_by_id(self, key_id: str) -> AccessKey | None:
224        """Get key by ID with early return optimization.
225
226        :param key_id: Access key ID
227        :return: Access key or None if not found
228        """
229        for key in self.access_keys:
230            if key.id == key_id:
231                return key
232        return None
233
234    def get_by_name(self, name: str) -> list[AccessKey]:
235        """Get keys by name with optimized list comprehension.
236
237        :param name: Key name
238        :return: List of matching keys (may be multiple)
239        """
240        return [key for key in self.access_keys if key.name == name]
241
242    def filter_with_limits(self) -> list[AccessKey]:
243        """Get keys with data limits (optimized comprehension).
244
245        :return: List of keys with limits
246        """
247        return [key for key in self.access_keys if key.has_data_limit]
248
249    def filter_without_limits(self) -> list[AccessKey]:
250        """Get keys without data limits (optimized comprehension).
251
252        :return: List of keys without limits
253        """
254        return [key for key in self.access_keys if not key.has_data_limit]

443class AccessKeyMetric(BaseValidatedModel):
444    """Per-key experimental metrics.
445
446    SCHEMA: Based on experimental metrics accessKeys array item
447    """
448
449    access_key_id: str = Field(alias="accessKeyId")
450    tunnel_time: TunnelTime = Field(alias="tunnelTime")
451    data_transferred: DataTransferred = Field(alias="dataTransferred")
452    connection: ConnectionInfo

531class AccessKeyNameRequest(BaseValidatedModel):
532    """Request model for renaming access key.
533
534    SCHEMA: Based on PUT /access-keys/{id}/name request body
535    """
536
537    name: str = Field(min_length=1, max_length=255)

 46class AsyncOutlineClient(
 47    BaseHTTPClient,
 48    ServerMixin,
 49    AccessKeyMixin,
 50    DataLimitMixin,
 51    MetricsMixin,
 52):
 53    """High-performance async client for Outline VPN Server API."""
 54
 55    __slots__ = (
 56        "_audit_logger_instance",
 57        "_config",
 58        "_default_json_format",
 59    )
 60
 61    def __init__(
 62        self,
 63        config: OutlineClientConfig | None = None,
 64        *,
 65        api_url: str | None = None,
 66        cert_sha256: str | None = None,
 67        audit_logger: AuditLogger | None = None,
 68        metrics: MetricsCollector | None = None,
 69        **overrides: Unpack[ConfigOverrides],
 70    ) -> None:
 71        """Initialize Outline client with modern configuration approach.
 72
 73        Uses structural pattern matching for configuration resolution.
 74
 75        :param config: Client configuration object
 76        :param api_url: API URL (alternative to config)
 77        :param cert_sha256: Certificate fingerprint (alternative to config)
 78        :param audit_logger: Custom audit logger
 79        :param metrics: Custom metrics collector
 80        :param overrides: Configuration overrides (timeout, retry_attempts, etc.)
 81        :raises ConfigurationError: If configuration is invalid
 82
 83        Example:
 84            >>> async with AsyncOutlineClient.from_env() as client:
 85            ...     info = await client.get_server_info()
 86        """
 87        # Build config_kwargs using utility function (DRY)
 88        config_kwargs = build_config_overrides(**overrides)
 89
 90        # Validate configuration using pattern matching
 91        resolved_config = self._resolve_configuration(
 92            config, api_url, cert_sha256, config_kwargs
 93        )
 94
 95        self._config = resolved_config
 96        self._audit_logger_instance = audit_logger
 97        self._default_json_format = resolved_config.json_format
 98
 99        # Initialize base HTTP client
100        super().__init__(
101            api_url=resolved_config.api_url,
102            cert_sha256=resolved_config.cert_sha256,
103            timeout=resolved_config.timeout,
104            retry_attempts=resolved_config.retry_attempts,
105            max_connections=resolved_config.max_connections,
106            user_agent=resolved_config.user_agent,
107            enable_logging=resolved_config.enable_logging,
108            circuit_config=resolved_config.circuit_config,
109            rate_limit=resolved_config.rate_limit,
110            allow_private_networks=resolved_config.allow_private_networks,
111            resolve_dns_for_ssrf=resolved_config.resolve_dns_for_ssrf,
112            audit_logger=audit_logger,
113            metrics=metrics,
114        )
115
116        # Cache instance for weak reference tracking (automatic cleanup)
117        _client_cache[id(self)] = self
118
119        if resolved_config.enable_logging and logger.isEnabledFor(logging.INFO):
120            safe_url = Validators.sanitize_url_for_logging(self.api_url)
121            logger.info("Client initialized for %s", safe_url)
122
123    @staticmethod
124    def _resolve_configuration(
125        config: OutlineClientConfig | None,
126        api_url: str | None,
127        cert_sha256: str | None,
128        kwargs: dict[str, Any],
129    ) -> OutlineClientConfig:
130        """Resolve and validate configuration using pattern matching.
131
132        :param config: Configuration object
133        :param api_url: Direct API URL
134        :param cert_sha256: Direct certificate
135        :param kwargs: Additional kwargs
136        :return: Resolved configuration
137        :raises ConfigurationError: If configuration is invalid
138        """
139        match config, api_url, cert_sha256:
140            # Pattern 1: Direct parameters provided (most common case)
141            case None, str(url), str(cert) if url and cert:
142                return OutlineClientConfig.create_minimal(url, cert, **kwargs)
143
144            # Pattern 2: Config object provided
145            case OutlineClientConfig() as cfg, None, None:
146                return cfg
147
148            # Pattern 3: Missing required parameters
149            case None, None, _:
150                raise ConfigurationError(
151                    "Missing required 'api_url'",
152                    field="api_url",
153                    security_issue=False,
154                )
155            case None, _, None:
156                raise ConfigurationError(
157                    "Missing required 'cert_sha256'",
158                    field="cert_sha256",
159                    security_issue=True,
160                )
161
162            # Pattern 4: Conflicting parameters
163            case OutlineClientConfig(), str() | None, str() | None:
164                raise ConfigurationError(
165                    "Cannot specify both 'config' and direct parameters"
166                )
167
168            # Pattern 5: Invalid combination (catch-all)
169            case _:
170                raise ConfigurationError("Invalid parameter combination")
171
172    @property
173    def config(self) -> OutlineClientConfig:
174        """Get immutable copy of configuration.
175
176        :return: Deep copy of configuration
177        """
178        return self._config.model_copy_immutable()
179
180    @property
181    def get_sanitized_config(self) -> dict[str, Any]:
182        """Delegate to config's sanitized representation.
183
184        See: OutlineClientConfig.get_sanitized_config().
185
186        :return: Sanitized configuration from underlying config object
187        """
188        return self._config.get_sanitized_config
189
190    @property
191    def json_format(self) -> bool:
192        """Get JSON format preference.
193
194        :return: True if raw JSON format is preferred
195        """
196        return self._default_json_format
197
198    # ===== Factory Methods =====
199
200    @classmethod
201    @asynccontextmanager
202    async def create(
203        cls,
204        api_url: str | None = None,
205        cert_sha256: str | None = None,
206        *,
207        config: OutlineClientConfig | None = None,
208        audit_logger: AuditLogger | None = None,
209        metrics: MetricsCollector | None = None,
210        **overrides: Unpack[ConfigOverrides],
211    ) -> AsyncGenerator[AsyncOutlineClient, None]:
212        """Create and initialize client as async context manager.
213
214        Automatically handles initialization and cleanup.
215        Recommended way to create clients in async contexts.
216
217        :param api_url: API URL
218        :param cert_sha256: Certificate fingerprint
219        :param config: Configuration object
220        :param audit_logger: Custom audit logger
221        :param metrics: Custom metrics collector
222        :param overrides: Configuration overrides (timeout, retry_attempts, etc.)
223        :yield: Initialized client instance
224        :raises ConfigurationError: If configuration is invalid
225
226        Example:
227            >>> async with AsyncOutlineClient.from_env() as client:
228            ...     keys = await client.get_access_keys()
229        """
230        if config is not None:
231            client = cls(config=config, audit_logger=audit_logger, metrics=metrics)
232        else:
233            client = cls(
234                api_url=api_url,
235                cert_sha256=cert_sha256,
236                audit_logger=audit_logger,
237                metrics=metrics,
238                **overrides,
239            )
240
241        async with client:
242            yield client
243
244    @classmethod
245    def from_env(
246        cls,
247        *,
248        env_file: str | Path | None = None,
249        audit_logger: AuditLogger | None = None,
250        metrics: MetricsCollector | None = None,
251        **overrides: Unpack[ConfigOverrides],
252    ) -> AsyncOutlineClient:
253        """Create client from environment variables.
254
255        Reads configuration from environment or .env file.
256        Modern approach using **overrides for runtime configuration.
257
258        :param env_file: Path to environment file (.env)
259        :param audit_logger: Custom audit logger
260        :param metrics: Custom metrics collector
261        :param overrides: Configuration overrides (timeout, enable_logging, etc.)
262        :return: Configured client instance
263        :raises ConfigurationError: If environment configuration is invalid
264
265        Example:
266            >>> async with AsyncOutlineClient.from_env(
267            ...     env_file=".env.production",
268            ...     timeout=20,
269            ... ) as client:
270            ...     info = await client.get_server_info()
271        """
272        config = OutlineClientConfig.from_env(env_file=env_file, **overrides)
273        return cls(config=config, audit_logger=audit_logger, metrics=metrics)
274
275    # ===== Context Manager Methods =====
276
277    async def __aexit__(
278        self,
279        exc_type: type[BaseException] | None,
280        exc_val: BaseException | None,
281        exc_tb: object | None,
282    ) -> None:
283        """Async context manager exit with comprehensive cleanup.
284
285        Ensures graceful shutdown even on exceptions. Uses ordered cleanup
286        sequence for proper resource deallocation.
287
288        Cleanup order:
289        1. Audit logger shutdown (drain queue)
290        2. HTTP client shutdown (close connections)
291        3. Emergency cleanup if steps 1-2 failed
292
293        :param exc_type: Exception type if error occurred
294        :param exc_val: Exception instance if error occurred
295        :param exc_tb: Exception traceback
296        :return: False to propagate exceptions
297        """
298        cleanup_errors: list[str] = []
299
300        # Step 1: Graceful audit logger shutdown
301        if self._audit_logger_instance is not None:
302            try:
303                if hasattr(self._audit_logger_instance, "shutdown"):
304                    shutdown_method = self._audit_logger_instance.shutdown
305                    if asyncio.iscoroutinefunction(shutdown_method):
306                        await shutdown_method()
307            except Exception as e:
308                error_msg = f"Audit logger shutdown error: {e}"
309                cleanup_errors.append(error_msg)
310                if logger.isEnabledFor(logging.WARNING):
311                    logger.warning(error_msg)
312
313        # Step 2: Shutdown HTTP client
314        try:
315            await self.shutdown(timeout=30.0)
316        except Exception as e:
317            error_msg = f"HTTP client shutdown error: {e}"
318            cleanup_errors.append(error_msg)
319            if logger.isEnabledFor(logging.ERROR):
320                logger.error(error_msg)
321
322        # Step 3: Emergency cleanup if shutdown failed
323        if cleanup_errors and hasattr(self, "_session"):
324            try:
325                if self._session and not self._session.closed:
326                    await self._session.close()
327                    if logger.isEnabledFor(logging.DEBUG):
328                        logger.debug("Emergency session cleanup completed")
329            except Exception as e:
330                if logger.isEnabledFor(logging.DEBUG):
331                    logger.debug("Emergency cleanup error: %s", e)
332
333        # Log summary of cleanup issues
334        if cleanup_errors and logger.isEnabledFor(logging.WARNING):
335            logger.warning(
336                "Cleanup completed with %d error(s): %s",
337                len(cleanup_errors),
338                "; ".join(cleanup_errors),
339            )
340
341        # Always propagate the original exception
342        return None
343
344    # ===== Utility Methods =====
345
346    async def health_check(self) -> dict[str, Any]:
347        """Perform basic health check.
348
349        Non-intrusive check that tests server connectivity without
350        modifying any state. Returns comprehensive health metrics.
351
352        :return: Health check result dictionary with response time
353
354        Example result:
355            {
356                "timestamp": 1234567890.123,
357                "healthy": True,
358                "response_time_ms": 45.2,
359                "connected": True,
360                "circuit_state": "closed",
361                "active_requests": 2,
362                "rate_limit_available": 98
363            }
364        """
365        import time
366
367        health_data: dict[str, Any] = {
368            "timestamp": time.time(),
369            "connected": self.is_connected,
370            "circuit_state": self.circuit_state,
371            "active_requests": self.active_requests,
372            "rate_limit_available": self.available_slots,
373        }
374
375        try:
376            start_time = time.monotonic()
377            await self.get_server_info()
378            duration = time.monotonic() - start_time
379
380            health_data["healthy"] = True
381            health_data["response_time_ms"] = round(duration * 1000, 2)
382
383        except Exception as e:
384            health_data["healthy"] = False
385            health_data["error"] = str(e)
386            health_data["error_type"] = type(e).__name__
387
388        return health_data
389
390    async def get_server_summary(self) -> dict[str, Any]:
391        """Get comprehensive server overview.
392
393        Aggregates multiple API calls into a single summary.
394        Continues on partial failures to return maximum information.
395        Executes non-dependent calls concurrently for performance.
396
397        :return: Server summary dictionary with aggregated data
398
399        Example result:
400            {
401                "timestamp": 1234567890.123,
402                "healthy": True,
403                "server": {...},
404                "access_keys_count": 10,
405                "metrics_enabled": True,
406                "transfer_metrics": {...},
407                "client_status": {...},
408                "errors": []
409            }
410        """
411        import time
412
413        summary: dict[str, Any] = {
414            "timestamp": time.time(),
415            "healthy": True,
416            "errors": [],
417        }
418
419        server_task = self.get_server_info(as_json=True)
420        keys_task = self.get_access_keys(as_json=True)
421        metrics_status_task = self.get_metrics_status(as_json=True)
422
423        server_result, keys_result, metrics_status_result = await asyncio.gather(
424            server_task, keys_task, metrics_status_task, return_exceptions=True
425        )
426
427        # Process server info
428        if isinstance(server_result, Exception):
429            summary["healthy"] = False
430            summary["errors"].append(f"Server info error: {server_result}")
431            if logger.isEnabledFor(logging.DEBUG):
432                logger.debug("Failed to fetch server info: %s", server_result)
433        else:
434            summary["server"] = server_result
435
436        # Process access keys
437        if isinstance(keys_result, Exception):
438            summary["healthy"] = False
439            summary["errors"].append(f"Access keys error: {keys_result}")
440            if logger.isEnabledFor(logging.DEBUG):
441                logger.debug("Failed to fetch access keys: %s", keys_result)
442        elif isinstance(keys_result, dict):
443            keys_list = keys_result.get("accessKeys", [])
444            summary["access_keys_count"] = (
445                len(keys_list) if isinstance(keys_list, list) else 0
446            )
447        elif isinstance(keys_result, AccessKeyList):
448            summary["access_keys_count"] = len(keys_result.access_keys)
449        else:
450            summary["access_keys_count"] = 0
451
452        # Process metrics status
453        if isinstance(metrics_status_result, Exception):
454            summary["errors"].append(f"Metrics status error: {metrics_status_result}")
455            if logger.isEnabledFor(logging.DEBUG):
456                logger.debug(
457                    "Failed to fetch metrics status: %s", metrics_status_result
458                )
459        elif isinstance(metrics_status_result, dict):
460            metrics_enabled = bool(metrics_status_result.get("metricsEnabled", False))
461            summary["metrics_enabled"] = metrics_enabled
462
463            # Fetch transfer metrics if enabled (dependent call - sequential)
464            if metrics_enabled:
465                try:
466                    transfer = await self.get_transfer_metrics(as_json=True)
467                    summary["transfer_metrics"] = transfer
468                except Exception as e:
469                    summary["errors"].append(f"Transfer metrics error: {e}")
470                    if logger.isEnabledFor(logging.DEBUG):
471                        logger.debug("Failed to fetch transfer metrics: %s", e)
472        elif isinstance(metrics_status_result, MetricsStatusResponse):
473            summary["metrics_enabled"] = metrics_status_result.metrics_enabled
474            if metrics_status_result.metrics_enabled:
475                try:
476                    transfer = await self.get_transfer_metrics(as_json=True)
477                    summary["transfer_metrics"] = transfer
478                except Exception as e:
479                    summary["errors"].append(f"Transfer metrics error: {e}")
480                    if logger.isEnabledFor(logging.DEBUG):
481                        logger.debug("Failed to fetch transfer metrics: %s", e)
482        else:
483            summary["metrics_enabled"] = False
484
485        # Add client status (synchronous, no API call)
486        summary["client_status"] = {
487            "connected": self.is_connected,
488            "circuit_state": self.circuit_state,
489            "active_requests": self.active_requests,
490            "rate_limit": {
491                "limit": self.rate_limit,
492                "available": self.available_slots,
493            },
494        }
495
496        return summary
497
498    def get_status(self) -> dict[str, Any]:
499        """Get current client status (synchronous).
500
501        Returns immediate status without making API calls.
502        Useful for monitoring and debugging.
503
504        :return: Status dictionary with all client metrics
505
506        Example result:
507            {
508                "connected": True,
509                "circuit_state": "closed",
510                "active_requests": 2,
511                "rate_limit": {
512                    "limit": 100,
513                    "available": 98,
514                    "active": 2
515                },
516                "circuit_metrics": {...}
517            }
518        """
519        return {
520            "connected": self.is_connected,
521            "circuit_state": self.circuit_state,
522            "active_requests": self.active_requests,
523            "rate_limit": {
524                "limit": self.rate_limit,
525                "available": self.available_slots,
526                "active": self.active_requests,
527            },
528            "circuit_metrics": self.get_circuit_metrics(),
529        }
530
531    def __repr__(self) -> str:
532        """Safe string representation without secrets.
533
534        Does not expose any sensitive information (URLs, certificates, tokens).
535
536        :return: String representation
537        """
538        status = "connected" if self.is_connected else "disconnected"
539        parts = [f"status={status}"]
540
541        if self.circuit_state:
542            parts.append(f"circuit={self.circuit_state}")
543
544        if self.active_requests:
545            parts.append(f"requests={self.active_requests}")
546
547        return f"AsyncOutlineClient({', '.join(parts)})"