54 lines
2.2 KiB
Python
54 lines
2.2 KiB
Python
"""Access policy: persisted confirmed mode, monotonic trial deadline."""
|
|
import secrets
|
|
import time
|
|
|
|
class Policy:
|
|
MODES = ('lan', 'tunnel', 'both')
|
|
|
|
def __init__(self, mode='lan', clock=time.monotonic, persist=lambda mode: None):
|
|
self.mode = mode if mode in self.MODES else 'lan'
|
|
self.clock = clock
|
|
self.persist = persist
|
|
self.pending = None
|
|
|
|
def expire(self):
|
|
if self.pending and self.clock() >= self.pending['deadline']:
|
|
self.mode = self.pending['previous']
|
|
self.pending = None
|
|
|
|
def allowed(self, ingress):
|
|
self.expire()
|
|
return self.mode == 'both' or self.mode == ingress
|
|
|
|
def propose(self, mode, connected):
|
|
self.expire()
|
|
if mode not in self.MODES:
|
|
raise ValueError('Ungültiger Zugriffsmodus.')
|
|
if self.pending:
|
|
raise ValueError('Zuerst den laufenden Zugriffsversuch bestätigen oder zurücknehmen.')
|
|
if mode == self.mode:
|
|
return
|
|
if mode in ('tunnel', 'both') and not connected:
|
|
raise ValueError('Zuerst WireGuard verbinden und einen aktuellen Handshake abwarten.')
|
|
self.pending = dict(previous=self.mode, target=mode, deadline=self.clock()+120, id=secrets.token_urlsafe(24))
|
|
self.mode = mode
|
|
|
|
def confirm(self, trial_id, ingress):
|
|
self.expire()
|
|
if not self.pending or not secrets.compare_digest(self.pending['id'], str(trial_id)):
|
|
raise ValueError('Kein passender laufender Zugriffsversuch.')
|
|
needed = 'tunnel' if self.mode in ('tunnel', 'both') else 'lan'
|
|
if ingress != needed:
|
|
raise ValueError('Bestätigung muss über den ausgewählten Netzwerkzugang erfolgen.')
|
|
self.persist(self.mode)
|
|
self.pending = None
|
|
|
|
def cancel(self):
|
|
if self.pending:
|
|
self.mode = self.pending['previous']
|
|
self.pending = None
|
|
|
|
def status(self):
|
|
self.expire()
|
|
return dict(mode=self.mode, pending=None if not self.pending else dict(id=self.pending['id'], target=self.pending['target'], previous=self.pending['previous'], seconds_remaining=max(0, int(self.pending['deadline']-self.clock()))))
|