box/bin/exec-constrained.py

#!/usr/bin/env python3
"""
exec-constrained.py — Constrained HTTPS script-execution endpoint for bl.

SSH-fallback: when SSH to bl is down, operators can still run a constrained
set of scripts (dm.py, job-dispatch, chromebox control) over HTTPS.

This is the hardened replacement for exec-server.py's /exec endpoint, which
executes ARBITRARY shell commands (subprocess.run(cmd, shell=True)). This
server NEVER takes a command string. Clients request a named OPERATION with
validated ARGUMENTS; the server maps op -> fixed argv. No shell. No
interpolation. No RCE.

Usage:
    python3 exec-constrained.py --port 8444

Client:
    curl -sk -X POST https://127.0.0.1:8444/exec \\
      -H 'Authorization: Bearer <token>' \\
      -H 'Content-Type: application/json' \\
      -d '{"op": "dm.send", "args": {"agent": "opm", "to": "646",
           "target": "main", "message": "hello"}}'

Or signature auth (no secret in transit):
    payload=$(python3 -c "import json,time,secrets; print(json.dumps({
      'op': 'dm.send',
      'args': {'agent':'opm','to':'646','target':'main','message':'hi'},
      'ts': time.time(), 'nonce': secrets.token_hex(16)}))")
    sig=$(printf '%s' "$payload" | ssh-keygen -Y sign -f ~/.ssh/id_frontdoor -n exec-constrained)
    curl -sk -X POST https://127.0.0.1:8444/exec \\
      -H 'Content-Type: application/json' \\
      -d "$(python3 -c "import json,sys; print(json.dumps({
        'identity': 'operator-main',
        'payload': sys.argv[1], 'signature': sys.argv[2]}))" "$payload" "$sig")"

Auth model: master bearer token + per-agent bearer tokens (same files as
exec-server.py, so migration is drop-in) OR ssh-keygen -Y signatures over
the op envelope (namespace 'exec-constrained'). See SECURITY.md.
"""
import argparse
import hashlib
import hmac
import json
import os
import re
import secrets
import subprocess
import sys
import tempfile
import threading
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
import ssl

# ---------------------------------------------------------------- config

BIN_DIR = '/home/super/Projects/NetVM/bin'
JOBS_DIR = '/home/super/Projects/NetVM/jobs'
TOKEN_FILE = '/home/super/.exec-server-token'
TOKEN_DIR = '/home/super/.exec-tokens'
SIGNERS_FILE = '/home/super/.exec-signers'
NONCE_FILE = '/home/super/.exec-constrained-nonces'
AUDIT_LOG = '/home/super/.exec-constrained-audit.jsonl'
SIG_NAMESPACE = 'exec-constrained'
SIG_MAX_SKEW = 300

AGENTS = ('muse', 'pip', '646', 'opm', 'def', 'dev')
TARGET_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9/_.-]{0,63}$')
JOB_RE = re.compile(r'^[a-z0-9][a-z0-9-]{0,63}$')
NAME_RE = re.compile(r'^[A-Za-z0-9_.-]{1,64}$')
IDENT_RE = re.compile(r'^[a-z0-9-]+$')
NONCE_RE = re.compile(r'^[0-9a-fA-F]{16,128}$')
HEX_RE = re.compile(r'^[0-9a-f]{8,128}$')

MAX_BODY = 64 * 1024          # 64KB request cap
MAX_MESSAGE = 2000            # dm.py message cap
MAX_OUTPUT = 256 * 1024       # per-stream output cap
WORK_DIR = '/home/super'


# ------------------------------------------------------- rate limiting

class RateLimiter:
    """Token bucket per identity. Thread-safe."""
    def __init__(self, rate_per_min=60, burst=30):
        self.rate = rate_per_min / 60.0
        self.burst = burst
        self._buckets = {}
        self._lock = threading.Lock()

    def allow(self, ident):
        now = time.monotonic()
        with self._lock:
            tokens, last = self._buckets.get(ident, (self.burst, now))
            tokens = min(self.burst, tokens + (now - last) * self.rate)
            if tokens >= 1.0:
                self._buckets[ident] = (tokens - 1.0, now)
                return True
            self._buckets[ident] = (tokens, now)
            return False


LIMITER = RateLimiter()


# ------------------------------------------------------------- auth

def get_token():
    try:
        with open(TOKEN_FILE) as f:
            return f.read().strip()
    except FileNotFoundError:
        return None


def check_token(token):
    """Bearer token -> identity label, or None. Never logs the token."""
    master = get_token()
    if token and master and hmac.compare_digest(token, master):
        return 'master'
    try:
        names = os.listdir(TOKEN_DIR)
    except FileNotFoundError:
        return None
    for name in names:
        if not IDENT_RE.fullmatch(name):
            continue
        p = os.path.join(TOKEN_DIR, name)
        if not os.path.isfile(p):
            continue
        try:
            with open(p) as f:
                t = f.read().strip()
        except OSError:
            continue
        if t and hmac.compare_digest(token, t):
            return name
    return None


def check_nonce(nonce):
    now = time.time()
    fresh = []
    try:
        with open(NONCE_FILE) as f:
            for line in f:
                parts = line.split()
                if len(parts) != 2:
                    continue
                n, t = parts
                try:
                    if now - float(t) < 2 * SIG_MAX_SKEW:
                        fresh.append((n, t))
                except ValueError:
                    pass
    except FileNotFoundError:
        pass
    if any(n == nonce for n, _ in fresh):
        return False
    fresh.append((nonce, str(now)))
    try:
        with open(NONCE_FILE, 'w') as f:
            for n, t in fresh:
                f.write(f'{n} {t}\n')
        os.chmod(NONCE_FILE, 0o600)
    except OSError:
        return False
    return True


