100 lines
4.2 KiB
Python
100 lines
4.2 KiB
Python
"""Independent Tor lifecycle observer; never expands the FD6 HostWatch budget."""
|
|
import array
|
|
import json
|
|
import os
|
|
import socket
|
|
import struct
|
|
from c01_host_watch import now
|
|
|
|
|
|
def exact_object(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise ValueError('duplicate Tor progress key')
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
class Watch:
|
|
def __init__(self, control, application, pid):
|
|
self.control, self.application, self.pid = control, application, pid
|
|
control.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
control.setblocking(False)
|
|
self.state, self.sequence, self.last_mutation = 'starting', 0, 0
|
|
self.deadline = now() + 2_000_000_000
|
|
self.reason = None
|
|
|
|
def revoke(self, reason):
|
|
if self.state != 'failed':
|
|
self.state, self.reason = 'failed', reason
|
|
try:
|
|
self.application.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
self.control.close()
|
|
|
|
def release_reaped(self):
|
|
self.revoke('reaped')
|
|
self.application.close()
|
|
|
|
def receive(self):
|
|
if self.state == 'failed':
|
|
return False
|
|
stamp = now()
|
|
if self.deadline is not None and stamp >= self.deadline:
|
|
self.revoke('deadline')
|
|
return False
|
|
try:
|
|
raw, anc, flags, _ = self.control.recvmsg(
|
|
4097, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(128),
|
|
socket.MSG_DONTWAIT | socket.MSG_CMSG_CLOEXEC)
|
|
except BlockingIOError:
|
|
return True
|
|
except OSError:
|
|
self.revoke('closed')
|
|
return False
|
|
credentials, unexpected = [], False
|
|
for level, kind, data in anc:
|
|
if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS):
|
|
rights = array.array('i')
|
|
rights.frombytes(data[:len(data) - len(data) % rights.itemsize])
|
|
for fd in rights:
|
|
os.close(fd)
|
|
unexpected = True
|
|
elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS) and len(data) == 12:
|
|
credentials.append(struct.unpack('3i', data))
|
|
else:
|
|
unexpected = True
|
|
try:
|
|
if unexpected or credentials != [(self.pid, 0, 0)] or flags & ~socket.MSG_CMSG_CLOEXEC or not 0 < len(raw) <= 4096:
|
|
raise ValueError('Tor progress peer/framing')
|
|
value = json.loads(raw, object_pairs_hook=exact_object)
|
|
if type(value) is not dict or set(value) != {'version', 'sequence', 'deadline_boottime_ns', 'phase'}:
|
|
raise ValueError('Tor progress fields')
|
|
version, sequence, deadline, phase = (value[k] for k in
|
|
('version', 'sequence', 'deadline_boottime_ns', 'phase'))
|
|
if type(version) is not int or version != 1 or type(sequence) is not int or not 0 <= sequence < 2**64 or type(deadline) is not int or not 0 <= deadline < 2**64:
|
|
raise ValueError('Tor progress numeric bounds')
|
|
if phase == 'Ready':
|
|
if self.state != 'starting' or sequence != 0 or deadline != 0:
|
|
raise ValueError('Tor readiness order')
|
|
self.state, self.deadline = 'idle', None
|
|
elif phase == 'Begin':
|
|
if self.state != 'idle' or not stamp < deadline <= stamp + 60_000_000_000:
|
|
raise ValueError('Tor admission deadline/order')
|
|
if sequence != 0 and sequence < self.last_mutation:
|
|
raise ValueError('Tor mutation sequence regression')
|
|
self.state, self.sequence, self.deadline = 'busy', sequence, deadline
|
|
elif phase == 'Complete':
|
|
if self.state != 'busy' or (sequence, deadline) != (self.sequence, self.deadline):
|
|
raise ValueError('Tor completion mismatch')
|
|
if sequence:
|
|
self.last_mutation = sequence
|
|
self.state, self.deadline = 'idle', None
|
|
else:
|
|
raise ValueError('Tor uncertain or unknown phase')
|
|
except (ValueError, TypeError, UnicodeError):
|
|
self.revoke('protocol or uncertain outcome')
|
|
return self.state != 'failed'
|