shawn/pi-multifx-pedal
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: remove rogue patch files from tracking
CI / test (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Shawn
2026-06-17 22:51:30 -04:00
parent ad5b369a10
commit a3ec86b814
4 changed files with 1 additions and 683 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -20,3 +20,4 @@ test_write.txt
test-drc.rpt
test_empty-drc.rptpatch_*.py
tests/test_issue_*.py
patch_*.py
patch_all.py
-378
View File
@@ -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: <pin>`` 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_hex>$<hash_hex>``
"""
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: <pin>``.
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!")
patch_final.py
-100
View File
@@ -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')}")
patch_server.py
-205
View File
@@ -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: <pin>``.
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}")