def check_signature(identity, payload, signature):
    """ssh-keygen -Y signature over {"op","args","ts","nonce"} -> identity/None."""
    if not IDENT_RE.fullmatch(identity or ''):
        return None
    try:
        data = json.loads(payload)
    except Exception:
        return None
    if not isinstance(data, dict):
        return None
    op = data.get('op')
    args = data.get('args')
    ts = data.get('ts')
    nonce = data.get('nonce')
    if not isinstance(op, str) or op not in OPS:
        return None
    if not isinstance(args, dict):
        return None
    if not isinstance(nonce, str) or not NONCE_RE.fullmatch(nonce):
        return None
    try:
        ts = float(ts)
    except (TypeError, ValueError):
        return None
    if abs(time.time() - ts) > SIG_MAX_SKEW:
        return None
    if not check_nonce(nonce):
        return None
    sig_path = None
    try:
        with tempfile.NamedTemporaryFile('w', delete=False, suffix='.sig') as f:
            f.write(signature if signature.endswith('\n') else signature + '\n')
            sig_path = f.name
        p = subprocess.run(
            ['ssh-keygen', '-Y', 'verify', '-f', SIGNERS_FILE, '-I', identity,
             '-n', SIG_NAMESPACE, '-s', sig_path],
            input=payload.encode(), capture_output=True, timeout=15)
        return identity if p.returncode == 0 else None
    except Exception:
        return None
    finally:
        if sig_path:
            try:
                os.unlink(sig_path)
            except OSError:
                pass

# ------------------------------------------------------- allowlist
#
# Each op maps to a FIXED script with VALIDATED arguments. The client can
# never influence: the executable path, the subcommand, or any flag name.
# Only whitelisted argument VALUES flow through, each checked below.
#
# To add an op: add an entry here with a build() function. build() receives
# the validated args dict and returns an argv list (no shell). Arg spec is
# enforced by validate() before build() runs.

class OpError(Exception):
    pass


def _clean_message(s):
    if not isinstance(s, str) or not s.strip():
        raise OpError('message must be a non-empty string')
    if len(s) > MAX_MESSAGE:
        raise OpError(f'message too long (max {MAX_MESSAGE})')
    if any(ord(c) < 32 and c not in '\n\t' for c in s):
        raise OpError('message contains control characters')
    return s


def _agent(v):
    if v not in AGENTS:
        raise OpError(f'agent must be one of {AGENTS}')
    return v


def _target(v):
    if not isinstance(v, str) or not TARGET_RE.fullmatch(v):
        raise OpError('target must match ^[A-Za-z0-9][A-Za-z0-9/_.-]{0,63}$')
    return v


def _opt_int(v, lo, hi, name):
    if v is None:
        return None
    if isinstance(v, bool) or not isinstance(v, int):
        raise OpError(f'{name} must be an integer')
    if not (lo <= v <= hi):
        raise OpError(f'{name} must be {lo}..{hi}')
    return v


def _job_name(v):
    # Must exist in JOBS_DIR and match the safe pattern (no path traversal).
    if not isinstance(v, str) or not JOB_RE.fullmatch(v):
        raise OpError('job must match ^[a-z0-9][a-z0-9-]{0,63}$')
    path = os.path.join(JOBS_DIR, v + '.json')
    if not os.path.isfile(path):
        raise OpError('unknown job')
    return v


def _dm_send_build(a):
    argv = [sys.executable, os.path.join(BIN_DIR, 'dm.py'), 'send',
            '--agent', a['agent'], '--target', a['target']]
    if a.get('to'):
        argv += ['--to', a['to']]
    if a.get('expect_reply'):
        argv += ['--expect-reply']
    if a.get('reply_timeout') is not None:
        argv += ['--reply-timeout', str(a['reply_timeout'])]
    if a.get('reply_nudges') is not None:
        argv += ['--reply-nudges', str(a['reply_nudges'])]
    if a.get('reply_escalate'):
        argv += ['--reply-escalate', a['reply_escalate']]
    if a.get('route'):
        argv += ['--route', a['route']]
    for t in a.get('tags', []):
        argv += ['--tag', t]
    argv.append(a['message'])
    return argv


def _dm_send_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'agent', 'to', 'target', 'message', 'expect_reply',
               'reply_timeout', 'reply_nudges', 'reply_escalate',
               'route', 'tags'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    a = {}
    a['agent'] = _agent(raw.get('agent'))
    if 'to' in raw and raw['to'] is not None:
        a['to'] = _agent(raw['to'])
    a['target'] = _target(raw.get('target'))
    a['message'] = _clean_message(raw.get('message'))
    a['expect_reply'] = bool(raw.get('expect_reply', False))
    a['reply_timeout'] = _opt_int(raw.get('reply_timeout'), 60, 604800,
                                 'reply_timeout')
    a['reply_nudges'] = _opt_int(raw.get('reply_nudges'), 0, 10,
                                 'reply_nudges')
    esc = raw.get('reply_escalate')
    if esc is not None:
        a['reply_escalate'] = _agent(esc)
    route = raw.get('route')
    if route is not None:
        if not isinstance(route, str) or not TARGET_RE.fullmatch(route):
            raise OpError('route must match target pattern')
        a['route'] = route
    tags = raw.get('tags', [])
    if not isinstance(tags, list) or len(tags) > 8:
        raise OpError('tags must be a list of at most 8 strings')
    clean_tags = []
    for t in tags:
        if not isinstance(t, str) or len(t) > 120:
            raise OpError('tag must be a string <= 120 chars')
        # Canonical tag vocabulary only (see DEPLOY-DECISIONS.md).
        if not re.fullmatch(r'[a-z0-9_:-]+=[a-zA-Z0-9_.:/-]*', t) and \
           not re.fullmatch(r'[a-z0-9_:-]+', t):
            raise OpError(f'malformed tag: {t}')
        clean_tags.append(t)
    a['tags'] = clean_tags
    return a


