fix(runtime): restore authorized local control paths

routes/cookbook_routes.py

@@ -408,8 +408,8 @@ def setup_cookbook_routes() -> APIRouter:
     async def protect_native_control(request: Request):
         if request.method in {"GET", "HEAD"}:
             return
-        # Cookbook's UI records and session strings are not an application
-        # process registry. No loopback caller can use them as local authority.
+        # UI records/session strings are not process authority. Local tool
+        # launches require a one-use capability from their admitted producer.
         path = request.url.path
         from routes.shell_routes import _require_admin
         if path in {"/api/cookbook/kill-pid", "/api/cookbook/state", "/api/cookbook/ssh-key"}:
@@ -417,7 +417,14 @@ def setup_cookbook_routes() -> APIRouter:
         if path in {"/api/model/download", "/api/model/serve"}:
             payload = await request.json()
             if not payload.get("remote_host"):
-                _require_admin(request)
+                from src.agent_runtime.local_model_control import consume_model_control
+                from src.agent_runtime.resources import ResourceIdentityError
+                try:
+                    claimed = consume_model_control(request, payload)
+                except (ResourceIdentityError, ValueError, TypeError):
+                    raise HTTPException(403, "Local model capability denied") from None
+                if not claimed:
+                    _require_admin(request)
     router = APIRouter(tags=["cookbook"], dependencies=[Depends(protect_native_control)])
     _cookbook_state_path = Path(COOKBOOK_STATE_FILE)
     _state_get_cache = {"ts": 0.0, "mtime": 0.0, "value": None}

routes/shell_routes.py

