#!/usr/bin/env python3 """LAN mixer UI; commands are allowlisted and volume endpoints are input-only.""" import argparse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import ipaddress import json from pathlib import Path import secrets import socket import subprocess import threading import time from urllib.parse import urlsplit import input_controls as audio ROOT = Path(__file__).resolve().parent / 'web' TOKEN = secrets.token_urlsafe(32) JOBS = {} JOB_LOCK = threading.Lock() BUSY = False def validate_action(data): if not isinstance(data, dict): raise ValueError('요청 형식이 잘못됐습니다.') action=data.get('action') if action == 'volume': key=data.get('input'); level=data.get('volume') if key!='airplay' and (not isinstance(key,str) or not audio.MAC.fullmatch(key.upper())): raise ValueError('입력 장비 주소가 잘못됐습니다.') if type(level) is not int or not 0<=level<=100: raise ValueError('볼륨 범위: 0~100') return ['volume',key,str(level)] if action == 'mute': key=data.get('input'); mute=data.get('mute') if key!='airplay' and (not isinstance(key,str) or not audio.MAC.fullmatch(key.upper())): raise ValueError('입력 장비 주소가 잘못됐습니다.') if type(mute) is not bool: raise ValueError('음소거 값이 잘못됐습니다.') return ['mute',key,'on' if mute else 'off'] if action == 'output': mode=data.get('mode') if mode=='dac': pcm=data.get('pcm') if not isinstance(pcm,str) or not __import__('re').fullmatch(r'hw:CARD=[A-Za-z0-9_-]+,DEV=[0-9]+',pcm): raise ValueError('DAC 주소가 잘못됐습니다.') return ['output','dac','--pcm',pcm] if mode=='bluetooth': addr=data.get('address') if not isinstance(addr,str) or not audio.MAC.fullmatch(addr.upper()): raise ValueError('장비 주소가 잘못됐습니다.') return ['output','bluetooth',addr.upper()] raise ValueError('출력 종류가 잘못됐습니다.') if action=='scan': return ['scan','--seconds','15'] if action=='discoverable': on=data.get('on'); seconds=data.get('seconds') if type(on) is not bool: raise ValueError('검색 허용 값이 잘못됐습니다.') if not on: return ['discoverable','off'] if seconds not in (120,300,900) or type(seconds) is not int: raise ValueError('검색 허용 시간: 2, 5, 15분') return ['discoverable','on','--seconds',str(seconds)] if action=='pair-output': addr=data.get('address') if not isinstance(addr,str) or not audio.MAC.fullmatch(addr.upper()): raise ValueError('장비 주소가 잘못됐습니다.') return ['pair',addr.upper(),'--output','--bond'] raise ValueError('지원하지 않는 작업입니다.') def launch(args): global BUSY with JOB_LOCK: if BUSY: raise ValueError('진행 중인 작업이 있습니다. 완료 후 다시 시도해 주세요.') BUSY=True; key=secrets.token_hex(12) JOBS[key]={'done':False,'ok':False,'message':'처리 중입니다.'} if len(JOBS)>100: for old in list(JOBS)[:-50]: JOBS.pop(old,None) def work(): global BUSY try: # DAC selection must switch to DAC mode in the same serialized command. p=subprocess.run(['/usr/local/bin/radxa-audio',*args],capture_output=True,text=True,timeout=180) result={'done':True,'ok':p.returncode==0,'message':p.stdout[-2000:] if p.returncode==0 else (p.stderr+p.stdout)[-2000:]} except Exception as e: result={'done':True,'ok':False,'message':str(e)} with JOB_LOCK: JOBS[key]=result; BUSY=False threading.Thread(target=work,daemon=True).start() return key def allowed_host(host): try: name=urlsplit('//'+host).hostname if not name: return False if name in ('localhost',socket.gethostname(),socket.gethostname()+'.local'): return True addr=ipaddress.ip_address(name) ips={'127.0.0.1','::1'} data=json.loads(subprocess.check_output(['ip','-j','address'],text=True,timeout=3)) ips.update(a['local'] for link in data for a in link.get('addr_info',[])) return str(addr) in ips except (ValueError, OSError, subprocess.SubprocessError): return False class Handler(BaseHTTPRequestHandler): def reply(self,status,data,ctype='application/json; charset=utf-8'): body=json.dumps(data,ensure_ascii=False).encode() if isinstance(data,(dict,list)) else data self.send_response(status) self.send_header('Content-Type',ctype);self.send_header('Content-Length',str(len(body))) self.send_header('Cache-Control','no-store');self.send_header('X-Content-Type-Options','nosniff') self.send_header('Content-Security-Policy',"default-src 'self'; style-src 'self'; script-src 'self'; connect-src 'self'; frame-ancestors 'none'") self.end_headers(); self.wfile.write(body) def do_GET(self): if not allowed_host(self.headers.get('Host','')): return self.reply(403,{'error':'접속 주소가 허용되지 않습니다.'}) path=urlsplit(self.path).path try: if path=='/api/state': state=audio.state();state['csrf']=TOKEN;state['busy']=BUSY return self.reply(200,state) if path.startswith('/api/jobs/'): with JOB_LOCK: job=JOBS.get(path.rsplit('/',1)[-1]) return self.reply(200 if job else 404,job or {'error':'작업을 찾을 수 없습니다.'}) files={'/':'index.html','/app.js':'app.js','/style.css':'style.css'} if path not in files: return self.reply(404,{'error':'페이지가 없습니다.'}) mime={'/':'text/html; charset=utf-8','/app.js':'text/javascript; charset=utf-8','/style.css':'text/css; charset=utf-8'}[path] self.reply(200,(ROOT/files[path]).read_bytes(),mime) except Exception as e: self.reply(503,{'error':str(e)}) def do_POST(self): host=self.headers.get('Host','') origin=self.headers.get('Origin') if not allowed_host(host) or (origin and origin!='http://'+host): return self.reply(403,{'error':'동일한 접속 화면에서 요청해 주세요.'}) if not secrets.compare_digest(self.headers.get('X-Mixer-Token',''),TOKEN): return self.reply(403,{'error':'화면을 새로고침해 주세요.'}) if urlsplit(self.path).path!='/api/action': return self.reply(404,{'error':'없는 API입니다.'}) try: if not self.headers.get('Content-Type','').startswith('application/json'): raise ValueError('JSON 요청이 필요합니다.') size=int(self.headers.get('Content-Length','0')) if not 0