31 lines
1.9 KiB
Python
31 lines
1.9 KiB
Python
"""Authenticated Deck administration through a native host Unix socket only."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
import time
|
|
from network.rpc import request
|
|
|
|
class NetworkClient:
|
|
def __init__(self):
|
|
self.native=os.environ.get('DECK_NETWORK_MODE')=='native'
|
|
self.disabled=not self.native
|
|
self.local=False
|
|
self.job=None;self.lock=threading.Lock();self.cached=None;self.cached_at=0
|
|
def call(self,action,values=None,ingress='management'):
|
|
if self.disabled:raise ValueError('Nativen WireGuard-Systemdienst zuerst auf dem Debian-Host einrichten.')
|
|
token=Path(os.environ.get('DECK_PROXY_TOKEN_FILE','/run/athena-deck-network/proxy-token')).read_text().strip()
|
|
result=request(dict(values or {},action=action,ingress=ingress,proxy_token=token),path='/run/athena-deck-network/control.sock')
|
|
self.cached=None
|
|
return result
|
|
def status(self,ingress='management'):
|
|
if self.disabled:return dict(installed=False,install_supported=False,state='disabled-module',ingress=ingress,backend='native-systemd',error='Nativer WireGuard-Systemdienst ist noch nicht angebunden. Einrichtung siehe Installationsanleitung.')
|
|
with self.lock:
|
|
if self.cached is not None and time.monotonic()-self.cached_at<3:return dict(self.cached,ingress=ingress)
|
|
try:result=self.call('status',ingress=ingress);result['reachable']=True
|
|
except (ValueError,OSError,json.JSONDecodeError):result=dict(installed=False,install_supported=False,reachable=False,state='unavailable',error='Nativer WireGuard-Systemdienst nicht erreichbar.')
|
|
self.cached=result;self.cached_at=time.monotonic()
|
|
return dict(result,ingress=ingress)
|
|
def install(self,password=None,api_token=None):
|
|
raise ValueError('WireGuard wird nativ mit deploy/setup_network_helper.py eingerichtet. Kein Netzwerk-Container und keine zweite Deck-Instanz.')
|