01b9f8097b
Page views and relay joins/session-lengths appended as JSON lines. Uniques via daily-salted hash — no IPs, no cookies, no client-side code, no third parties. tools/stats_report.py prints a per-day table. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
323 lines
11 KiB
Python
323 lines
11 KiB
Python
"""backnforth server — static site + WebSocket relay, Python stdlib only.
|
|
|
|
python3 server.py [--no-browser]
|
|
|
|
Serves the therapist page on http://<ip>:8000/ and the session relay on
|
|
ws://<ip>:8787. No pip installs, no virtualenv — any Python 3.7+ works,
|
|
including old machines that will never see a package index.
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import socket
|
|
import socketserver
|
|
import struct
|
|
import sys
|
|
import threading
|
|
import time
|
|
import webbrowser
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
HTTP_PORT = 8000
|
|
WS_PORT = 8787
|
|
_WS_GUID = b'258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
|
|
|
|
SESSIONS = {} # session id -> set of live WSHandler
|
|
LOCK = threading.Lock()
|
|
_SERVERS = []
|
|
TOKENS = None # None = open relay; a set = hosts must present a valid access code
|
|
STATS = None # path to a JSONL file; None = no metrics at all
|
|
|
|
|
|
def anon(ip):
|
|
"""Daily-salted hash: counts uniques per day without ever storing an IP."""
|
|
day = time.strftime('%Y-%m-%d')
|
|
return hashlib.sha1((day + '|' + (ip or '?')).encode()).hexdigest()[:10]
|
|
|
|
|
|
def log_stat(event, **fields):
|
|
if not STATS:
|
|
return
|
|
fields.update(t=int(time.time()), e=event)
|
|
try:
|
|
with open(STATS, 'a', encoding='utf-8') as f:
|
|
f.write(json.dumps(fields) + '\n')
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def local_ip():
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
s.connect(('8.8.8.8', 80))
|
|
ip = s.getsockname()[0]
|
|
s.close()
|
|
return ip
|
|
except OSError:
|
|
try:
|
|
return socket.gethostbyname(socket.gethostname())
|
|
except OSError:
|
|
return '127.0.0.1'
|
|
|
|
|
|
def shutdown_all():
|
|
for srv in _SERVERS:
|
|
try:
|
|
srv.shutdown()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class WSHandler(socketserver.BaseRequestHandler):
|
|
"""Minimal RFC6455 server: unfragmented text frames, ping/pong, close.
|
|
|
|
ponytail: no fragmentation/binary support — the state payloads are a few
|
|
hundred bytes of JSON and browsers never fragment at that size.
|
|
"""
|
|
|
|
def setup(self):
|
|
self.session_id = None
|
|
self.send_lock = threading.Lock()
|
|
self.hdrs = {}
|
|
self.role = None
|
|
self.joined_at = None
|
|
|
|
def _client_ip(self):
|
|
xff = self.hdrs.get(b'x-forwarded-for')
|
|
if xff:
|
|
return xff.decode('utf-8', 'replace').split(',')[0].strip()
|
|
return self.client_address[0]
|
|
|
|
def handle(self):
|
|
if not self._handshake():
|
|
return
|
|
try:
|
|
while True:
|
|
msg = self._read_message()
|
|
if msg is None:
|
|
break
|
|
self._dispatch(msg)
|
|
except (ConnectionError, OSError):
|
|
pass
|
|
|
|
def finish(self):
|
|
if self.session_id:
|
|
with LOCK:
|
|
SESSIONS.get(self.session_id, set()).discard(self)
|
|
if self.joined_at:
|
|
log_stat('session', role=self.role or '?',
|
|
secs=int(time.time() - self.joined_at))
|
|
|
|
# ---- relay protocol (same JSON messages as the old websockets server) ----
|
|
|
|
def _dispatch(self, text):
|
|
try:
|
|
data = json.loads(text)
|
|
except ValueError:
|
|
return
|
|
t = data.get('type')
|
|
if t == 'join':
|
|
sid = data.get('sessionId') or 'default'
|
|
if TOKENS is not None:
|
|
# hosts need an access code; patients may only join a LIVE session
|
|
# (they got the random room id from their therapist's link)
|
|
if data.get('role') == 'patient':
|
|
with LOCK:
|
|
live = bool(SESSIONS.get(sid))
|
|
if not live:
|
|
self._send_json({'type': 'join_denied', 'reason': 'session not live'})
|
|
return
|
|
elif data.get('token') not in TOKENS:
|
|
self._send_json({'type': 'join_denied', 'reason': 'access code required'})
|
|
return
|
|
self.session_id = sid
|
|
self.role = data.get('role') or '?'
|
|
self.joined_at = time.time()
|
|
with LOCK:
|
|
SESSIONS.setdefault(self.session_id, set()).add(self)
|
|
self._send_json({'type': 'joined', 'sessionId': self.session_id})
|
|
log_stat('join', role=self.role, u=anon(self._client_ip()))
|
|
elif t in ('state', 'patientViewport'):
|
|
if TOKENS is not None and not self.session_id:
|
|
return # hardened relay only forwards for joined connections
|
|
sid = (self.session_id if TOKENS is not None
|
|
else data.get('sessionId') or self.session_id or 'default')
|
|
out = json.dumps({'type': t, 'payload': data.get('payload')})
|
|
with LOCK:
|
|
peers = list(SESSIONS.get(sid, ()))
|
|
for peer in peers:
|
|
if peer is not self:
|
|
try:
|
|
peer._send_text(out)
|
|
except Exception:
|
|
pass
|
|
elif t in ('whoami', 'detect_ip'):
|
|
self._send_json({'type': 'detect_ip', 'ip': local_ip()})
|
|
elif t == 'ping':
|
|
self._send_json({'type': 'pong'})
|
|
elif t == 'shutdown':
|
|
if self.client_address[0] in ('127.0.0.1', '::1', local_ip()):
|
|
self._send_json({'type': 'shutdown_ack'})
|
|
threading.Thread(target=shutdown_all, daemon=True).start()
|
|
|
|
# ---- websocket plumbing ----
|
|
|
|
def _handshake(self):
|
|
self.request.settimeout(10)
|
|
data = b''
|
|
while b'\r\n\r\n' not in data:
|
|
chunk = self.request.recv(4096)
|
|
if not chunk or len(data) > 65536:
|
|
return False
|
|
data += chunk
|
|
headers = {}
|
|
for line in data.split(b'\r\n')[1:]:
|
|
if b':' in line:
|
|
k, v = line.split(b':', 1)
|
|
headers[k.strip().lower()] = v.strip()
|
|
self.hdrs = headers
|
|
key = headers.get(b'sec-websocket-key')
|
|
if not key:
|
|
return False
|
|
accept = base64.b64encode(hashlib.sha1(key + _WS_GUID).digest()).decode()
|
|
self.request.sendall((
|
|
'HTTP/1.1 101 Switching Protocols\r\n'
|
|
'Upgrade: websocket\r\n'
|
|
'Connection: Upgrade\r\n'
|
|
'Sec-WebSocket-Accept: ' + accept + '\r\n\r\n'
|
|
).encode())
|
|
self.request.settimeout(None)
|
|
return True
|
|
|
|
def _recv_exact(self, n):
|
|
buf = b''
|
|
while len(buf) < n:
|
|
chunk = self.request.recv(n - len(buf))
|
|
if not chunk:
|
|
return None
|
|
buf += chunk
|
|
return buf
|
|
|
|
def _read_message(self):
|
|
while True:
|
|
head = self._recv_exact(2)
|
|
if head is None:
|
|
return None
|
|
b1, b2 = head
|
|
opcode = b1 & 0x0F
|
|
length = b2 & 0x7F
|
|
if length == 126:
|
|
ext = self._recv_exact(2)
|
|
if ext is None:
|
|
return None
|
|
length = struct.unpack('>H', ext)[0]
|
|
elif length == 127:
|
|
ext = self._recv_exact(8)
|
|
if ext is None:
|
|
return None
|
|
length = struct.unpack('>Q', ext)[0]
|
|
if length > 1_000_000:
|
|
return None
|
|
mask = self._recv_exact(4) if (b2 & 0x80) else b'\x00\x00\x00\x00'
|
|
if mask is None:
|
|
return None
|
|
payload = self._recv_exact(length) if length else b''
|
|
if payload is None:
|
|
return None
|
|
payload = bytes(c ^ mask[i % 4] for i, c in enumerate(payload))
|
|
if opcode == 0x1: # text
|
|
try:
|
|
return payload.decode('utf-8')
|
|
except UnicodeDecodeError:
|
|
return None
|
|
elif opcode == 0x9: # ping -> pong
|
|
self._send_frame(0xA, payload)
|
|
elif opcode == 0x8: # close
|
|
try:
|
|
self._send_frame(0x8, b'')
|
|
except Exception:
|
|
pass
|
|
return None
|
|
# anything else (binary/continuation): skip
|
|
|
|
def _send_frame(self, opcode, payload):
|
|
n = len(payload)
|
|
head = bytes([0x80 | opcode])
|
|
if n < 126:
|
|
head += bytes([n])
|
|
elif n < 65536:
|
|
head += bytes([126]) + struct.pack('>H', n)
|
|
else:
|
|
head += bytes([127]) + struct.pack('>Q', n)
|
|
with self.send_lock:
|
|
self.request.sendall(head + payload)
|
|
|
|
def _send_text(self, text):
|
|
self._send_frame(0x1, text.encode('utf-8'))
|
|
|
|
def _send_json(self, obj):
|
|
self._send_text(json.dumps(obj))
|
|
|
|
|
|
class _QuietHTTP(SimpleHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def log_request(self, code='-', size='-'):
|
|
path = self.path.split('?')[0].split('#')[0]
|
|
if path.endswith('/') or path.endswith('.html'):
|
|
ip = (self.headers.get('X-Forwarded-For') or
|
|
self.client_address[0]).split(',')[0].strip()
|
|
log_stat('page', p=path, u=anon(ip))
|
|
|
|
|
|
def _argval(flag, default):
|
|
args = sys.argv[1:]
|
|
return args[args.index(flag) + 1] if flag in args else default
|
|
|
|
|
|
def main():
|
|
global TOKENS
|
|
os.chdir(os.path.dirname(os.path.abspath(__file__)))
|
|
bind = _argval('--bind', '0.0.0.0') # e.g. a tailnet IP behind a reverse proxy
|
|
http_port = int(_argval('--http-port', HTTP_PORT))
|
|
tokens_file = _argval('--tokens', None) # one access code per line, # comments ok
|
|
if tokens_file:
|
|
with open(tokens_file, encoding='utf-8') as f:
|
|
TOKENS = {l.strip() for l in f if l.strip() and not l.startswith('#')}
|
|
print('relay hardened: %d access codes loaded' % len(TOKENS))
|
|
global STATS
|
|
STATS = _argval('--stats', None) # JSONL metrics: pages, joins, session lengths
|
|
if STATS:
|
|
print('metrics -> %s (anonymous: daily-salted hashes, no IPs stored)' % STATS)
|
|
|
|
socketserver.ThreadingTCPServer.allow_reuse_address = True
|
|
ThreadingHTTPServer.allow_reuse_address = True
|
|
httpd = ThreadingHTTPServer((bind, http_port), _QuietHTTP)
|
|
wsd = socketserver.ThreadingTCPServer((bind, WS_PORT), WSHandler)
|
|
httpd.daemon_threads = True
|
|
wsd.daemon_threads = True
|
|
_SERVERS.extend([httpd, wsd])
|
|
|
|
ip = bind if bind not in ('0.0.0.0', '') else local_ip()
|
|
print('backnforth running:')
|
|
print(' Therapist page: http://%s:%d/' % (ip, http_port))
|
|
print(' Session relay: ws://%s:%d' % (ip, WS_PORT))
|
|
print('Press Ctrl+C to stop.')
|
|
|
|
threading.Thread(target=wsd.serve_forever, daemon=True).start()
|
|
if '--no-browser' not in sys.argv:
|
|
try:
|
|
webbrowser.open('http://%s:%d/easy.html' % (ip, http_port))
|
|
except Exception:
|
|
pass
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print('\nStopped.')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|