def _dm_thread_validate(raw):
    # Same shape as dm.send but uses the thread subcommand.
    return _dm_send_validate(raw)


def _dm_thread_build(a):
    argv = _dm_send_build(a)
    argv[2] = 'thread'
    return argv


def _dm_read_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'agent', 'target', 'limit'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {
        'agent': _agent(raw.get('agent')),
        'target': _target(raw.get('target')),
        'limit': _opt_int(raw.get('limit', 20), 1, 100, 'limit') or 20,
    }


def _dm_read_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'dm.py'), 'read',
            '--agent', a['agent'], '--target', a['target'],
            '--limit', str(a['limit'])]


def _job_run_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'job'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {'job': _job_name(raw.get('job'))}


def _job_run_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'job-dispatch.py'), a['job']]


def _chat_messages_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'account', 'limit'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {
        'account': _agent(raw.get('account')),
        'limit': _opt_int(raw.get('limit', 20), 1, 100, 'limit') or 20,
    }


def _chat_messages_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'muse-chat-api.py'),
            '--account', a['account'], 'messages', '--limit', str(a['limit'])]


def _chat_send_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'account', 'message', 'thread'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    a = {
        'account': _agent(raw.get('account')),
        'message': _clean_message(raw.get('message')),
    }
    if raw.get('thread') is not None:
        th = raw['thread']
        if not isinstance(th, str) or not HEX_RE.fullmatch(th):
            raise OpError('thread must be a hex uuid')
        a['thread'] = th
    return a


def _chat_send_build(a):
    argv = [sys.executable, os.path.join(BIN_DIR, 'muse-chat-api.py'),
            '--account', a['account'], 'send', '--message', a['message']]
    if a.get('thread'):
        argv += ['--thread', a['thread']]
    return argv


def _safe_str(v, max_len=120, name='string'):
    if v is None:
        return ''
    if not isinstance(v, str):
        raise OpError(f'{name} must be a string')
    if len(v) > max_len:
        raise OpError(f'{name} exceeds max length {max_len}')
    if any(ord(c) < 32 and c not in '\n\t' for c in v):
        raise OpError(f'{name} contains control characters')
    return v.strip()


def _health_validate(raw):
    if raw not in ({}, None):
        raise OpError('health.check takes no args')
    return {}


def _health_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'super-cli.py'), 'fleet', 'status', '--json']


def _cdp_latency_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'cdp-latency']


def _chrome_errors_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'chrome-errors']


def _watchdog_alerts_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'watchdog-alerts']


def _relay_health_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'relay-health']


def _quality_check_validate(raw):
    if raw not in ({}, None):
        if isinstance(raw, dict):
            allowed = {'scope'}
            for k in raw:
                if k not in allowed:
                    raise OpError(f'unknown arg: {k}')
        else:
            raise OpError('quality.check args must be an object or empty')
    return {}


def _quality_check_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'quality-check']


def _subagent_spawn_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'agent', 'title', 'prompt', 'wait'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {
        'agent': _agent(raw.get('agent')),
        'title': _safe_str(raw.get('title', 'subagent'), 120, 'title') or 'subagent',
        'prompt': _clean_message(raw.get('prompt')),
        'wait': _opt_int(raw.get('wait', 0), 0, 60, 'wait') or 0,
    }


def _subagent_spawn_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'super-cli.py'),
            'deploy', 'subagent',
            '--agent', a['agent'],
            '--title', a['title'],
            '--wait', str(a['wait']),
            a['prompt']]


def _thread_list_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'agent'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {'agent': _agent(raw.get('agent'))}


def _thread_list_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'super-cli.py'),
            'thread', 'list', a['agent'], '--json']


def _thread_view_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'agent', 'thread', 'limit'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    th = raw.get('thread')
    if not isinstance(th, str) or not TARGET_RE.fullmatch(th):
        raise OpError('thread must match target pattern')
    return {
        'agent': _agent(raw.get('agent')),
        'thread': th,
        'limit': _opt_int(raw.get('limit', 15), 1, 100, 'limit') or 15,
    }


def _thread_view_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'super-cli.py'),
            'thread', 'view', a['agent'], a['thread'],
            '--limit', str(a['limit']), '--json']


def _pipeline_run_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {'name': _job_name(raw.get('name'))}


def _pipeline_run_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'super-cli.py'), 'deploy', 'pipeline', a['name']]


def _cron_status_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {'name': _job_name(raw.get('name')) if raw.get('name') else 'heartbeat'}


def _cron_status_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'timer-status', a['name']]


def _cron_view_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {'name': _job_name(raw.get('name'))}


def _cron_view_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'job-get', a['name']]


def _cron_runs_validate(raw):
    if raw not in ({}, None):
        raise OpError('cron.runs takes no required args')
    return {}


def _cron_runs_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'job-list']


def _cron_timer_start_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {'name': _job_name(raw.get('name'))}


def _cron_timer_start_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'timer-start', a['name']]


def _cron_timer_create_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    return {'name': _job_name(raw.get('name'))}


def _cron_timer_create_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'timer-create', a['name']]


