routes/static_files.py
Rama main · réplica de lectura del repositorio original.
"""
routes/static_files.py — Static Asset Serving Blueprint (P2-T8)
Extracted from server.py during Phase 2 blueprint split.
Registers routes:
GET /sounds/<path:filepath> — sound effect files
GET /uploads/<path:filename> — uploaded user files
GET /src/<path:filepath> — frontend JS/CSS source modules
GET /known_faces/<name>/<filename> — face recognition photos
GET /api/dj-sound — DJ soundboard API (list/play)
"""
import hmac
import json
import logging
import os
import random
import re
from datetime import datetime, timezone
from pathlib import Path
from flask import Blueprint, Response, jsonify, redirect, request, send_file
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Blueprint
# ---------------------------------------------------------------------------
static_files_bp = Blueprint('static_files', __name__)
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
from services.paths import APP_ROOT, SOUNDS_DIR, UPLOADS_DIR, KNOWN_FACES_DIR, STATIC_DIR
BASE_DIR = APP_ROOT
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
# ---------------------------------------------------------------------------
# Uploads index — ONE ledger of every client-provided file, shared with the
# host-side intake helper (scripts/uploads/jambot-upload-intake.py writes the
# same schema for SMS/email media). Dotfile so serve_upload never exposes it.
# The agent reads it at workspace/uploads/.uploads-index.jsonl.
# ---------------------------------------------------------------------------
UPLOADS_INDEX = UPLOADS_DIR / '.uploads-index.jsonl'
def append_upload_index(record: dict) -> None:
"""Append one entry to the uploads ledger. Fail-soft: an index problem
must never fail the upload itself."""
try:
import json as _json
with open(UPLOADS_INDEX, 'a', encoding='utf-8') as fh:
fh.write(_json.dumps(record, ensure_ascii=False) + '\n')
try:
UPLOADS_INDEX.chmod(0o666) # host-side writers run as a different uid
except OSError:
pass
except Exception as exc: # noqa: BLE001
logger.warning('uploads-index append failed: %s', exc)
# ---------------------------------------------------------------------------
# DJ Sounds catalogue
# ---------------------------------------------------------------------------
DJ_SOUNDS = {
'air_horn': {
'description': 'Classic stadium air horn - ba ba baaaa!',
'when_to_use': 'Before drops, hype moments, celebrating wins, hip-hop DJ style'
},
'scratch_long': {
'description': 'Extended DJ scratch solo - wicka wicka',
'when_to_use': 'Transitions, hip-hop moments, showing off DJ skills'
},
'rewind': {
'description': 'DJ rewind - pull up selecta!',
'when_to_use': 'Going back, replaying something fire, dancehall pull-ups'
},
'record_stop': {
'description': 'Record stopping abruptly',
'when_to_use': 'Stopping everything, dramatic pause, cutting the music'
},
'impact': {
'description': 'Punchy cinematic impact hit',
'when_to_use': 'Punctuating statements, transitions, emphasis'
},
'crowd_cheer': {
'description': 'Nightclub crowd cheering and going wild',
'when_to_use': 'Big wins, amazing moments, festival energy, applause'
},
'crowd_hype': {
'description': 'Hyped up rave crowd losing their minds',
'when_to_use': 'Peak energy moments, party atmosphere'
},
'yeah': {
'description': 'Hype man YEAH! vocal shot',
'when_to_use': 'Hyping up, agreement, energy boost'
},
'lets_go': {
'description': 'LETS GO! vocal chant',
'when_to_use': 'Starting something, getting pumped, motivation'
},
'laser': {
'description': 'Retro arcade laser zap - pew pew',
'when_to_use': 'Sci-fi moments, gaming references, 80s vibes'
},
'gunshot': {
'description': 'Dancehall gunshot sound - gun finger!',
'when_to_use': 'Reggae/dancehall vibes, shooting down bad ideas'
},
'bruh': {
'description': 'Classic bruh sound effect',
'when_to_use': 'Facepalm moments, disappointment, when someone says something dumb'
},
'sad_trombone': {
'description': 'Sad trombone wah wah wah - womp womp',
'when_to_use': 'Fails, disappointments, when things go wrong'
},
}
# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
def _safe_path(base_dir: Path, *parts) -> Path | None:
"""
Resolve a path within base_dir, rejecting any traversal outside it.
Returns the resolved Path on success, or None if traversal is detected.
"""
try:
resolved = (base_dir / Path(*parts)).resolve()
base_resolved = base_dir.resolve()
if resolved == base_resolved or base_resolved in resolved.parents:
return resolved
except Exception:
pass
return None
@static_files_bp.route('/sounds/<path:filepath>')
def serve_sound(filepath):
"""Serve sound effect files (including subdirectories like DJ-clips/)"""
sound_path = _safe_path(SOUNDS_DIR, filepath)
if sound_path is None:
return jsonify({"error": "Invalid path"}), 400
if sound_path.exists():
return send_file(sound_path, mimetype='audio/mpeg')
return jsonify({"error": "Sound not found"}), 404
# ---------------------------------------------------------------------------
# Upload constants & helpers
# ---------------------------------------------------------------------------
# Hard limit enforced before writing to disk (100 MB)
_MAX_UPLOAD_BYTES = 100 * 1024 * 1024
# Maximum characters returned to the AI as content_preview for text files.
# Files at or below this threshold are inlined verbatim into the conversation.
# Larger files set content_preview_truncated=True and the agent is directed to
# use its Read tool on the saved path instead (which lives in a prunable
# tool-result block rather than an un-prunable user message).
_MAX_PREVIEW_CHARS = 50_000
# Maximum characters for extracted text from binary documents (PDF/DOCX/XLSX/PPTX).
# Binary extractors sanitize their own output, so they share this cap via
# _sanitize_text below.
_MAX_BINARY_PREVIEW_CHARS = 20_000
# Only block executables — accept everything else (weird exports, unknown formats, etc.)
_BLOCKED_EXTENSIONS = {
'.exe', '.bat', '.cmd', '.com', '.scr', '.pif',
'.msi', '.dll', '.sys', '.vbs', '.vbe', '.wsh',
'.wsf', '.ps1', '.sh', '.cpl', '.inf', '.reg',
}
# Control characters to strip from extracted text (keeps \t \n \r)
_CTRL_RE = re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]')
def _sanitize_text(text: str) -> str:
"""Strip control chars, collapse excessive blank lines, cap at binary preview limit.
Used by PDF/DOCX/XLSX/PPTX extractors — they don't signal truncation back to
the upload handler so use the tighter cap."""
text = _CTRL_RE.sub('', text)
text = re.sub(r'\n{4,}', '\n\n\n', text) # no more than 3 consecutive blank lines
return text[:_MAX_BINARY_PREVIEW_CHARS].strip()
def _read_text_preview(path: Path) -> tuple[str, bool]:
"""Read a text file for inline preview. Returns (sanitized_text, truncated).
Never loads more than _MAX_PREVIEW_CHARS + 1 chars into memory, so large
uploads don't spike RSS."""
with open(path, 'r', errors='replace') as fh:
raw = fh.read(_MAX_PREVIEW_CHARS + 1)
truncated = len(raw) > _MAX_PREVIEW_CHARS
cleaned = _CTRL_RE.sub('', raw[:_MAX_PREVIEW_CHARS])
cleaned = re.sub(r'\n{4,}', '\n\n\n', cleaned)
return cleaned.strip(), truncated
def _extract_pdf(path: Path) -> str:
"""Extract text from a PDF using pypdf. Returns sanitized string."""
from pypdf import PdfReader
reader = PdfReader(str(path))
pages = []
for i, page in enumerate(reader.pages):
try:
text = page.extract_text() or ''
pages.append(text)
except Exception:
pages.append(f'[Page {i + 1}: extraction failed]')
return _sanitize_text('\n\n'.join(pages))
def _extract_docx(path: Path) -> str:
"""Extract text from a .docx using python-docx. Returns sanitized string."""
from docx import Document
doc = Document(str(path))
parts = []
for para in doc.paragraphs:
t = para.text.strip()
if t:
parts.append(t)
for table in doc.tables:
for row in table.rows:
cells = [c.text.strip() for c in row.cells if c.text.strip()]
if cells:
parts.append(' | '.join(cells))
return _sanitize_text('\n'.join(parts))
def _extract_xlsx(path: Path) -> str:
"""Extract cell values from a .xlsx using openpyxl. Returns sanitized string."""
import openpyxl
wb = openpyxl.load_workbook(str(path), read_only=True, data_only=True)
parts = []
try:
for sheet in wb.worksheets:
parts.append(f'[Sheet: {sheet.title}]')
for row in sheet.iter_rows(values_only=True):
cells = [str(c) for c in row if c is not None and str(c).strip()]
if cells:
parts.append('\t'.join(cells))
finally:
wb.close()
return _sanitize_text('\n'.join(parts))
def _extract_pptx(path: Path) -> str:
"""Extract text from a .pptx using python-pptx. Returns sanitized string."""
from pptx import Presentation
prs = Presentation(str(path))
parts = []
for i, slide in enumerate(prs.slides, 1):
slide_texts = []
for shape in slide.shapes:
if shape.has_text_frame:
for para in shape.text_frame.paragraphs:
t = para.text.strip()
if t:
slide_texts.append(t)
if slide_texts:
parts.append(f'[Slide {i}]')
parts.extend(slide_texts)
return _sanitize_text('\n'.join(parts))
def _call_skill_runner(path: Path, original_name: str) -> str | None:
"""
Try to extract document text via the shared skill-runner service.
Returns extracted text on success, None if the service is unavailable.
Falls back gracefully so local extractors can take over.
"""
try:
import requests
with open(path, 'rb') as fh:
resp = requests.post(
'http://skill-runner:8900/extract',
files={'file': (original_name, fh)},
data={'filename': original_name},
timeout=30,
)
if resp.ok:
data = resp.json()
return data.get('text', '')
logger.warning('skill-runner /extract returned %d for %s', resp.status_code, original_name)
except Exception as exc:
logger.debug('skill-runner unavailable, using local extractors: %s', exc)
return None
@static_files_bp.route('/api/upload', methods=['POST'])
def upload_file():
"""Accept a file upload from the text panel and save to uploads/."""
import mimetypes
import uuid
if 'file' not in request.files:
import io as _io
import time as _time
ct = request.content_type or ''
# iOS/Cloudflare sometimes sends camera photo as raw binary body (no multipart).
# Detect: Content-Type is an image MIME, body has data, no other fields.
if (ct.startswith('image/') or ct == 'application/octet-stream') and request.content_length and request.content_length > 0:
raw = request.get_data()
if raw:
ext_guess = mimetypes.guess_extension(ct) or '.jpg'
if ext_guess in ('.jpe', '.jpeg'):
ext_guess = '.jpg'
fname_raw = f'capture_{int(_time.time())}{ext_guess}'
logger.info('upload: raw-body fallback — ct=%s len=%s fname=%s', ct, len(raw), fname_raw)
from werkzeug.datastructures import FileStorage
f_raw = FileStorage(stream=_io.BytesIO(raw), filename=fname_raw, content_type=ct)
# Jump straight to save logic via a local alias
_raw_file = f_raw
else:
_raw_file = None
else:
_raw_file = None
if _raw_file is None:
logger.warning(
'upload: no file field — content_type=%r content_length=%r '
'files_keys=%r form_keys=%r user_agent=%r',
ct,
request.content_length,
list(request.files.keys()),
list(request.form.keys()),
request.headers.get('User-Agent', '')[:120],
)
return jsonify({'error': 'No file provided'}), 400
# Use the raw-body file as `f` for the rest of the route
f = _raw_file
else:
f = request.files['file']
if not f.filename:
# iOS camera blobs arrive with no filename — generate one from content-type
import time as _time
ct = f.content_type or 'image/jpeg'
ext_guess = mimetypes.guess_extension(ct) or '.jpg'
# mimetypes sometimes returns .jpe or .jpeg — normalize to .jpg
if ext_guess in ('.jpe', '.jpeg'):
ext_guess = '.jpg'
f.filename = f'capture_{int(_time.time())}{ext_guess}'
logger.info('upload: empty filename from client; generated fallback %s (content_type=%s)', f.filename, ct)
# --- Sanitize filename, validate extension ---
original_name = Path(f.filename).name
ext = Path(original_name).suffix.lower()
if ext in _BLOCKED_EXTENSIONS:
return jsonify({'error': f'File type "{ext}" is not allowed'}), 415
# --- Size check before writing to disk ---
# Seek to end to get byte length without reading into memory
f.stream.seek(0, 2)
file_size = f.stream.tell()
f.stream.seek(0)
if file_size > _MAX_UPLOAD_BYTES:
return jsonify({'error': 'File too large (100 MB max)'}), 413
# --- Save with original-stem prefix + short uuid suffix so the agent
# can find files by the name the user uploaded them as. UUID suffix
# prevents collisions when the same name is uploaded twice. ---
stem_safe = re.sub(r'[^A-Za-z0-9._-]+', '_', Path(original_name).stem)[:80].strip('._-') or 'file'
safe_name = f"{stem_safe}__{uuid.uuid4().hex[:8]}{ext}"
dest = UPLOADS_DIR / safe_name
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
f.save(str(dest))
# --- HEIC/HEIF → JPG auto-conversion ---
# Browsers can't display HEIC natively. Convert on ingest so the file is
# immediately usable. Uses pillow-heif (in requirements.txt).
if ext in ('.heic', '.heif'):
jpg_name = f"{stem_safe}__{uuid.uuid4().hex[:8]}.jpg"
jpg_dest = UPLOADS_DIR / jpg_name
try:
import pillow_heif
from PIL import Image as _Image
pillow_heif.register_heif_opener()
with _Image.open(str(dest)) as _img:
_img.convert('RGB').save(str(jpg_dest), 'JPEG', quality=90)
dest.rename(dest.parent / (dest.stem + '.heic.original'))
dest = jpg_dest
safe_name = jpg_name
ext = '.jpg'
logger.info(f'### HEIC→JPG conversion ok: {jpg_name}')
except Exception as _e:
logger.warning(f'### HEIC→JPG conversion error: {_e}')
mime = f.mimetype or mimetypes.guess_type(original_name)[0] or ''
# Re-derive mime from converted extension so HEIC→JPG is flagged as image
if ext == '.jpg':
mime = 'image/jpeg'
is_image = mime.startswith('image/')
# Full public URL — the agent (and the index) always get the complete location
url_full = f"https://{request.host}/uploads/{safe_name}"
result = {
'original_name': original_name,
'path': str(dest),
'filename': safe_name,
'url': f'/uploads/{safe_name}',
'url_full': url_full,
'agent_path': f'workspace/uploads/{safe_name}',
'file_size': file_size,
}
append_upload_index({
'at': datetime.now(timezone.utc).isoformat(timespec='seconds'),
'filename': safe_name,
'original_name': original_name,
'url': url_full,
'agent_path': f'workspace/uploads/{safe_name}',
'source': 'chat',
'from': '',
'mime': mime,
'size': file_size,
'note': '',
})
if is_image:
result['type'] = 'image'
return jsonify(result)
result['type'] = 'file'
# --- Extract readable content by type ---
_BINARY_EXTS = {'.pdf', '.docx', '.xlsx', '.pptx'}
try:
if ext in _BINARY_EXTS:
# Try shared skill-runner first (preferred — keeps this container lean)
text = _call_skill_runner(dest, original_name)
# Fall back to local extractors if skill-runner unavailable
if text is None:
if ext == '.pdf':
text = _extract_pdf(dest)
elif ext == '.docx':
text = _extract_docx(dest)
elif ext == '.xlsx':
text = _extract_xlsx(dest)
elif ext == '.pptx':
text = _extract_pptx(dest)
if text:
result['content_preview'] = text
result['extracted_type'] = ext.lstrip('.')
else:
result['extraction_error'] = (
f'Could not extract text from {ext} file. '
'Install skill-runner or document packages to enable this.'
)
else:
# Plain text / code / CSV — stream the first _MAX_PREVIEW_CHARS only
text_types = {'text/', 'application/json', 'application/xml', 'application/javascript'}
if any(mime.startswith(t) for t in text_types) or ext in {
'.txt', '.md', '.csv', '.log',
'.py', '.js', '.ts', '.json', '.yaml', '.yml',
'.html', '.css',
}:
preview, truncated = _read_text_preview(dest)
result['content_preview'] = preview
if truncated:
result['content_preview_truncated'] = True
except Exception as exc:
logger.warning('Document extraction failed for %s: %s', original_name, exc)
result['extraction_error'] = f'Could not extract text from {ext} file'
return jsonify(result)
@static_files_bp.route('/static/emulators/<path:filepath>')
def serve_emulator(filepath):
"""Serve bundled emulator files (js-dos, etc.) from /app/static/emulators/."""
emulators_dir = STATIC_DIR / 'emulators'
path = _safe_path(emulators_dir, filepath)
if path is None:
return jsonify({"error": "Invalid path"}), 400
if not path.exists():
return jsonify({"error": "File not found"}), 404
mime_types = {'.js': 'application/javascript', '.css': 'text/css', '.wasm': 'application/wasm'}
mime = mime_types.get(path.suffix, 'application/octet-stream')
response = send_file(path, mimetype=mime)
response.headers['Cache-Control'] = 'public, max-age=86400'
response.headers['Access-Control-Allow-Origin'] = '*'
return response
# ---------------------------------------------------------------------------
# Uploads auth (2026-10-06): /uploads/ was served to anyone with the URL, and search engines index it.
# A plain login gate would BREAK legitimate machine consumers (scripts fetching files over HTTP, MMS
# providers fetching media), so this ships in phases, controlled by UPLOADS_AUTH_MODE:
# off — no check (self-hosted default when CANVAS_REQUIRE_AUTH is off)
# shadow — serve as before, but log every request that WOULD be denied (the default)
# enforce — deny unless the request carries a grant (401, or a login redirect for browsers)
# A request is granted by ANY of:
# session — a valid Clerk token (the same check canvas pages use)
# signature — ?exp=&sig= minted by services/upload_links.py with this instance's UPLOADS_SIGNING_KEY
# agent_key — the internal X-Agent-Key header matching AGENT_API_KEY (same check as app.require_auth)
# Each would-deny row records the state of all three grants, so enforcement can be decided per consumer.
# The would-deny log is a dotfile, so serve_upload never exposes it.
# ---------------------------------------------------------------------------
UPLOADS_AUTH_MODE = os.getenv('UPLOADS_AUTH_MODE', 'shadow').strip().lower()
UPLOADS_REQUIRE_AUTH = os.getenv('CANVAS_REQUIRE_AUTH', 'false').lower() == 'true'
UPLOADS_SHADOW_LOG = UPLOADS_DIR / '.uploads-auth-shadow.jsonl'
_UPLOADS_SHADOW_MAX_BYTES = 5 * 1024 * 1024
# Always-private areas (2026-11-09). The phased mode above still leaves /uploads/ public on every
# instance (shadow), but some areas are client intake that is never a page asset: Field Dock photos
# are receipts / work orders / invoices with a Bill-To name, address and phone, and uploads/sms/ is
# raw inbound MMS. These, and the /api/uploads listing (the index of every filename), are ENFORCED
# whenever CANVAS_REQUIRE_AUTH is on, whatever UPLOADS_AUTH_MODE says, with the same three grants.
# Prefixes are relative to UPLOADS_DIR and matched on the RESOLVED path (so '../', './', '//' and
# symlinks cannot walk around them). An instance adds its own with UPLOADS_PRIVATE_PREFIXES
# (comma-separated, e.g. "receipts/,leads/"); the built-in ones cannot be removed by env.
# Top-level sms-*.jpg (sms-router MMS saves) are NOT here: they are deliberately public website
# assets (sms-router app.py) and are used by public canvas pages; they wait for enforce mode.
_BUILTIN_PRIVATE_PREFIXES = ('field-dock/', 'sms/')
def _private_prefixes():
extra = os.getenv('UPLOADS_PRIVATE_PREFIXES', '')
out = list(_BUILTIN_PRIVATE_PREFIXES)
for p in extra.split(','):
p = p.strip().strip('/')
if p and not p.startswith('.') and '..' not in p:
out.append(p + '/')
return tuple(out)
def _upload_is_private(rel_path):
"""True when a path relative to UPLOADS_DIR (posix, already resolved) is in an always-private
area. Case-folded so 'Field-Dock/' cannot slip past on a case-insensitive filesystem. The area's
own directory ('field-dock', no trailing slash) counts as inside it."""
rel = rel_path.strip('/').lower() + '/'
return any(rel.startswith(p.lower()) for p in _private_prefixes())
def _upload_rel(path):
"""The resolved path of `path` relative to the resolved UPLOADS_DIR, posix; '' if outside."""
try:
return Path(os.path.realpath(path)).relative_to(Path(os.path.realpath(UPLOADS_DIR))).as_posix()
except (ValueError, OSError):
return ''
def _upload_has_session():
from services.auth import get_token_from_request, verify_clerk_token
try:
token = get_token_from_request()
return bool(token and verify_clerk_token(token))
except Exception:
return False
def _upload_session_state():
"""'ok' | 'absent' | why a presented token failed. The grant is decided by verify_clerk_token
alone; the failure reason is a diagnostic. 'expired' is the common one: the __session cookie
lives 30 days but carries a ~60 s Clerk token that only the app shell refreshes, so a browser
that navigated away from the shell sends a stale token."""
from services.auth import get_token_from_request
if _upload_has_session():
return 'ok'
try:
token = get_token_from_request()
except Exception:
return 'invalid'
if not token:
return 'absent'
try:
from services.auth import token_failure_reason
why = token_failure_reason(token)
except Exception:
return 'invalid'
# 'ok' here means it verified a moment after verify_clerk_token said no; the decision stands.
return why if why in ('expired', 'not-allowed', 'no-keys') else 'invalid'
def _upload_agent_key_state():
"""The internal agent key, checked exactly as app.require_auth does (X-Agent-Key vs
AGENT_API_KEY, no grant when the key is unset), with a constant-time comparison."""
presented = request.headers.get('X-Agent-Key')
if not presented:
return 'absent'
expected = os.getenv('AGENT_API_KEY', '').strip()
if not expected:
return 'unconfigured'
return 'ok' if hmac.compare_digest(presented.encode('utf-8'), expected.encode('utf-8')) else 'invalid'
def _upload_signature_state(filename):
from services import upload_links
return upload_links.verify('/uploads/' + filename, request.args.get('exp'), request.args.get('sig'))
def _upload_grants(filename, listing=False):
"""Evaluate grants cheapest-first, stopping at the first that holds. Returns
(granted_by or None, {grant: state}) — states are only complete when nothing granted.
A listing (/api/uploads) has no signed-link grant: a signature opens one file, never the index."""
states = {}
checks = [('agent_key', _upload_agent_key_state)]
if not listing:
checks.append(('signature', lambda: _upload_signature_state(filename)))
checks.append(('session', _upload_session_state))
for name, check in checks:
states[name] = check()
if states[name] == 'ok':
return name, states
return None, states
def _upload_deny_reason(states):
"""The most specific reason: the first PRESENTED credential that failed, else no credential."""
for name in ('signature', 'agent_key', 'session'):
if states.get(name) not in (None, 'absent'):
return f'{name}-{states[name]}'
return 'no-credential'
def _upload_wants_html():
return request.headers.get('Accept', '').startswith('text/html')
def _upload_enforce_action(listing=False):
"""What enforce would answer this request: a browser navigation is sent through the login
shell, which refreshes a stale session and comes straight back (the canvas-page bounce);
anything else gets a bare 401 and does not recover on its own."""
return 'login-redirect' if (not listing and _upload_wants_html()) else '401'
def _upload_shadow_record(filename, exists, states=None, reason=None, listing=False, mode=None):
try:
if UPLOADS_SHADOW_LOG.exists() and UPLOADS_SHADOW_LOG.stat().st_size > _UPLOADS_SHADOW_MAX_BYTES:
return
ref = request.headers.get('Referer', '')
row = {
'ts': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
'file': filename[:200], 'exists': bool(exists), 'mode': mode or UPLOADS_AUTH_MODE,
'ua': request.headers.get('User-Agent', '')[:160],
'ref_host': re.sub(r'^https?://([^/]+).*$', r'\1', ref)[:120] if ref else '',
'accept': request.headers.get('Accept', '')[:60],
'reason': reason or 'no-credential',
'grants': states or {},
'enforce_action': _upload_enforce_action(listing),
}
fd = os.open(str(UPLOADS_SHADOW_LOG), os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o640)
try:
os.write(fd, (json.dumps(row) + '\n').encode('utf-8'))
finally:
os.close(fd)
except Exception as e: # measurement must never break serving
logger.warning('[uploads-auth] shadow record failed: %s', e)
def _upload_access_denied(filename, exists, listing=False, private=False):
"""None = serve. Otherwise the deny response.
Every route that serves bytes from UPLOADS_DIR calls this BEFORE checking existence:
serve_upload, routes/albums.py serve_album_photo, and the /api/uploads listing (listing=True,
which takes no signed-link grant and always answers JSON).
The listing and private=True (an always-private area, see _BUILTIN_PRIVATE_PREFIXES) are
enforced whenever auth is required, in every UPLOADS_AUTH_MODE; everything else is denied in
enforce mode only. Their denial rows are logged with mode 'always-enforce' (a real denial,
not a would-deny)."""
if not UPLOADS_REQUIRE_AUTH:
return None
# Every caller gets the private-area rule on the path as requested; callers that know the
# RESOLVED path (serve_upload, albums) also pass private= for it.
private = private or (not listing and _upload_is_private(filename))
always = listing or private
if not always and UPLOADS_AUTH_MODE not in ('shadow', 'enforce'):
return None
granted_by, states = _upload_grants(filename, listing=listing)
if granted_by:
return None
reason = _upload_deny_reason(states)
_upload_shadow_record(filename, exists, states, reason, listing=listing,
mode='always-enforce' if always else None)
if not always and UPLOADS_AUTH_MODE != 'enforce':
return None
logger.warning('[uploads-auth] DENIED %s (%s)', filename[:200], reason)
if _upload_enforce_action(listing) == 'login-redirect':
return redirect('/?redirect=/uploads/' + filename)
if listing:
return jsonify({'error': 'Unauthorized', 'code': 'auth_required'}), 401
return 'Unauthorized', 401
_UPLOADS_LISTING = '(listing) /api/uploads'
@static_files_bp.route('/api/uploads', methods=['GET'])
def list_uploads():
"""List uploaded files. Optional ?type=image|text filter.
Moved here from server.py (uploads auth phase 3). Since 2026-11-09 the listing is ALWAYS gated
when auth is required (every UPLOADS_AUTH_MODE): it is the index that makes every filename on the
instance discoverable. app.require_auth also gates it now ('/api/upload' is public as an exact
path for the POST only); this route gate stays as defense in depth. Grants: Clerk session
(Bearer or __session cookie) or X-Agent-Key; a signed link never opens the index.
"""
denied = _upload_access_denied(_UPLOADS_LISTING, True, listing=True)
if denied is not None:
return denied
file_type = request.args.get("type") # "image", "text", or None for all
image_exts = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
text_exts = {".pdf", ".txt", ".md", ".json", ".csv", ".html", ".js", ".py", ".ts", ".css"}
files = []
for f in sorted(UPLOADS_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True):
if f.is_dir() or f.name.startswith("."):
continue
ext = f.suffix.lower()
is_image = ext in image_exts
kind = "image" if is_image else ("text" if ext in text_exts else "other")
if file_type and kind != file_type:
continue
stat = f.stat()
files.append({
"filename": f.name,
"url": f"/uploads/{f.name}",
"type": kind,
"size": stat.st_size,
"modified": stat.st_mtime,
})
return jsonify({"files": files, "count": len(files)})
@static_files_bp.route('/uploads/<path:filename>')
def serve_upload(filename):
"""Serve uploaded files (path traversal guarded).
NO-CACHE POLICY (do not change without reading docs/jambot/no-cache-policy.md):
Uploads contain icons, wallpapers, agent-generated assets, and canvas page
media that agents and admins regenerate live. Caching breaks the "live
updates always visible" guarantee of the canvas system. The ONLY assets
that may be cached are known_faces photos (handled by a separate route).
Optimize for size (smaller files), not for browser cache hits.
"""
# dotfiles (e.g. the .uploads-index.jsonl ledger) are internal — never served
if any(part.startswith('.') for part in Path(filename).parts):
return jsonify({"error": "File not found"}), 404
upload_path = _safe_path(UPLOADS_DIR, filename)
if upload_path is None:
return jsonify({"error": "Invalid path"}), 400
# Uploads auth: decide access BEFORE answering whether the file exists (no 401/404 existence oracle).
# Private areas are judged on the RESOLVED path, so './', '//', '../' and symlinks cannot dodge them.
private = _upload_is_private(_upload_rel(upload_path)) or _upload_is_private(filename)
denied = _upload_access_denied(filename, upload_path.exists(), private=private)
if denied is not None:
return denied
if not upload_path.exists():
return jsonify({"error": "File not found"}), 404
resp = send_file(upload_path)
resp.headers['X-Robots-Tag'] = 'noindex, nofollow, noarchive'
# F-5 (2026-07-15): neutralize STORED XSS. Uploads are user/agent-supplied; an
# .html or .svg served with a rendering content-type executes in OUR origin and can
# steal the Clerk session. Serve only known-safe media types inline with their real
# mime; serve EVERYTHING else (html, svg, xml, js, ...) as a non-rendering
# text/plain attachment. Agent code-file uploads still store + download fine — they
# just never execute in-browser. nosniff + a locked-down CSP are belt-and-suspenders.
_ext = upload_path.suffix.lower()
_SAFE_INLINE = {
'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.ico', '.avif',
'.mp4', '.webm', '.ogg', '.mov', '.mp3', '.wav', '.m4a', '.pdf',
}
# Inert DOWNLOAD formats (2026-10-12): office documents, CSV and zip never execute in a
# browser, so they keep their REAL mime — still an attachment, still nosniff + the
# sandbox CSP below. Downgrading them to text/plain made iOS Safari show a client's .xlsx
# as mojibake and save it as "<name>.xlsx.txt". Anything not listed (html, svg, js, xml,
# unknown) still falls back to text/plain, so the F-5 guard fails closed as before.
_DOWNLOAD_MIME = {
'.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
'.xls': 'application/vnd.ms-excel',
'.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'.doc': 'application/msword',
'.pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation',
'.ppt': 'application/vnd.ms-powerpoint',
'.ods': 'application/vnd.oasis.opendocument.spreadsheet',
'.odt': 'application/vnd.oasis.opendocument.text',
'.csv': 'text/csv; charset=utf-8',
'.zip': 'application/zip',
}
if _ext not in _SAFE_INLINE:
resp.headers['Content-Type'] = _DOWNLOAD_MIME.get(_ext, 'text/plain; charset=utf-8')
resp.headers['Content-Disposition'] = (
'attachment; filename="%s"' % Path(filename).name.replace('"', ''))
resp.headers['X-Content-Type-Options'] = 'nosniff'
# CSP by type (2026-08-02): the strict `sandbox; default-src 'none'` neutralizes stored
# XSS in html/svg/js, but `sandbox` with no allowlist ALSO stops the browser's built-in
# media/pdf viewer from loading — so direct links to an uploaded .mp4/.mp3/.pdf served 200
# but played nothing. Inert media served inline with
# its real mime + nosniff has no script surface, so it gets a CSP that permits only
# self-origin media/images/pdf to render while still blocking all script + framing, and
# crucially drops `sandbox`. Non-safe types (already downgraded to a text/plain attachment
# above) keep the maximally-locked CSP.
if _ext in _SAFE_INLINE:
resp.headers['Content-Security-Policy'] = (
"default-src 'none'; img-src 'self' data: blob:; media-src 'self' blob:; "
"object-src 'self'; style-src 'unsafe-inline'; frame-ancestors 'none'")
else:
resp.headers['Content-Security-Policy'] = "default-src 'none'; sandbox; frame-ancestors 'none'"
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
resp.headers['Pragma'] = 'no-cache'
resp.headers['Expires'] = '0'
return resp
@static_files_bp.route('/src/<path:filepath>')
def serve_src(filepath):
"""Serve frontend source files (JS, CSS modules)"""
# P7-T3 security: prevent path traversal (same guard used by serve_sound)
src_path = _safe_path(APP_ROOT / 'src', filepath)
if src_path is None:
return jsonify({"error": "Invalid path"}), 400
if not src_path.exists():
return jsonify({"error": "File not found"}), 404
mime_types = {
'.js': 'application/javascript',
'.css': 'text/css',
'.html': 'text/html',
'.json': 'application/json',
}
mime_type = mime_types.get(src_path.suffix.lower(), 'text/plain')
resp = send_file(src_path, mimetype=mime_type)
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
return resp
@static_files_bp.route('/known_faces/<name>/<filename>')
def serve_face_photo(name, filename):
"""Serve face photos for the My Face section.
CACHE ALLOWED — face photos are the documented exception to the
no-cache policy (docs/jambot/no-cache-policy.md). Face content does not
update live and is identity-stable. A long browser cache here is a
deliberate perf win and does NOT violate the canvas/icons rule.
"""
photo_path = _safe_path(KNOWN_FACES_DIR, name, filename)
if photo_path is None:
return jsonify({"error": "Invalid path"}), 400
if photo_path.exists():
resp = send_file(photo_path)
resp.headers['Cache-Control'] = 'public, max-age=86400'
return resp
return jsonify({"error": "Photo not found"}), 404
@static_files_bp.route('/api/dj-sound', methods=['GET'])
def handle_dj_sound():
"""
DJ Soundboard endpoint.
Query params:
- action: 'list' or 'play'
- sound: sound name (e.g., 'air_horn', 'scratch', 'siren_rise')
Returns sound info or triggers playback.
"""
action = request.args.get('action', 'list')
sound = request.args.get('sound', '')
if action == 'list':
sounds_list = [
{
'name': name,
'description': info['description'],
'when_to_use': info['when_to_use'],
'available': (SOUNDS_DIR / f"{name}.mp3").exists(),
}
for name, info in DJ_SOUNDS.items()
]
return jsonify({
'action': 'list',
'sounds': sounds_list,
'count': len(sounds_list),
'response': (
f"Soundboard loaded! {len(sounds_list)} effects ready. "
"I got air horns, sirens, scratches, crowd effects, and more!"
),
})
if action == 'play':
if not sound:
sound = random.choice(list(DJ_SOUNDS.keys()))
sound_lower = sound.lower().replace(' ', '_').replace('-', '_')
matched = next(
(name for name in DJ_SOUNDS if sound_lower in name or name in sound_lower),
None,
)
if not matched:
matched = next(
(name for name in DJ_SOUNDS
if any(word in name for word in sound_lower.split('_'))),
None,
)
if not matched:
return jsonify({
'action': 'error',
'response': (
f"No sound matching '{sound}'. "
"Try: air_horn, siren, scratch, applause, bass_drop, rewind..."
),
})
sound_file = SOUNDS_DIR / f"{matched}.mp3"
if not sound_file.exists():
return jsonify({
'action': 'error',
'response': f"Sound file for '{matched}' not found. Need to generate it first!",
})
info = DJ_SOUNDS[matched]
return jsonify({
'action': 'play',
'sound': matched,
'description': info['description'],
'url': f"/sounds/{matched}.mp3",
'response': f"*{info['description'].upper()}* 🎵",
})
return jsonify({'error': 'Unknown action'}), 400
@static_files_bp.route('/manifest.json')
def serve_manifest():
"""PWA Web App Manifest — dynamically injects CLIENT_NAME for per-tenant PWA identity"""
import json as _json, os as _os
client_name = _os.environ.get("CLIENT_NAME", "").strip()
path = STATIC_DIR / 'manifest.json'
manifest = _json.loads(path.read_text())
if client_name:
manifest["name"] = client_name
manifest["short_name"] = client_name
resp = Response(
_json.dumps(manifest, indent=2),
mimetype='application/manifest+json'
)
resp.headers['Cache-Control'] = 'public, max-age=86400'
return resp
@static_files_bp.route('/sw.js')
def serve_sw():
"""PWA Service Worker — must be served from root scope"""
path = STATIC_DIR / 'sw.js'
resp = send_file(path, mimetype='application/javascript')
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
resp.headers['Service-Worker-Allowed'] = '/'
return resp
@static_files_bp.route('/static/icons/<filename>')
def serve_icon(filename):
"""PWA icons.
NO-CACHE: icons across the system are not cached. See
docs/jambot/no-cache-policy.md.
"""
icon_path = _safe_path(STATIC_DIR / 'icons', filename)
if icon_path is None or not icon_path.exists():
return jsonify({"error": "Icon not found"}), 404
resp = send_file(icon_path, mimetype='image/png')
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
resp.headers['Pragma'] = 'no-cache'
resp.headers['Expires'] = '0'
return resp
@static_files_bp.route('/install')
def serve_install():
"""PWA install landing page"""
path = STATIC_DIR / 'install.html'
resp = send_file(path, mimetype='text/html')
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
return resp
@static_files_bp.route('/admin')
def serve_admin():
"""Serve the OpenUI admin dashboard"""
admin_path = APP_ROOT / 'src' / 'admin.html'
if not admin_path.exists():
return jsonify({"error": "Admin dashboard not found"}), 404
resp = send_file(admin_path, mimetype='text/html')
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
return resp