Claude (homelab agent)andClaude Opus 5.5
8318d550ec
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
106 lines
3.7 KiB
Python
Executable File
106 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Read-only HTTP server for archived SDR recordings (runs on home-server, stdlib only).
|
|
|
|
GET/HEAD /YYYY-MM-DD/sdr_YYYYMMDD_HHMMSS_FREQ.mp3 (Range supported)
|
|
GET /health -> "ok"
|
|
Nothing else: no directory listing, no other file types, no path traversal.
|
|
Only the Pi's sdr_web talks to it (firewall); it proxies audio to the radio page.
|
|
"""
|
|
import argparse
|
|
import os
|
|
import re
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
PATH_RE = re.compile(r"^/(\d{4}-\d{2}-\d{2})/([A-Za-z0-9-]+_\d{8}_\d{6}_\d+\.mp3)$")
|
|
ROOT = "/srv/storage/sdr-archive"
|
|
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
server_version = "sdr-archive/1"
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def log_message(self, fmt, *a):
|
|
pass
|
|
|
|
def small(self, code, text):
|
|
b = text.encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(b)))
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(b)
|
|
|
|
def do_POST(self):
|
|
self.small(405, "method not allowed")
|
|
|
|
do_PUT = do_DELETE = do_PATCH = do_OPTIONS = do_POST
|
|
|
|
def do_HEAD(self):
|
|
self.do_GET()
|
|
|
|
def do_GET(self):
|
|
if "\x00" in self.path or ".." in self.path:
|
|
return self.small(404, "not found")
|
|
if self.path == "/health":
|
|
return self.small(200, "ok")
|
|
m = PATH_RE.match(self.path.split("?", 1)[0])
|
|
if not m:
|
|
return self.small(404, "not found")
|
|
path = os.path.join(ROOT, m[1], m[2])
|
|
try:
|
|
fh = open(path, "rb")
|
|
except OSError:
|
|
return self.small(404, "not found")
|
|
with fh:
|
|
size = os.fstat(fh.fileno()).st_size
|
|
start, end, code = 0, size - 1, 200
|
|
rng = self.headers.get("Range")
|
|
if rng:
|
|
r = re.match(r"bytes=(\d*)-(\d*)$", rng.strip())
|
|
if not r or (r[1] == "" and r[2] == ""):
|
|
return self.small(416, "")
|
|
if r[1] == "":
|
|
start = max(0, size - int(r[2]))
|
|
else:
|
|
start = int(r[1])
|
|
if r[2] != "":
|
|
end = min(int(r[2]), size - 1)
|
|
if start >= size or start > end:
|
|
return self.small(416, "")
|
|
code = 206
|
|
length = end - start + 1
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "audio/mpeg")
|
|
self.send_header("Accept-Ranges", "bytes")
|
|
self.send_header("Content-Length", str(length))
|
|
self.send_header("Cache-Control", "public, max-age=86400")
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
if code == 206:
|
|
self.send_header("Content-Range", f"bytes {start}-{end}/{size}")
|
|
self.end_headers()
|
|
if self.command == "HEAD":
|
|
return
|
|
fh.seek(start)
|
|
left = length
|
|
try:
|
|
while left > 0:
|
|
chunk = fh.read(min(65536, left))
|
|
if not chunk:
|
|
break
|
|
self.wfile.write(chunk)
|
|
left -= len(chunk)
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--bind", default="127.0.0.1")
|
|
ap.add_argument("--port", type=int, default=18091)
|
|
ap.add_argument("--root", default=ROOT)
|
|
a = ap.parse_args()
|
|
ROOT = a.root
|
|
ThreadingHTTPServer.daemon_threads = True
|
|
ThreadingHTTPServer((a.bind, a.port), H).serve_forever()
|