# ------------------------------------------------------- tmux ops
def _tmux_session_name(s):
    if not isinstance(s, str) or not re.fullmatch(r'^[a-zA-Z0-9_.-]{1,64}$', s):
        raise OpError('session must match ^[a-zA-Z0-9_.-]{1,64}$')
    return s


def _tmux_send_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'session', 'keys', 'no_enter'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    if not raw.get('session'):
        raise OpError('session is required')
    if 'keys' not in raw:
        raise OpError('keys is required')
    return {
        'session': _tmux_session_name(raw['session']),
        'keys': str(raw['keys']),
        'no_enter': bool(raw.get('no_enter', False)),
    }


def _tmux_send_build(a):
    cmd = [sys.executable, os.path.join(BIN_DIR, 'muse-tmux.py'), 'send', a['session'], a['keys']]
    if a.get('no_enter'):
        cmd.append('--no-enter')
    return cmd


def _tmux_capture_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'session', 'lines'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    if not raw.get('session'):
        raise OpError('session is required')
    lines = raw.get('lines', 30)
    try:
        lines = int(lines)
        if lines < 1 or lines > 500:
            lines = 30
    except Exception:
        lines = 30
    return {
        'session': _tmux_session_name(raw['session']),
        'lines': lines,
    }


def _tmux_capture_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'muse-tmux.py'), 'capture', a['session'], '--lines', str(a['lines'])]


def _tmux_list_validate(raw):
    return {}


def _tmux_list_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'muse-tmux.py'), 'list']


def _tmux_new_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'session', 'window', 'command'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    if not raw.get('session'):
        raise OpError('session is required')
    return {
        'session': _tmux_session_name(raw['session']),
        'window': str(raw.get('window', '')) if raw.get('window') else None,
        'command': str(raw.get('command', '')) if raw.get('command') else None,
    }


def _tmux_new_build(a):
    cmd = [sys.executable, os.path.join(BIN_DIR, 'muse-tmux.py'), 'new', a['session']]
    if a.get('window'):
        cmd.extend(['--window', a['window']])
    if a.get('command'):
        cmd.extend(['--command', a['command']])
    return cmd


def _tmux_kill_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'session'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    if not raw.get('session'):
        raise OpError('session is required')
    return {'session': _tmux_session_name(raw['session'])}


def _tmux_kill_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'muse-tmux.py'), 'kill', a['session']]


def _tmux_prune_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'ttl'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    ttl = raw.get('ttl', 7200)
    try:
        ttl = int(ttl)
        if ttl < 60:
            ttl = 60
    except Exception:
        ttl = 7200
    return {'ttl': ttl}


def _tmux_prune_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'muse-tmux.py'), 'prune', '--ttl', str(a['ttl'])]


def _vars_list_validate(raw):
    if raw not in ({}, None):
        raise OpError('vars.list takes no required args')
    return {}


def _vars_list_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'vars-list']


def _vars_get_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    name = raw.get('name')
    if not isinstance(name, str) or not NAME_RE.fullmatch(name):
        raise OpError('name must match safe identifier')
    return {'name': name}


def _vars_get_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'vars-get', a['name']]


def _vars_set_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'name', 'value'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    name = raw.get('name')
    if not isinstance(name, str) or not NAME_RE.fullmatch(name):
        raise OpError('name must match safe identifier')
    val = raw.get('value')
    if val is None or not isinstance(val, (str, int, float, bool)):
        raise OpError('value must be a scalar')
    return {'name': name, 'value': str(val)}


def _vars_set_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'vars-set', a['name'], a['value']]


def _files_read_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'path', 'lines'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    p = raw.get('path')
    if not isinstance(p, str) or not p.strip() or '..' in p:
        raise OpError('path must be a safe relative or repo path without ".."')
    return {
        'path': p.strip(),
        'lines': _opt_int(raw.get('lines', 100), 1, 1000, 'lines') or 100
    }


def _files_read_build(a):
    # Pass JSON args via stdin to box-sys-op.py
    return [sys.executable, os.path.join(BIN_DIR, 'box-sys-op.py'), 'files.read']


def _files_write_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'path', 'content'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    p = raw.get('path')
    if not isinstance(p, str) or not p.strip() or '..' in p:
        raise OpError('path must be a safe relative or repo path without ".."')
    content = raw.get('content')
    if not isinstance(content, str):
        raise OpError('content must be a string')
    if len(content.encode('utf-8')) > 64 * 1024:
        raise OpError('content exceeds max size 64KB')
    return {'path': p.strip(), 'content': content}


def _files_write_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-sys-op.py'), 'files.write']


def _web_fetch_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'url'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    u = raw.get('url')
    if not isinstance(u, str) or not u.strip():
        raise OpError('url must be a string')
    if not (u.startswith('http://') or u.startswith('https://')):
        raise OpError('url must start with http:// or https://')
    return {'url': u.strip()}


def _web_fetch_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-sys-op.py'), 'web.fetch']


def _service_status_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'unit', 'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    unit = raw.get('unit') or raw.get('name')
    if not isinstance(unit, str) or not NAME_RE.fullmatch(unit):
        raise OpError('unit must match safe identifier')
    return {'unit': unit}


def _service_status_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-sys-op.py'), 'service.status']


def _service_restart_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'unit', 'name'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    unit = raw.get('unit') or raw.get('name')
    if not isinstance(unit, str) or not NAME_RE.fullmatch(unit):
        raise OpError('unit must match safe identifier')
    return {'unit': unit}


def _service_restart_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-sys-op.py'), 'service.restart']