@@ -59,12 +59,17 @@ from core.platform_compat import (
 def _require_admin(request: Request):
     """Reject non-admin callers. Shell exec is admin-only — never expose to
     regular users; that's RCE-after-signup."""
-    # Anonymous loopback is also reachable from an admitted native workload.
-    # It cannot be treated as a human admin or as process creation authority.
+    # Tool authentication is never human administration. Operator-disabled
+    # login has a separate direct-local transport contract below; it supplies
+    # no resource grant to model producers.
     from src.agent_runtime.authority import is_internal_tool_request
-    if is_internal_tool_request(request):
+    from core.middleware import INTERNAL_TOOL_HEADER
+    if is_internal_tool_request(request) or request.headers.get(INTERNAL_TOOL_HEADER):
         raise HTTPException(403, "Internal shell execution requires a dedicated resource-bound producer")
     if _auth_disabled():
+        from src.auth_helpers import is_direct_loopback_request
+        if is_direct_loopback_request(request):
+            return
         raise HTTPException(403, "Anonymous native process control has no resource authority")
     auth_manager = getattr(request.app.state, "auth_manager", None)
     if not auth_manager:

specs/auth-security.md

@@ -74,6 +74,13 @@ Missing-owner values remain state-dependent at legacy call sites, but new storag
 
 - Auth-enabled, configured auth with no `current_user` is unauthenticated and should fail closed at route dependencies.
 - `AUTH_ENABLED=false` is an explicit local single-user/no-login mode. Existing route dependencies can still return `""`, and admin gates allow the local operator. `effective_storage_owner()` and `storage_owner_for_request()` normalize an absent owner to `__odysseus_local__` only in this mode.
+  Native shell and local Cookbook administration additionally require a direct
+  loopback connection without proxy forwarding, cross-site indicators, or an
+  internal-tool header. Remote/proxied anonymous traffic remains denied at
+  these controls. Auth-enabled administration still requires a human admin.
+  Local Cookbook tools use a separate one-use capability for an admitted exact
+  request/operation/native backend and resolved launch body; that capability
+  cannot administer shell, PID, SSH-key, or arbitrary Cookbook state routes.
 - Chat/agent code that reads `get_current_user(request)` directly gets `None` when auth middleware is disabled, because no middleware stamps request state.
 - SQL `NULL`/JSON missing owners remain legacy/shared compatibility data, not the same thing as a logged-out authenticated caller.
 - `"api"` and `"internal-tool"` are request sentinels. They must not be persisted as normal storage owners unless a route explicitly defines that behavior.

src/agent_runtime/authority.py

@@ -523,6 +523,13 @@ def seal_task_authority(prompt, task_type, action, *, owner=None, parent_authori
     if operation is not None:
         authority = replace(authority, grants=(OperationGrant(operation.tool,
             inputs=frozenset({operation.input})),))
+        if task_type == "action" and action == "cookbook_serve":
+            # The direct admin scheduling ingress selects the native Cookbook
+            # producer. Restore never infers this from task names/availability.
+            # Any model-created task still intersects with its parent's ceiling.
+            backend = NativeBackendResource("serve_model")
+            authority = replace(authority, backend_resources=tuple(dict.fromkeys(
+                (*authority.backend_resources, backend))))
     parent = active_request_authority() if parent_authority is MISSING_AUTHORITY else parent_authority
     if parent_authority is None:
         parent = RequestAuthority.empty(owner=owner)

src/agent_runtime/local_model_control.py

@@ -0,0 +1,105 @@
+"""One-use transport capabilities for admitted local Cookbook producers.
+
+The internal HTTP token authenticates transport only. A capability bridges one
+server-owned request/operation/backend to one exact resolved local launch body.
+It is never persisted, returned to the model, or usable for shell/job control.
+"""
+from contextlib import contextmanager
+from dataclasses import dataclass
+import hashlib
+import json
+import secrets
+import threading
+import time
+
+from src.agent_runtime.resources import NativeBackendResource, ResourceIdentityError
+
+CAPABILITY_HEADER = "X-Odysseus-Local-Model-Capability"
+_ROUTES = {"download_model": "/api/model/download", "serve_model": "/api/model/serve",
+           "serve_preset": "/api/model/serve"}
+_PENDING = {}
+_LOCK = threading.Lock()
+
+
+def _digest(payload):
+    return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":"),
+                                     allow_nan=False).encode()).hexdigest()
+
+
+@dataclass(frozen=True)
+class _Capability:
+    authority: object
+    operation: object
+    backend: NativeBackendResource
+    path: str
+    payload_digest: str
+    deadline: float
+
+
+@contextmanager
+def model_control_headers(tool, content, owner, payload, *, scheduled=False):
+    from src.tools._common import _internal_headers
+    headers = _internal_headers(owner)
+    if payload.get("remote_host"):
+        yield headers  # Remote workload authority/transport is unchanged.
+        return
+    from src.agent_runtime.authority import active_request_authority, ExactOperation
+    from src.agent_runtime.remote_resources import active_backend_operation
+    from src.tool_security import owner_is_admin_or_single_user
+    authority = active_request_authority()
+    operation = ExactOperation.normalize(tool, content)
+    backend = active_backend_operation()
+    if (authority is None or authority.owner != str(owner or "").strip().casefold()
+            or tool not in _ROUTES or not owner_is_admin_or_single_user(owner)):
+        raise ResourceIdentityError("Local model producer has no matching server authority")
+    if scheduled:
+        # Called only by the server-owned scheduled action, after restoration of
+        # its immutable input ceiling. A task name or owner alone is not enough.
+        if tool != "serve_model" or not authority.permits(operation):
+            raise ResourceIdentityError("Scheduled local model input is outside authority")
+        resource = NativeBackendResource(tool)
+        if resource not in authority.backend_resources:
+            raise ResourceIdentityError("Scheduled local model backend is outside authority")
+    else:
+        # This binding exists only after dispatch admission (including one-use
+        # exact approval). A generic tool grant/header cannot create it over HTTP.
+        if (backend is None or backend.resource != NativeBackendResource(tool)
+                or (backend.request_id, backend.owner, backend.session_id,
+                    backend.transport_tool, backend.exact_input) !=
+                (authority.request_id, authority.owner, authority.session_id, tool, operation.input)):
+            raise ResourceIdentityError("Local model producer operation or backend changed")
+        resource = backend.resource
+    capability = _Capability(authority, operation, resource, _ROUTES[tool], _digest(payload), time.monotonic() + 60)
+    token = secrets.token_urlsafe(32)
+    headers.update({CAPABILITY_HEADER: token, "X-Odysseus-Owner": authority.owner})
+    with _LOCK:
+        _PENDING[token] = capability
+    try:
+        yield headers
+    finally:
+        with _LOCK:
+            _PENDING.pop(token, None)
+
+
+def consume_model_control(request, payload):
+    """Claim exactly once at the local route, before any producer effect."""
+    from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN, INTERNAL_TOOL_USER
+    from src.auth_helpers import is_direct_loopback_request
+    token = request.headers.get(CAPABILITY_HEADER)
+    if not token:
+        return False
+    if (not is_direct_loopback_request(request)
+            or not secrets.compare_digest(request.headers.get(INTERNAL_TOOL_HEADER, ""), INTERNAL_TOOL_TOKEN)):
+        raise ResourceIdentityError("Local model transport is untrusted")
+    with _LOCK:
+        capability = _PENDING.get(token)
+        if (capability is None or capability.deadline < time.monotonic()
+                or request.method != "POST" or request.url.path != capability.path
+                or payload.get("remote_host") or _digest(payload) != capability.payload_digest
+                or request.headers.get("X-Odysseus-Owner", "") != capability.authority.owner
+                or getattr(request.state, "current_user", None) not in
+                    (None, INTERNAL_TOOL_USER, capability.authority.owner)):
+            raise ResourceIdentityError("Local model capability binding changed or expired")
+        del _PENDING[token]
+    request.state.local_model_authority = capability.authority
+    return True

