132 lines
5.5 KiB
Python
132 lines
5.5 KiB
Python
import requests
|
|
import logging
|
|
import time
|
|
from typing import Dict, Any, Optional
|
|
|
|
class VPNClient:
|
|
"""Client for monitoring VPN server status and public IP information"""
|
|
|
|
def __init__(self, base_url: str = 'http://192.168.4.36:8000',
|
|
logger: logging.Logger = None):
|
|
"""
|
|
Initialize VPN client
|
|
|
|
Args:
|
|
base_url: Base URL for VPN monitoring API
|
|
logger: Optional logger instance (uses module logger if not provided)
|
|
"""
|
|
self.base_url = base_url.rstrip('/')
|
|
self.logger = logger or logging.getLogger(__name__)
|
|
|
|
# API endpoints
|
|
self.vpn_status_url = f'{self.base_url}/v1/vpn/status'
|
|
self.public_ip_url = f'{self.base_url}/v1/publicip/ip'
|
|
|
|
# API request retry configuration
|
|
self.api_retry_attempts = 3
|
|
self.api_retry_delay = 2
|
|
self.api_retry_backoff = 2
|
|
|
|
def get_vpn_status(self, verbose_debug: bool = True) -> Dict[str, Any]:
|
|
"""
|
|
Retrieve VPN status from VPN monitoring API with retry logic
|
|
|
|
Args:
|
|
verbose_debug: Whether to log detailed debug information
|
|
|
|
Returns:
|
|
VPN status dictionary with 'status' field
|
|
"""
|
|
last_exception = None
|
|
|
|
for attempt in range(self.api_retry_attempts):
|
|
try:
|
|
response = requests.get(self.vpn_status_url, timeout=10)
|
|
response.raise_for_status()
|
|
status_data = response.json()
|
|
|
|
# Log the VPN status for debugging (optional)
|
|
if verbose_debug:
|
|
vpn_status = status_data.get('status', 'unknown')
|
|
self.logger.debug(f"VPN status: {vpn_status}")
|
|
|
|
return status_data
|
|
except Exception as e:
|
|
last_exception = e
|
|
if attempt < self.api_retry_attempts - 1: # Not the last attempt
|
|
delay = self.api_retry_delay * (self.api_retry_backoff ** attempt)
|
|
self.logger.warning(f"VPN status request attempt {attempt + 1}/{self.api_retry_attempts} failed: {type(e).__name__}: {e}. Retrying in {delay} seconds...")
|
|
time.sleep(delay)
|
|
else:
|
|
# More detailed error logging for final failure
|
|
error_type = type(last_exception).__name__
|
|
error_details = str(last_exception)
|
|
self.logger.error(f"VPN status request failed after {self.api_retry_attempts} attempts: {error_type}: {error_details}")
|
|
|
|
# Return error details for better debugging
|
|
return {
|
|
'status': 'error',
|
|
'error_type': error_type,
|
|
'error_details': error_details,
|
|
'api_url': self.vpn_status_url
|
|
}
|
|
|
|
# Fallback return if all attempts fail (shouldn't normally reach here)
|
|
return {
|
|
'status': 'error',
|
|
'error_type': 'Unknown',
|
|
'error_details': 'All retry attempts exhausted',
|
|
'api_url': self.vpn_status_url
|
|
}
|
|
|
|
def get_public_ip_info(self, verbose_debug: bool = True) -> Dict[str, Any]:
|
|
"""
|
|
Retrieve public IP information from VPN monitoring API with retry logic
|
|
|
|
Args:
|
|
verbose_debug: Whether to log detailed debug information
|
|
|
|
Returns:
|
|
Public IP information dictionary with 'public_ip' and geographic details
|
|
"""
|
|
last_exception = None
|
|
|
|
for attempt in range(self.api_retry_attempts):
|
|
try:
|
|
response = requests.get(self.public_ip_url, timeout=10)
|
|
response.raise_for_status()
|
|
ip_data = response.json()
|
|
|
|
# Log the public IP for debugging (optional)
|
|
if verbose_debug:
|
|
public_ip = ip_data.get('public_ip', 'unknown')
|
|
self.logger.debug(f"Public IP: {public_ip}")
|
|
|
|
return ip_data
|
|
except Exception as e:
|
|
last_exception = e
|
|
if attempt < self.api_retry_attempts - 1: # Not the last attempt
|
|
delay = self.api_retry_delay * (self.api_retry_backoff ** attempt)
|
|
self.logger.warning(f"Public IP request attempt {attempt + 1}/{self.api_retry_attempts} failed: {type(e).__name__}: {e}. Retrying in {delay} seconds...")
|
|
time.sleep(delay)
|
|
else:
|
|
# More detailed error logging for final failure
|
|
error_type = type(last_exception).__name__
|
|
error_details = str(last_exception)
|
|
self.logger.error(f"Public IP request failed after {self.api_retry_attempts} attempts: {error_type}: {error_details}")
|
|
|
|
# Return error details for better debugging
|
|
return {
|
|
'public_ip': 'error',
|
|
'error_type': error_type,
|
|
'error_details': error_details,
|
|
'api_url': self.public_ip_url
|
|
}
|
|
|
|
# Fallback return if all attempts fail (shouldn't normally reach here)
|
|
return {
|
|
'public_ip': 'error',
|
|
'error_type': 'Unknown',
|
|
'error_details': 'All retry attempts exhausted',
|
|
'api_url': self.public_ip_url
|
|
} |