def _followup_create_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'agent', 'in_m', 'prompt', 'thread', 'sidechat', 'sender'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    agent = _agent(raw.get('agent', 'opm'))
    sender = raw.get('sender')
    if sender:
        sender = _agent(sender)
    try:
        in_m = float(raw.get('in_m', 1))
    except (TypeError, ValueError):
        raise OpError('in_m must be a number')
    if in_m < 0.05 or in_m > 1440:
        raise OpError('in_m must be between 0.05 and 1440 minutes')
    prompt = _clean_message(raw.get('prompt'))
    sidechat = raw.get('thread') or raw.get('sidechat')
    if sidechat and not TARGET_RE.fullmatch(str(sidechat)):
        raise OpError('thread/sidechat must match safe identifier')
    return {
        'agent': agent,
        'sender': sender or agent,
        'in_m': in_m,
        'prompt': prompt,
        'thread': str(sidechat) if sidechat else None
    }


def _followup_create_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-sys-op.py'), 'followup.create']


def _swarm_spawn_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'count', 'task', 'label'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    count = _opt_int(raw.get('count', 1), 1, 50, 'count') or 1
    task = _clean_message(raw.get('task'))
    label = raw.get('label')
    if label and not TARGET_RE.fullmatch(str(label)):
        raise OpError('label must match safe identifier')
    return {'count': count, 'task': task, 'label': str(label) if label else None}


def _swarm_spawn_build(a):
    cmd = [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'swarm-spawn', str(a['count']), a['task']]
    if a.get('label'):
        cmd.extend(['--label', a['label']])
    return cmd


def _swarm_status_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'swarm_id', 'id'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    sid = raw.get('swarm_id') or raw.get('id')
    if not isinstance(sid, str) or not TARGET_RE.fullmatch(sid):
        raise OpError('swarm_id must match safe identifier')
    return {'swarm_id': sid}


def _swarm_status_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'swarm-status', a['swarm_id']]


def _swarm_list_validate(raw):
    if raw not in ({}, None):
        raise OpError('swarm.list takes no required args')
    return {}


def _swarm_list_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'swarm-list']


def _swarm_results_validate(raw):
    if not isinstance(raw, dict):
        raise OpError('args must be an object')
    allowed = {'swarm_id', 'id'}
    for k in raw:
        if k not in allowed:
            raise OpError(f'unknown arg: {k}')
    sid = raw.get('swarm_id') or raw.get('id')
    if not isinstance(sid, str) or not TARGET_RE.fullmatch(sid):
        raise OpError('swarm_id must match safe identifier')
    return {'swarm_id': sid}


def _swarm_results_build(a):
    return [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), 'swarm-results', a['swarm_id']]


