network.md 20 KB
trixy_core.network
Netzwerk-Protokoll, Verschlüsselung, Security und TLS.
Übersicht
from trixy_core.network import (
# Protocol
TrixyProtocol, ProtocolMessage, ProtocolFlags,
# Encryption
TrixyEncryption,
# Commands
SatelliteConnect, SatelliteAccepted, WakewordDetected,
# Security
AccessController, ConnectionRateLimiter, DDoSProtection,
# TLS
TLSConfig, TLSManager, CertificateGenerator,
# Pool
ConnectionPool, Connection, PoolConfig,
# Retry
NetworkRetry, RetryConfig, RetryStrategy, with_retry,
# KeepAlive
KeepAliveManager, KeepAliveSession, KeepAliveConfig, ConnectionState,
# Compression
Compressor, CompressionLevel, ZlibCompressor, GzipCompressor,
CompressionRegistry, get_compressor,
# RPC
CorrelationManager, CorrelationConfig, PendingRequest, RequestState,
)
Trixy-Protokoll
Binäres Nachrichtenprotokoll für Satellite-Kommunikation.
Nachrichtenformat
┌────────────────────────────────────────────────────────────┐
│ TRIXY PROTOCOL MESSAGE │
├──────────────┬──────────────┬──────────────┬───────────────┤
│ Magic (4B) │ Version (1B) │ Flags (4B) │ Timestamp (8B)│
│ "TRXI" │ 0x01 │ │ │
├──────────────┴──────────────┴──────────────┴───────────────┤
│ Checksum (16B) │
│ MD5 Hash │
├─────────────────────────────────────────────────────────────┤
│ Class Name Length (4B) │
├─────────────────────────────────────────────────────────────┤
│ Class Name (var) │
├─────────────────────────────────────────────────────────────┤
│ Payload Length (4B) │
├─────────────────────────────────────────────────────────────┤
│ Payload (var) │
│ (Serialized Message Data) │
└─────────────────────────────────────────────────────────────┘
TrixyProtocol
from trixy_core.network import TrixyProtocol, ProtocolMessage
protocol = TrixyProtocol()
# Nachricht erstellen
message = ProtocolMessage(
class_name="SatelliteConnect",
payload={"room_id": "living_room", "alias": "Wohnzimmer"},
)
# Serialisieren
data = protocol.serialize(message)
# Deserialisieren
received = protocol.deserialize(data)
print(received.class_name) # "SatelliteConnect"
print(received.payload) # {"room_id": "living_room", ...}
ProtocolFlags
from trixy_core.network import ProtocolFlags
class ProtocolFlags(IntFlag):
NONE = 0
ENCRYPTED = 1 # Nachricht ist verschlüsselt
COMPRESSED = 2 # Nachricht ist komprimiert
FRAGMENTED = 4 # Teil einer größeren Nachricht
REQUIRES_ACK = 8 # Bestätigung erforderlich
PRIORITY = 16 # Hohe Priorität
Verschlüsselung
AES-256-GCM Verschlüsselung für sichere Kommunikation.
from trixy_core.network import TrixyEncryption
# Verschlüsselung erstellen
encryption = TrixyEncryption(key="32_byte_encryption_key_here!")
# Verschlüsseln
plaintext = b"Geheime Nachricht"
ciphertext = encryption.encrypt(plaintext)
# Entschlüsseln
decrypted = encryption.decrypt(ciphertext)
assert decrypted == plaintext
# Mit Protokoll verwenden
protocol = TrixyProtocol(encryption=encryption)
Command-Nachrichten
Vordefinierte Nachrichtentypen für Satellite-Kommunikation.
Satellite-Befehle
from trixy_core.network.cmd import (
SatelliteConnect,
SatelliteAccepted,
SatelliteDenied,
SatelliteDisconnect,
)
# Verbindungsanfrage
connect = SatelliteConnect(
room_id="living_room",
mac_address="AA:BB:CC:DD:EE:FF",
alias="Wohnzimmer",
capabilities=["audio_in", "audio_out", "wakeword"],
)
# Verbindung akzeptiert
accepted = SatelliteAccepted(
session_id="abc123",
server_version="1.0.0",
)
# Verbindung abgelehnt
denied = SatelliteDenied(
reason="Nicht registriert",
code=403,
)
Wakeword-Befehle
from trixy_core.network.cmd import (
WakewordDetected,
WakewordSelected,
WakewordAbort,
)
# Wakeword erkannt
detected = WakewordDetected(
wakeword="trixy",
confidence=0.95,
audio_level=0.7,
timestamp=datetime.now(),
)
# Wakeword ausgewählt (nach Arbitration)
selected = WakewordSelected(
satellite_id="living_room",
session_id="sess_123",
)
# Wakeword abgebrochen
abort = WakewordAbort(
reason="Anderer Satellite ausgewählt",
)
Recording-Befehle
from trixy_core.network.cmd import (
RecordingStart,
RecordingStop,
RecordingDone,
)
# Aufnahme starten
start = RecordingStart(
session_id="sess_123",
max_duration=60.0,
)
# Aufnahme stoppen
stop = RecordingStop(
reason="silence",
duration=5.2,
)
# Aufnahme abgeschlossen
done = RecordingDone(
audio_data=audio_bytes,
duration=5.2,
sample_rate=16000,
)
Security
AccessController
Zugriffskontrolle basierend auf IP, MAC und Regeln.
from trixy_core.network.security import (
AccessController,
AccessRule,
IPMatcher,
CIDRMatcher,
MACMatcher,
)
controller = AccessController()
# Regeln hinzufügen
controller.add_rule(AccessRule(
name="local_network",
matcher=CIDRMatcher("192.168.1.0/24"),
action="allow",
))
controller.add_rule(AccessRule(
name="known_devices",
matcher=MACMatcher(["AA:BB:CC:DD:EE:FF", "11:22:33:44:55:66"]),
action="allow",
))
controller.add_rule(AccessRule(
name="block_all",
matcher=IPMatcher("0.0.0.0/0"),
action="deny",
))
# Zugriff prüfen
decision = controller.check(
ip="192.168.1.100",
mac="AA:BB:CC:DD:EE:FF",
)
if decision.allowed:
accept_connection()
else:
reject_connection(decision.reason)
ConnectionRateLimiter
Rate-Limiting für Verbindungen.
from trixy_core.network.security import (
ConnectionRateLimiter,
RateLimitConfig,
)
config = RateLimitConfig(
max_connections_per_ip=10,
max_connections_per_second=5,
burst_size=20,
)
limiter = ConnectionRateLimiter(config)
# Verbindung prüfen
result = limiter.check(ip="192.168.1.100")
if result.allowed:
accept_connection()
else:
reject_connection(
reason="Rate limit exceeded",
retry_after=result.retry_after,
)
DDoSProtection
Schutz vor DDoS-Angriffen.
from trixy_core.network.security import (
DDoSProtection,
DDoSConfig,
ThreatLevel,
)
config = DDoSConfig(
connection_threshold=100, # Verbindungen pro Sekunde
request_threshold=1000, # Requests pro Sekunde
block_duration=300, # Blockdauer in Sekunden
whitelist=["192.168.1.0/24"],
)
protection = DDoSProtection(config)
# Verbindung verarbeiten
async def handle_connection(reader, writer):
ip = writer.get_extra_info("peername")[0]
threat = protection.assess(ip)
if threat.level >= ThreatLevel.HIGH:
writer.close()
return
if threat.level >= ThreatLevel.MEDIUM:
# Rate-Limiting aktivieren
await asyncio.sleep(1.0)
# Normal verarbeiten
await process_connection(reader, writer)
# Statistiken
stats = protection.get_stats()
print(f"Blocked IPs: {stats.blocked_count}")
print(f"Current threat level: {stats.threat_level}")
TLS/SSL
TLSConfig
from trixy_core.network.tls import TLSConfig, TLSVersion
config = TLSConfig(
cert_file="certs/server.crt",
key_file="certs/server.key",
ca_file="certs/ca.crt",
min_version=TLSVersion.TLS_1_2,
verify_client=True,
ciphers=[
"ECDHE-ECDSA-AES256-GCM-SHA384",
"ECDHE-RSA-AES256-GCM-SHA384",
],
)
TLSManager
from trixy_core.network.tls import TLSManager
manager = TLSManager(config)
# SSL-Context erstellen
server_ctx = manager.create_server_context()
client_ctx = manager.create_client_context()
# Mit asyncio verwenden
server = await asyncio.start_server(
handle_client,
host="0.0.0.0",
port=2101,
ssl=server_ctx,
)
# Zertifikat-Info
info = manager.get_certificate_info()
print(f"CN: {info.common_name}")
print(f"Gültig bis: {info.not_after}")
CertificateGenerator
Selbstsignierte Zertifikate generieren (für Entwicklung).
from trixy_core.network.tls import (
CertificateGenerator,
CertificateRequest,
KeyType,
)
generator = CertificateGenerator()
request = CertificateRequest(
common_name="trixy.local",
organization="Trixy",
country="DE",
validity_days=365,
key_type=KeyType.RSA,
key_size=4096,
san=["localhost", "192.168.1.100"],
)
# Generieren
result = generator.generate(request)
# Speichern
result.save_cert("certs/server.crt")
result.save_key("certs/server.key")
Connection Pool
Wiederverwendbare Verbindungen.
from trixy_core.network.pool import (
ConnectionPool,
PoolConfig,
Connection,
ConnectionState,
)
config = PoolConfig(
min_connections=5,
max_connections=50,
idle_timeout=300,
max_lifetime=3600,
health_check_interval=30,
)
pool = ConnectionPool(config)
# Verbindung abrufen
async with pool.acquire() as conn:
await conn.send(data)
response = await conn.receive()
# Oder manuell
conn = await pool.acquire()
try:
await conn.send(data)
finally:
await pool.release(conn)
# Pool-Statistiken
stats = pool.get_statistics()
print(f"Active: {stats.active}")
print(f"Idle: {stats.idle}")
print(f"Total: {stats.total}")
# Pool schließen
await pool.close()
Health-Checks
from trixy_core.network.pool import HealthChecker, HealthCheckConfig
config = HealthCheckConfig(
interval=30.0,
timeout=5.0,
unhealthy_threshold=3,
healthy_threshold=2,
)
checker = HealthChecker(config)
@checker.check
async def ping(conn: Connection) -> bool:
try:
await conn.send(b"PING")
response = await asyncio.wait_for(conn.receive(), timeout=5.0)
return response == b"PONG"
except Exception:
return False
# Health-Check starten
await checker.start(pool)
Retry
Netzwerk-Retry mit verschiedenen Backoff-Strategien.
RetryConfig
from trixy_core.network.retry import (
NetworkRetry,
RetryConfig,
RetryStrategy,
with_retry,
)
config = RetryConfig(
max_attempts=5,
initial_delay=1.0,
max_delay=60.0,
backoff_multiplier=2.0,
strategy=RetryStrategy.EXPONENTIAL,
jitter=True,
retryable_exceptions=[ConnectionError, TimeoutError],
)
RetryStrategy
from trixy_core.network.retry import RetryStrategy
class RetryStrategy(Enum):
CONSTANT = "constant" # Immer gleiche Verzögerung
LINEAR = "linear" # delay * attempt
EXPONENTIAL = "exponential" # delay * (multiplier ^ attempt)
FIBONACCI = "fibonacci" # Fibonacci-Sequenz
DECORRELATED = "decorrelated" # Randomisiert (AWS-Stil)
NetworkRetry
from trixy_core.network.retry import NetworkRetry
retry = NetworkRetry(config)
# Mit Context-Manager
async with retry:
response = await make_request()
# Mit execute()
result = await retry.execute(make_request)
# Statistiken
stats = retry.stats
print(f"Versuche: {stats.total_attempts}")
print(f"Erfolge: {stats.successful_attempts}")
print(f"Fehler: {stats.failed_attempts}")
@with_retry Decorator
from trixy_core.network.retry import with_retry, RetryConfig
@with_retry(max_attempts=3, strategy=RetryStrategy.EXPONENTIAL)
async def fetch_data(url: str) -> dict:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
# Oder mit vollständiger Config
config = RetryConfig(max_attempts=5, initial_delay=0.5)
@with_retry(config=config)
async def reliable_operation():
...
Vordefinierte Configs
from trixy_core.network.retry import (
AGGRESSIVE_RETRY, # Viele Versuche, kurze Delays
CONSERVATIVE_RETRY, # Wenige Versuche, lange Delays
FAST_FAIL_RETRY, # Schnelles Aufgeben
)
retry = NetworkRetry(AGGRESSIVE_RETRY)
KeepAlive
Verbindungs-Monitoring mit Ping/Pong.
KeepAliveConfig
from trixy_core.network.keepalive import (
KeepAliveManager,
KeepAliveSession,
KeepAliveConfig,
ConnectionState,
ConnectionStats,
)
config = KeepAliveConfig(
ping_interval=30.0, # Sekunden zwischen Pings
pong_timeout=10.0, # Timeout für Pong
max_missed_pongs=3, # Max verpasste Pongs bis Disconnect
initial_delay=5.0, # Verzögerung vor erstem Ping
adaptive_interval=True, # Interval an RTT anpassen
)
KeepAliveManager
from trixy_core.network.keepalive import KeepAliveManager
manager = KeepAliveManager(default_config=config)
# Verbindung registrieren
async def ping_function():
await connection.send(b"PING")
session = await manager.register(
connection_id="conn-001",
ping_func=ping_function,
auto_start=True,
)
# Pong empfangen
manager.handle_pong("conn-001")
# Callbacks registrieren
async def on_timeout(conn_id: str):
print(f"Timeout: {conn_id}")
async def on_disconnect(conn_id: str):
print(f"Disconnected: {conn_id}")
manager.on_timeout(on_timeout)
manager.on_disconnect(on_disconnect)
# Statistiken
stats = manager.get_stats("conn-001")
print(f"Pings gesendet: {stats.pings_sent}")
print(f"Pongs empfangen: {stats.pongs_received}")
print(f"RTT: {stats.last_rtt_ms}ms")
# Alle Sessions
all_stats = manager.get_all_stats()
summary = manager.get_summary()
# Cleanup
stale = manager.get_stale_connections()
disconnected = manager.get_disconnected()
removed = await manager.cleanup()
# Beenden
await manager.unregister("conn-001")
await manager.stop_all()
ConnectionState
from trixy_core.network.keepalive import ConnectionState
class ConnectionState(Enum):
CONNECTED = "connected" # Normal verbunden
DISCONNECTED = "disconnected" # Getrennt
STALE = "stale" # Keine Antwort
RECONNECTING = "reconnecting" # Wiederverbinden
Compression
Datenkomprimierung für Netzwerk-Traffic.
CompressionLevel
from trixy_core.network.compression import (
Compressor,
CompressionLevel,
CompressionStats,
ZlibCompressor,
GzipCompressor,
CompressionRegistry,
get_compressor,
)
class CompressionLevel(IntEnum):
NONE = 0 # Keine Komprimierung
FASTEST = 1 # Schnellste
DEFAULT = 6 # Standard
BEST = 9 # Beste Kompression
ZlibCompressor
from trixy_core.network.compression import ZlibCompressor
compressor = ZlibCompressor(level=CompressionLevel.DEFAULT)
# Komprimieren
data = b"Hello World! " * 100
compressed = compressor.compress(data)
# Dekomprimieren
decompressed = compressor.decompress(compressed)
assert decompressed == data
# Mit Header (Größen-Präfix)
with_header = compressor.compress_with_header(data)
original = compressor.decompress_with_header(with_header)
# Prüfen ob Komprimierung sinnvoll
if compressor.should_compress(data):
result, was_compressed = compressor.compress_if_beneficial(data)
# Statistiken
stats = compressor.stats
print(f"Komprimiert: {stats.total_compressed}")
print(f"Ratio: {stats.compression_ratio:.2%}")
print(f"Ersparnis: {stats.space_saved_percent:.1f}%")
compressor.reset_stats()
GzipCompressor
from trixy_core.network.compression import GzipCompressor
compressor = GzipCompressor(level=CompressionLevel.BEST)
# Gleiche API wie ZlibCompressor
compressed = compressor.compress(data)
decompressed = compressor.decompress(compressed)
CompressionRegistry
from trixy_core.network.compression import (
CompressionRegistry,
get_default_registry,
get_compressor,
)
# Default-Registry mit zlib und gzip
registry = get_default_registry()
# Compressor abrufen
zlib = registry.get("zlib")
gzip = registry.get("gzip")
# Nach Content-Type
compressor = registry.get_by_content_type("application/gzip")
# Verfügbare auflisten
available = registry.list_available() # ["zlib", "gzip"]
# Convenience-Funktion
compressor = get_compressor("zlib", CompressionLevel.BEST)
# Eigenen Compressor registrieren
registry.register("custom", CustomCompressor, "application/x-custom")
RPC Correlation
Request-Response Matching für asynchrone Kommunikation.
CorrelationConfig
from trixy_core.network.rpc import (
CorrelationManager,
CorrelationConfig,
PendingRequest,
RequestState,
)
config = CorrelationConfig(
default_timeout=30.0, # Standard-Timeout
cleanup_interval=60.0, # Cleanup-Intervall
max_pending=10000, # Max. ausstehende Requests
id_generator=None, # Eigener ID-Generator
)
CorrelationManager
from trixy_core.network.rpc import CorrelationManager
manager = CorrelationManager[ResponseType](config=config)
# Request erstellen
correlation_id = await manager.create_request(
timeout=60.0,
metadata={"user": "test"},
)
# Nachricht senden
await send_message(correlation_id, payload)
# Auf Antwort warten
try:
response = await manager.wait_for_response(correlation_id)
print(f"Antwort: {response}")
except TimeoutError:
print("Timeout!")
# Request abschließen (bei Antwort-Empfang)
manager.complete(correlation_id, response_data)
# Request fehlschlagen lassen
manager.fail(correlation_id, ValueError("Fehler"))
# Request abbrechen
manager.cancel(correlation_id)
# Statistiken
stats = manager.get_stats()
print(f"Ausstehend: {stats['total_pending']}")
print(f"States: {stats['states']}")
# Cleanup
removed = await manager.cleanup_expired()
cancelled = await manager.cancel_all()
# Cleanup-Task
await manager.start_cleanup()
await manager.stop_cleanup()
RequestState
from trixy_core.network.rpc import RequestState
class RequestState(Enum):
PENDING = "pending" # Wartet auf Antwort
COMPLETED = "completed" # Erfolgreich abgeschlossen
TIMEOUT = "timeout" # Timeout
CANCELLED = "cancelled" # Abgebrochen
ERROR = "error" # Fehler
PendingRequest
from trixy_core.network.rpc import PendingRequest
request = manager.get(correlation_id)
print(request.correlation_id) # ID
print(request.state) # RequestState
print(request.is_complete) # True wenn fertig
print(request.elapsed_seconds) # Verstrichene Zeit
print(request.is_expired()) # Timeout überschritten?
print(request.metadata) # Metadaten
Custom ID-Generator
import uuid
def custom_generator() -> str:
return f"req-{uuid.uuid4().hex[:8]}"
config = CorrelationConfig(id_generator=custom_generator)
manager = CorrelationManager[str](config=config)