362 Zeilen
12 KiB
Python
362 Zeilen
12 KiB
Python
"""
|
|
API Client für die Kommunikation mit dem License Server.
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import logging
|
|
from typing import Dict, Any, Optional
|
|
from urllib.parse import urljoin
|
|
|
|
logger = logging.getLogger("license_api_client")
|
|
logger.setLevel(logging.DEBUG)
|
|
# Füge Console Handler hinzu falls noch nicht vorhanden
|
|
if not logger.handlers:
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
|
|
logger.addHandler(handler)
|
|
|
|
|
|
class LicenseAPIClient:
|
|
"""Client für die Kommunikation mit dem License Server API."""
|
|
|
|
def __init__(self, base_url: str = "https://api-software-undso.intelsight.de",
|
|
api_key: str = "AF-2025-8E57CA6A97E257C5FA3E7778B8B44413"): # TODO: Update with valid API key
|
|
"""
|
|
Initialisiert den API Client.
|
|
|
|
Args:
|
|
base_url: Basis-URL des License Servers
|
|
api_key: API Key für die Authentifizierung
|
|
"""
|
|
self.base_url = base_url.rstrip('/')
|
|
self.api_key = api_key
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'X-API-Key': self.api_key,
|
|
'Content-Type': 'application/json',
|
|
'User-Agent': 'AccountForger/1.0.0'
|
|
})
|
|
|
|
def _make_request(self, method: str, endpoint: str,
|
|
data: Optional[Dict[str, Any]] = None,
|
|
params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""
|
|
Führt eine API-Anfrage aus.
|
|
|
|
Args:
|
|
method: HTTP-Methode (GET, POST, etc.)
|
|
endpoint: API-Endpunkt (z.B. '/api/license/activate')
|
|
data: Request-Body (für POST/PUT)
|
|
params: Query-Parameter (für GET)
|
|
|
|
Returns:
|
|
Response als Dictionary
|
|
|
|
Raises:
|
|
requests.exceptions.RequestException: Bei Netzwerkfehlern
|
|
"""
|
|
url = urljoin(self.base_url, endpoint)
|
|
|
|
try:
|
|
logger.debug(f"API Request: {method} {url}")
|
|
logger.debug(f"Headers: {self.session.headers}")
|
|
if data:
|
|
logger.debug(f"Request Body: {data}")
|
|
|
|
response = self.session.request(
|
|
method=method,
|
|
url=url,
|
|
json=data,
|
|
params=params,
|
|
timeout=30
|
|
)
|
|
|
|
logger.debug(f"API Response: {response.status_code}")
|
|
|
|
# Bei 401 ist der API Key ungültig
|
|
if response.status_code == 401:
|
|
error_data = response.json() if response.text else {}
|
|
logger.error(f"API Key ungültig: {error_data.get('error', 'Unauthorized')}")
|
|
return {
|
|
'success': False,
|
|
'error': error_data.get('error', 'Invalid or missing API key'),
|
|
'code': 'INVALID_API_KEY',
|
|
'status': 401
|
|
}
|
|
|
|
# Erfolgreiche Responses
|
|
if response.status_code in [200, 201]:
|
|
return {
|
|
'success': True,
|
|
'data': response.json(),
|
|
'status': response.status_code
|
|
}
|
|
|
|
# Andere Fehler
|
|
try:
|
|
error_data = response.json()
|
|
return {
|
|
'success': False,
|
|
'error': error_data.get('error', f'HTTP {response.status_code}'),
|
|
'code': error_data.get('code', 'UNKNOWN_ERROR'),
|
|
'status': response.status_code,
|
|
'data': error_data
|
|
}
|
|
except:
|
|
return {
|
|
'success': False,
|
|
'error': f'HTTP {response.status_code}: {response.text}',
|
|
'code': 'HTTP_ERROR',
|
|
'status': response.status_code
|
|
}
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error(f"Timeout bei Anfrage an {url}")
|
|
return {
|
|
'success': False,
|
|
'error': 'Request timeout',
|
|
'code': 'TIMEOUT',
|
|
'status': 0
|
|
}
|
|
except requests.exceptions.ConnectionError:
|
|
logger.error(f"Verbindungsfehler bei Anfrage an {url}")
|
|
return {
|
|
'success': False,
|
|
'error': 'Connection error',
|
|
'code': 'CONNECTION_ERROR',
|
|
'status': 0
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Unerwarteter Fehler bei API-Anfrage: {e}")
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'code': 'UNKNOWN_ERROR',
|
|
'status': 0
|
|
}
|
|
|
|
def activate_license(self, license_key: str, hardware_hash: str,
|
|
machine_name: str, app_version: str = "1.0.0") -> Dict[str, Any]:
|
|
"""
|
|
Aktiviert eine Lizenz auf einem neuen System.
|
|
|
|
Args:
|
|
license_key: Lizenzschlüssel (XXXX-XXXX-XXXX-XXXX)
|
|
hardware_hash: Eindeutiger Hardware-Identifier
|
|
machine_name: Name des Computers
|
|
app_version: Version der Anwendung
|
|
|
|
Returns:
|
|
API Response
|
|
"""
|
|
data = {
|
|
'license_key': license_key,
|
|
'hardware_fingerprint': hardware_hash, # Neuer Parameter-Name
|
|
'machine_name': machine_name, # Neuer Parameter-Name
|
|
'app_version': app_version
|
|
}
|
|
|
|
return self._make_request('POST', '/api/license/activate', data=data)
|
|
|
|
def verify_license(self, license_key: str, hardware_hash: str,
|
|
machine_id: str, activation_id: int,
|
|
app_version: str = "1.0.0") -> Dict[str, Any]:
|
|
"""
|
|
Verifiziert eine aktive Lizenz.
|
|
|
|
Args:
|
|
license_key: Lizenzschlüssel
|
|
hardware_hash: Hardware-Identifier
|
|
machine_id: Maschinen-ID
|
|
activation_id: Aktivierungs-ID (von activate_license)
|
|
app_version: Version der Anwendung
|
|
|
|
Returns:
|
|
API Response
|
|
"""
|
|
data = {
|
|
'license_key': license_key,
|
|
'hardware_fingerprint': hardware_hash, # Neuer Parameter-Name
|
|
'machine_name': machine_id, # Neuer Parameter-Name
|
|
'activation_id': activation_id,
|
|
'app_version': app_version
|
|
}
|
|
|
|
return self._make_request('POST', '/api/license/verify', data=data)
|
|
|
|
def get_license_info(self, license_key: str) -> Dict[str, Any]:
|
|
"""
|
|
Holt Informationen zu einer Lizenz.
|
|
|
|
Args:
|
|
license_key: Lizenzschlüssel
|
|
|
|
Returns:
|
|
API Response
|
|
"""
|
|
return self._make_request('GET', f'/api/license/info/{license_key}')
|
|
|
|
def start_session(self, license_key: str, machine_id: str,
|
|
hardware_hash: str, version: str = "1.0.0", ip_address: str = None) -> Dict[str, Any]:
|
|
"""
|
|
Startet eine neue Session für eine Lizenz.
|
|
|
|
Args:
|
|
license_key: Lizenzschlüssel
|
|
machine_id: Maschinen-ID (z.B. Computername)
|
|
hardware_hash: Hardware-Identifier
|
|
version: App-Version
|
|
ip_address: IP-Adresse des Clients (NEU)
|
|
|
|
Returns:
|
|
API Response mit session_token
|
|
"""
|
|
data = {
|
|
'license_key': license_key,
|
|
# Neue Parameter-Namen
|
|
'machine_name': machine_id,
|
|
'hardware_fingerprint': hardware_hash,
|
|
# Alte Parameter-Namen für Backward Compatibility
|
|
'machine_id': machine_id,
|
|
'hardware_id': hardware_hash, # Server erwartet beide Hardware-Felder
|
|
'hardware_hash': hardware_hash,
|
|
'version': version
|
|
}
|
|
|
|
# IP-Adresse hinzufügen wenn vorhanden
|
|
if ip_address:
|
|
data['ip_address'] = ip_address
|
|
|
|
return self._make_request('POST', '/api/license/session/start', data=data)
|
|
|
|
def session_heartbeat(self, session_token: str, license_key: str) -> Dict[str, Any]:
|
|
"""
|
|
Sendet einen Heartbeat für eine aktive Session.
|
|
|
|
Args:
|
|
session_token: Session Token von start_session
|
|
license_key: Lizenzschlüssel
|
|
|
|
Returns:
|
|
API Response
|
|
"""
|
|
data = {
|
|
'session_token': session_token,
|
|
'license_key': license_key
|
|
}
|
|
|
|
return self._make_request('POST', '/api/license/session/heartbeat', data=data)
|
|
|
|
def end_session(self, session_token: str) -> Dict[str, Any]:
|
|
"""
|
|
Beendet eine aktive Session.
|
|
|
|
Args:
|
|
session_token: Session Token
|
|
|
|
Returns:
|
|
API Response
|
|
"""
|
|
data = {
|
|
'session_token': session_token
|
|
}
|
|
|
|
return self._make_request('POST', '/api/license/session/end', data=data)
|
|
|
|
def check_version(self, current_version: str, license_key: str) -> Dict[str, Any]:
|
|
"""
|
|
Prüft auf verfügbare Updates.
|
|
|
|
Args:
|
|
current_version: Aktuelle Version der App
|
|
license_key: Lizenzschlüssel
|
|
|
|
Returns:
|
|
API Response mit Update-Info
|
|
"""
|
|
data = {
|
|
'current_version': current_version,
|
|
'license_key': license_key
|
|
}
|
|
|
|
return self._make_request('POST', '/api/version/check', data=data)
|
|
|
|
def get_latest_version(self) -> Dict[str, Any]:
|
|
"""
|
|
Holt Informationen zur neuesten Version.
|
|
|
|
Returns:
|
|
API Response
|
|
"""
|
|
return self._make_request('GET', '/api/version/latest')
|
|
|
|
def test_connection(self) -> Dict[str, Any]:
|
|
"""
|
|
Testet die Verbindung zum Server.
|
|
|
|
Returns:
|
|
API Response
|
|
"""
|
|
try:
|
|
response = self.session.get(
|
|
urljoin(self.base_url, '/health'),
|
|
timeout=10
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return {
|
|
'success': True,
|
|
'data': response.json(),
|
|
'status': 200
|
|
}
|
|
else:
|
|
return {
|
|
'success': False,
|
|
'error': f'Server returned status {response.status_code}',
|
|
'status': response.status_code
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'status': 0
|
|
}
|
|
|
|
|
|
# Test-Funktion
|
|
if __name__ == "__main__":
|
|
# Logging konfigurieren
|
|
logging.basicConfig(
|
|
level=logging.DEBUG,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
# API Client erstellen
|
|
client = LicenseAPIClient()
|
|
|
|
print("=== License Server API Client Test ===\n")
|
|
|
|
# 1. Verbindung testen
|
|
print("1. Teste Verbindung zum Server...")
|
|
result = client.test_connection()
|
|
print(f" Ergebnis: {'✓' if result['success'] else '✗'}")
|
|
if result['success']:
|
|
print(f" Server Status: {result['data']}")
|
|
else:
|
|
print(f" Fehler: {result['error']}")
|
|
|
|
print("\n2. Teste API Key Authentifizierung...")
|
|
# Versuche License Info zu holen (benötigt gültigen API Key)
|
|
test_license = "TEST-TEST-TEST-TEST"
|
|
result = client.get_license_info(test_license)
|
|
print(f" Ergebnis: {'✓' if result['success'] else '✗'}")
|
|
if not result['success']:
|
|
print(f" Status: {result['status']}")
|
|
print(f" Fehler: {result['error']}")
|
|
if result['status'] == 401:
|
|
print(" → API Key wird abgelehnt!")
|
|
else:
|
|
print(f" → API Key wird akzeptiert!")
|
|
|
|
print("\n=== Test abgeschlossen ===") |