From a3ec86b814c4d37f04227b5cce8e742b5fdfd423 Mon Sep 17 00:00:00 2001 From: Shawn Date: Wed, 17 Jun 2026 22:51:30 -0400 Subject: [PATCH] chore: remove rogue patch files from tracking --- .gitignore | 1 + patch_all.py | 378 ------------------------------------------------ patch_final.py | 100 ------------- patch_server.py | 205 -------------------------- 4 files changed, 1 insertion(+), 683 deletions(-) delete mode 100644 patch_all.py delete mode 100644 patch_final.py delete mode 100644 patch_server.py diff --git a/.gitignore b/.gitignore index fe838ff..5d73515 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,4 @@ test_write.txt test-drc.rpt test_empty-drc.rptpatch_*.py tests/test_issue_*.py +patch_*.py diff --git a/patch_all.py b/patch_all.py deleted file mode 100644 index 8e2f084..0000000 --- a/patch_all.py +++ /dev/null @@ -1,378 +0,0 @@ -#!/usr/bin/env python3 -"""Apply ALL auth + HTTPS patches to the pedal codebase in one shot.""" -import hashlib, logging, os, re -from pathlib import Path - -BASE = Path("/home/oplabs/projects/pi-multifx-pedal") - -# ── 1. Create src/web/auth.py ───────────────────────────────────────── -auth_py = '''"""Authentication layer for the Pi Multi-FX Pedal web interface. - -Provides a shared-secret PIN authentication model: - -- PIN stored in ``~/.pedal/auth.pin`` as a PBKDF2-SHA256 hash. -- On first boot (no PIN file), the ``/api/auth/setup`` endpoint is open - for exactly one call to set an initial PIN. -- All other API endpoints require ``X-Auth-Token: `` in the request - header (or ``token`` query parameter for WebSocket). -- Failed auth returns 401 with a generic message. -""" - -from __future__ import annotations - -import hashlib -import hmac -import logging -import secrets -from pathlib import Path - -from fastapi import HTTPException, Request, status - -logger = logging.getLogger(__name__) - -# ── Paths ───────────────────────────────────────────────────────────────────── - -AUTH_DIR=*** / ".pedal" -AUTH_FILE=*** / "auth.pin" - -# ── PIN hashing (PBKDF2-SHA256, no bcrypt dependency needed) ───────────────── - - -def _hash_pin(pin: str) -> str: - """Return a PBKDF2-SHA256 hex digest with random salt. - Format: ``$`` - """ - salt = secrets.token_hex(16) - h = hashlib.pbkdf2_hmac("sha256", pin.encode(), salt.encode(), 100_000) - return f"{salt}${h.hex()}" - - -def _verify_pin(pin: str, stored: str) -> bool: - """Verify *pin* against a stored ``salt$hash`` string.""" - if "$" not in stored: - return False - salt, expected = stored.split("$", 1) - h = hashlib.pbkdf2_hmac("sha256", pin.encode(), salt.encode(), 100_000) - return hmac.compare_digest(h.hex(), expected) - - -# ── Public API ─────────────────────────────────────────────────────────────── - - -def is_auth_configured() -> bool: - """Return ``True`` if a PIN has been set (auth is active).""" - return AUTH_FILE.exists() - - -def set_pin(pin: str) -> None: - """Store a new PIN hash. - Args: - pin: The PIN to store (4-32 characters). - Raises: - ValueError: If the PIN is too short or too long. - """ - if len(pin) < 4: - raise ValueError("PIN must be at least 4 characters") - if len(pin) > 64: - raise ValueError("PIN must be at most 64 characters") - AUTH_DIR.mkdir(parents=True, exist_ok=True) - hashed = _hash_pin(pin) - AUTH_FILE.write_text(hashed) - AUTH_FILE.chmod(0o600) - logger.info("Auth PIN set (file: %s)", AUTH_FILE) - - -def verify_token(token: str) -> bool: - """Check whether *token* matches the stored PIN. - Returns ``True`` if *token* is the correct PIN. - If no PIN is configured yet, returns ``True`` (no auth required). - """ - if not is_auth_configured(): - return True - try: - stored = AUTH_FILE.read_text().strip() - return _verify_pin(token, stored) - except Exception: - return False - - -# ── FastAPI dependency (for per-endpoint use, kept for backward compat) ─────── - - -async def require_auth(request: Request) -> None: - """FastAPI dependency — reject with 401 if the request lacks a valid token.""" - if not is_auth_configured(): - return - 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): - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Authentication required. Set X-Auth-Token header to your pedal PIN.", - headers={"WWW-Authenticate": "Bearer"}, - ) -''' - -(BASE / "src" / "web" / "auth.py").write_text(auth_py) -print("1. Created src/web/auth.py") - -# ── 2. Patch server.py ───────────────────────────────────────────────── -sp = (BASE / "src" / "web" / "server.py").read_text() - -changes = 0 - -# 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 sp, "2a: pattern not found" -sp = sp.replace(old, new, 1) -changes += 1 -print("2a. Import + _safe_detail added") - -# 2b. Add 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 sp, "2b: pattern not found" -sp = sp.replace(old, new, 1) -changes += 1 -print("2b. Auth middleware added") - -# 2c. Add auth endpoints before REST API section -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 sp, "2c: pattern not found" -sp = sp.replace(old, new, 1) -changes += 1 -print("2c. Auth endpoints added") - -# 2d. Add 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 sp, "2d: pattern not found" -sp = sp.replace(old, new, 1) -changes += 1 -print("2d. WebSocket auth added") - -# 2e. Replace all detail=str(e) with _safe_detail -count = sp.count("detail=str(e)") -assert count > 0, "2e: no detail=str(e) found" -sp = sp.replace("detail=str(e)", "detail=_safe_detail(logger, e)") -changes += 1 -print(f"2e. {count} detail=str(e) replaced with _safe_detail") - -# 2f. Update __init__ for 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 sp, "2f: pattern not found" -sp = sp.replace(old, new, 1) -changes += 1 -print("2f. __init__ SSL params added") - -# 2g. Update start() for 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 sp, "2g: pattern not found" -sp = sp.replace(old, new, 1) -changes += 1 -print("2g. start() SSL support added") - -# 2h. Update 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 sp, "2h: pattern not found" -sp = sp.replace(old, new, 1) -changes += 1 -print("2h. Log message updated") - -(BASE / "src" / "web" / "server.py").write_text(sp) -print(f"\nserver.py: {changes} patches applied") - -# ── 3. Patch main.py ────────────────────────────────────────────────── -mp = (BASE / "main.py").read_text() - -old = """ self.web = WebServer( - deps=WebServerDeps( - presets=self.presets, - pipeline=self.pipeline, - nam_host=self.nam_host, - ir_loader=self.ir_loader, - audio_system=self.audio_system, - jack_audio=self.jack_audio, - config=self._config, - config_path=self._config_path, - ), - host="0.0.0.0", - port=80, - ) - self.web.start() - logger.info("Web UI server started on http://0.0.0.0:80")""" - -new = """ web_cfg = self._config.get("web", {}) - web_host = web_cfg.get("host", "0.0.0.0") - web_port = int(web_cfg.get("port", 80)) - ssl_cert = web_cfg.get("ssl_certfile") or None - ssl_key = web_cfg.get("ssl_keyfile") or None - self.web = WebServer( - deps=WebServerDeps( - presets=self.presets, - pipeline=self.pipeline, - nam_host=self.nam_host, - ir_loader=self.ir_loader, - audio_system=self.audio_system, - jack_audio=self.jack_audio, - config=self._config, - config_path=self._config_path, - ), - host=web_host, - port=web_port, - ssl_certfile=ssl_cert, - ssl_keyfile=ssl_key, - ) - self.web.start() - proto = "https" if ssl_cert else "http" - logger.info("Web UI server started on %s://%s:%d", proto, web_host, web_port)""" - -assert old in mp, "3: pattern not found" -mp = mp.replace(old, new, 1) -(BASE / "main.py").write_text(mp) -print("3. main.py: web config + SSL support added") - -# ── Final summary ────────────────────────────────────────────────────── -print("\n═══ VERIFICATION ═══") -for f in ["src/web/auth.py", "src/web/server.py", "main.py"]: - p = BASE / f - m = hashlib.md5(p.read_bytes()).hexdigest() - sz = p.stat().st_size - print(f" {f}: {sz} bytes, md5={m}") - -# Check for expected content -sp_check = (BASE / "src/web/server.py").read_text() -assert "from .auth import" in sp_check, "FAIL: auth import missing" -assert "_safe_detail" in sp_check, "FAIL: _safe_detail missing" -assert "auth_middleware" in sp_check, "FAIL: auth_middleware missing" -assert "is_auth_configured()" in sp_check, "FAIL: is_auth_configured missing" -assert "detail=_safe_detail(logger, e)" in sp_check, "FAIL: _safe_detail calls missing" -assert "ssl_certfile" in sp_check, "FAIL: ssl_certfile missing" -assert sp_check.count("detail=str(e)") == 0, "FAIL: remaining str(e) leaks" -print("\n✅ All patches verified and persisted!") diff --git a/patch_final.py b/patch_final.py deleted file mode 100644 index e4d947f..0000000 --- a/patch_final.py +++ /dev/null @@ -1,100 +0,0 @@ -#!/usr/bin/env python3 -"""Apply all remaining patches to server.py""" -import os, re - -path = "/home/oplabs/projects/pi-multifx-pedal/src/web/server.py" -content = open(path).read() -orig_md5 = __import__('hashlib').md5(content.encode()).hexdigest() -print(f"File: {len(content)} chars, md5={orig_md5}") - -# 1. Replace detail=str(e) with _safe_detail -old = "detail=str(e)" -new = "detail=_safe_detail(logger, e)" -count = content.count(old) -content = content.replace(old, new) -print(f"1. Replaced {count} detail=str(e) -> _safe_detail") - -# 2. Add SSL params to __init__ -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, "2: init pattern not found" -content = content.replace(old, new, 1) -print("2. SSL params added to __init__") - -# 3. Update start() for 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, "3: start pattern not found" -content = content.replace(old, new, 1) -print("3. SSL support added to start()") - -# 4. Update 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, "4: log pattern not found" -content = content.replace(old, new, 1) -print("4. Log message updated") - -# Write to a temp file first -tmp = "/tmp/server_patched.py" -open(tmp, "w").write(content) - -# Verify tmp has the changes -with open(tmp) as f: - tmp_content = f.read() -print(f"Written to {tmp}: {len(tmp_content)} chars") -print(f"Has _safe_detail: {tmp_content.count('_safe_detail')}") -print(f"Has ssl_certfile: {tmp_content.count('ssl_certfile')}") -print(f"Still has detail=str(e): {tmp_content.count('detail=str(e)')}") - -# Now copy over the original -import shutil -shutil.copy2(tmp, path) -print(f"Copied to {path}") - -# Final check -with open(path) as f: - final = f.read() -print(f"Final md5={__import__('hashlib').md5(final.encode()).hexdigest()}") -print(f"Final _safe_detail count: {final.count('_safe_detail')}") -print(f"Final ssl_certfile count: {final.count('ssl_certfile')}") diff --git a/patch_server.py b/patch_server.py deleted file mode 100644 index 34ff0e1..0000000 --- a/patch_server.py +++ /dev/null @@ -1,205 +0,0 @@ -#!/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}")