src/auth_helpers.py

@@ -7,6 +7,27 @@ from fastapi import Request, HTTPException
 from src.owner_identity import auth_disabled, effective_storage_owner
 
 
+def is_direct_loopback_request(request: Request) -> bool:
+    """Local operator transport, excluding reverse proxies and cross-site calls.
+
+    Locality supplies no model/tool authority. Native administration uses this
+    only in the operator's explicit auth-disabled single-user mode.
+    """
+    client = getattr(request, "client", None)
+    if not client or client.host not in {"127.0.0.1", "::1"}:
+        return False
+    forwarding = ("cf-connecting-ip", "cf-ray", "cf-visitor", "x-forwarded-for",
+                  "x-forwarded-host", "x-forwarded-proto", "x-real-ip", "forwarded")
+    if any(request.headers.get(name) for name in forwarding):
+        return False
+    if request.headers.get("sec-fetch-site") in {"cross-site", "same-site"}:
+        return False
+    origin = request.headers.get("origin")
+    if origin and origin != str(request.base_url).rstrip("/"):
+        return False
+    return True
+
+
 def get_current_user(request: Request) -> Optional[str]:
     """Get current username from request state (set by auth middleware)."""
     return getattr(request.state, 'current_user', None)

src/builtin_actions.py

@@ -3387,10 +3387,12 @@ async def action_cookbook_serve(
     if srv.get("platform"): body["platform"] = srv["platform"]
 
     try:
-        async with httpx.AsyncClient(timeout=30) as client:
-            r = await client.post(f"{internal_api_base()}/api/model/serve",
-                                  json=body, headers=headers)
-            data = r.json() if r.content else {}
+        from src.agent_runtime.local_model_control import model_control_headers
+        with model_control_headers("serve_model", command, owner, body, scheduled=True) as launch_headers:
+            async with httpx.AsyncClient(timeout=30) as client:
+                r = await client.post(f"{internal_api_base()}/api/model/serve",
+                                      json=body, headers=launch_headers)
+                data = r.json() if r.content else {}
     except Exception as e:
         return f"Launch HTTP failed: {e}", False
     if not data.get("ok"):

src/tools/cookbook.py

@@ -772,9 +772,11 @@ async def do_download_model(content: str, owner: Optional[str] = None) -> Dict:
     if env_cfg.get("platform"):   payload["platform"]   = env_cfg["platform"]
     if env_cfg.get("ssh_port"):   payload["ssh_port"]   = env_cfg["ssh_port"]
     try:
-        async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(f"{_INTERNAL_BASE}/api/model/download",
-                                     json=payload, headers=_internal_headers())
+        from src.agent_runtime.local_model_control import model_control_headers
+        with model_control_headers("download_model", content, owner, payload) as launch_headers:
+            async with httpx.AsyncClient(timeout=30) as client:
+                resp = await client.post(f"{_INTERNAL_BASE}/api/model/download",
+                                         json=payload, headers=launch_headers)
             data = resp.json()
         if data.get("ok"):
             sid = data.get("session_id", "?")
@@ -857,9 +859,11 @@ async def do_serve_model(content: str, owner: Optional[str] = None) -> Dict:
     if env_cfg.get("platform"):   payload["platform"]   = env_cfg["platform"]
     if env_cfg.get("ssh_port"):   payload["ssh_port"]   = env_cfg["ssh_port"]
     try:
-        async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(f"{_INTERNAL_BASE}/api/model/serve",
-                                     json=payload, headers=_internal_headers())
+        from src.agent_runtime.local_model_control import model_control_headers
+        with model_control_headers("serve_model", content, owner, payload) as launch_headers:
+            async with httpx.AsyncClient(timeout=30) as client:
+                resp = await client.post(f"{_INTERNAL_BASE}/api/model/serve",
+                                         json=payload, headers=launch_headers)
             data = resp.json()
         if data.get("ok"):
             sid = data.get("session_id", "?")
@@ -1908,9 +1912,11 @@ async def do_serve_preset(content: str, owner: Optional[str] = None) -> Dict:
         payload["ssh_port"] = env_cfg["ssh_port"]
 
     try:
-        async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(f"{_INTERNAL_BASE}/api/model/serve",
-                                     json=payload, headers=_internal_headers())
+        from src.agent_runtime.local_model_control import model_control_headers
+        with model_control_headers("serve_preset", content, owner, payload) as launch_headers:
+            async with httpx.AsyncClient(timeout=30) as client:
+                resp = await client.post(f"{_INTERNAL_BASE}/api/model/serve",
+                                         json=payload, headers=launch_headers)
             data = resp.json()
         if data.get("ok"):
             sid = data.get("session_id", "?")

tests/test_runtime_resource_integration.py

@@ -330,7 +330,7 @@ async def test_anonymous_native_cookbook_control_rejected_before_producer(monkey
     monkeypatch.setattr(asyncio, "create_subprocess_shell", lambda *a, **k: pytest.fail("Anonymous producer reached"))
     app = FastAPI()
     app.include_router(cookbook_routes.setup_cookbook_routes())
-    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://local") as client:
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, client=("192.0.2.1", 123)), base_url="http://local") as client:
         result = await client.post(path, json=payload)
     assert result.status_code == 403
 

