#!/usr/bin/env python3 from __future__ import annotations import configparser import base64 import hashlib import hmac import io import json import math import os import random import re import secrets import socket import struct import sys import time import urllib.error import urllib.parse import urllib.request import webbrowser import wave from html import escape as html_escape from pathlib import Path from urllib.parse import urlparse from threading import Timer from flask import Flask, Response, abort, g, jsonify, redirect, render_template_string, request, url_for def bundle_dir() -> Path: if getattr(sys, "frozen", False): return Path(getattr(sys, "_MEIPASS", Path(sys.executable).resolve().parent)) return Path(__file__).resolve().parent def writable_app_dir() -> Path: configured = ( os.environ.get("BLAME_MACHINE_DATA_DIR", "").strip() or os.environ.get("DECISION_ARCADE_DATA_DIR", "").strip() ) if configured: return Path(configured).expanduser() if getattr(sys, "frozen", False): return Path.home() / ".blame_machine" return Path(__file__).resolve().parent BUNDLE_DIR = bundle_dir() APP_DIR = writable_app_dir() APP_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) try: APP_DIR.chmod(0o700) except PermissionError: # Bind-mounted directories may be managed by the host instead. pass DEFAULT_PORT = 5000 DEFAULT_LAN_HOST = "0.0.0.0" DEFAULT_BROWSER_HOST = "127.0.0.1" DEFAULT_PUBLIC_URL = "" IDENTITY_USER_HEADER = os.environ.get("IDENTITY_USER_HEADER", "X-Access-User").strip() or "X-Access-User" IDENTITY_EMAIL_HEADER = os.environ.get("IDENTITY_EMAIL_HEADER", "X-Access-Email").strip() or "X-Access-Email" IDENTITY_NAME_HEADER = os.environ.get("IDENTITY_NAME_HEADER", "X-Access-Name").strip() or "X-Access-Name" IDENTITY_SECRET_HEADER = os.environ.get("IDENTITY_SECRET_HEADER", "X-Access-Proxy-Secret").strip() or "X-Access-Proxy-Secret" IDENTITY_SHARED_SECRET = os.environ.get("IDENTITY_SHARED_SECRET", "").strip() LOCAL_USER = os.environ.get("LOCAL_USER", "local").strip() or "local" GUEST_COOKIE_NAME = os.environ.get("BLAME_MACHINE_GUEST_COOKIE", "blame_machine_guest").strip() or "blame_machine_guest" GUEST_COOKIE_MAX_AGE = max(3600, int(os.environ.get("BLAME_MACHINE_GUEST_COOKIE_MAX_AGE", str(60 * 60 * 24 * 365)).strip() or str(60 * 60 * 24 * 365))) TURNSTILE_SITE_KEY = os.environ.get("TURNSTILE_SITE_KEY", "").strip() TURNSTILE_SECRET_KEY = os.environ.get("TURNSTILE_SECRET_KEY", "").strip() TURNSTILE_ACTION = os.environ.get("TURNSTILE_ACTION", "start_play").strip() or "start_play" TURNSTILE_COOKIE_NAME = os.environ.get("TURNSTILE_COOKIE_NAME", "blame_machine_human").strip() or "blame_machine_human" TURNSTILE_COOKIE_MAX_AGE = max(300, int(os.environ.get("TURNSTILE_COOKIE_MAX_AGE", str(60 * 60 * 12)).strip() or str(60 * 60 * 12))) TURNSTILE_VERIFY_TIMEOUT = max(1.0, float(os.environ.get("TURNSTILE_VERIFY_TIMEOUT", "5").strip() or "5")) AUDIO_DEBUG_ENABLED = ( os.environ.get("BLAME_MACHINE_AUDIO_DEBUG", "").strip().lower() in {"1", "true", "yes", "on"} ) BUILD_LABEL = os.environ.get("APP_COMMIT", "").strip()[:7].lower() SOCIAL_DESCRIPTION = "Blame me for bad decisions." APPLE_TOUCH_ICON_PATH = BUNDLE_DIR / "apple-touch-icon.png" SOCIAL_CARD_PNG_PATH = BUNDLE_DIR / "social-card.png" app = Flask(__name__) GLOBAL_WILDCARDS = [ "pick none of these", "reroll later", "ask a friend", "do something completely different", "sleep on it", "flip a coin and trust your reaction", "focus on yourself for now", "take a break and come back to it", ] CATEGORY_DEFAULT_WILDCARDS = { "dessert": [ "switch it up with a cupcake", "plot twist: brownie", "go rogue with cookies", "wildcard says cheesecake", ], "beverage": [ "surprise yourself with something fizzy", "plot twist: water first", "switch it up with lemonade", "wildcard says iced coffee", ], "task": [ "do the easiest win first", "start with the most annoying thing", "take a 20-minute reset, then decide", "clean something small first", ], "car": [ "test drive both first", "keep the money today", "sleep on it", "check insurance before deciding", ], } BANNED_WORDS_PATH = BUNDLE_DIR / "banned_words.txt" ALLOWED_WORDS_PATH = BUNDLE_DIR / "allowed_words.txt" COIN_PRIMER_MP3_PATH = BUNDLE_DIR / "coin-primer.mp3" COIN_PRIMER_MP3 = COIN_PRIMER_MP3_PATH.read_bytes() if COIN_PRIMER_MP3_PATH.exists() else b"" FAST_REROLL_SECONDS = 10 RESULT_RETURN_SECONDS = 60 WILDCARD_CHANCE = 0.2 RECENT_HISTORY_LIMIT = 3 CATEGORY_NAME_RE = re.compile(r"^[^\[\]\r\n]{1,80}$") CHOICE_TEMPLATES = [ "you should totally go with {choice}", "you'd be nuts to choose anything but {choice}", "{choice} is the move", "come on now, it's obviously {choice}", "trust the machine: {choice}", "{choice} and don't overthink it", "the cabinet has spoken: {choice}", "all signs point to {choice}", "save yourself the spiral and pick {choice}", "this has {choice} written all over it", "don't make it weird, just pick {choice}", "the joystick says {choice}", "c'mon, sweetheart, it's {choice}", "ay, quit stalling, go with {choice}", "use your head for once and pick {choice}", "listen, pal, {choice} is the answer", "you're really gonna argue with {choice}?", "forget the drama, take {choice}", "be serious, you're taking {choice}", "quit playin' and go with {choice}", "that's the move, kid: {choice}", "don't overcook it, pick {choice}", "oh brother, it's obviously {choice}", "what are we even doing here? it's {choice}", "yeah, no, you're picking {choice}", "please. the answer is {choice}", ] CLEAR_ARCADE_DESCRIPTIONS = [ "WIPE MY CATEGORIES BEFORE ANYONE SEES THIS.", "RESET MY ARCADE. DON'T ASK QUESTIONS.", "START OVER LIKE NONE OF THIS HAPPENED.", ] BASE_STYLES = """ """ MUSIC_UI = """ """ MUSIC_SCRIPT = """ """ APP_SHELL_SCRIPT = """ """ GAME_SHELL_SCRIPT = """ """ GAME_HTML_BODY = """
""" + MUSIC_UI + """
""" + BUILD_LABEL + """
""" + MUSIC_SCRIPT + GAME_SHELL_SCRIPT + """ """ def current_public_base_url() -> str: configured = configured_public_url() if configured: return configured return request.url_root.strip().rstrip("/") def absolute_app_url(path: str) -> str: return f"{current_public_base_url()}{path}" def render_game_html() -> str: title = "Blame Machine" game_url = absolute_app_url("/game") social_image = absolute_app_url("/social-card.png") favicon = absolute_app_url("/favicon.svg") apple_touch_icon = absolute_app_url("/apple-touch-icon.png") head = f""" {html_escape(title)} """ + BASE_STYLES + """ """ return head + GAME_HTML_BODY def load_word_list(path: Path) -> set[str]: if not path.exists(): return set() words = set() with path.open("r", encoding="utf-8") as handle: for line in handle: line = line.strip().lower() if not line or line.startswith("#"): continue words.add(line) return words BANNED_WORDS = load_word_list(BANNED_WORDS_PATH) ALLOWED_WORDS = load_word_list(ALLOWED_WORDS_PATH) app.config["ACCESS_URL"] = None app.config["ACCESS_MESSAGE"] = None app.config["BROWSER_URL"] = None def debug_beep_bytes() -> bytes: rate = 44100 duration_seconds = 0.25 frequency_hz = 1320 buffer = io.BytesIO() with wave.open(buffer, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(rate) frames: list[bytes] = [] total = int(rate * duration_seconds) for index in range(total): if index < total * 0.8: envelope = 1.0 else: envelope = max(0.0, (total - index) / (total * 0.2)) sample = int( 32767 * 0.65 * envelope * math.sin(2 * math.pi * frequency_hz * index / rate) ) frames.append(struct.pack(" str: encoded = base64.b64encode(AUDIO_DEBUG_BEEP_WAV).decode("ascii") return f"data:audio/wav;base64,{encoded}" def synth_wav_bytes( duration_seconds: float, events: list[dict[str, float | str]], sample_rate: int = 44100, master_gain: float = 0.7, sharpness: float = 1.0, ) -> bytes: total_frames = max(1, int(sample_rate * duration_seconds)) buffer = io.BytesIO() frames: list[bytes] = [] def osc_value(kind: str, phase: float) -> float: cycle = phase % 1.0 if kind == "square": return 1.0 if cycle < 0.5 else -1.0 if kind == "triangle": return 1.0 - 4.0 * abs(cycle - 0.5) if kind == "saw": return (2.0 * cycle) - 1.0 return math.sin(2 * math.pi * cycle) for index in range(total_frames): t = index / sample_rate sample_value = 0.0 for event in events: start = float(event.get("start", 0.0)) end = float(event.get("end", duration_seconds)) if t < start or t >= end: continue local = t - start span = max(end - start, 0.001) shaped = max(0.5, min(2.0, sharpness)) attack = min(0.010 / (shaped * shaped), span * 0.18) release = min(0.050 / (shaped * shaped), span * 0.35) if local < attack: envelope = local / max(attack, 0.001) else: if shaped >= 1.0: decay_rate = 2.0 + ((shaped - 1.0) * 10.0) sustain_floor = max(0.12, 1.0 - ((shaped - 1.0) * 0.86)) envelope = sustain_floor + ((1.0 - sustain_floor) * math.exp(-decay_rate * (local / span))) else: envelope = 1.0 if end - t < release: envelope *= max(0.0, (end - t) / max(release, 0.001)) freq = float(event.get("freq", 440.0)) sweep_to = event.get("sweep_to") if sweep_to is not None: freq = freq + (float(sweep_to) - freq) * min(max(local / span, 0.0), 1.0) phase = local * freq amp = float(event.get("amp", 0.2)) kind = str(event.get("wave", "sine")) transient_boost = 1.0 if shaped > 1.0 and local < 0.018: transient_boost += (shaped - 1.0) * 0.45 sample_value += osc_value(kind, phase) * amp * envelope * transient_boost sample_value = max(-1.0, min(1.0, sample_value * master_gain)) frames.append(struct.pack(" bytes: pitch = 0.85 beat = 0.36 pattern = [220.0, 220.0, 261.63, 220.0, 293.66, 261.63, 220.0, 196.0] arp = [659.25, 783.99, 880.0, 783.99, 698.46, 783.99, 1046.5, 783.99] events: list[dict[str, float | str]] = [] for step, freq in enumerate(pattern): start = step * beat events.append({"start": start, "end": start + 0.22, "freq": freq * pitch, "wave": "triangle", "amp": 0.12}) events.append({"start": start, "end": start + 0.08, "freq": freq * 2 * pitch, "wave": "square", "amp": 0.08}) events.append({"start": start, "end": start + 0.15, "freq": arp[step] * pitch, "wave": "saw", "amp": 0.10}) if step % 4 == 3: events.append({"start": start, "end": start + 0.04, "freq": 1320.0 * pitch, "wave": "square", "amp": 0.05}) return synth_wav_bytes(beat * 8, events, master_gain=0.74, sharpness=2.0) def build_category_loop_wav() -> bytes: beat = 60 / 96 / 2 bass = [196.0, 196.0, 220.0, 196.0, 196.0, 174.61, 196.0, 220.0] lead = [392.0, 440.0, 392.0, 523.25, 440.0, 392.0, 349.23, 392.0] events: list[dict[str, float | str]] = [] for step, freq in enumerate(bass): start = step * beat events.append({"start": start, "end": start + 0.20, "freq": freq, "wave": "triangle", "amp": 0.12}) events.append({"start": start + 0.015, "end": start + 0.14, "freq": freq * 2, "wave": "triangle", "amp": 0.08}) events.append({"start": start + 0.04, "end": start + 0.18, "freq": lead[step], "wave": "triangle", "amp": 0.07}) return synth_wav_bytes(beat * 8, events, master_gain=0.72) def build_coin_wav() -> bytes: return synth_wav_bytes( 0.22, [ {"start": 0.00, "end": 0.06, "freq": 1760.0, "sweep_to": 1620.0, "wave": "square", "amp": 0.35}, {"start": 0.06, "end": 0.16, "freq": 1318.51, "sweep_to": 920.0, "wave": "square", "amp": 0.42}, ], master_gain=0.9, ) def build_reveal_wav() -> bytes: return synth_wav_bytes( 0.55, [ {"start": 0.00, "end": 0.16, "freq": 523.25, "sweep_to": 659.25, "wave": "triangle", "amp": 0.18}, {"start": 0.11, "end": 0.30, "freq": 659.25, "sweep_to": 783.99, "wave": "triangle", "amp": 0.22}, {"start": 0.24, "end": 0.50, "freq": 783.99, "sweep_to": 1046.5, "wave": "triangle", "amp": 0.28}, ], master_gain=0.95, ) def build_wonk_wav() -> bytes: return synth_wav_bytes( 0.42, [ {"start": 0.00, "end": 0.18, "freq": 330.0, "sweep_to": 220.0, "wave": "saw", "amp": 0.22}, {"start": 0.16, "end": 0.38, "freq": 246.94, "sweep_to": 164.81, "wave": "saw", "amp": 0.20}, ], master_gain=0.95, ) AUDIO_DEBUG_BEEP_WAV = debug_beep_bytes() AUDIO_DEBUG_BEEP_DATA_URI = debug_beep_data_uri() AUDIO_LOOP_HOME_WAV = build_home_loop_wav() AUDIO_LOOP_CATEGORY_WAV = build_category_loop_wav() AUDIO_COIN_WAV = build_coin_wav() AUDIO_REVEAL_WAV = build_reveal_wav() AUDIO_WONK_WAV = build_wonk_wav() def slugify_identity(value: str) -> str: value = value.strip().lower() value = re.sub(r"[^a-z0-9._-]+", "-", value) value = re.sub(r"-{2,}", "-", value).strip("-") return value or "local" def identity_storage_key(value: str) -> str: normalized = value.strip().lower().encode("utf-8") digest = hashlib.sha256(normalized).hexdigest()[:24] return f"user-{digest}" def trusted_identity_headers_present() -> bool: if not IDENTITY_SHARED_SECRET: return False secret = request.headers.get(IDENTITY_SECRET_HEADER, "").strip() return bool(secret) and hmac.compare_digest(secret, IDENTITY_SHARED_SECRET) def turnstile_enabled() -> bool: return bool(TURNSTILE_SITE_KEY and TURNSTILE_SECRET_KEY) def valid_guest_cookie_value(value: str) -> bool: return bool(re.fullmatch(r"[A-Za-z0-9_-]{24,128}", value or "")) def ensure_guest_cookie_value() -> str: existing = request.cookies.get(GUEST_COOKIE_NAME, "").strip() if valid_guest_cookie_value(existing): return existing pending = getattr(g, "pending_guest_cookie", "").strip() if valid_guest_cookie_value(pending): return pending pending = secrets.token_urlsafe(24) g.pending_guest_cookie = pending return pending def make_signed_value(secret_key: str, payload: str) -> str: digest = hmac.new(secret_key.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest() return digest def issue_turnstile_clearance() -> str: expires_at = int(time.time()) + TURNSTILE_COOKIE_MAX_AGE nonce = secrets.token_urlsafe(12) payload = f"{expires_at}:{nonce}" signature = make_signed_value(TURNSTILE_SECRET_KEY, payload) return f"{payload}:{signature}" def has_valid_turnstile_clearance() -> bool: if not turnstile_enabled(): return False raw = request.cookies.get(TURNSTILE_COOKIE_NAME, "").strip() try: expires_raw, nonce, signature = raw.split(":", 2) expires_at = int(expires_raw) except ValueError: return False if expires_at <= int(time.time()): return False payload = f"{expires_at}:{nonce}" expected_signature = make_signed_value(TURNSTILE_SECRET_KEY, payload) return hmac.compare_digest(signature, expected_signature) def client_ip() -> str: cf_ip = request.headers.get("CF-Connecting-IP", "").strip() if cf_ip: return cf_ip forwarded_for = request.headers.get("X-Forwarded-For", "").strip() if forwarded_for: return forwarded_for.split(",", 1)[0].strip() return request.remote_addr or "" def verify_turnstile_token(token: str) -> tuple[bool, list[str]]: payload = { "secret": TURNSTILE_SECRET_KEY, "response": token, "remoteip": client_ip(), } encoded = json.dumps(payload).encode("utf-8") req = urllib.request.Request( "https://challenges.cloudflare.com/turnstile/v0/siteverify", data=encoded, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=TURNSTILE_VERIFY_TIMEOUT) as response: result = json.loads(response.read().decode("utf-8")) except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError): return False, ["internal-error"] if not result.get("success"): return False, [str(code) for code in result.get("error-codes", [])] action = str(result.get("action", "") or "").strip() if TURNSTILE_ACTION and action and action != TURNSTILE_ACTION: return False, ["action-mismatch"] return True, [] def turnstile_required_response() -> Response: return ( jsonify( { "turnstile_required": True, "error": "human_check_required", "turnstile_site_key": TURNSTILE_SITE_KEY, "turnstile_action": TURNSTILE_ACTION, } ), 403, ) def current_identity() -> dict[str, str]: if trusted_identity_headers_present(): user = request.headers.get(IDENTITY_USER_HEADER, "").strip() email = request.headers.get(IDENTITY_EMAIL_HEADER, "").strip() name = request.headers.get(IDENTITY_NAME_HEADER, "").strip() if user: login_value = email or user return { "user": user, "slug": slugify_identity(user), "storage_key": identity_storage_key(user), "email": email, "name": name, "label": f"You are logged in as: {login_value}", "source_message": "Private profile loaded through Access SSO.", } guest_cookie = ensure_guest_cookie_value() guest_label = f"GAME TOKEN: {guest_cookie[:8].upper()}" return { "user": f"guest-{guest_cookie[:12]}", "slug": slugify_identity(guest_label), "storage_key": identity_storage_key(f"guest:{guest_cookie}"), "email": "", "name": "", "label": guest_label, "source_message": "Public mode. This browser keeps its own arcade cabinet.", } def config_path_for_identity(identity: dict[str, str]) -> Path: user_dir = APP_DIR / "users" user_dir.mkdir(parents=True, exist_ok=True, mode=0o700) try: user_dir.chmod(0o700) except PermissionError: # Bind-mounted directories may be owned by the host. pass return user_dir / f"{identity['storage_key']}.ini" def load_config(identity: dict[str, str]) -> configparser.ConfigParser: config = configparser.ConfigParser() ini_path = config_path_for_identity(identity) if ini_path.exists(): config.read(ini_path) return config def save_config(config: configparser.ConfigParser, identity: dict[str, str]) -> None: ini_path = config_path_for_identity(identity) fd = os.open(ini_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as handle: config.write(handle) try: ini_path.chmod(0o600) except PermissionError: # Bind-mounted files may be owned by the host. pass def normalize_for_filter(text: str) -> str: text = text.lower().strip() text = re.sub(r"[^a-z0-9\s]", "", text) text = re.sub(r"\s+", " ", text) return text def is_blocked_entry(text: str) -> bool: normalized = normalize_for_filter(text) if not normalized: return False if normalized in ALLOWED_WORDS: return False return normalized in BANNED_WORDS def norm(text: str) -> str: return text.strip().lower() def titleize(text: str) -> str: return text.strip().title() def parse_csv(raw: str) -> list[str]: items = re.split(r"[\n,]+", raw) return [item.strip() for item in items if item.strip()] def dedupe_choices(items: list[str]) -> list[str]: seen: set[str] = set() unique: list[str] = [] for item in items: key = norm(item) if key in seen: continue seen.add(key) unique.append(item.strip()) return unique def duplicate_choices(items: list[str]) -> list[str]: seen: set[str] = set() duplicate_keys: set[str] = set() duplicates: list[str] = [] for item in items: key = norm(item) if key in seen: if key not in duplicate_keys: duplicates.append(item.strip()) duplicate_keys.add(key) continue seen.add(key) return duplicates def category_wildcards(category: str) -> list[str]: return CATEGORY_DEFAULT_WILDCARDS.get(category, GLOBAL_WILDCARDS) def random_wildcard(category: str) -> str: return random.choice(category_wildcards(category)) def funny_choice_result(choice: str) -> str: return random.choice(CHOICE_TEMPLATES).format(choice=choice) def random_clear_arcade_description() -> str: return random.choice(CLEAR_ARCADE_DESCRIPTIONS) def is_valid_category_name(category: str) -> bool: return bool(category) and bool(CATEGORY_NAME_RE.fullmatch(category)) def ensure_category(config: configparser.ConfigParser, category: str) -> None: if category not in config: config[category] = {} config[category].setdefault("choices", "") config[category].setdefault("last_choice", "") config[category].setdefault("last_ts", "0") config[category].setdefault("recent_choices", "") def last_pick_info(config: configparser.ConfigParser, category: str) -> tuple[str, float]: ensure_category(config, category) last_choice = config[category].get("last_choice", "").strip() last_ts_raw = config[category].get("last_ts", "0").strip() try: last_ts = float(last_ts_raw) except ValueError: last_ts = 0.0 return last_choice, last_ts def recent_choice_history(config: configparser.ConfigParser, category: str) -> list[str]: ensure_category(config, category) return [norm(item) for item in parse_csv(config[category].get("recent_choices", ""))] def update_recent_history(config: configparser.ConfigParser, category: str, choice: str) -> None: history = recent_choice_history(config, category) normalized_choice = norm(choice) history = [item for item in history if item != normalized_choice] history.insert(0, normalized_choice) trimmed = history[:RECENT_HISTORY_LIMIT] config[category]["recent_choices"] = ", ".join(trimmed) def choice_pool(choices: list[str], recent_history: list[str]) -> list[str]: if len(choices) <= 1: return choices available = [choice for choice in choices if norm(choice) not in recent_history] return available or choices def category_context(config: configparser.ConfigParser, category: str) -> dict: ensure_category(config, category) choices = config[category].get("choices", "") choices_list = parse_csv(choices) return { "category": category, "category_display": titleize(category), "choices": choices, "choices_list": choices_list, "choice_count": len(choices_list), "can_decide": len(choices_list) >= 2, "wildcard_percent": int(WILDCARD_CHANCE * 100), } def category_subtitle(mode: str) -> str: subtitles = { "edit": "LOAD THE MACHINE.", "ready": "LOCKED IN. READY TO SPIN.", "result": "THE ARCADE HAS SPOKEN.", } return subtitles.get(mode, "DECIDE SOMETHING.") def home_state(identity: dict[str, str], error: str | None = None) -> dict: config = load_config(identity) return { "view": "home", "scene": "home", "player_label": identity["label"], "player_source_message": identity["source_message"], "categories": sorted(config.sections()), "error": error, "access_url": app.config.get("ACCESS_URL"), "access_message": app.config.get("ACCESS_MESSAGE"), "clear_arcade_description": random_clear_arcade_description(), } def clear_state(identity: dict[str, str]) -> dict: return { "view": "clear", "scene": "home", "player_label": identity["label"], "player_source_message": identity["source_message"], "clear_arcade_description": random_clear_arcade_description(), } def category_state( identity: dict[str, str], category: str, mode: str, result: str | None = None, used_wildcard: bool = False, error: str | None = None, result_meta: str | None = None, ) -> dict: config = load_config(identity) category = norm(category) ctx = category_context(config, category) ctx["view"] = "category" ctx["scene"] = "result" if mode == "result" else "category" ctx["mode"] = mode ctx["result"] = result ctx["used_wildcard"] = used_wildcard ctx["error"] = error ctx["result_meta"] = result_meta ctx["result_kind"] = "normal" ctx["auto_return_seconds"] = RESULT_RETURN_SECONDS if mode == "result" else 0 ctx["subtitle"] = category_subtitle(mode) ctx["player_label"] = identity["label"] ctx["player_source_message"] = identity["source_message"] return ctx def payload_value(name: str, default: str = "") -> str: data = request.get_json(silent=True) if isinstance(data, dict) and name in data: value = data.get(name, default) return "" if value is None else str(value) return request.form.get(name, default) def expected_request_origin() -> str: proto = request.headers.get("X-Forwarded-Proto", "").strip() host = request.headers.get("X-Forwarded-Host", "").strip() if proto and host: return f"{proto}://{host}".rstrip("/") return request.host_url.rstrip("/") def origin_host_matches(origin_like: str, expected_origin: str) -> bool: if not origin_like: return False try: parsed_origin = urlparse(origin_like) parsed_expected = urlparse(expected_origin) except ValueError: return False return bool(parsed_origin.netloc) and parsed_origin.netloc == parsed_expected.netloc @app.after_request def disable_shell_caching(response): if request.path in {"/", "/game"} or request.path.startswith("/api/"): response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" pending_guest_cookie = getattr(g, "pending_guest_cookie", "").strip() if pending_guest_cookie: response.set_cookie( GUEST_COOKIE_NAME, pending_guest_cookie, max_age=GUEST_COOKIE_MAX_AGE, httponly=True, secure=bool(request.is_secure or request.headers.get("X-Forwarded-Proto", "").strip() == "https"), samesite="Lax", path="/", ) pending_turnstile_cookie = getattr(g, "pending_turnstile_cookie", "").strip() if pending_turnstile_cookie: response.set_cookie( TURNSTILE_COOKIE_NAME, pending_turnstile_cookie, max_age=TURNSTILE_COOKIE_MAX_AGE, httponly=True, secure=bool(request.is_secure or request.headers.get("X-Forwarded-Proto", "").strip() == "https"), samesite="Lax", path="/", ) return response def same_origin_api_request() -> bool: expected_origin = expected_request_origin() origin = request.headers.get("Origin", "").strip().rstrip("/") referer = request.headers.get("Referer", "").strip() fetch_site = request.headers.get("Sec-Fetch-Site", "").strip().lower() if origin: return origin == expected_origin or origin_host_matches(origin, expected_origin) if referer: return ( referer == expected_origin or referer.startswith(expected_origin + "/") or origin_host_matches(referer, expected_origin) ) return fetch_site in {"", "same-origin", "same-site"} def audio_debug_allowed() -> bool: return AUDIO_DEBUG_ENABLED def bounded_log_value(value: object, limit: int = 240) -> str: text = str(value or "").replace("\r", " ").replace("\n", " ") return text[:limit] def svg_response(markup: str, cache_seconds: int = 3600) -> Response: response = Response(markup, mimetype="image/svg+xml") response.headers["Cache-Control"] = f"public, max-age={cache_seconds}" return response def favicon_svg_markup() -> str: return """ BM """ def social_card_svg_markup() -> str: return """ ★ BLAME MACHINE ★ INSERT COIN. BLAME THE MACHINE. BLAME ME FOR BAD DECISIONS. THE JOYSTICK SAYS BURGERS. """ @app.before_request def protect_api_posts(): if request.method != "POST" or not request.path.startswith("/api/"): return None if same_origin_api_request(): if ( turnstile_enabled() and not trusted_identity_headers_present() and request.path in {"/api/category", "/api/save", "/api/decide", "/api/clear"} and not has_valid_turnstile_clearance() ): token = payload_value("turnstile_token").strip() if not token: return turnstile_required_response() verified, error_codes = verify_turnstile_token(token) if not verified: app.logger.warning( "TURNSTILE_REJECT path=%r ip=%r errors=%r ua=%r", request.path, client_ip(), error_codes, request.headers.get("User-Agent", ""), ) return turnstile_required_response() g.pending_turnstile_cookie = issue_turnstile_clearance() return None identity = current_identity() app.logger.warning( "API_POST_REJECT path=%r expected_origin=%r origin=%r referer=%r sec_fetch_site=%r ua=%r", request.path, expected_request_origin(), request.headers.get("Origin", ""), request.headers.get("Referer", ""), request.headers.get("Sec-Fetch-Site", ""), request.headers.get("User-Agent", ""), ) return jsonify(home_state(identity, error="forbidden")), 403 @app.route("/") @app.route("/game") def home(): return render_template_string(render_game_html()) @app.route("/favicon.svg") def favicon_svg(): return svg_response(favicon_svg_markup(), cache_seconds=86400) @app.route("/favicon.ico") def favicon_ico(): return redirect(url_for("favicon_svg"), code=302) @app.route("/social-card.svg") def social_card(): return svg_response(social_card_svg_markup(), cache_seconds=86400) @app.route("/apple-touch-icon.png") def apple_touch_icon(): if APPLE_TOUCH_ICON_PATH.exists(): response = Response(APPLE_TOUCH_ICON_PATH.read_bytes(), mimetype="image/png") response.headers["Cache-Control"] = "public, max-age=86400" return response return redirect(url_for("favicon_svg"), code=302) @app.route("/social-card.png") def social_card_png(): if SOCIAL_CARD_PNG_PATH.exists(): response = Response(SOCIAL_CARD_PNG_PATH.read_bytes(), mimetype="image/png") response.headers["Cache-Control"] = "public, max-age=86400" return response return redirect(url_for("social_card"), code=302) @app.route("/api/state", methods=["GET"]) def api_state(): identity = current_identity() view = request.args.get("view", "").strip().lower() category = norm(request.args.get("category", "")) mode = request.args.get("mode", "ready").strip().lower() if view == "clear": return jsonify(clear_state(identity)) if view == "category" and category: if not is_valid_category_name(category): return jsonify(home_state(identity, error="invalid")), 400 if mode not in {"edit", "ready", "result"}: mode = "ready" return jsonify(category_state(identity, category, mode)) return jsonify(home_state(identity)) @app.route("/api/category", methods=["POST"]) def api_category(): identity = current_identity() config = load_config(identity) category = norm(payload_value("category")) if not category: return jsonify(home_state(identity, error="empty_category")), 400 if not is_valid_category_name(category): return jsonify(home_state(identity, error="invalid")), 400 if is_blocked_entry(category): return jsonify(home_state(identity, error="blocked")), 400 if config.has_section(category): return jsonify(home_state(identity, error="duplicate")), 409 ensure_category(config, category) save_config(config, identity) return jsonify(category_state(identity, category, mode="edit")) @app.route("/api/save", methods=["POST"]) def api_save(): identity = current_identity() config = load_config(identity) category = norm(payload_value("category")) if not is_valid_category_name(category): return jsonify(home_state(identity, error="invalid")), 400 ensure_category(config, category) raw_choices = payload_value("choices").strip() parsed_choices = parse_csv(raw_choices) duplicates = duplicate_choices(parsed_choices) if duplicates: return jsonify(category_state(identity, category, mode="edit", error="duplicate_choices")) choices = dedupe_choices(parsed_choices) for choice in choices: if is_blocked_entry(choice): return jsonify(category_state(identity, category, mode="edit", error="blocked")) config[category]["choices"] = ", ".join(choices) save_config(config, identity) return jsonify(category_state(identity, category, mode="ready")) @app.route("/api/decide", methods=["POST"]) def api_decide(): identity = current_identity() config = load_config(identity) category = norm(payload_value("category")) if not is_valid_category_name(category): return jsonify(home_state(identity, error="invalid")), 400 ensure_category(config, category) choices = parse_csv(config[category].get("choices", "")) now = time.time() if len(choices) < 2: return jsonify( category_state( identity, category, mode="result", result="ONE CHOICE? REALLY?", used_wildcard=False, result_meta="COME BACK WHEN YOU'VE GIVEN THE MACHINE AN ACTUAL DECISION.", ) ) last_choice, last_ts = last_pick_info(config, category) if last_choice and (now - last_ts) <= FAST_REROLL_SECONDS: state = category_state( identity, category, mode="result", result=f"{last_choice} - did you think i'd change my mind so quickly!", used_wildcard=False, result_meta=f"FAST REROLL LOCK: {FAST_REROLL_SECONDS} SECONDS", ) state["result_kind"] = "too_soon" return jsonify(state) use_wildcard = random.random() < WILDCARD_CHANCE if use_wildcard: state = category_state( identity, category, mode="result", result=random_wildcard(category), used_wildcard=True, result_meta=f"{len(choices)} PICKS IN THE MACHINE", ) state["result_kind"] = "wildcard" return jsonify(state) recent_history = recent_choice_history(config, category) available_choices = choice_pool(choices, recent_history) selected = random.choice(available_choices) config[category]["last_choice"] = selected config[category]["last_ts"] = str(now) update_recent_history(config, category, selected) save_config(config, identity) meta = f"{len(choices)} UNIQUE PICKS | AVOIDING LAST {min(len(recent_history), RECENT_HISTORY_LIMIT)}" state = category_state( identity, category, mode="result", result=funny_choice_result(selected), used_wildcard=False, result_meta=meta, ) state["result_kind"] = "normal" return jsonify(state) @app.route("/api/clear", methods=["POST"]) def api_clear(): identity = current_identity() ini_path = config_path_for_identity(identity) if ini_path.exists(): ini_path.unlink() return jsonify(home_state(identity)) @app.route("/sound.wav") def sound_wav(): return Response( AUDIO_DEBUG_BEEP_WAV, mimetype="audio/wav", headers={ "Content-Disposition": "inline; filename=blame-machine-debug.wav", "Cache-Control": "no-store, max-age=0", }, ) @app.route("/audio/coin-primer.mp3") def audio_coin_primer(): if not COIN_PRIMER_MP3: return Response(status=404) return Response( COIN_PRIMER_MP3, mimetype="audio/mpeg", headers={ "Content-Disposition": "inline; filename=blame-machine-coin-primer.mp3", "Cache-Control": "no-store, max-age=0", }, ) @app.route("/audio/loop-home.wav") def audio_loop_home(): return Response( AUDIO_LOOP_HOME_WAV, mimetype="audio/wav", headers={"Cache-Control": "public, max-age=300"}, ) @app.route("/audio/loop-category.wav") def audio_loop_category(): return Response( AUDIO_LOOP_CATEGORY_WAV, mimetype="audio/wav", headers={"Cache-Control": "public, max-age=300"}, ) @app.route("/audio/cue-coin.wav") def audio_cue_coin(): return Response( AUDIO_COIN_WAV, mimetype="audio/wav", headers={"Cache-Control": "public, max-age=300"}, ) @app.route("/audio/cue-reveal.wav") def audio_cue_reveal(): return Response( AUDIO_REVEAL_WAV, mimetype="audio/wav", headers={"Cache-Control": "public, max-age=300"}, ) @app.route("/audio/cue-wonk.wav") def audio_cue_wonk(): return Response( AUDIO_WONK_WAV, mimetype="audio/wav", headers={"Cache-Control": "public, max-age=300"}, ) @app.route("/sound") def sound_page(): if not audio_debug_allowed(): return ("Not found", 404) return render_template_string( """ Blame Machine Sound File
★ BLAME MACHINE SOUND FILE ★
""" + BUILD_LABEL + """
THIS IS THE RAW WAV TEST. NO WEB AUDIO. JUST A FILE.
OPEN RAW SOUND FILE
""" ) @app.route("/audio-debug") def audio_debug(): if not audio_debug_allowed(): abort(404) return render_template_string(""" Blame Machine Audio Debug
★ BLAME MACHINE AUDIO DEBUG ★
""" + BUILD_LABEL + """
A/B AUDIO HARNESS.
SAME HOST. SAME CONTAINER. NO GAME FLOW.
EVENT TAIL
READY
LIVE STATE
ctx: none
"""); @app.route("/audio-debug/log", methods=["POST"]) def audio_debug_log(): if not audio_debug_allowed(): abort(404) if not same_origin_api_request(): return ("", 403) data = request.get_json(silent=True) or {} print( "AUDIO_DEBUG " f"client={request.headers.get('CF-Connecting-IP') or request.remote_addr!r} " f"message={bounded_log_value(data.get('message'))!r} " f"ctx_state={data.get('ctxState')!r} " f"ctx_time={data.get('ctxTime')!r} " f"html_paused={data.get('htmlPaused')!r} " f"html_ready={data.get('htmlReadyState')!r} " f"html_time={data.get('htmlCurrentTime')!r} " f"ua={bounded_log_value(data.get('ua'), limit=180)!r}", flush=True, ) return ("", 204) def open_browser() -> None: browser_url = app.config.get("BROWSER_URL") if browser_url: webbrowser.open(browser_url) def local_ip_address() -> str: candidates: list[str] = [] try: hostname_ips = socket.gethostbyname_ex(socket.gethostname())[2] candidates.extend(hostname_ips) except OSError: pass probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: # This does not require external internet access; it only asks the OS # which interface would route outward if one exists. probe.connect(("10.255.255.255", 1)) candidates.append(probe.getsockname()[0]) except OSError: pass finally: probe.close() for candidate in candidates: if candidate and not candidate.startswith("127."): return candidate return "127.0.0.1" def configured_port() -> int: raw_port = os.environ.get("PORT", str(DEFAULT_PORT)).strip() try: return int(raw_port) except ValueError: return DEFAULT_PORT def configured_public_url() -> str: return os.environ.get("PUBLIC_URL", DEFAULT_PUBLIC_URL).strip().rstrip("/") def should_open_browser() -> bool: if os.environ.get("BLAME_MACHINE_NO_BROWSER", "").strip() == "1": return False return not bool(configured_public_url()) def configure_access() -> tuple[str, str, str]: port = configured_port() public_url = configured_public_url() ip_address = local_ip_address() if public_url: browser_url = public_url access_url = public_url access_message = "OPEN THIS FROM ANY PHONE, ANYWHERE." else: browser_url = f"http://{DEFAULT_BROWSER_HOST}:{port}" if ip_address == "127.0.0.1": access_url = browser_url access_message = "NO LAN DETECTED. PLAY ON THIS COMPUTER." else: access_url = f"http://{ip_address}:{port}" access_message = "PLAY ON THIS COMPUTER OR A PHONE/TABLET ON THE SAME WI-FI." app.config["ACCESS_URL"] = access_url app.config["ACCESS_MESSAGE"] = access_message app.config["BROWSER_URL"] = browser_url return ip_address, access_url, access_message def print_access_banner(ip_address: str, access_url: str, access_message: str) -> None: print("") print("Blame Machine is running.") print(f"Open on this computer: {app.config['BROWSER_URL']}") print(access_message) print(f"Access URL: {access_url}") print("") if __name__ == "__main__": ip_address, access_url, access_message = configure_access() print_access_banner(ip_address, access_url, access_message) if should_open_browser(): Timer(1.0, open_browser).start() app.run(host=DEFAULT_LAN_HOST, port=configured_port(), debug=False, use_reloader=False)