7 Commits

Author SHA1 Message Date
Afonso Coutinho 88191d17fb fix: auto-spam move/delete targets the wrong message (seqnum vs UID) (#1874) 2026-07-02 10:40:19 +01:00
Ashvin dff91efb10 fix(agent): skip deny-listed sensitive files in glob (#5094) 2026-07-02 10:28:33 +01:00
Ashvin b26ebbda95 fix(security): match the sensitive-file deny-list case-insensitively (#5097) 2026-07-02 10:11:51 +01:00
lekt8 260f432332 fix(session): use utcnow_naive across session routes (#1116) (#5003)
Replace remaining datetime.utcnow() call sites in session CRUD, incognito
purge cutoff, and webhook payloads with core.database.utcnow_naive.
2026-07-02 11:04:22 +02:00
Mazen Tamer Salah e157f1e63d fix(cookbook): stop Ollama runner from executing the install one-liner (#3926)
The generated bash runner printed the missing-ollama hint with the install
one-liner wrapped in backticks inside a double-quoted echo. Backticks in
double quotes are command substitution, so on any serve target without
ollama the script downloaded and ran the system-wide installer (including
remote SSH hosts) instead of printing the hint. _validate_serve_cmd rejects
backticks in user-supplied commands for exactly this reason; the app's own
generated script never goes through that validator.

Move the hint into OLLAMA_MISSING_HINT in cookbook_helpers (no substitution
tokens) and emit it single-quoted via _bash_squote. Tests assert the hint
has no expansion tokens, that no generated echo line carries backticks
inside double quotes, and that bash prints the line literally.

Fixes #3816
2026-07-02 10:01:57 +01:00
pewdiepie-archdaemon dc3530b8fa Show fallback model in picker 2026-07-01 13:53:51 +00:00
pewdiepie-archdaemon 2918739489 Fix merged test regressions 2026-07-01 11:12:55 +00:00
23 changed files with 288 additions and 34 deletions
+5 -3
View File
@@ -1359,9 +1359,11 @@ def setup_chat_routes(
elif chunk.startswith("event: "):
yield chunk
elif chunk == "data: [DONE]\n\n":
if full_response:
_has_tool_events = bool((last_metrics or {}).get("tool_events"))
if full_response or _has_tool_events:
_response_to_save = full_response or "Done."
_saved_id = save_assistant_response(
sess, session_manager, session, full_response, last_metrics,
sess, session_manager, session, _response_to_save, last_metrics,
character_name=ctx.preset.character_name,
web_sources=web_sources,
rag_sources=ctx.rag_sources,
@@ -1371,7 +1373,7 @@ def setup_chat_routes(
if _saved_id:
yield f'data: {json.dumps({"type": "message_saved", "id": _saved_id})}\n\n'
run_post_response_tasks(
sess, session_manager, session, message, full_response,
sess, session_manager, session, message, _response_to_save,
last_metrics, ctx.uprefs, memory_manager, memory_vector, webhook_manager,
incognito=incognito, compare_mode=compare_mode,
character_name=ctx.preset.character_name,
+12
View File
@@ -558,6 +558,18 @@ def _bash_squote(v: str) -> str:
return v.replace("'", "'\\''")
# Shown by generated runner scripts when the ollama binary is missing on the
# target host. Must stay free of backticks/$( ) and be emitted single-quoted:
# an earlier version wrapped the install one-liner in backticks inside a
# double-quoted echo, which bash executed as command substitution and ran the
# system-wide installer (including on remote SSH hosts) instead of printing
# the hint.
OLLAMA_MISSING_HINT = (
"ERROR: Ollama not found on this server. Install it from "
"https://ollama.com/download or run: curl -fsSL https://ollama.com/install.sh | sh"
)
# Allow-list of binaries permitted as the leading token of `req.cmd` for /api/model/serve.
# Anything else is rejected before the cmd is interpolated into a tmux/PowerShell wrapper.
_SERVE_CMD_ALLOWLIST = {
+5 -2
View File
@@ -49,7 +49,7 @@ logger = logging.getLogger(__name__)
from routes.cookbook_helpers import (
_SESSION_ID_RE, _validate_repo_id, _validate_serve_model_id, _validate_include, _validate_token,
_validate_local_dir, _validate_gpus, _shell_path,
_ps_squote, _bash_squote, _validate_serve_cmd, _parse_serve_phase,
_ps_squote, _bash_squote, _validate_serve_cmd, _parse_serve_phase, OLLAMA_MISSING_HINT,
_safe_env_prefix, _local_tooling_path_export, _append_serve_preflight_exit_lines,
_append_serve_exit_code_lines, _append_llama_cpp_linux_accel_build_lines, _cached_model_scan_script,
load_stored_hf_token,
@@ -1861,7 +1861,10 @@ def setup_cookbook_routes() -> APIRouter:
runner_lines.append(' exec 3<&-; exec 3>&-')
runner_lines.append('done')
runner_lines.append('if ! command -v ollama &>/dev/null; then')
runner_lines.append(' echo "ERROR: Ollama not found on this server. Install it from https://ollama.com/download or `curl -fsSL https://ollama.com/install.sh | sh`."')
# Single-quoted on purpose: backticks inside a double-quoted
# echo are command substitution, and this line used to run the
# curl|sh installer on the target host instead of printing it.
runner_lines.append(f" echo '{_bash_squote(OLLAMA_MISSING_HINT)}'")
runner_lines.append(' echo')
runner_lines.append(' echo "=== Process exited with code 127 ==="')
runner_lines.append(' exec bash -i')
+7 -2
View File
@@ -1273,10 +1273,15 @@ def _imap_move(uid, dest, src="INBOX", account_id: str | None = None, owner: str
try:
c = _imap_connect(account_id, owner=owner)
c.select(_q(src))
status, _ = c.copy(uid, _q(dest))
# Callers pass a real IMAP UID (from conn.uid("SEARCH", ...)). copy()
# and store() operate on message SEQUENCE NUMBERS, so addressing them
# with a UID moved/deleted the wrong message (or silently no-oped when
# the UID exceeded the message count). Use the UID commands, matching
# the move/delete path in email_routes.py.
status, _ = c.uid("COPY", uid, _q(dest))
if status != "OK":
return False
c.store(uid, "+FLAGS", "\\Deleted")
c.uid("STORE", uid, "+FLAGS", "\\Deleted")
c.expunge()
return True
except Exception as e:
+2
View File
@@ -2388,6 +2388,8 @@ def setup_email_routes():
owner: str = Depends(require_owner),
):
"""Read email body. Cached for 30m, sync IMAP work runs in a thread."""
mark_seen = True if mark_seen is True or str(mark_seen).lower() == "true" else False
full = True if full is True or str(full).lower() == "true" else False
fixture_result = _fixture_email_read(uid, folder, owner)
if fixture_result is not None:
return fixture_result
+2
View File
@@ -218,6 +218,8 @@ def _default_endpoint_needs_assignment(
return True
if current_default_id not in enabled_endpoint_ids:
return True
if current_default_endpoint is None:
return False
if not (current_default_model or "").strip():
return True
visible = _endpoint_visible_model_ids(current_default_endpoint)
+10 -10
View File
@@ -162,7 +162,7 @@ def _persist_session_headers(session_id: str, headers: dict | None) -> None:
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
if db_session:
db_session.headers = headers or {}
db_session.updated_at = datetime.utcnow()
db_session.updated_at = utcnow_naive()
db.commit()
except Exception:
db.rollback()
@@ -223,8 +223,8 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
# purge exists only to catch ghosts the frontend missed (tab close,
# crash). Only clean up rows old enough to be definitely orphaned.
try:
from datetime import datetime as _dt, timedelta as _td
_cutoff = _dt.utcnow() - _td(minutes=10)
from datetime import timedelta as _td
_cutoff = utcnow_naive() - _td(minutes=10)
_purge_db = SessionLocal()
try:
from core.database import ChatMessage as _DbMsg
@@ -470,7 +470,7 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
db_session = db.query(DbSession).filter(DbSession.id == sid).first()
if db_session:
db_session.folder = folder if folder else None
db_session.updated_at = datetime.utcnow()
db_session.updated_at = utcnow_naive()
db.commit()
result["folder"] = folder if folder else None
finally:
@@ -517,7 +517,7 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
db_session.model = model
db_session.endpoint_url = endpoint_url
db_session.headers = session.headers or {}
db_session.updated_at = datetime.utcnow()
db_session.updated_at = utcnow_naive()
db.commit()
finally:
db.close()
@@ -646,7 +646,7 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
db_session = db.query(DbSession).filter(DbSession.id == sid).first()
if db_session:
db_session.archived = True
db_session.updated_at = datetime.utcnow()
db_session.updated_at = utcnow_naive()
db.commit()
# Update in memory if it exists
@@ -680,7 +680,7 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
if not db_session:
raise HTTPException(404, f"Session {sid} not found")
db_session.archived = False
db_session.updated_at = datetime.utcnow()
db_session.updated_at = utcnow_naive()
db.commit()
# Reload into session manager so it appears in the active list
try:
@@ -890,7 +890,7 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
if db_session:
db_session.is_important = important
db_session.updated_at = datetime.utcnow()
db_session.updated_at = utcnow_naive()
db.commit()
# Update in memory if it exists
@@ -979,7 +979,7 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
metadata={
"compacted": True,
"summarized_count": len(older),
"timestamp": datetime.utcnow().isoformat(),
"timestamp": utcnow_naive().isoformat(),
},
)
new_history = [summary_msg] + recent
@@ -1256,7 +1256,7 @@ def setup_session_routes(session_manager: SessionManager, config: dict, webhook_
db_session = db_session_q.first()
if db_session:
db_session.folder = folder_name
db_session.updated_at = datetime.utcnow()
db_session.updated_at = utcnow_naive()
updated += 1
db.commit()
except Exception as e:
+2
View File
@@ -137,6 +137,8 @@ def setup_upload_routes(upload_handler):
session_id: Optional[str] = Form(None),
):
"""Upload files with enhanced security and organization."""
if not isinstance(session_id, str):
session_id = None
if not files:
raise HTTPException(400, "No files uploaded")
+22 -3
View File
@@ -281,7 +281,13 @@ class LsTool:
class GlobTool:
async def execute(self, content: str, ctx: dict) -> dict:
from src.tool_execution import _resolve_tool_path, _resolve_search_root, _truncate
from src.tool_execution import (
_SENSITIVE_BASENAMES,
_is_sensitive_path,
_resolve_tool_path,
_resolve_search_root,
_truncate,
)
args = {}
_s = (content or "").strip()
if _s.startswith("{"):
@@ -322,7 +328,11 @@ class GlobTool:
) == nbase
except ValueError:
inside = False
if inside and os.path.exists(cand):
# A literal that names a deny-listed sensitive file (.env,
# .ssh/id_rsa, …) falls through to the walk, which skips it —
# otherwise glob would surface secret paths that read_file /
# grep already refuse to touch.
if inside and os.path.exists(cand) and not _is_sensitive_path(cand):
return [cand], None
# Literal not at exact path — fall through to walk so
# e.g. "foo.py" still matches at any depth (like rglob).
@@ -334,11 +344,20 @@ class GlobTool:
for dp, dns, fns in os.walk(base):
# Prune skipped dirs before descending (unlike rglob which
# descends first then filters — fatal on large node_modules).
dns[:] = [d for d in dns if d not in _CODENAV_SKIP_DIRS]
# Sensitive dirs (.ssh, .gnupg, …) are pruned too so glob
# never enumerates the keys/tokens inside them.
dns[:] = [
d for d in dns
if d not in _CODENAV_SKIP_DIRS and d not in _SENSITIVE_BASENAMES
]
for name in fns + dns:
full = os.path.join(dp, name)
rel = os.path.relpath(full, base).replace(os.sep, "/")
if regex.fullmatch(rel) or regex.fullmatch(name):
# Skip deny-listed sensitive files (.env, id_rsa,
# known_hosts, …) the same way grep does.
if _is_sensitive_path(os.path.realpath(full)):
continue
try:
mtime = os.stat(full).st_mtime
except OSError:
+18 -8
View File
@@ -71,25 +71,35 @@ _SENSITIVE_FILE_PATTERNS: tuple[str, ...] = (
"known_hosts",
)
# Case-folded views used for matching. On a case-insensitive filesystem
# (Windows, default macOS) ".SSH/AUTHORIZED_KEYS" and ".env" resolve to the
# same protected files as their lowercase forms, so the deny-list has to fold
# case before comparing — the sibling resolver already normcases paths for the
# same reason. casefold (not os.path.normcase) because normcase is a no-op on
# POSIX, which is exactly where the macOS read-exfil path lives.
_SENSITIVE_BASENAMES_CF: frozenset[str] = frozenset(b.casefold() for b in _SENSITIVE_BASENAMES)
_SENSITIVE_FILE_PATTERNS_CF: frozenset[str] = frozenset(p.casefold() for p in _SENSITIVE_FILE_PATTERNS)
def _is_sensitive_path(resolved: str) -> bool:
"""Return True if *resolved* falls under a sensitive directory or
matches a sensitive filename regardless of what root it sits under.
Matching is case-insensitive: on Windows / default macOS a case-variant
name (``.SSH``, ``AUTHORIZED_KEYS``, ``Id_Rsa``) points at the same file as
the lowercase form, so a case-sensitive check would let it slip past the
deny-list in every file tool that relies on it.
"""
parts = resolved.split(os.sep)
filenames: set[str] = {parts[-1]} if parts else set()
parts = [p.casefold() for p in resolved.split(os.sep)]
filename = parts[-1] if parts else ""
# Check if any path component is a sensitive directory.
for part in parts:
if part in _SENSITIVE_BASENAMES:
if part in _SENSITIVE_BASENAMES_CF:
return True
# Check filename against known sensitive files.
for pat in _SENSITIVE_FILE_PATTERNS:
if pat in filenames:
return True
return False
return filename in _SENSITIVE_FILE_PATTERNS_CF
def _tool_path_roots() -> list[str]:
+7
View File
@@ -254,6 +254,7 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
"calendar_href": ev.calendar_id,
"event_type": ev.event_type or "",
"importance": ev.importance or "normal",
"rrule": ev.rrule or "",
})
if not events:
response_text = f"No events between {start_dt.date().isoformat()} and {end_dt.date().isoformat()}."
@@ -268,6 +269,8 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
line += f" #{ev['event_type']}"
if ev.get("importance") and ev["importance"] != "normal":
line += f" !{ev['importance']}"
if ev.get("rrule"):
line += f" repeats({ev['rrule']})"
if ev.get("location"):
line += f" @ {ev['location']}"
if ev.get("calendar"):
@@ -480,6 +483,10 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
ev.event_type = _tag or None
if args.get("importance") is not None:
ev.importance = args["importance"]
if args.get("rrule") is not None:
ev.rrule = args.get("rrule") or ""
elif str(args.get("repeat") or "").strip().lower() in {"none", "no", "off", "false", "single"}:
ev.rrule = ""
is_caldav = ev.calendar and ev.calendar.source == "caldav"
if is_caldav:
ev.caldav_sync_pending = "update"
+1
View File
@@ -4051,6 +4051,7 @@ function startOdysseusApp() {
// Non-critical: load in parallel, resolve silently
modelsModule.refreshModels(false).then(() => {
try { sessionModule.updateModelPicker(); } catch (_) {}
const modelsBox = document.getElementById('models');
const hasModels = modelsBox && modelsBox.querySelector('.models-row');
if (!hasModels) {
+12
View File
@@ -754,6 +754,18 @@ export function updateModelPicker() {
// silently pre-populate the chatbox of the next user that signed in. If
// we have no session model and no pending-chat pick, fall through to
// the "Select model" placeholder below.
//
// But if the server model cache already has an online endpoint, make the
// same safe fallback visible in the picker immediately. The send path can
// already resolve a usable model; the UI should not sit on "Select model"
// and make it look broken.
if (!modelId && !currentSessionId && window.modelsModule && window.modelsModule.getCachedItems) {
const fallback = _firstAvailableModel();
if (fallback) {
_deps.setPendingChat(fallback);
modelId = fallback.modelId;
}
}
// Check if selected model is still available — fall back ONLY for pending chats with no user selection
// Never override an existing session's model — the user explicitly chose it
+1 -1
View File
@@ -1017,7 +1017,7 @@ function _showTaskDropdown(anchor, items) {
const dd = document.createElement('div');
dd.className = 'task-dropdown';
dd._anchor = anchor;
dd.style.cssText = 'position:fixed;z-index:100000;background:var(--panel);border:1px solid var(--border);border-radius:6px;box-shadow:0 4px 12px rgba(0,0,0,0.3);padding:4px;min-width:120px;';
dd.style.cssText = `position:fixed;z-index:${topPortalZ()};background:var(--panel);border:1px solid var(--border);border-radius:6px;box-shadow:0 4px 12px rgba(0,0,0,0.3);padding:4px;min-width:120px;`;
items.forEach(item => {
const btn = document.createElement('button');
btn.style.cssText = 'display:flex;align-items:center;gap:8px;width:100%;text-align:left;padding:6px 10px;border:none;background:none;color:var(--fg);font-size:11px;font-family:inherit;cursor:pointer;border-radius:4px;transition:background 0.1s;';
+1 -1
View File
@@ -86,7 +86,7 @@ def test_expand_non_recurring_returns_single():
def test_expand_rrule_skips_deleted_occurrence_exdate():
cal = _import_calendar_helpers()
cal = import_calendar_routes()
ev = _make_event(
dtstart=datetime(2026, 7, 1, 14, 0),
dtend=datetime(2026, 7, 1, 15, 0),
+1 -1
View File
@@ -102,7 +102,7 @@ def test_local_windows_platform_comes_from_backend_host_state():
assert "platform: _envState.hostPlatform || _envState.platform || ''" not in text
assert 'return "windows" if IS_WINDOWS else ""' in routes
assert 'env["hostPlatform"] = _client_host_platform()' in routes
assert "return _state_for_client({})" in routes
assert "client_state = _state_for_client({})" in routes
assert 'env.pop("hostPlatform", None)' in routes
assert "delete env.hostPlatform;" in running
@@ -35,8 +35,8 @@ def test_windows_session_commands_use_shared_powershell_wrapper_and_local_log_di
assert "host ? '$env:TEMP\\\\odysseus-sessions' : '$env:TEMP\\\\odysseus-tmux'" in source
assert "function _winPowerShellCmd(task, ps)" in source
assert "const command = `powershell -Command \"${ps}\"`;" in source
assert "if (!task.remoteHost) return command;" in source
assert "return `ssh ${_sshPrefix(_getPort(task))}${task.remoteHost} ${_shQuote(command)}`;" in source
assert "if (!host) return command;" in source
assert "return `ssh ${_sshPrefix(_getPort(task))}${host} ${_shQuote(command)}`;" in source
def test_dep_install_success_recognized_from_exit_sentinel():
+56
View File
@@ -0,0 +1,56 @@
"""_imap_move must address messages by UID, not sequence number.
The auto-spam poller passes a real IMAP UID (from conn.uid("SEARCH", ...))
to _imap_move, but the function used conn.copy()/conn.store(), which operate
on message SEQUENCE NUMBERS. So a UID like 90521 was interpreted as sequence
number 90521 moving/deleting the wrong message or silently no-oping. It
must use the UID commands.
"""
import sys
import types
import pytest
@pytest.fixture
def email_helpers(monkeypatch, tmp_path):
# Keep _init_scheduled_db (run at import) off the real data dir.
monkeypatch.setenv("ODYSSEUS_DATA_DIR", str(tmp_path))
import routes.email_helpers as eh
return eh
class _FakeIMAP:
def __init__(self):
self.calls = []
def select(self, mbox):
self.calls.append(("select", mbox)); return ("OK", [b""])
def copy(self, *a):
self.calls.append(("copy",) + a); return ("OK", [b""])
def store(self, *a):
self.calls.append(("store",) + a); return ("OK", [b""])
def uid(self, *a):
self.calls.append(("uid",) + a); return ("OK", [b""])
def expunge(self):
self.calls.append(("expunge",)); return ("OK", [b""])
def logout(self):
pass
def test_move_uses_uid_commands_not_seqnum(email_helpers, monkeypatch):
fake = _FakeIMAP()
monkeypatch.setattr(email_helpers, "_imap_connect", lambda *a, **k: fake)
ok = email_helpers._imap_move(b"90521", "Spam", src="INBOX")
assert ok is True
verbs = [c[0] for c in fake.calls]
uid_ops = [c[1] for c in fake.calls if c[0] == "uid"]
assert "COPY" in uid_ops and "STORE" in uid_ops
# the sequence-number commands must NOT be used to address a UID
assert "copy" not in verbs
assert "store" not in verbs
+57
View File
@@ -0,0 +1,57 @@
"""The generated Ollama runner must print the install hint, not execute it.
The runner script emitted by /api/model/serve contained:
echo "ERROR: Ollama not found ... or `curl -fsSL .../install.sh | sh`."
Backticks inside double quotes are bash command substitution, so on any host
without ollama the script downloaded and ran the system-wide installer
(including remote SSH serve targets) instead of printing the hint. The hint
now lives in OLLAMA_MISSING_HINT, contains no substitution tokens, and is
emitted single-quoted.
"""
import os
import shutil
import subprocess
import pytest
from routes.cookbook_helpers import OLLAMA_MISSING_HINT, _bash_squote
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def test_hint_has_no_shell_expansion_tokens():
assert "`" not in OLLAMA_MISSING_HINT
assert "$(" not in OLLAMA_MISSING_HINT
def test_hint_still_tells_the_user_how_to_install():
assert "https://ollama.com/download" in OLLAMA_MISSING_HINT
assert "install.sh" in OLLAMA_MISSING_HINT
def test_no_runner_echo_line_uses_backticks_in_double_quotes():
# Source-level guard: generated-script echo lines must never carry
# backticks inside a double-quoted bash string again.
src = open(os.path.join(ROOT, "routes", "cookbook_routes.py"), encoding="utf-8").read()
offenders = [
line.strip()
for line in src.splitlines()
if "append(" in line and 'echo "' in line and "`" in line.split('echo "', 1)[1]
]
assert offenders == []
def test_single_quoted_echo_prints_hint_literally():
bash = shutil.which("bash")
if not bash:
pytest.skip("bash not available")
out = subprocess.run(
[bash, "-c", f"echo '{_bash_squote(OLLAMA_MISSING_HINT)}'"],
capture_output=True,
text=True,
timeout=30,
)
assert out.returncode == 0
assert out.stdout.strip() == OLLAMA_MISSING_HINT
+11
View File
@@ -0,0 +1,11 @@
"""Regression: session routes must not call datetime.utcnow() (#1116)."""
import inspect
import routes.session_routes as sr
def test_session_routes_module_does_not_reference_utcnow():
source = inspect.getsource(sr)
assert "datetime.utcnow()" not in source
assert "_dt.utcnow()" not in source
+21
View File
@@ -57,6 +57,27 @@ def test_non_sensitive_path():
assert not _is_sensitive_path("/home/user/projects/file.py")
def test_sensitive_case_insensitive():
"""On case-insensitive filesystems (Windows, default macOS) a case-variant
name resolves to the same protected file, so the deny-list must match
regardless of case. Built with os.path.join so the separator is right on
both POSIX and Windows.
"""
from src.tool_execution import _is_sensitive_path
# sensitive directory, varied case
assert _is_sensitive_path(os.path.join("home", "u", ".SSH", "authorized_keys"))
assert _is_sensitive_path(os.path.join("home", "u", ".Gnupg", "pubring.kbx"))
# sensitive filename, varied case
assert _is_sensitive_path(os.path.join("ws", "AUTHORIZED_KEYS"))
assert _is_sensitive_path(os.path.join("ws", "Id_Rsa"))
assert _is_sensitive_path(os.path.join("ws", ".ENV"))
assert _is_sensitive_path(os.path.join("ws", ".Env"))
# both dir and file varied
assert _is_sensitive_path(os.path.join("home", "u", ".SSH", "AUTHORIZED_KEYS"))
# an ordinary file with none of the sensitive names is still allowed
assert not _is_sensitive_path(os.path.join("ws", "Readme.md"))
# ── Unit tests on _resolve_tool_path ─────────────────────────────────
def test_blocks_etc_shadow():
+1 -1
View File
@@ -17,7 +17,7 @@ SRC = Path(__file__).resolve().parent.parent / "static/js/fileHandler.js"
def _upload_pending_body() -> str:
text = SRC.read_text(encoding="utf-8")
start = text.index("export async function uploadPending()")
start = text.index("export async function uploadPending(")
rest = text[start:]
m = re.search(r"\n(export |function )", rest[1:])
return rest[: m.start() + 1] if m else rest
+32
View File
@@ -167,6 +167,38 @@ async def test_glob_confined_e2e(ws, admin):
assert r["exit_code"] == 0 and "No files" in r["output"]
@pytest.mark.asyncio
async def test_glob_skips_sensitive_files_in_workspace(ws, admin):
"""glob must not enumerate deny-listed sensitive files that live inside the
workspace. read_file/write_file/edit_file refuse them and grep skips them,
so glob surfacing their paths is an enumeration oracle for prompt-injection.
"""
with open(os.path.join(ws, "keep.py"), "w") as f:
f.write("x")
with open(os.path.join(ws, ".env"), "w") as f:
f.write("AWS_SECRET=xxx")
with open(os.path.join(ws, "id_rsa"), "w") as f: # non-dotfile key at root
f.write("KEY")
os.makedirs(os.path.join(ws, ".ssh"), exist_ok=True)
with open(os.path.join(ws, ".ssh", "authorized_keys"), "w") as f:
f.write("ssh-rsa AAAA")
# A recursive wildcard returns ordinary files but none of the sensitive
# ones. The pattern "**/*" contains no secret names, so a secret basename
# appearing in the output is a real leak (not the echoed not-found pattern).
_, r = await execute_tool_block(_block("glob", json.dumps({"pattern": "**/*"})), owner="a", workspace=ws)
assert r["exit_code"] == 0
assert "keep.py" in r["output"]
for leak in (".env", "id_rsa", "authorized_keys"):
assert leak not in r["output"], f"glob leaked sensitive file: {leak}"
# Directly targeting a sensitive file (literal fast-path and wildcard) must
# come back as the not-found message, never a match with the file's path.
for pat in (".env", "**/id_rsa", "**/authorized_keys"):
_, r = await execute_tool_block(_block("glob", json.dumps({"pattern": pat})), owner="a", workspace=ws)
assert r["exit_code"] == 0 and "No files" in r["output"]
@pytest.mark.asyncio
async def test_subprocess_cwd_is_workspace_e2e(ws, admin):
"""python tool runs with cwd = workspace (OS-agnostic probe)."""