diff --git a/api/routes.py b/api/routes.py index e1c3118ab..395ca5101 100644 --- a/api/routes.py +++ b/api/routes.py @@ -7535,7 +7535,8 @@ def _load_branch_source_or_refuse(handler, sid: str): return get_session(sid) except KeyError: _foreign_session, _reason = _claim_or_synthesize_cli_session(sid) - if _reason == "not_claimable" and _foreign_session is not None and is_cron_session(getattr(_foreign_session, "session_id", ""), str((getattr(_foreign_session, "source_tag", None) or getattr(_foreign_session, "raw_source", None) or getattr(_foreign_session, "source", None) or "")).strip().lower()): + _source_kind = str((getattr(_foreign_session, "source_tag", None) or getattr(_foreign_session, "raw_source", None) or getattr(_foreign_session, "source", None) or "")).strip().lower() if _foreign_session is not None else "" + if _reason == "not_claimable" and _foreign_session is not None and _source_kind == "cron": _foreign_session._branch_source_readonly = True; return _foreign_session if _reason == "not_claimable": bad(handler, "Read-only sessions cannot be branched from WebUI", 403); return None bad(handler, "Session not found", 404) diff --git a/static/sessions.js b/static/sessions.js index 53d6228a5..eeccd0b98 100644 --- a/static/sessions.js +++ b/static/sessions.js @@ -2012,11 +2012,8 @@ function _isBranchableReadOnlySession(session) { session && session.source_tag, session && session.raw_source, session && session.source, - session && session.session_source, - session && session.source_label, ].map(v => String(v || '').trim().toLowerCase()); - const sid = String(session && session.session_id || '').toLowerCase(); - return sources.includes('cron') || sid.startsWith('cron_'); + return sources.includes('cron'); } function _sourceKeyForSession(session) { diff --git a/tests/test_465_session_branching.py b/tests/test_465_session_branching.py index a9d8a6461..1146405e2 100644 --- a/tests/test_465_session_branching.py +++ b/tests/test_465_session_branching.py @@ -218,8 +218,10 @@ def test_branch_endpoint_consults_foreign_session_guard_on_missing_sidecar(): "Helper should classify missing-sidecar foreign sessions before returning" assert 'if _reason == "not_claimable":' in helper, \ "Helper should branch on not_claimable foreign ownership" - assert 'is_cron_session(' in helper, \ - "Helper should narrow read-only branch sources to canonical cron sessions" + assert '_source_kind == "cron"' in helper, \ + "Helper should narrow read-only branch sources to resolved cron source metadata" + assert 'is_cron_session(' not in helper, \ + "Helper should not use the cron_ session-id prefix as a branch permission gate" assert '_foreign_session._branch_source_readonly = True' in helper, \ "Helper should mark synthesized read-only sources so branch does not save them" assert 'return _foreign_session' in helper, \ @@ -537,7 +539,7 @@ def test_branchable_read_only_helper_accepts_cron_sources(): assert "session && session.source_tag" in src assert "session && session.raw_source" in src assert "sources.includes('cron')" in src - assert "sid.startsWith('cron_')" in src + assert "sid.startsWith('cron_')" not in src assert "sid.startsWith('cron-')" not in src @@ -595,7 +597,7 @@ def test_forkFromMessage_rejects_read_only_non_cron_sessions_without_loading_or_ def test_forkFromMessage_allows_read_only_cron_sessions_to_post(): """Read-only cron message forks should reach the existing keep_count path.""" result = _commands_harness( - "S.session = { session_id: 'cron_1', read_only: true };\n" + "S.session = { session_id: 'cron_1', raw_source: 'cron', read_only: true };\n" "_oldestIdx = 2;\n" "await forkFromMessage(4);\n" "console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));" @@ -611,15 +613,15 @@ def test_forkFromMessage_allows_read_only_cron_sessions_to_post(): assert payload["renderCalls"] == 1, "cron forkFromMessage should refresh the session list" -def test_cmdBranch_rejects_non_canonical_cron_dash_id_without_posting(): - """Only canonical cron source fields or cron_ ids should unlock read-only branching.""" +def test_cmdBranch_rejects_cron_prefixed_id_without_canonical_source(): + """Only canonical cron source fields should unlock read-only branching.""" result = _commands_harness( - "S.session = { session_id: 'cron-1', session_source: 'other', read_only: true };\n" + "S.session = { session_id: 'cron_spoof_messaging', session_source: 'other', read_only: true };\n" "await cmdBranch('');\n" "console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));" ) payload = json.loads(result) - assert payload["calls"] == [], "cron- id alone should not unlock read-only /branch" + assert payload["calls"] == [], "cron_ id alone should not unlock read-only /branch" assert payload["toasts"][0][0] == "Read-only sessions cannot be forked." diff --git a/tests/test_chat_start_claim_cli_session.py b/tests/test_chat_start_claim_cli_session.py index 9b5733844..81fd6fb33 100644 --- a/tests/test_chat_start_claim_cli_session.py +++ b/tests/test_chat_start_claim_cli_session.py @@ -1101,6 +1101,34 @@ def test_branch_still_refuses_non_cron_not_claimable_sources( assert not (isolated_state_db["sessions_dir"] / f"{SID}.json").exists() +def test_branch_refuses_cron_prefixed_non_cron_not_claimable_source( + routes_module, monkeypatch, isolated_state_db +): + SID = "cron_spoof_messaging" + _make_state_db( + isolated_state_db["db"], SID, message_count=1, + title="Messaging chat", source="messaging", cwd="/root", + ) + monkeypatch.setattr( + routes_module, + "_lookup_cli_session_metadata", + lambda _sid: { + "session_id": SID, + "source_tag": "messaging", + "raw_source": "messaging", + "session_source": "other", + }, + ) + monkeypatch.setattr(routes_module, "_check_csrf", lambda _handler: True) + monkeypatch.setattr(routes_module, "_guard_request_session_visibility", lambda *args, **kwargs: True) + handler = _FakePostHandler({"session_id": SID}, path="/api/session/branch") + routes_module.handle_post(handler, SimpleNamespace(path="/api/session/branch", query="")) + assert handler.status == 403 + payload = _response_json(handler) + assert "read-only" in payload["error"].lower() + assert not (isolated_state_db["sessions_dir"] / f"{SID}.json").exists() + + def test_branch_from_claimable_tui_still_creates_fork( routes_module, monkeypatch, isolated_state_db ):