#!/usr/bin/env python3 """Apply auth patches to server.py — run this script.""" import hashlib, os path = "/home/oplabs/projects/pi-multifx-pedal/src/web/server.py" content = open(path).read() print(f"BEFORE: md5={hashlib.md5(content.encode()).hexdigest()}, size={len(content)}") changes = [] # 2a. Add auth import + _safe_detail after MIDIHandler import old = "from ..midi.handler import MIDIHandler\n\nlogger = logging.getLogger(__name__)" new = """from ..midi.handler import MIDIHandler from .auth import is_auth_configured, set_pin, verify_token, require_auth logger = logging.getLogger(__name__) # ── Exception sanitization ──────────────────────────────────────────────────── def _safe_detail(log: logging.Logger, exc: Exception, status_code: int = 500) -> str: \"\"\"Log full exception server-side, return generic message to client.\"\"\" log.warning("API error (HTTP %%d): %%s", status_code, exc, exc_info=True) return "Internal server error\"""" assert old in content, "2a: pattern not found" content = content.replace(old, new, 1) changes.append("2a. Import + _safe_detail") print(f"2a OK. size={len(content)}") # 2b. Auth middleware after app = FastAPI(...) old = ' app = FastAPI(title="Pi Multi-FX Pedal", version="0.1.0", lifespan=lifespan)' new = ''' app = FastAPI(title="Pi Multi-FX Pedal", version="0.1.0", lifespan=lifespan) # ── Auth middleware ─────────────────────────────────────────── @app.middleware("http") async def auth_middleware(request: Request, call_next): public_paths = {"/", "/api/auth/setup", "/api/auth/status"} if request.url.path in public_paths or request.url.path.startswith("/static/"): return await call_next(request) if not request.url.path.startswith("/api/"): return await call_next(request) if not is_auth_configured(): return await call_next(request) token: str | None = request.headers.get("X-Auth-Token") if not token: token = request.query_params.get("token") if not token or not verify_token(token): return JSONResponse( status_code=401, content={"detail": "Authentication required. Set X-Auth-Token header to your pedal PIN."}, ) return await call_next(request)''' assert old in content, "2b: pattern not found" content = content.replace(old, new, 1) changes.append("2b. Auth middleware") print(f"2b OK. size={len(content)}") # 2c. Auth endpoints old = " # ── REST API ────────────────────────────────────────────────" new = ''' # ── Auth endpoints (no token required) ───────────────────── @app.get("/api/auth/status") async def auth_status(): """Check whether authentication is configured.""" return {"auth_configured": is_auth_configured()} @app.post("/api/auth/setup") async def auth_setup(data: dict): """Set the pedal PIN on first boot. This endpoint is open exactly once — after the PIN is set, all API calls require ``X-Auth-Token: ``. Body: { "pin": "1234" } """ if is_auth_configured(): raise HTTPException(status_code=400, detail="PIN already configured") pin = data.get("pin", "") try: set_pin(pin) return {"ok": True, "message": "Authentication PIN set"} except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) # ── REST API ────────────────────────────────────────────────''' assert old in content, "2c: pattern not found" content = content.replace(old, new, 1) changes.append("2c. Auth endpoints") print(f"2c OK. size={len(content)}") # 2d. WebSocket auth old = ''' @app.websocket("/ws") async def websocket_endpoint(ws: WebSocket): await self._manager.connect(ws) logger.info("WebSocket client connected (%d active)", self._manager.active_count) # Track per-connection channel preference''' new = ''' @app.websocket("/ws") async def websocket_endpoint(ws: WebSocket): # Require auth for WebSocket if PIN is configured if is_auth_configured(): ws_token = ws.query_params.get("token") if not ws_token or not verify_token(ws_token): await ws.close(code=4001, reason="Authentication required") return await self._manager.connect(ws) logger.info("WebSocket client connected (%d active)", self._manager.active_count) # Track per-connection channel preference''' assert old in content, "2d: pattern not found" content = content.replace(old, new, 1) changes.append("2d. WebSocket auth") print(f"2d OK. size={len(content)}") # 2e. Replace detail=str(e) count = content.count("detail=str(e)") assert count > 0, "2e: no detail=str(e)" content = content.replace("detail=str(e)", "detail=_safe_detail(logger, e)") changes.append(f"2e. {count}x detail=str(e) replaced") print(f"2e OK. size={len(content)}") # 2f. __init__ SSL params old = """ def __init__( self, deps: WebServerDeps, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, ) -> None: self.deps = deps self.host = host self.port = port""" new = """ def __init__( self, deps: WebServerDeps, host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, ssl_certfile: Optional[str] = None, ssl_keyfile: Optional[str] = None, ) -> None: self.deps = deps self.host = host self.port = port self.ssl_certfile = ssl_certfile self.ssl_keyfile = ssl_keyfile""" assert old in content, "2f: pattern not found" content = content.replace(old, new, 1) changes.append("2f. __init__ SSL params") print(f"2f OK. size={len(content)}") # 2g. start() SSL old = """ import threading config = uvicorn.Config( self._app, host=self.host, port=self.port, log_level="info", access_log=False, )""" new = """ import threading ssl_kw = {} if self.ssl_certfile and self.ssl_keyfile: ssl_kw["ssl_certfile"] = self.ssl_certfile ssl_kw["ssl_keyfile"] = self.ssl_keyfile config = uvicorn.Config( self._app, host=self.host, port=self.port, log_level="info", access_log=False, **ssl_kw, )""" assert old in content, "2g: pattern not found" content = content.replace(old, new, 1) changes.append("2g. start() SSL") print(f"2g OK. size={len(content)}") # 2h. Log message old = ' logger.info("Web server starting on http://%s:%d", self.host, self.port)' new = """ proto = "https" if self.ssl_certfile else "http" logger.info("Web server starting on %s://%s:%d", proto, self.host, self.port)""" assert old in content, "2h: pattern not found" content = content.replace(old, new, 1) changes.append("2h. Log message") print(f"2h OK. size={len(content)}") # Write to a NEW file to test write capability test_path = "/home/oplabs/projects/pi-multifx-pedal/src/web/server_patched.py" open(test_path, "w").write(content) print(f"\nWritten to {test_path}") print(f"AFTER: md5={hashlib.md5(content.encode()).hexdigest()}") print(f"Changes: {', '.join(changes)}") # Verify it compiled import py_compile try: py_compile.compile(test_path, doraise=True) print("✅ Syntax check passed!") except py_compile.PyCompileError as e: print(f"❌ Syntax error: {e}")