diff --git a/routes/chat_helpers.py b/routes/chat_helpers.py index ddc193f45..8b7ebff5a 100644 --- a/routes/chat_helpers.py +++ b/routes/chat_helpers.py @@ -19,6 +19,7 @@ from src.model_context import estimate_tokens, get_context_length from src.auth_helpers import ( RequestCapability, effective_user, + is_bearer_principal, request_capability as build_request_capability, ) from src.tool_approval_scopes import CHAT_SESSION_APPROVAL_CONTEXT_MARKER @@ -204,6 +205,11 @@ def _allowed_models_from_privileges(privs: dict) -> Optional[frozenset[str]]: def _allowed_models_for_request(request) -> Optional[frozenset[str]]: """Return the caller's model allowlist, or ``None`` when unrestricted.""" + # ``effective_user`` is an attribution/storage identity for bearers, not a + # browser privilege principal. In particular, an admin-owned token must + # not inherit the owner's ADMIN_PRIVILEGES map through this lookup. + if is_bearer_principal(request): + return None try: user = effective_user(request) except Exception: @@ -226,6 +232,12 @@ def _enforce_chat_privileges(request, sess) -> None: (single-user mode). Admins receive ADMIN_PRIVILEGES from get_privileges, which means unrestricted allowed_models / zero cap -> no-op for them. """ + # Bearer authority is defined by the token scope at the route boundary. + # Do not turn its owner attribution back into a browser privilege lookup; + # that would make an admin-owned token inherit the admin model/cap policy. + if is_bearer_principal(request): + return + try: user = effective_user(request) except Exception: @@ -817,11 +829,16 @@ async def build_chat_context( # Normalize model ID. Prefer cached endpoint models so group chat does not # re-hit slow local /models endpoints on every participant turn. - norm = _normalize_model_id_from_cache(sess) or normalize_model_id( - sess.endpoint_url, - sess.model, - owner=getattr(sess, "owner", None), - ) + norm = _normalize_model_id_from_cache(sess) + # Model normalization falls back to a live /models or /tags request on a + # cache miss. A bearer chat request may use the stored model as-is, but it + # must not implicitly refresh an endpoint catalogue while building context. + if norm is None and capability.allow_live_probes: + norm = normalize_model_id( + sess.endpoint_url, + sess.model, + owner=getattr(sess, "owner", None), + ) if norm: sess.model = norm @@ -860,11 +877,22 @@ async def build_chat_context( # session history before we know which route can answer and would make a # later larger-context candidate unable to recover discarded history. if defer_context_shaping: - context_length = get_context_length(sess.endpoint_url, sess.model) + context_kwargs = {} + if not capability.allow_live_probes: + context_kwargs["allow_live_probes"] = False + context_length = get_context_length(sess.endpoint_url, sess.model, **context_kwargs) was_compacted = False else: + compact_kwargs = {"owner": user} + if not capability.allow_live_probes: + compact_kwargs["allow_live_probes"] = False messages, context_length, was_compacted = await maybe_compact( - sess, sess.endpoint_url, sess.model, messages, sess.headers, owner=user, + sess, + sess.endpoint_url, + sess.model, + messages, + sess.headers, + **compact_kwargs, ) _before_trim_messages = len(messages) _before_trim_tokens = estimate_tokens(messages) diff --git a/routes/chat_routes.py b/routes/chat_routes.py index 6a33d00fe..21d3484e6 100644 --- a/routes/chat_routes.py +++ b/routes/chat_routes.py @@ -167,6 +167,7 @@ def _chat_candidate_request_factory( *, session=None, owner: Optional[str] = None, + allow_live_probes: bool = True, ): """Shape one route-neutral Chat prompt for each candidate window.""" @@ -180,15 +181,20 @@ def _chat_candidate_request_factory( async def factory(index, candidate_url, candidate_model, candidate_headers): compaction_state = {} + compact_kwargs = { + "owner": owner, + "persist": False, + "compaction_state": compaction_state, + } + if not allow_live_probes: + compact_kwargs["allow_live_probes"] = False candidate_messages, context_length, was_compacted = await maybe_compact( session, candidate_url, candidate_model, list(messages), candidate_headers, - owner=owner, - persist=False, - compaction_state=compaction_state, + **compact_kwargs, ) if not context_length: context_length = fallback_context_length @@ -878,17 +884,23 @@ def setup_chat_routes( selected_context_length, session=sess, owner=owner, + allow_live_probes=request_capability.allow_live_probes, ) requested_model = sess.model + llm_kwargs = { + "fallback_statuses": foreground_policy.eligible_statuses, + "candidate_request_factory": candidate_request_factory, + "temperature": ctx.preset.temperature, + "max_tokens": ctx.preset.max_tokens, + "prompt_type": preset_id, + "session_id": session, + } + if not request_capability.allow_live_probes: + llm_kwargs["allow_live_probes"] = False reply, actual_candidate, actual_model = await llm_call_async_with_route_fallback( foreground_candidates, request_messages, - fallback_statuses=foreground_policy.eligible_statuses, - candidate_request_factory=candidate_request_factory, - temperature=ctx.preset.temperature, - max_tokens=ctx.preset.max_tokens, - prompt_type=preset_id, - session_id=session, + **llm_kwargs, ) actual_index = _candidate_index(foreground_candidates, actual_candidate) apply_compaction_state( @@ -1605,7 +1617,12 @@ def setup_chat_routes( # Enforce per-user privileges _privs = {} _user = ctx.user - if _user and hasattr(request.app.state, 'auth_manager') and request.app.state.auth_manager: + if ( + not api_token_request + and _user + and hasattr(request.app.state, 'auth_manager') + and request.app.state.auth_manager + ): _privs = request.app.state.auth_manager.get_privileges(_user) if _privs: if not _privs.get("can_use_bash", True): @@ -1897,6 +1914,7 @@ def setup_chat_routes( _selected_context_length, session=sess, owner=_user, + allow_live_probes=request_capability.allow_live_probes, ) # Send model name early so the frontend can show it during streaming @@ -2030,23 +2048,28 @@ def setup_chat_routes( # ── Chat mode: call stream_llm directly, NO tools, NO document access ── try: - async for chunk in stream_llm_with_fallback( - _foreground_candidates, - messages, - temperature=ctx.preset.temperature, + stream_kwargs = { + "temperature": ctx.preset.temperature, # Respect the preset; 0/unset = let the server decide (no # cap), matching agent mode. The old hard 4096 fallback # truncated reasoning models mid- — they'd burn the # whole budget thinking and never emit the answer (seen in # Compare on heavy generation prompts). - max_tokens=ctx.preset.max_tokens, - prompt_type=preset_id, - tools=None, - session_id=session, - fallback_statuses=_foreground_policy.eligible_statuses, - fallback_on_empty=_foreground_policy.fallback_on_empty, - candidate_request_factory=_chat_request_factory, - candidate_route_descriptors=_foreground_route_descriptors, + "max_tokens": ctx.preset.max_tokens, + "prompt_type": preset_id, + "tools": None, + "session_id": session, + "fallback_statuses": _foreground_policy.eligible_statuses, + "fallback_on_empty": _foreground_policy.fallback_on_empty, + "candidate_request_factory": _chat_request_factory, + "candidate_route_descriptors": _foreground_route_descriptors, + } + if not request_capability.allow_live_probes: + stream_kwargs["allow_live_probes"] = False + async for chunk in stream_llm_with_fallback( + _foreground_candidates, + messages, + **stream_kwargs, ): if chunk.startswith("data: ") and not chunk.startswith("data: [DONE]"): try: diff --git a/routes/email_helpers.py b/routes/email_helpers.py index 257f5f921..67972ef8a 100644 --- a/routes/email_helpers.py +++ b/routes/email_helpers.py @@ -34,7 +34,7 @@ from fastapi import Query, HTTPException, Request from pydantic import BaseModel from typing import Optional, List -from src.auth_helpers import _auth_disabled, get_current_user +from src.auth_helpers import _auth_disabled, get_current_user, is_bearer_principal from src.secret_storage import decrypt as _decrypt logger = logging.getLogger(__name__) @@ -420,6 +420,15 @@ def _require_auth(request: Request) -> str: unconfigured mode are only honoured if they're coming from localhost; everyone else gets 401. """ + # The legacy email router uses one generic dependency for mailbox reads, + # drafts, AI helpers, and SMTP send. It has no per-route token-scope + # contract, so a bearer must not be allowed to enter it as the ``api`` + # pseudo-user. Otherwise owner-scoped lookup can miss the token owner and + # fall through to process-wide legacy settings credentials below. + # Scope-aware integrations must use their dedicated route boundary. + if is_bearer_principal(request): + raise HTTPException(403, "API tokens must use a scope-aware email route") + u = get_current_user(request) if u: return u diff --git a/routes/history/history_routes.py b/routes/history/history_routes.py index f43123c58..05337c00f 100644 --- a/routes/history/history_routes.py +++ b/routes/history/history_routes.py @@ -10,8 +10,11 @@ from fastapi import APIRouter, Depends, Request, HTTPException from core.models import ChatMessage from core.database import SessionLocal, ChatMessage as DbChatMessage, Session as DbSession -from src.auth_helpers import effective_user, require_chat_scope -from src.message_metadata import sanitize_client_message_metadata +from src.auth_helpers import effective_user, is_bearer_principal, require_chat_scope +from src.message_metadata import ( + sanitize_client_message_metadata, + sanitize_projected_message_metadata, +) from src.topic_analyzer import analyze_topics from src.upload_handler import reserve_message_upload_references from routes.session_routes import ( @@ -142,11 +145,19 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: f"Referenced upload is no longer available: {missing_id}", ) - def _db_history_entry(m: DbChatMessage) -> Dict[str, Any]: + def _display_metadata(value: Any, *, sanitize: bool) -> dict: + meta = _metadata_dict(value) + if sanitize: + return sanitize_projected_message_metadata(meta) or {} + return dict(meta) + + def _db_history_entry( + m: DbChatMessage, + *, + sanitize: bool = False, + ) -> Dict[str, Any]: entry = {"role": m.role, "content": _history_display_content(m.content)} - meta = _metadata_dict(m.meta_data) - if meta: - meta = dict(meta) + meta = _display_metadata(m.meta_data, sanitize=sanitize) if m.timestamp and "timestamp" not in meta: meta["timestamp"] = m.timestamp.isoformat() + "Z" if meta: @@ -161,6 +172,7 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: offset: Optional[int] = None, ) -> Dict[str, Any]: require_chat_scope(request) + sanitize_history = is_bearer_principal(request) _verify_session_owner(request, session_id) if limit is not None: page_limit = max(1, min(int(limit), 100)) @@ -188,7 +200,11 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: .all() ) history_dict = [ - entry for entry in (_db_history_entry(m) for m in rows) + entry + for entry in ( + _db_history_entry(m, sanitize=sanitize_history) + for m in rows + ) if not (entry.get("metadata") or {}).get("hidden") ] return { @@ -214,7 +230,10 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: for msg in session.history: if isinstance(msg, ChatMessage): # Skip hidden messages (e.g. compaction summaries for AI context) - msg_meta = _metadata_dict(msg.metadata) + msg_meta = _display_metadata( + msg.metadata, + sanitize=sanitize_history, + ) if msg_meta.get("hidden"): continue entry = {"role": msg.role, "content": _history_display_content(msg.content)} @@ -222,7 +241,10 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: entry["metadata"] = msg_meta history_dict.append(entry) elif isinstance(msg, dict): - msg_meta = _metadata_dict(msg.get("metadata")) + msg_meta = _display_metadata( + msg.get("metadata"), + sanitize=sanitize_history, + ) if msg_meta.get("hidden"): continue entry = { @@ -248,7 +270,11 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: ) # Response excludes hidden messages, matching the in-memory path. history_dict = [ - entry for entry in (_db_history_entry(m) for m in db_messages) + entry + for entry in ( + _db_history_entry(m, sanitize=sanitize_history) + for m in db_messages + ) if not (entry.get("metadata") or {}).get("hidden") ] except Exception as e: @@ -653,12 +679,15 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: # in-memory messages, corrupting their _db_id and breaking # edit/delete-by-id on the original conversation. meta = dict(msg.metadata) if isinstance(msg.metadata, dict) else None + if is_bearer_principal(request): + meta = sanitize_projected_message_metadata(meta) new_session.add_message(ChatMessage(msg.role, msg.content, meta)) - try: - from src.event_bus import fire_event - fire_event("session_created", getattr(source, 'owner', None)) - except Exception: - logger.debug("session_created event dispatch failed", exc_info=True) + if not is_bearer_principal(request): + try: + from src.event_bus import fire_event + fire_event("session_created", getattr(source, 'owner', None)) + except Exception: + logger.debug("session_created event dispatch failed", exc_info=True) return { "status": "ok", diff --git a/routes/session_routes.py b/routes/session_routes.py index f7447547c..8428469e3 100644 --- a/routes/session_routes.py +++ b/routes/session_routes.py @@ -11,7 +11,13 @@ from core.session_manager import SessionManager from core.models import ChatMessage from src.request_models import SessionResponse from core.database import Session as DbSession, SessionLocal, Document, GalleryImage, utcnow_naive -from src.auth_helpers import effective_user, _auth_disabled, owner_filter, require_chat_scope +from src.auth_helpers import ( + effective_user, + _auth_disabled, + is_bearer_principal, + owner_filter, + require_chat_scope, +) from src.message_metadata import sanitize_client_message_metadata from src.session_image_cleanup import _generated_image_path_for_cleanup, session_image_refs from src.session_actions import is_session_recently_active @@ -158,7 +164,10 @@ def _reject_raw_endpoint_url_for_non_admin( # Raw URLs make the server dial whatever host the request supplies. For # non-admin users, require a saved endpoint row so normal owner scoping and # endpoint validation have already happened. - if user and not _current_user_is_admin(request, user): + # A bearer may be attributed to an admin owner for storage and endpoint + # visibility, but it is still not an interactive admin principal. Raw + # endpoint URLs therefore remain unavailable to every bearer request. + if is_bearer_principal(request) or (user and not _current_user_is_admin(request, user)): raise HTTPException(403, "Choose a registered model endpoint") @@ -450,14 +459,17 @@ def setup_session_routes( from src.endpoint_resolver import build_headers session.headers = build_headers(resolved_key, resolved_base) _persist_session_headers(sid, session.headers) - # Fire webhook (sync-safe) - if webhook_manager: - webhook_manager.fire_and_forget("session.created", { - "session_id": sid, "name": session.name, "model": model_to_use, - }) - # Fire event for automation tasks - from src.event_bus import fire_event - fire_event("session_created", user) + # A bearer can create owner-attributed chat data, but must not cause + # owner lifecycle automation or webhook delivery as a side effect. + if not is_bearer_principal(request): + # Fire webhook (sync-safe) + if webhook_manager: + webhook_manager.fire_and_forget("session.created", { + "session_id": sid, "name": session.name, "model": model_to_use, + }) + # Fire event for automation tasks + from src.event_bus import fire_event + fire_event("session_created", user) return SessionResponse( id=sid, name=session.name, @@ -943,8 +955,9 @@ def setup_session_routes( ) session.headers = {"Authorization": f"Bearer {OPENAI_API_KEY}"} session_manager.save_sessions() - from src.event_bus import fire_event - fire_event("session_created", user) + if not is_bearer_principal(request): + from src.event_bus import fire_event + fire_event("session_created", user) return {"id": sid, "name": "", "model": model} @router.post("/session/{session_id}/important") diff --git a/routes/upload_routes.py b/routes/upload_routes.py index 8d6e1514d..47e83ac17 100644 --- a/routes/upload_routes.py +++ b/routes/upload_routes.py @@ -21,7 +21,7 @@ from core.database import ( Note, Session as DbSession, ) -from src.auth_helpers import effective_user, require_chat_scope +from src.auth_helpers import effective_user, require_chat_scope, require_non_bearer_request from src.attachment_refs import attachment_refs_from_metadata from src.constants import GENERATED_IMAGES_DIR from src.upload_handler import ( @@ -462,6 +462,7 @@ def setup_upload_routes(upload_handler): Cached under UPLOAD_DIR/.vision/{file_id}.txt — first call computes, subsequent loads are instant. Pass force=1 to recompute.""" require_chat_scope(request) + require_non_bearer_request(request) if not upload_handler.validate_upload_id(file_id): raise HTTPException(400, "Invalid file ID") info = _load_upload_info(file_id) @@ -507,6 +508,7 @@ def setup_upload_routes(upload_handler): """Persist a user-edited vision/OCR text for an attachment. Stored in the same cache file so the chat send picks it up as the override.""" require_chat_scope(request) + require_non_bearer_request(request) if not upload_handler.validate_upload_id(file_id): raise HTTPException(400, "Invalid file ID") info = _load_upload_info(file_id) diff --git a/routes/webhook/webhook_routes.py b/routes/webhook/webhook_routes.py index 066e26189..87f7f1440 100644 --- a/routes/webhook/webhook_routes.py +++ b/routes/webhook/webhook_routes.py @@ -9,7 +9,7 @@ from fastapi import APIRouter, HTTPException, Request, Form from pydantic import BaseModel, Field from core.database import SessionLocal, Webhook, ModelEndpoint -from src.auth_helpers import owner_filter, require_chat_scope +from src.auth_helpers import is_bearer_principal, owner_filter, require_chat_scope from src.url_security import validate_public_http_url from src.webhook_manager import WebhookManager, validate_webhook_url, validate_events @@ -385,10 +385,14 @@ def setup_webhook_routes( sess.add_message(ChatMessage("assistant", reply)) session_manager.save_sessions() - webhook_manager.fire_and_forget("chat.completed", { - "session_id": session_id, "model": sess.model, - "user_message": message[:2000], "response": reply[:2000], - }) + # /api/v1/chat remains a synchronous bearer integration: the response + # is returned normally, but the token must not fan that content out to + # an owner-configured asynchronous callback after authorization ends. + if not is_bearer_principal(request): + webhook_manager.fire_and_forget("chat.completed", { + "session_id": session_id, "model": sess.model, + "user_message": message[:2000], "response": reply[:2000], + }) return {"response": reply, "session_id": session_id, "model": sess.model} diff --git a/src/auth_helpers.py b/src/auth_helpers.py index 752355c2b..ae7a1b232 100644 --- a/src/auth_helpers.py +++ b/src/auth_helpers.py @@ -31,6 +31,7 @@ class RequestCapability: allow_detached_execution: bool allow_message_events: bool allow_auto_naming: bool + allow_live_probes: bool def is_bearer_principal(request: Request) -> bool: @@ -101,6 +102,7 @@ def request_capability(request: Request) -> RequestCapability: allow_detached_execution=not bearer, allow_message_events=not bearer, allow_auto_naming=not bearer, + allow_live_probes=not bearer, ) diff --git a/src/context_compactor.py b/src/context_compactor.py index d6adf5bb4..94e4b7638 100644 --- a/src/context_compactor.py +++ b/src/context_compactor.py @@ -330,12 +330,16 @@ async def maybe_compact( *, persist: bool = True, compaction_state: Optional[Dict[str, Any]] = None, + allow_live_probes: bool = True, ) -> tuple: """Check context usage and compact if above threshold. Returns (messages, context_length, was_compacted). """ - context_length = get_context_length(endpoint_url, model) + context_kwargs = {} + if not allow_live_probes: + context_kwargs["allow_live_probes"] = False + context_length = get_context_length(endpoint_url, model, **context_kwargs) used = estimate_tokens(messages) pct = (used / context_length) * 100 if context_length else 0 @@ -392,6 +396,9 @@ async def maybe_compact( ] try: + summary_kwargs = {} + if not allow_live_probes: + summary_kwargs["allow_live_probes"] = False summary = await llm_call_async( compact_url, compact_model, @@ -400,6 +407,7 @@ async def maybe_compact( max_tokens=SUMMARY_MAX_TOKENS, headers=compact_headers, timeout=30, + **summary_kwargs, ) except Exception as e: logger.error(f"Compaction summary failed: {e}") diff --git a/src/llm_core.py b/src/llm_core.py index cea829d45..699460e00 100644 --- a/src/llm_core.py +++ b/src/llm_core.py @@ -1903,6 +1903,7 @@ def list_model_ids( *, owner: Optional[str] = None, endpoint_id: Optional[str] = None, + allow_live_probes: bool = True, ) -> List[str]: """List available model IDs from an endpoint.""" cached = _configured_cached_model_ids(base_chat_url, owner=owner, endpoint_id=endpoint_id) @@ -1911,6 +1912,8 @@ def list_model_ids( provider = _detect_provider(base_chat_url) if provider == "anthropic": return list(ANTHROPIC_MODELS) + if not allow_live_probes: + return [] try: h = {} if headers: @@ -1952,9 +1955,16 @@ def normalize_model_id( *, owner: Optional[str] = None, endpoint_id: Optional[str] = None, + allow_live_probes: bool = True, ) -> Optional[str]: """Normalize a model ID to match available models.""" - avail = list_model_ids(endpoint_url, timeout, owner=owner, endpoint_id=endpoint_id) + avail = list_model_ids( + endpoint_url, + timeout, + owner=owner, + endpoint_id=endpoint_id, + allow_live_probes=allow_live_probes, + ) if not avail: return None if requested in avail: @@ -1968,7 +1978,8 @@ def normalize_model_id( def llm_call(url: str, model: str, messages: List[Dict], temperature: float = LLMConfig.DEFAULT_TEMPERATURE, max_tokens: int = LLMConfig.DEFAULT_MAX_TOKENS, headers: Optional[Dict] = None, - timeout: int = LLMConfig.DEFAULT_TIMEOUT, prompt_type: Optional[str] = None) -> str: + timeout: int = LLMConfig.DEFAULT_TIMEOUT, prompt_type: Optional[str] = None, + allow_live_probes: bool = True) -> str: """Synchronous LLM call with optional prompt type enhancement.""" h = _provider_headers(_detect_provider(url)) # Tolerate headers that arrive as a JSON string (some sessions stored them @@ -2012,9 +2023,12 @@ def llm_call(url: str, model: str, messages: List[Dict], temperature: float = LL payload = _build_anthropic_payload(model, messages_copy, temperature, max_tokens) elif provider == "ollama": target_url = _normalize_ollama_url(url) + context_kwargs = {} + if not allow_live_probes: + context_kwargs["allow_live_probes"] = False payload = _build_ollama_payload( model, messages_copy, temperature, max_tokens, - stream=False, num_ctx=get_context_length(url, model), + stream=False, num_ctx=get_context_length(url, model, **context_kwargs), ) else: target_url = _normalize_openai_chat_url(url) @@ -2273,6 +2287,7 @@ async def llm_call_async( workload: str = "foreground", availability_only_transport: bool = False, return_model_metadata: bool = False, + allow_live_probes: bool = True, ) -> str | tuple[str, str]: """Asynchronous LLM call using httpx with connection pooling, timeout, retry logic, and performance logging.""" provider = _detect_provider(url) @@ -2307,6 +2322,9 @@ async def llm_call_async( # Reuse stream_llm's validated Codex SSE path and collect deltas. parts: List[str] = [] actual_model = model + stream_kwargs = {"workload": workload} + if not allow_live_probes: + stream_kwargs["allow_live_probes"] = False async for chunk in stream_llm( url, model, @@ -2315,7 +2333,7 @@ async def llm_call_async( max_tokens=max_tokens, headers=headers, timeout=timeout, - workload=workload, + **stream_kwargs, ): event_is_error = False for line in str(chunk).splitlines(): @@ -2372,9 +2390,12 @@ async def llm_call_async( h = {"Content-Type": "application/json"} if headers: h.update(headers) + context_kwargs = {} + if not allow_live_probes: + context_kwargs["allow_live_probes"] = False payload = _build_ollama_payload( model, messages_copy, temperature, max_tokens, - stream=False, num_ctx=get_context_length(url, model), + stream=False, num_ctx=get_context_length(url, model, **context_kwargs), ) else: target_url = _normalize_openai_chat_url(url) @@ -2560,9 +2581,13 @@ async def stream_llm(url: str, model: str, messages: List[Dict], temperature: fl max_tokens: int = LLMConfig.DEFAULT_MAX_TOKENS, headers: Optional[Dict] = None, timeout: int = LLMConfig.STREAM_TIMEOUT, prompt_type: Optional[str] = None, tools: Optional[List[Dict]] = None, session_id: Optional[str] = None, - tool_choice_none: bool = False, workload: str = "foreground"): + tool_choice_none: bool = False, workload: str = "foreground", + allow_live_probes: bool = True): target_url = _stream_target_url(url) async with _local_model_slot(target_url, model, workload): + inner_kwargs = {} + if not allow_live_probes: + inner_kwargs["allow_live_probes"] = False async for chunk in _stream_llm_inner( url, model, @@ -2575,6 +2600,7 @@ async def stream_llm(url: str, model: str, messages: List[Dict], temperature: fl tools=tools, session_id=session_id, tool_choice_none=tool_choice_none, + **inner_kwargs, ): yield chunk @@ -2583,7 +2609,7 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat max_tokens: int = LLMConfig.DEFAULT_MAX_TOKENS, headers: Optional[Dict] = None, timeout: int = LLMConfig.STREAM_TIMEOUT, prompt_type: Optional[str] = None, tools: Optional[List[Dict]] = None, session_id: Optional[str] = None, - tool_choice_none: bool = False): + tool_choice_none: bool = False, allow_live_probes: bool = True): """Stream LLM responses with improved error handling. Yields SSE chunks: @@ -2618,9 +2644,14 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat h = {"Content-Type": "application/json"} if headers: h.update(headers) + context_kwargs = {} + if not allow_live_probes: + context_kwargs["allow_live_probes"] = False payload = _build_ollama_payload( model, messages_copy, temperature, max_tokens, - stream=True, tools=tools, num_ctx=get_context_length(url, model), + stream=True, + tools=tools, + num_ctx=get_context_length(url, model, **context_kwargs), ) elif provider == "chatgpt-subscription": target_url = _normalize_chatgpt_subscription_url(url) diff --git a/src/model_context.py b/src/model_context.py index b6afa801e..8c4c0e416 100644 --- a/src/model_context.py +++ b/src/model_context.py @@ -238,16 +238,31 @@ KNOWN_CONTEXT_WINDOWS = { _context_cache: Dict[Tuple[str, str], Tuple[int, bool]] = {} -def _get_context_length_cached(endpoint_url: str, model: str) -> Tuple[int, bool]: +def _get_context_length_cached( + endpoint_url: str, + model: str, + *, + allow_live_probes: bool = True, +) -> Tuple[int, bool]: """Return (context_length, known). ``known`` is False only when the value is a bare DEFAULT_CONTEXT fallback (no endpoint report and not in the known table).""" + cache_key = (endpoint_url, model) + if not allow_live_probes: + # A bearer may consume metadata already learned by an interactive or + # explicitly privileged refresh, but a context build must not cause a + # new /slots, /models, or catalog request or populate those caches. + cached = _context_cache.get(cache_key) + if cached: + return cached + known = _lookup_known(model) + return (known, True) if known else (DEFAULT_CONTEXT, False) + configured_kind = _configured_endpoint_kind(endpoint_url) is_local = is_local_endpoint(endpoint_url) # Key on (endpoint_url, model): the same model id can be served by two # different remote endpoints with different real context windows (e.g. a # capped proxy vs. the full provider), so caching by model id alone would # serve one endpoint's window for the other (issue #2603). - cache_key = (endpoint_url, model) if not is_local and cache_key in _context_cache: return _context_cache[cache_key] @@ -261,23 +276,41 @@ def _get_context_length_cached(endpoint_url: str, model: str) -> Tuple[int, bool return ctx, known -def get_context_length(endpoint_url: str, model: str) -> int: +def get_context_length( + endpoint_url: str, + model: str, + *, + allow_live_probes: bool = True, +) -> int: """Get the context window size for a model. Queries /v1/models on the endpoint and looks for context_length or context_window fields. Caches result per (endpoint, model). Falls back to DEFAULT_CONTEXT if unavailable. """ - return _get_context_length_cached(endpoint_url, model)[0] + return _get_context_length_cached( + endpoint_url, + model, + allow_live_probes=allow_live_probes, + )[0] -def get_context_length_known(endpoint_url: str, model: str) -> Tuple[int, bool]: +def get_context_length_known( + endpoint_url: str, + model: str, + *, + allow_live_probes: bool = True, +) -> Tuple[int, bool]: """Like ``get_context_length`` but also returns whether the window was actually discovered (endpoint-reported or in the known-models table) rather than the bare DEFAULT_CONTEXT fallback. Callers that *scale* a budget off the window must not trust an unknown value — a fallback 128K isn't proof the model holds 128K (review on #4122).""" - return _get_context_length_cached(endpoint_url, model) + return _get_context_length_cached( + endpoint_url, + model, + allow_live_probes=allow_live_probes, + ) def budget_context_for_model(endpoint_url: str, model: str, *, fallback: int = 0) -> int: diff --git a/tests/test_api_token_security_cycle3.py b/tests/test_api_token_security_cycle3.py new file mode 100644 index 000000000..031982592 --- /dev/null +++ b/tests/test_api_token_security_cycle3.py @@ -0,0 +1,558 @@ +"""Cycle-3 regressions for the bearer capability boundary. + +The tests intentionally call route endpoints directly as well as exercising +the shared helpers. FastAPI dependency execution is not a substitute for the +handler's own authorization checks when an endpoint can be called by another +in-process route or test harness. +""" + +import json +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from fastapi import HTTPException + + +def _request(*, bearer=True, owner="alice", scopes=("chat",), current_user="api", auth_manager=None): + return SimpleNamespace( + state=SimpleNamespace( + api_token=bearer, + api_token_owner=owner if bearer else None, + api_token_scopes=list(scopes), + current_user=current_user if bearer else current_user, + ), + app=SimpleNamespace(state=SimpleNamespace(auth_manager=auth_manager)), + headers={}, + ) + + +class _JsonRequest: + def __init__(self, body=None, *, bearer=True, owner="alice", scopes=("chat",)): + self.state = SimpleNamespace( + api_token=bearer, + api_token_owner=owner if bearer else None, + api_token_scopes=list(scopes), + current_user="api" if bearer else owner, + ) + self.app = SimpleNamespace(state=SimpleNamespace(auth_manager=None)) + self.headers = {} + self.body = body or {} + + async def json(self): + return self.body + + +def _latest_endpoint(router, path, method): + for route in reversed(router.routes): + if route.path == path and method in (route.methods or set()): + return route.endpoint + raise AssertionError(f"route not found: {method} {path}") + + +def test_admin_owned_bearer_does_not_inherit_raw_endpoint_or_model_privileges(): + from routes.chat_helpers import _allowed_models_for_request, _enforce_chat_privileges + from routes.session_routes import _reject_raw_endpoint_url_for_non_admin + + auth_manager = MagicMock() + auth_manager.is_admin.return_value = True + auth_manager.get_privileges.side_effect = AssertionError("bearer used owner privilege lookup") + request = _request(owner="admin", auth_manager=auth_manager) + + with pytest.raises(HTTPException) as exc: + _reject_raw_endpoint_url_for_non_admin( + request, + "admin", + endpoint_id="", + endpoint_url="http://127.0.0.1:9/private", + ) + assert exc.value.status_code == 403 + assert _allowed_models_for_request(request) is None + _enforce_chat_privileges(request, SimpleNamespace(model="admin-only")) + auth_manager.get_privileges.assert_not_called() + + cookie_request = _request( + bearer=False, + owner=None, + current_user="admin", + auth_manager=auth_manager, + ) + auth_manager.get_privileges.side_effect = None + auth_manager.get_privileges.return_value = { + "allowed_models": ["admin-only"], + "allowed_models_restricted": True, + } + _reject_raw_endpoint_url_for_non_admin( + cookie_request, + "admin", + endpoint_id="", + endpoint_url="http://127.0.0.1:9/private", + ) + assert _allowed_models_for_request(cookie_request) == frozenset({"admin-only"}) + + +class _SessionManager: + def __init__(self): + self.sessions = {} + self.saved = 0 + + def create_session(self, **kwargs): + session = SimpleNamespace( + id=kwargs["session_id"], + name=kwargs.get("name", ""), + endpoint_url=kwargs.get("endpoint_url", ""), + model=kwargs.get("model", ""), + rag=kwargs.get("rag", False), + owner=kwargs.get("owner"), + headers={}, + history=[], + ) + self.sessions[session.id] = session + return session + + def save_sessions(self): + self.saved += 1 + + +def test_bearer_session_lifecycle_routes_do_not_emit_webhook_or_event(monkeypatch): + import src.event_bus as event_bus + from routes import session_routes + + events = [] + monkeypatch.setattr(event_bus, "fire_event", lambda *args, **kwargs: events.append((args, kwargs))) + manager = _SessionManager() + webhook_manager = MagicMock() + router = session_routes.setup_session_routes( + manager, + { + "REQUEST_TIMEOUT": 1, + "OPENAI_API_KEY": "server-key", + "SESSIONS_FILE": "sessions.json", + }, + webhook_manager=webhook_manager, + ) + request = _request() + + create = _latest_endpoint(router, "/api/session", "POST") + with pytest.raises(HTTPException) as exc: + create( + request, + name="Private endpoint", + endpoint_url="http://127.0.0.1:9/private", + model="stored-model", + rag="false", + skip_validation="true", + api_key="", + endpoint_id="", + ) + assert exc.value.status_code == 403 + assert manager.sessions == {} + + result = create( + request, + name="API chat", + endpoint_url="", + model="stored-model", + rag="false", + skip_validation="true", + api_key="", + endpoint_id="", + ) + assert result.model == "stored-model" + + create_openai = _latest_endpoint(router, "/api/session/openai", "POST") + openai_result = create_openai(request, name="OpenAI", model="gpt-4o", rag="false") + assert openai_result["model"] == "gpt-4o" + webhook_manager.fire_and_forget.assert_not_called() + assert events == [] + + cookie_request = _request(bearer=False, owner=None, current_user="alice", scopes=()) + cookie_result = create( + cookie_request, + name="Browser chat", + endpoint_url="", + model="stored-model", + rag="false", + skip_validation="true", + api_key="", + endpoint_id="", + ) + assert cookie_result.model == "stored-model" + assert events and events[-1][0] == ("session_created", "alice") + + +def test_generic_email_dependency_rejects_bearer_before_legacy_fallback(monkeypatch): + from routes import email_helpers + + with pytest.raises(HTTPException) as exc: + email_helpers._require_auth(_request(owner="alice", scopes=("chat",))) + assert exc.value.status_code == 403 + + assert email_helpers._require_auth( + _request(bearer=False, owner=None, current_user="alice", scopes=()) + ) == "alice" + monkeypatch.setattr(email_helpers, "_auth_disabled", lambda: True) + assert email_helpers._require_auth( + _request(bearer=False, owner=None, current_user=None, scopes=()) + ) == "" + + +@pytest.mark.asyncio +async def test_bearer_context_builder_uses_no_live_model_or_context_probes(monkeypatch): + from routes import chat_helpers + from src.auth_helpers import request_capability + + calls = {"normalize": 0, "compact": []} + + class _ChatHandler: + def validate_and_extract_preset(self, _preset_id): + return 0.2, 64, None, None + + async def preprocess_message(self, message, att_ids, sess, **kwargs): + assert kwargs["allow_tool_preprocessing"] is False + return message, message, message, [], [] + + class _ChatProcessor: + def build_context_preface(self, **kwargs): + return [], [], [] + + def fail_normalize(*args, **kwargs): + calls["normalize"] += 1 + raise AssertionError("bearer context performed a live model probe") + + async def fake_compact(*args, **kwargs): + calls["compact"].append(kwargs) + return args[3], 128000, False + + monkeypatch.setattr(chat_helpers, "_normalize_model_id_from_cache", lambda _sess: None) + monkeypatch.setattr(chat_helpers, "normalize_model_id", fail_normalize) + monkeypatch.setattr(chat_helpers, "maybe_compact", fake_compact) + monkeypatch.setattr(chat_helpers, "load_prefs_for_user", lambda _owner: {}) + + session = SimpleNamespace( + endpoint_url="http://127.0.0.1:9999/v1/chat/completions", + model="uncached-model", + headers={}, + owner="alice", + history=[], + get_context_messages=lambda: [], + add_message=lambda _message: None, + ) + request = _request() + context = await chat_helpers.build_chat_context( + session, + request, + _ChatHandler(), + _ChatProcessor(), + message="hello", + session_id="session-1", + incognito=True, + allow_tool_preprocessing=False, + persist_user_message=False, + capability=request_capability(request), + ) + + assert context.context_length == 128000 + assert calls["normalize"] == 0 + assert calls["compact"] == [{"owner": "alice", "allow_live_probes": False}] + + +def test_model_and_context_no_live_probe_options_do_not_touch_endpoints(monkeypatch): + from src import llm_core, model_context + + monkeypatch.setattr(llm_core, "_configured_cached_model_ids", lambda *args, **kwargs: []) + monkeypatch.setattr( + llm_core, + "httpx_get_kimi_aware", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("model endpoint touched")), + ) + assert llm_core.list_model_ids( + "http://127.0.0.1:9999/v1", + allow_live_probes=False, + ) == [] + assert llm_core.normalize_model_id( + "http://127.0.0.1:9999/v1", + "uncached-model", + allow_live_probes=False, + ) is None + + monkeypatch.setattr(model_context, "_context_cache", {}) + monkeypatch.setattr( + model_context, + "httpx", + SimpleNamespace( + get=lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("context endpoint touched") + ) + ), + ) + assert model_context.get_context_length( + "http://127.0.0.1:9999/v1", + "uncached-model", + allow_live_probes=False, + ) == model_context.DEFAULT_CONTEXT + assert model_context.get_context_length( + "http://127.0.0.1:9999/v1", + "gpt-4o", + allow_live_probes=False, + ) == 128000 + assert model_context._context_cache == {} + + +def _forged_metadata(): + return { + "safe": {"label": "retain"}, + "_tool_approval_chat_session_granted": True, + "approval_id": "approval-1", + "approved_by_interactive_session": True, + "resolved": "approve", + "session_id": "session-1", + "tool_events": [ + { + "kind": "tool_approval", + "approval_id": "nested-1", + "resolved": "approve", + "ask_user": { + "approval_id": "nested-2", + "approved_by_interactive_session": True, + "label": "display-only", + }, + } + ], + } + + +@pytest.mark.asyncio +async def test_bearer_history_and_fork_scrub_legacy_approval_metadata(monkeypatch): + import src.event_bus as event_bus + from core.models import ChatMessage + from routes.history import history_routes + + display_source = SimpleNamespace( + id="source", + name="Source", + owner="alice", + endpoint_url="https://example.test/v1", + model="model", + history=[ + ChatMessage("user", "hello", _forged_metadata()), + {"role": "assistant", "content": "answer", "metadata": _forged_metadata()}, + ], + ) + fork_source = SimpleNamespace( + id="source", + name="Source", + owner="alice", + endpoint_url="https://example.test/v1", + model="model", + history=[ChatMessage("user", "hello", _forged_metadata())], + ) + + class _Forked: + def __init__(self): + self.history = [] + + def add_message(self, message): + self.history.append(message) + + class _Manager: + def __init__(self, source): + self.source = source + self.created = None + + def get_session(self, _session_id): + return self.source + + def create_session(self, **kwargs): + self.created = _Forked() + return self.created + + monkeypatch.setattr(history_routes, "_verify_session_owner", lambda *args, **kwargs: None) + events = [] + monkeypatch.setattr(event_bus, "fire_event", lambda *args, **kwargs: events.append(args)) + display_manager = _Manager(display_source) + display_router = history_routes.setup_history_routes(display_manager) + history_endpoint = _latest_endpoint(display_router, "/api/history/{session_id}", "GET") + request = _JsonRequest({"keep_count": 1}) + + displayed = await history_endpoint(request, "source") + assert len(displayed["history"]) == 2 + for entry in displayed["history"]: + metadata = entry.get("metadata", {}) + assert metadata.get("safe") == {"label": "retain"} + assert all( + field not in metadata + for field in ( + "_tool_approval_chat_session_granted", + "approval_id", + "approved_by_interactive_session", + "resolved", + "session_id", + ) + ) + assert metadata["tool_events"][0]["ask_user"] == {"label": "display-only"} + + fork_manager = _Manager(fork_source) + fork_router = history_routes.setup_history_routes(fork_manager) + fork_endpoint = _latest_endpoint(fork_router, "/api/session/{session_id}/fork", "POST") + forked = await fork_endpoint(request, "source") + assert forked["status"] == "ok" + assert events == [] + assert fork_manager.created.history[0].metadata["safe"] == {"label": "retain"} + assert "approval_id" not in fork_manager.created.history[0].metadata + + +class _ColumnQuery: + def __init__(self, result, *, count=0, rows=None): + self.result = result + self.count_value = count + self.rows = rows or [] + + def filter(self, *args): + return self + + def order_by(self, *args): + return self + + def offset(self, value): + return self + + def limit(self, value): + return self + + def first(self): + return self.result + + def count(self): + return self.count_value + + def all(self): + return self.rows + + +class _HistoryDb: + def __init__(self, history_routes, session_row, message_rows): + self.history_routes = history_routes + self.session_row = session_row + self.message_rows = message_rows + + def query(self, model): + if model is self.history_routes.DbSession: + return _ColumnQuery(self.session_row) + return _ColumnQuery(None, count=len(self.message_rows), rows=self.message_rows) + + def close(self): + return None + + +@pytest.mark.asyncio +async def test_bearer_paginated_history_scrubs_db_projection(monkeypatch): + from routes.history import history_routes + db_session = SimpleNamespace(model="model", endpoint_url="https://example.test/v1", name="Chat") + db_message = SimpleNamespace( + role="assistant", + content="answer", + meta_data=json.dumps(_forged_metadata()), + timestamp=datetime(2026, 1, 1, 0, 0, 0), + ) + db = _HistoryDb(history_routes, db_session, [db_message]) + monkeypatch.setattr(history_routes, "SessionLocal", lambda: db) + monkeypatch.setattr(history_routes, "_verify_session_owner", lambda *args, **kwargs: None) + router = history_routes.setup_history_routes(SimpleNamespace()) + endpoint = _latest_endpoint(router, "/api/history/{session_id}", "GET") + + payload = await endpoint(_JsonRequest(), "source", limit=10, offset=0) + metadata = payload["history"][0]["metadata"] + assert metadata["safe"] == {"label": "retain"} + assert "approval_id" not in metadata + assert "approved_by_interactive_session" not in metadata + assert metadata["tool_events"][0]["ask_user"] == {"label": "display-only"} + + +@pytest.mark.asyncio +async def test_sync_bearer_chat_keeps_response_but_suppresses_completion_webhook(monkeypatch): + from routes.webhook import webhook_routes + from src import llm_core + + class _Session: + def __init__(self, kwargs): + self.endpoint_url = kwargs["endpoint_url"] + self.model = kwargs["model"] + self.owner = kwargs["owner"] + self.headers = {} + self.history = [] + + def add_message(self, message): + self.history.append(message) + + class _Manager: + def __init__(self): + self.created = [] + + def create_session(self, **kwargs): + session = _Session(kwargs) + self.created.append(session) + return session + + def save_sessions(self): + return None + + class _Webhooks: + def __init__(self): + self.events = [] + + def fire_and_forget(self, event, payload): + self.events.append((event, payload)) + + async def fake_llm(*args, **kwargs): + return "answer" + + monkeypatch.setattr(webhook_routes, "validate_public_http_url", lambda value: value) + monkeypatch.setattr(llm_core, "llm_call_async", fake_llm) + webhook_manager = _Webhooks() + router = webhook_routes.setup_webhook_routes( + webhook_manager, + auth_manager=None, + session_manager=_Manager(), + ) + endpoint = _latest_endpoint(router, "/api/v1/chat", "POST") + body = SimpleNamespace( + message="hello", + model="gpt-4o", + session=None, + api_key="test-key", + base_url="https://api.example.com/v1", + provider=None, + ) + + result = await endpoint(_request(), body) + assert result["response"] == "answer" + assert webhook_manager.events == [] + + +@pytest.mark.asyncio +async def test_upload_vision_handlers_reject_bearer_before_ai_or_cache_work(): + from routes import upload_routes + + upload_handler = MagicMock() + router, _cleanup = upload_routes.setup_upload_routes(upload_handler) + get_vision = _latest_endpoint(router, "/api/upload/{file_id}/vision", "GET") + put_vision = _latest_endpoint(router, "/api/upload/{file_id}/vision", "PUT") + request = _request() + + for endpoint, kwargs in ((get_vision, {"force": 0}), (put_vision, {})): + with pytest.raises(HTTPException) as exc: + result = endpoint(request, "upload-1", **kwargs) + if hasattr(result, "__await__"): + await result + assert exc.value.status_code == 403 + upload_handler.validate_upload_id.assert_not_called() + + cookie_request = _request(bearer=False, owner=None, current_user="alice", scopes=()) + with pytest.raises(HTTPException) as exc: + result = get_vision(cookie_request, "upload-1", force=0) + if hasattr(result, "__await__"): + await result + assert exc.value.status_code == 404 + upload_handler.validate_upload_id.assert_called_once_with("upload-1") diff --git a/tests/test_chat_cached_model_normalization.py b/tests/test_chat_cached_model_normalization.py index b601f8779..84d23ca10 100644 --- a/tests/test_chat_cached_model_normalization.py +++ b/tests/test_chat_cached_model_normalization.py @@ -9,7 +9,9 @@ def test_chat_context_uses_cached_models_before_live_model_probe(): assert "def _normalize_model_id_from_cache" in source assert "cached_models" in source - assert "norm = _normalize_model_id_from_cache(sess) or normalize_model_id" in source + assert "norm = _normalize_model_id_from_cache(sess)" in source + assert "capability.allow_live_probes" in source + assert "normalize_model_id(" in source def test_cached_model_match_keeps_basename_normalization(): diff --git a/tests/test_chat_helpers.py b/tests/test_chat_helpers.py index 16cb38c01..02ac977e0 100644 --- a/tests/test_chat_helpers.py +++ b/tests/test_chat_helpers.py @@ -500,8 +500,9 @@ async def _build_context_owner_probe(monkeypatch, request_state): captured["preface_owner"] = kwargs["owner"] return [], [], [] - async def fake_maybe_compact(sess, endpoint_url, model, messages, headers, owner=None): + async def fake_maybe_compact(sess, endpoint_url, model, messages, headers, owner=None, **kwargs): captured["compact_owner"] = owner + captured["allow_live_probes"] = kwargs.get("allow_live_probes", True) return messages, 8192, False monkeypatch.setattr(chat_helpers, "preprocess", fake_preprocess) @@ -557,10 +558,12 @@ async def test_build_chat_context_uses_api_token_owner_for_compaction_scope(monk ) assert ctx.user == "alice" + assert captured["allow_live_probes"] is False assert captured == { "prefs_owner": "alice", "preface_owner": "alice", "compact_owner": "alice", + "allow_live_probes": False, } @@ -579,4 +582,5 @@ async def test_build_chat_context_keeps_cookie_user_owner_scope(monkeypatch): "prefs_owner": "bob", "preface_owner": "bob", "compact_owner": "bob", + "allow_live_probes": True, }