# op -> {validate, build, timeout, side_effecting, description}
OPS = {
    'dm.send': {
        'validate': _dm_send_validate, 'build': _dm_send_build,
        'timeout': 120, 'side_effecting': True,
        'desc': 'Send a DM via dm.py (verified delivery)',
    },
    'dm.thread': {
        'validate': _dm_thread_validate, 'build': _dm_thread_build,
        'timeout': 120, 'side_effecting': True,
        'desc': 'Send a threaded DM via dm.py',
    },
    'dm.read': {
        'validate': _dm_read_validate, 'build': _dm_read_build,
        'timeout': 60, 'side_effecting': False,
        'desc': 'Read recent DMs (read-only)',
    },
    'job.run': {
        'validate': _job_run_validate, 'build': _job_run_build,
        'timeout': 300, 'side_effecting': True,
        'desc': 'Run a job from the jobs directory',
    },
    'chat.messages': {
        'validate': _chat_messages_validate, 'build': _chat_messages_build,
        'timeout': 60, 'side_effecting': False,
        'desc': 'Read recent chat messages (read-only)',
    },
    'chat.send': {
        'validate': _chat_send_validate, 'build': _chat_send_build,
        'timeout': 120, 'side_effecting': True,
        'desc': 'Send a chat message via muse-chat-api.py',
    },
    'health.check': {
        'validate': _health_validate, 'build': _health_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Run fleet status health check (read-only)',
    },
    'cdp.latency': {
        'validate': _health_validate, 'build': _cdp_latency_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Run CDP latency probe across fleet profiles (read-only)',
    },
    'chrome.errors': {
        'validate': _health_validate, 'build': _chrome_errors_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Scan chrome log files for FATAL/crash/OOM errors since watermark (read-only)',
    },
    'watchdog.alerts': {
        'validate': _health_validate, 'build': _watchdog_alerts_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Check for new browser relaunch watchdog alerts since watermark (read-only)',
    },
    'relay.health': {
        'validate': _health_validate, 'build': _relay_health_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Check systemd relay service status across all node ports (read-only)',
    },
    'quality.check': {
        'validate': _quality_check_validate, 'build': _quality_check_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Run box quality gates and self-diagnostics suite (read-only)',
    },
    'subagent.spawn': {
        'validate': _subagent_spawn_validate, 'build': _subagent_spawn_build,
        'timeout': 120, 'side_effecting': True,
        'desc': 'Spawn an autonomous subagent session via muse-cli gateway',
    },
    'thread.list': {
        'validate': _thread_list_validate, 'build': _thread_list_build,
        'timeout': 60, 'side_effecting': False,
        'desc': 'List threads/sessions for an agent',
    },
    'thread.view': {
        'validate': _thread_view_validate, 'build': _thread_view_build,
        'timeout': 60, 'side_effecting': False,
        'desc': 'View messages in a thread/session',
    },
    'pipeline.run': {
        'validate': _pipeline_run_validate, 'build': _pipeline_run_build,
        'timeout': 120, 'side_effecting': True,
        'desc': 'Dispatch a multi-step pipeline across agents',
    },
    'cron.status': {
        'validate': _cron_status_validate, 'build': _cron_status_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Check systemd timer / scheduled check status',
    },
    'cron.view': {
        'validate': _cron_view_validate, 'build': _cron_view_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'View scheduled job definition and prompt',
    },
    'cron.runs': {
        'validate': _cron_runs_validate, 'build': _cron_runs_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'List all scheduled crons/jobs across the fleet',
    },
    'cron.run': {
        'validate': _job_run_validate, 'build': _job_run_build,
        'timeout': 300, 'side_effecting': True,
        'desc': 'Trigger on-demand execution of a scheduled job',
    },
    'cron.timer_create': {
        'validate': _cron_timer_create_validate, 'build': _cron_timer_create_build,
        'timeout': 30, 'side_effecting': True,
        'desc': 'Create systemd user timer unit for a job',
    },
    'cron.timer_start': {
        'validate': _cron_timer_start_validate, 'build': _cron_timer_start_build,
        'timeout': 30, 'side_effecting': True,
        'desc': 'Enable and start systemd user timer unit for a job',
    },
    'vars.list': {
        'validate': _vars_list_validate, 'build': _vars_list_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'List intrinsic loop strategy and timing variables',
    },
    'vars.get': {
        'validate': _vars_get_validate, 'build': _vars_get_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Get specific intrinsic loop variable',
    },
    'vars.set': {
        'validate': _vars_set_validate, 'build': _vars_set_build,
        'timeout': 30, 'side_effecting': True,
        'desc': 'Update intrinsic loop variable',
    },
    'files.read': {
        'validate': _files_read_validate, 'build': _files_read_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Read sandboxed file within repository tree',
    },
    'files.write': {
        'validate': _files_write_validate, 'build': _files_write_build,
        'timeout': 30, 'side_effecting': True,
        'desc': 'Write sandboxed file within repository tree',
    },
    'web.fetch': {
        'validate': _web_fetch_validate, 'build': _web_fetch_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Perform safe HTTP/HTTPS GET request with SSRF guard',
    },
    'service.status': {
        'validate': _service_status_validate, 'build': _service_status_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Inspect allowlisted fleet systemd service status',
    },
    'service.restart': {
        'validate': _service_restart_validate, 'build': _service_restart_build,
        'timeout': 30, 'side_effecting': True,
        'desc': 'Restart allowlisted fleet systemd service',
    },
    'swarm.spawn': {
        'validate': _swarm_spawn_validate, 'build': _swarm_spawn_build,
        'timeout': 30, 'side_effecting': True,
        'desc': 'Spawn autonomous subagent swarm slots managed by Box',
    },
    'swarm.status': {
        'validate': _swarm_status_validate, 'build': _swarm_status_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Inspect subagent swarm status and progress',
    },
    'swarm.list': {
        'validate': _swarm_list_validate, 'build': _swarm_list_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'List all subagent swarms and their counts',
    },
    'swarm.results': {
        'validate': _swarm_results_validate, 'build': _swarm_results_build,
        'timeout': 30, 'side_effecting': False,
        'desc': 'Retrieve results and summaries for a completed swarm',
    },
    'followup.create': {
        'validate': _followup_create_validate, 'build': _followup_create_build,
        'timeout': 30, 'side_effecting': True,
        'desc': 'Schedule a delayed autonomous self-followup reminder via transient systemd timer',
    },
    'tmux.send': {
        'validate': _tmux_send_validate, 'build': _tmux_send_build,
        'timeout': 15, 'side_effecting': True,
        'desc': 'Send keystrokes to a shared Muse tmux session (/tmp/tmux-muse.sock)',
    },
    'tmux.capture': {
        'validate': _tmux_capture_validate, 'build': _tmux_capture_build,
        'timeout': 15, 'side_effecting': False,
        'desc': 'Capture pane output from a shared Muse tmux session (/tmp/tmux-muse.sock)',
    },
    'tmux.list': {
        'validate': _tmux_list_validate, 'build': _tmux_list_build,
        'timeout': 10, 'side_effecting': False,
        'desc': 'List sessions on shared Muse tmux socket (/tmp/tmux-muse.sock)',
    },
    'tmux.new': {
        'validate': _tmux_new_validate, 'build': _tmux_new_build,
        'timeout': 15, 'side_effecting': True,
        'desc': 'Create a new session on shared Muse tmux socket (/tmp/tmux-muse.sock)',
    },
    'tmux.kill': {
        'validate': _tmux_kill_validate, 'build': _tmux_kill_build,
        'timeout': 15, 'side_effecting': True,
        'desc': 'Kill a session on shared Muse tmux socket (/tmp/tmux-muse.sock)',
    },
    'tmux.prune': {
        'validate': _tmux_prune_validate, 'build': _tmux_prune_build,
        'timeout': 15, 'side_effecting': True,
        'desc': 'Reap stale unattached sessions inactive for >TTL (default 2h)',
    },
    'exec.ping': {
        'validate': _health_validate,
        'build': lambda a: ['/bin/echo', 'PONG'],
        'timeout': 10, 'side_effecting': False,
        'desc': 'Canary no-op for watchdogs',
    },
}

# identity -> set of ops. 'master' may invoke everything. Unknown identities
# get the read-only subset. Per-agent tokens inherit their agent name as the
# identity; tighten per agent here as needed.
# ---- read-only tolerance + hyphenated aliases used by job templates ----
_ARG_SYNONYMS = {'job': 'name', 'id': 'name', 'job_name': 'name', 'unit': 'name'}


