Something went wrong. Try again.
GET /xrpc/tech.waow.typeahead.searchActors typeahead.waow.tech
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196#!/usr/bin/env python3"""Authenticated, bounded shadow comparisons. Never returns search results."""from concurrent.futures import ThreadPoolExecutorimport hmacfrom http.server import BaseHTTPRequestHandler, HTTPServerimport jsonimport osimport sqlite3import socketfrom pathlib import Pathimport threadingimport time
from compare import compare_query
class Canary: def __init__(self, binary, overlay, snapshot, state_file, expires_at, evaluator=compare_query, clock=time.time): self.binary, self.overlay, self.snapshot = binary, overlay, snapshot self.state_file, self.expires_at = state_file, expires_at self.evaluate, self.clock = evaluator, clock self.lock = threading.Lock() self.pool = ThreadPoolExecutor(max_workers=1) self.busy = False self.last_accepted = 0.0 self.state = {'paused': True, 'reason': 'initial', 'accepted': 0, 'passed': 0, 'failed': 0, 'recent': []} if state_file.exists(): saved = json.loads(state_file.read_text()) # Fail closed on restart, retaining the total sample budget. if not isinstance(saved.get('accepted'), int) or saved['accepted'] < 0: raise ValueError('Invalid persisted canary state') self.state = saved self.state.update(paused=True, reason='restart') self.save()
def save(self): temp = self.state_file.with_suffix('.tmp') temp.write_text(json.dumps(self.state)) temp.replace(self.state_file)
def status(self): with self.lock: return {**self.state, 'busy': self.busy, 'expires_at': self.expires_at, 'expired': self.clock() >= self.expires_at, 'max_samples': 120, 'min_interval_s': 15}
def control(self, enabled): with self.lock: if enabled and (self.clock() >= self.expires_at or self.state['accepted'] >= 120): return False self.state.update(paused=not enabled, reason='operator') self.save() return True
def submit(self, query): with self.lock: now = self.clock() if self.state['paused'] or now >= self.expires_at: return 503 if self.busy or now - self.last_accepted < 15 or self.state['accepted'] >= 120: return 429 self.busy = True self.last_accepted = now self.state['accepted'] += 1 self.save() self.pool.submit(self.run, query) return 202
def run(self, query): try: result = self.evaluate(self.binary, self.overlay, self.snapshot, query) except Exception: result = {'passed': False, 'error': 'supervisor_error'} # No query text, actor records, caller identifiers, or credentials retained. with self.lock: self.busy = False result['completed_at'] = self.clock() self.state['recent'] = (self.state['recent'] + [result])[-20:] if result.get('passed') is True: self.state['passed'] += 1 else: self.state['failed'] += 1 self.state.update(paused=True, reason='comparison_failed') self.save()
class CanaryHTTPServer(HTTPServer): address_family = socket.AF_INET6
class Handler(BaseHTTPRequestHandler): def setup(self): super().setup() self.connection.settimeout(1)
def log_message(self, *args): pass
def reply(self, code, value): body = json.dumps(value).encode() self.send_response(code) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(body))) self.send_header('Cache-Control', 'no-store') self.end_headers() self.wfile.write(body)
def authorized(self): expected = 'Bearer ' + self.server.token if not hmac.compare_digest(self.headers.get('Authorization', '').encode(), expected.encode()): self.reply(401, {'error': 'unauthorized'}) return False return True
def do_GET(self): if self.path == '/health': self.reply(200, {'status': 'ok', 'role': 'shadow-canary'}) elif self.authorized(): if self.path == '/status': self.reply(200, self.server.canary.status()) else: self.reply(404, {'error': 'not_found'})
def do_POST(self): if not self.authorized(): return try: size = int(self.headers.get('Content-Length', '0')) if not 0 < size <= 2048 or self.headers.get('Transfer-Encoding'): self.reply(400, {'error': 'invalid_length'}) return value = json.loads(self.rfile.read(size)) if not isinstance(value, dict): raise ValueError('object required') except (ValueError, TimeoutError): self.reply(400, {'error': 'invalid_json'}) return if self.path == '/compare': query = value.get('q') if not isinstance(query, str) or '\0' in query: self.reply(400, {'error': 'invalid_query'}) return try: valid = len(query.encode('utf-8')) <= 256 except UnicodeEncodeError: valid = False if not valid: self.reply(400, {'error': 'invalid_query'}) return code = self.server.canary.submit(query) self.reply(code, {'accepted': code == 202}) elif self.path == '/control' and type(value.get('enabled')) is bool: ok = self.server.canary.control(value['enabled']) self.reply(200 if ok else 409, {'updated': ok}) else: self.reply(404, {'error': 'not_found'})
def main(): token = os.environ['CANARY_TOKEN'] if len(token) < 32: raise ValueError('A dedicated canary token is required') expiry = int(os.environ['CANARY_EXPIRES_AT']) if not time.time() < expiry <= time.time() + 24 * 3600: raise ValueError('Canary expiry must be within the next 24 hours') paths = [Path(os.environ[k]) for k in ('CANARY_BINARY', 'CANARY_OVERLAY', 'CANARY_SNAPSHOT')] if any(not p.is_file() for p in paths): raise ValueError('Missing canary fixture or executable') with sqlite3.connect(paths[1].as_uri() + '?mode=ro', uri=True) as overlay: if not overlay.execute("SELECT 1 FROM sqlite_master WHERE type='index' AND name='canary_rank'").fetchone(): raise ValueError('Ranking index is missing') watermark = overlay.execute("SELECT value FROM sync_meta WHERE key='last_sync'").fetchone()[0] with sqlite3.connect(paths[2].as_uri() + '?mode=ro', uri=True) as snapshot: build = snapshot.execute("SELECT value FROM index_meta WHERE key='build_id'").fetchone()[0] state_file = Path(os.environ['CANARY_STATE_FILE']) canary = Canary(*paths, state_file, expiry) canary.state['fixture'] = {'build_id': build, 'overlay_watermark': watermark, 'frozen': True} canary.save() server = CanaryHTTPServer(('::', 8080), Handler) server.canary, server.token = canary, token server.timeout = 1 try: while time.time() < expiry: server.handle_request() finally: canary.control(False) server.server_close() canary.pool.shutdown(wait=True)
if __name__ == '__main__': main()