tests/test_wave3_local_control.py

@@ -0,0 +1,209 @@
+"""Local administration and exact Cookbook caller-to-route regressions."""
+import asyncio
+import json
+from dataclasses import replace
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from fastapi import FastAPI, HTTPException
+from starlette.requests import Request
+
+from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN, INTERNAL_TOOL_USER
+from routes import cookbook_routes, shell_routes
+from src import builtin_actions, tool_execution
+from src.agent_runtime.authority import ExactOperation, OperationGrant, RequestAuthority, bind_request_authority, seal_task_authority, restore_task_authority
+from src.agent_runtime.remote_resources import bind_backend_for_operation, bind_backend_operation
+from src.agent_runtime.local_model_control import CAPABILITY_HEADER, model_control_headers
+from src.agent_runtime.resources import ResourceIdentityError
+from src.tools import cookbook
+from src.tool_capabilities import ToolRunSecurityContext
+from src.tool_types import ToolBlock
+
+
+def request(host='127.0.0.1', headers=None, user=None):
+    req = Request({'type': 'http', 'method': 'POST', 'scheme': 'http', 'path': '/api/shell/exec',
+                   'server': ('127.0.0.1', 7000), 'client': (host, 1234),
+                   'headers': [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()],
+                   'app': SimpleNamespace(state=SimpleNamespace(auth_manager=SimpleNamespace(is_admin=lambda u: u == 'alice')))})
+    req.state.current_user = user
+    return req
+
+
+@pytest.mark.parametrize('host,headers,allowed', [
+    ('127.0.0.1', {}, True), ('::1', {}, True), ('192.0.2.1', {}, False),
+    ('127.0.0.1', {'x-forwarded-for': '192.0.2.1'}, False),
+    ('127.0.0.1', {'forwarded': 'for=192.0.2.1'}, False),
+    ('127.0.0.1', {'cf-ray': 'proxy'}, False),
+    ('127.0.0.1', {'x-forwarded-proto': 'https'}, False),
+    ('127.0.0.1', {'sec-fetch-site': 'cross-site'}, False),
+    ('127.0.0.1', {'origin': 'https://evil.example'}, False),
+    ('127.0.0.1', {INTERNAL_TOOL_HEADER: 'forged'}, False),
+    ('127.0.0.1', {INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN}, False),
+])
+def test_auth_disabled_operator_transport(monkeypatch, host, headers, allowed):
+    monkeypatch.setenv('AUTH_ENABLED', 'false')
+    req = request(host, headers)
+    if allowed:
+        shell_routes._require_admin(req)
+    else:
+        with pytest.raises(HTTPException) as error:
+            shell_routes._require_admin(req)
+        assert error.value.status_code == 403
+
+
+@pytest.mark.parametrize('user,allowed', [('alice', True), ('bob', False), (None, False), ('api', False), (INTERNAL_TOOL_USER, False)])
+def test_auth_enabled_administration(monkeypatch, user, allowed):
+    monkeypatch.setenv('AUTH_ENABLED', 'true')
+    if allowed:
+        shell_routes._require_admin(request('192.0.2.1', user=user))
+    else:
+        with pytest.raises(HTTPException):
+            shell_routes._require_admin(request(user=user))
+
+
+@pytest.fixture
+def control_app(tmp_path, monkeypatch):
+    monkeypatch.setenv('AUTH_ENABLED', 'true')
+    manager = SimpleNamespace(is_configured=True, users={'alice': {}}, is_admin=lambda u: u == 'alice')
+    import core.auth
+    monkeypatch.setattr(core.auth, 'AuthManager', lambda: manager)
+    monkeypatch.setattr(tool_execution, '_owner_is_admin', lambda u: u == 'alice')
+    monkeypatch.setattr(cookbook_routes, 'TMUX_LOG_DIR', tmp_path / 'tmux')
+    state = tmp_path / 'cookbook.json'
+    state.write_text(json.dumps({'presets': [{'name': 'preset', 'model': 'samplepkg', 'cmd': 'python -m pip install samplepkg'}]}))
+    monkeypatch.setattr(cookbook_routes, 'COOKBOOK_STATE_FILE', str(state))
+    monkeypatch.setattr(builtin_actions, 'COOKBOOK_STATE_FILE', str(state))
+    spawned = []
+    async def spawn(command, **kwargs):
+        spawned.append(command)
+        async def wait(): return 0
+        async def read(): return b''
+        return SimpleNamespace(returncode=0, wait=wait, stderr=SimpleNamespace(read=read))
+    monkeypatch.setattr(asyncio, 'create_subprocess_shell', spawn)
+    async def remote_probe(*args, **kwargs):
+        async def communicate(): return b'tmux', b''
+        return SimpleNamespace(returncode=0, communicate=communicate)
+    monkeypatch.setattr(asyncio, 'create_subprocess_exec', remote_probe)
+    import src.assistant_log
+    monkeypatch.setattr(src.assistant_log, 'log_to_assistant', lambda *a, **k: None)
+    async def endpoint(**kwargs): return {'added': True, 'endpoint_id': 'endpoint'}
+    monkeypatch.setattr(cookbook, '_ensure_served_endpoint', endpoint)
+    app = FastAPI()
+    app.state.auth_manager = manager
+    @app.middleware('http')
+    async def attribution(req, next):
+        if req.headers.get(INTERNAL_TOOL_HEADER) == INTERNAL_TOOL_TOKEN:
+            req.state.current_user = req.headers.get('X-Odysseus-Owner') or INTERNAL_TOOL_USER
+        return await next(req)
+    app.include_router(cookbook_routes.setup_cookbook_routes())
+    app.include_router(shell_routes.setup_shell_routes())
+    real_client = httpx.AsyncClient
+    def client_factory(*args, **kwargs):
+        kwargs.setdefault('transport', httpx.ASGITransport(app=app))
+        return real_client(*args, **kwargs)
+    monkeypatch.setattr(httpx, 'AsyncClient', client_factory)
+    return app, spawned, tmp_path
+
+
+@pytest.mark.parametrize('tool,args', [
+    ('download_model', {'repo_id': 'org/model', 'local': True}),
+    ('serve_model', {'repo_id': 'samplepkg', 'cmd': 'python -m pip install samplepkg', 'local': True}),
+    ('serve_preset', {'name': 'preset'}),
+])
+async def test_real_local_tool_dispatch_reaches_real_model_route(control_app, tool, args):
+    app, spawned, work = control_app
+    authority = RequestAuthority('request', 'alice', 'thread', str(work), (OperationGrant(tool),))
+    _, result = await tool_execution.execute_tool_block(ToolBlock(tool, json.dumps(args)), owner='alice',
+        session_id='thread', workspace=str(work), request_authority=authority, security_context=ToolRunSecurityContext())
+    assert result['exit_code'] == 0, result
+    assert result['session_id'].startswith('cookbook-' if tool == 'download_model' else 'serve-')
+    assert len(spawned) == 1 and 'tmux new-session' in spawned[0]
+
+
+async def test_real_scheduled_local_action_uses_restored_exact_authority(control_app):
+    app, spawned, work = control_app
+    command = json.dumps({'repo_id': 'samplepkg', 'cmd': 'python -m pip install samplepkg', 'set_default': False})
+    snapshot = seal_task_authority(command, 'action', 'cookbook_serve', owner='alice')
+    authority = restore_task_authority(snapshot, command, 'action', 'cookbook_serve', owner='alice')
+    with bind_request_authority(authority):
+        message, ok = await builtin_actions.action_cookbook_serve('alice', command=command)
+    assert ok, message
+    assert len(spawned) == 1
+    with bind_request_authority(authority):
+        _, ok = await builtin_actions.action_cookbook_serve('alice', command=command.replace('samplepkg', 'changedpkg'))
+    assert not ok and len(spawned) == 1
+
+
+@pytest.mark.parametrize('host,headers', [
+    ('127.0.0.1', {INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN}),
+    ('192.0.2.1', {INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN}),
+    ('127.0.0.1', {INTERNAL_TOOL_HEADER: 'forged'}),
+    ('127.0.0.1', {CAPABILITY_HEADER: 'forged', INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN}),
+])
+async def test_header_only_cannot_launch(control_app, host, headers):
+    app, spawned, _ = control_app
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, client=(host, 123)), base_url='http://127.0.0.1') as client:
+        for path in ('/api/model/download', '/api/model/serve', '/api/shell/exec'):
+            r = await client.post(path, json={'repo_id': 'org/model', 'cmd': 'printf nope', 'command': 'printf nope'}, headers=headers)
+            assert r.status_code == 403
+    assert not spawned
+
+
+@pytest.mark.parametrize('substitute', ['body', 'route', 'owner', 'remote', 'proxy', 'replay'])
+async def test_capability_exact_transport_binding(control_app, substitute):
+    app, spawned, work = control_app
+    content = json.dumps({'repo_id': 'org/model', 'local': True})
+    authority = RequestAuthority('request', 'alice', 'thread', str(work), (OperationGrant('download_model'),))
+    operation = ExactOperation.normalize('download_model', content)
+    backend = bind_backend_for_operation(authority, operation)
+    body = {'repo_id': 'org/model'}
+    with bind_request_authority(authority), bind_backend_operation(backend), model_control_headers('download_model', content, 'alice', body) as headers:
+        changed = dict(headers); payload = dict(body); path = '/api/model/download'; host = '127.0.0.1'
+        if substitute == 'body': payload['repo_id'] = 'org/changed'
+        if substitute == 'route': path = '/api/model/serve'
+        if substitute == 'owner': changed['X-Odysseus-Owner'] = 'bob'
+        if substitute == 'remote': host = '192.0.2.1'
+        if substitute == 'proxy': changed['x-forwarded-for'] = '192.0.2.1'
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, client=(host, 123)), base_url='http://127.0.0.1') as client:
+            if substitute == 'replay':
+                assert (await client.post(path, json=payload, headers=changed)).status_code == 200
+            r = await client.post(path, json=payload, headers=changed)
+            assert r.status_code == 403
+    assert len(spawned) == (1 if substitute == 'replay' else 0)
+
+
+@pytest.mark.parametrize('field', ['owner', 'request_id', 'session_id'])
+def test_producer_wrong_application_binding(control_app, field):
+    _, _, work = control_app
+    content = '{"repo_id":"org/model","local":true}'
+    authority = RequestAuthority('request', 'alice', 'thread', str(work), (OperationGrant('download_model'),))
+    backend = bind_backend_for_operation(authority, ExactOperation.normalize('download_model', content))
+    changed = replace(authority, **{field: 'replacement'}, resource_roots=None, backend_resources=None, owned_scopes=None)
+    with bind_request_authority(changed), bind_backend_operation(backend), pytest.raises(ResourceIdentityError):
+        with model_control_headers('download_model', content, 'alice', {'repo_id': 'org/model'}):
+            pytest.fail('Substituted producer obtained a capability')
+
+
+async def test_remote_route_semantics_remain_unchanged(control_app):
+    app, spawned, _ = control_app
+    async with httpx.AsyncClient(base_url='http://127.0.0.1') as client:
+        r = await client.post('/api/model/download', json={'repo_id': 'org/model', 'remote_host': 'gpu.example'},
+                              headers={INTERNAL_TOOL_HEADER: INTERNAL_TOOL_TOKEN})
+    assert r.status_code == 200 and r.json()['ok'], r.text
+    assert len(spawned) == 1 and 'ssh ' in spawned[0]
+
+
+async def test_auth_disabled_local_route_usage(control_app, monkeypatch):
+    app, spawned, _ = control_app
+    monkeypatch.setenv('AUTH_ENABLED', 'false')
+    async with httpx.AsyncClient(base_url='http://127.0.0.1') as client:
+        shell = await client.post('/api/shell/exec', json={'command': ''})
+        state = await client.post('/api/cookbook/state', json={'tasks': []})
+        launch = await client.post('/api/model/download', json={'repo_id': 'org/model'})
+    assert shell.status_code == state.status_code == launch.status_code == 200
+    assert launch.json()['ok'] and len(spawned) == 1
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, client=('192.0.2.1', 1)), base_url='http://127.0.0.1') as client:
+        for path, body in [('/api/shell/exec', {'command': ''}), ('/api/cookbook/state', {'tasks': []}), ('/api/model/download', {'repo_id': 'org/model'})]:
+            assert (await client.post(path, json=body)).status_code == 403