def _tolerant_args(spec, args):
    """Read-only ops: drop unknown args (mapping common synonyms) instead of failing.
    Returns (args, dropped_names)."""
    if args is None:
        return {}, []
    if not isinstance(args, dict):
        return args, []
    args, dropped = dict(args), []
    for _ in range(len(args) + 1):
        try:
            spec['validate'](args)
            break
        except OpError as e:
            m = re.match(r'unknown arg: (\S+)', str(e))
            if m and m.group(1) in args:
                k = m.group(1)
                v = args.pop(k)
                syn = _ARG_SYNONYMS.get(k)
                if syn and syn not in args:
                    args[syn] = v
                else:
                    dropped.append(k)
                continue
            if 'takes no' in str(e):
                dropped.extend(args)
                args = {}
                continue
            break
    return args, dropped


def _register_aliases():
    alias = {'job-list': 'cron.runs', 'quality.validate': 'quality.check',
             'vars-list': 'vars.list', 'fleet-status': 'health.check'}
    for new, old in alias.items():
        if new not in OPS and old in OPS:
            OPS[new] = dict(OPS[old], desc=OPS[old]['desc'] + f' (alias of {old})')
    def _ctl(*words):
        return lambda a: [sys.executable, os.path.join(BIN_DIR, 'box-ctl.py'), *words]
    def _noargs(raw):
        return {}
    for new, words, desc in (('loop-status', ('loop-status',), 'Fleet loop health (read-only)'),
                             ('timer-list', ('timer', 'list'), 'List box timers (read-only)')):
        if new not in OPS:
            OPS[new] = {'validate': _noargs, 'build': _ctl(*words), 'timeout': 30,
                        'side_effecting': False, 'desc': desc}


_register_aliases()

PERMISSIONS = {
    'master': set(OPS),
    'operator-main': set(OPS),
    'operator-646': set(OPS),
    'operator-muse': set(OPS),
    'operator-pip': set(OPS),
    'operator-opm': set(OPS),
    'operator-dev': set(OPS),
    'operator-def': set(OPS),
    'operator-super': set(OPS),
    'super': set(OPS),
    '646': set(OPS),
    'pip': set(OPS),
    'muse': set(OPS),
    'opm': set(OPS),
    'dev': set(OPS),
    'def': set(OPS),
    'exec-canary': {'exec.ping'},
}
DEFAULT_PERMS = {'dm.read', 'chat.messages', 'health.check', 'thread.list', 'thread.view', 'exec.ping'}


def permitted(ident, op):
    perms = PERMISSIONS.get(ident, DEFAULT_PERMS)
    return op in perms

# ------------------------------------------------------- audit log

_audit_lock = threading.Lock()


def audit(entry):
    """Append-only JSONL audit. Never includes tokens or signatures."""
    entry = dict(entry)
    entry['ts'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
    line = json.dumps(entry, separators=(',', ':')) + '\n'
    with _audit_lock:
        try:
            with open(AUDIT_LOG, 'a') as f:
                f.write(line)
            os.chmod(AUDIT_LOG, 0o600)
        except OSError:
            pass


# ------------------------------------------------------- HTTP handler

class Handler(BaseHTTPRequestHandler):
    server_version = 'exec-constrained/1.0'

    def log_message(self, format, *args):
        sys.stderr.write(f'{self.client_address[0]} - {format % args}\n')

    def _send(self, code, obj):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == '/health':
            self._send(200, {'status': 'ok', 'mode': 'constrained'})
        elif self.path == '/ops':
            # List ops (names + descriptions only; no arg details needed).
            self._send(200, {'ops': [
                {'op': k, 'desc': v['desc'],
                 'side_effecting': v['side_effecting']}
                for k, v in sorted(OPS.items())]})
        elif self.path in ('/box', '/box-relay.sh'):
            relay_path = os.path.join(BIN_DIR, 'box-relay.sh')
            if os.path.isfile(relay_path):
                try:
                    with open(relay_path, 'rb') as f:
                        data = f.read()
                    self.send_response(200)
                    self.send_header('Content-Type', 'text/x-shellscript')
                    self.send_header('Content-Length', str(len(data)))
                    self.end_headers()
                    self.wfile.write(data)
                    return
                except Exception:
                    pass
            self._send(404, {'error': 'not found'})
        else:
            self._send(404, {'error': 'not found'})

    def do_POST(self):
        if self.path == '/exec':
            self._handle_exec()
        else:
            self._send(404, {'error': 'not found'})

    def _handle_exec(self):
        peer = self.client_address[0]
        try:
            length = int(self.headers.get('Content-Length', 0))
        except ValueError:
            length = 0
        if length <= 0 or length > MAX_BODY:
            self._send(413 if length > MAX_BODY else 400,
                       {'error': 'bad content length'})
            return
        try:
            data = json.loads(self.rfile.read(length))
        except Exception:
            self._send(400, {'error': 'invalid json'})
            return
        if not isinstance(data, dict):
            self._send(400, {'error': 'body must be an object'})
            return

        # --- auth: Bearer <redacted> header (preferred) or legacy body token,
        # --- or ssh-keygen signature envelope.
        ident = None
        op = data.get('op')
        args = data.get('args', {})
        authz = self.headers.get('Authorization', '')
        if authz.startswith('Bearer '):
            token = authz[7:].strip()
            ident = check_token(token)
        elif data.get('token'):
            ident = check_token(data['token'])
        elif data.get('identity') and data.get('payload') \
                and data.get('signature'):
            # Signature envelope carries op/args; ignore body's op/args.
            ident = check_signature(data['identity'], data['payload'],
                                    data['signature'])
            if ident:
                try:
                    env = json.loads(data['payload'])
                    op, args = env.get('op'), env.get('args', {})
                except Exception:
                    op, args = None, {}
        if not ident:
            audit({'event': 'auth_failed', 'peer': peer,
                   'op': str(op)[:64]})
            self._send(401, {'error': 'unauthorized'})
            return

        # --- rate limit
        if not LIMITER.allow(ident):
            audit({'event': 'rate_limited', 'ident': ident, 'peer': peer,
                   'op': str(op)[:64]})
            self._send(429, {'error': 'rate limited'})
            return

        # --- op + permission + args
        if not isinstance(op, str) or op not in OPS:
            self._send(400, {'error': 'unknown op',
                             'hint': 'GET /ops for the allowlist'})
            return
        if not permitted(ident, op):
            audit({'event': 'forbidden', 'ident': ident, 'peer': peer,
                   'op': op})
            self._send(403, {'error': 'forbidden for this identity'})
            return
        dropped = []
        try:
            spec = OPS[op]
            if not spec.get('side_effecting'):
                args, dropped = _tolerant_args(spec, args)
            clean = spec['validate'](args)
            argv = spec['build'](clean)
        except OpError as e:
            self._send(400, {'error': f'bad args: {e}', 'op': op,
                             'desc': OPS[op].get('desc', ''),
                             'hint': 'GET /ops for the allowlist; see desc for usage'})
            return
        except Exception as e:
            sys.stderr.write(f'Validation/build exception for {op}: {e}\n')
            import traceback; traceback.print_exc()
            self._send(500, {'error': f'internal: {e}'})
            return

        # --- execute (NO shell, argv only)
        audit({'event': 'exec_start', 'ident': ident, 'peer': peer,
               'op': op,
               'args_sha': hashlib.sha256(
                   json.dumps(clean, sort_keys=True).encode()
               ).hexdigest()[:16]})
        t0 = time.monotonic()
        try:
            stdin_input = json.dumps(clean) if op.startswith(('files.', 'web.', 'service.', 'followup.')) else None
            p = subprocess.run(argv, input=stdin_input, capture_output=True, text=True,
                               timeout=spec['timeout'], cwd=WORK_DIR)
            rc = p.returncode
            out = p.stdout[-MAX_OUTPUT:]
            err = p.stderr[-MAX_OUTPUT:]
            ok = True
        except subprocess.TimeoutExpired:
            rc, out, err, ok = -1, '', 'timeout', False
        except Exception as e:
            rc, out, err, ok = -1, '', str(e)[:500], False
        dt = round(time.monotonic() - t0, 2)
        audit({'event': 'exec_done', 'ident': ident, 'peer': peer,
               'op': op, 'rc': rc, 'duration_s': dt, 'ok': ok})
        if ok:
            resp = {'rc': rc, 'stdout': out, 'stderr': err, 'duration_s': dt}
            if dropped:
                resp['dropped_args'] = dropped
            self._send(200, resp)
        else:
            self._send(500, {'error': err, 'rc': rc})



# ------------------------------------------------------- main

def main():
    global TOKEN_FILE, TOKEN_DIR, SIGNERS_FILE, NONCE_FILE, AUDIT_LOG
    global BIN_DIR, JOBS_DIR, WORK_DIR
    ap = argparse.ArgumentParser()
    ap.add_argument('--port', type=int, default=8444)
    ap.add_argument('--host', default='127.0.0.1')
    ap.add_argument('--token-file', default=TOKEN_FILE)
    ap.add_argument('--token-dir', default=TOKEN_DIR)
    ap.add_argument('--signers-file', default=SIGNERS_FILE)
    ap.add_argument('--nonce-file', default=NONCE_FILE)
    ap.add_argument('--audit-log', default=AUDIT_LOG)
    ap.add_argument('--bin-dir', default=BIN_DIR)
    ap.add_argument('--jobs-dir', default=JOBS_DIR)
    ap.add_argument('--cert-file',
                    default='/home/super/.exec-constrained-cert.pem')
    ap.add_argument('--key-file',
                    default='/home/super/.exec-constrained-key.pem')
    ap.add_argument('--work-dir', default='/home/super')
    args = ap.parse_args()

    TOKEN_FILE, TOKEN_DIR = args.token_file, args.token_dir
    SIGNERS_FILE, NONCE_FILE = args.signers_file, args.nonce_file
    AUDIT_LOG = args.audit_log
    BIN_DIR, JOBS_DIR = args.bin_dir, args.jobs_dir
    WORK_DIR = args.work_dir

    server = HTTPServer((args.host, args.port), Handler)

    cert_file = args.cert_file
    key_file = args.key_file
    if not os.path.exists(cert_file):
        print('Generating self-signed cert...', file=sys.stderr)
        subprocess.run([
            'openssl', 'req', '-x509', '-newkey', 'rsa:2048',
            '-keyout', key_file, '-out', cert_file,
            '-days', '3650', '-nodes',
            '-subj', '/CN=bl-exec-constrained',
        ], check=True, capture_output=True)
        os.chmod(key_file, 0o600)

    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(cert_file, key_file)
    server.socket = ctx.wrap_socket(server.socket, server_side=True)

    print(f'Constrained exec listening on https://{args.host}:{args.port}/exec',
          file=sys.stderr)
    print(f'Allowlist: {", ".join(sorted(OPS))}', file=sys.stderr)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print('\nShutting down', file=sys.stderr)


if __name__ == '__main__':
    main()