Skip to content

Commit fbd9ba1

Browse files
clean6378-max-itclean6378-max-it
committed
fix(#126): harden workspace filter validation and tooltip a11y
1 parent 4e8320e commit fbd9ba1

4 files changed

Lines changed: 236 additions & 186 deletions

File tree

‎api/search.py‎

Lines changed: 209 additions & 183 deletions
Original file line numberDiff line numberDiff line change
@@ -1,183 +1,209 @@
1-
"""
2-
API route for search — mirrors src/app/api/search/route.ts
3-
GET /api/search?q=...&type=all|chat|composer&all_history=1&workspace=<id>
4-
"""
5-
6-
from __future__ import annotations
7-
8-
import logging
9-
import os
10-
import sqlite3
11-
from typing import Any
12-
13-
from flask import Blueprint, Response, current_app, request
14-
15-
from api.flask_config import json_response
16-
17-
from models import ParseWarningCollector, SearchResult
18-
from services.search import (
19-
DEFAULT_SEARCH_WINDOW_DAYS,
20-
rank_results,
21-
resolve_search_since_ms,
22-
search_cli_sessions,
23-
search_global_storage,
24-
search_legacy_workspaces,
25-
)
26-
from utils.cli_chat_reader import list_cli_projects
27-
from utils.workspace_path import get_cli_chats_path, resolve_workspace_path
28-
29-
bp = Blueprint("search", __name__)
30-
_logger = logging.getLogger(__name__)
31-
32-
_MAX_SEARCH_SINCE_DAYS = 36_500 # ~100 years; avoids timedelta overflow on bad input
33-
_MAX_SEARCH_QUERY_LEN = 500
34-
_VALID_SEARCH_TYPES = frozenset({"all", "chat", "composer"})
35-
36-
37-
def _parse_since_days_param(raw: str | None) -> int | None:
38-
if raw is None or not str(raw).strip():
39-
return None
40-
try:
41-
days = int(raw)
42-
except ValueError:
43-
return None
44-
if days <= 0 or days > _MAX_SEARCH_SINCE_DAYS:
45-
return None
46-
return days
47-
48-
49-
def _search_error(
50-
message: str,
51-
code: str,
52-
status: int,
53-
) -> tuple[Response, int]:
54-
return json_response({"error": message, "code": code}, status)
55-
56-
57-
def _workspace_exists(workspace_id: str, workspace_path: str) -> bool:
58-
if workspace_id == "global":
59-
return True
60-
if workspace_id.startswith("cli:"):
61-
project_id = workspace_id[4:]
62-
return any(
63-
cp.get("project_id") == project_id
64-
for cp in list_cli_projects(get_cli_chats_path())
65-
)
66-
return os.path.isdir(os.path.join(workspace_path, workspace_id))
67-
68-
69-
def _filter_results_by_workspace(
70-
results: list[SearchResult],
71-
workspace_id: str,
72-
) -> list[SearchResult]:
73-
return [r for r in results if r.get("workspaceId") == workspace_id]
74-
75-
76-
@bp.route("/api/search")
77-
def search() -> tuple[Response, int] | Response:
78-
"""Search chats, composers, and CLI sessions across Cursor storage.
79-
80-
Args:
81-
q: Search query string (required; 400 when empty).
82-
type: Filter scope — ``all`` (default), ``chat``, or ``composer``.
83-
workspace: Optional workspace folder hash; 404 when unknown.
84-
85-
Returns:
86-
JSON ``{"results": [...]}`` with optional ``warnings``. Structured
87-
``{"error", "code"}`` bodies for 400/404/503/500 failures.
88-
"""
89-
query = request.args.get("q", "").strip()
90-
if not query:
91-
return _search_error("No search query provided", "empty_query", 400)
92-
if len(query) > _MAX_SEARCH_QUERY_LEN:
93-
return _search_error("Search query is too long", "query_too_long", 400)
94-
95-
search_type = request.args.get("type", "all")
96-
if search_type not in _VALID_SEARCH_TYPES:
97-
return _search_error("Invalid search type", "invalid_type", 400)
98-
99-
since_days_raw = request.args.get("since_days")
100-
if (
101-
since_days_raw is not None
102-
and str(since_days_raw).strip()
103-
and _parse_since_days_param(since_days_raw) is None
104-
):
105-
return _search_error("Invalid since_days parameter", "invalid_since_days", 400)
106-
107-
workspace_filter = request.args.get("workspace", "").strip() or None
108-
workspace_path = resolve_workspace_path()
109-
if workspace_filter and not _workspace_exists(workspace_filter, workspace_path):
110-
return _search_error("Workspace not found", "workspace_not_found", 404)
111-
112-
try:
113-
rules = current_app.config.get("EXCLUSION_RULES") or []
114-
all_history = request.args.get("all_history") in ("1", "true")
115-
since_ms = resolve_search_since_ms(
116-
all_history=all_history,
117-
since_days=_parse_since_days_param(since_days_raw),
118-
)
119-
120-
parse_warnings = ParseWarningCollector()
121-
query_lower = query.lower()
122-
123-
results: list[SearchResult] = []
124-
if search_type != "chat":
125-
results.extend(
126-
search_global_storage(
127-
workspace_path,
128-
query,
129-
query_lower,
130-
rules,
131-
parse_warnings,
132-
since_ms=since_ms,
133-
)
134-
)
135-
results.extend(
136-
search_legacy_workspaces(
137-
workspace_path,
138-
query,
139-
query_lower,
140-
search_type,
141-
rules,
142-
since_ms=since_ms,
143-
)
144-
)
145-
if search_type == "all":
146-
results.extend(
147-
search_cli_sessions(
148-
get_cli_chats_path(),
149-
query,
150-
query_lower,
151-
rules,
152-
parse_warnings,
153-
since_ms=since_ms,
154-
)
155-
)
156-
157-
ranked = rank_results(results)
158-
if workspace_filter:
159-
ranked = _filter_results_by_workspace(ranked, workspace_filter)
160-
161-
payload: dict[str, Any] = {
162-
"results": ranked,
163-
"allHistory": since_ms is None,
164-
"searchWindowDays": (
165-
None if since_ms is None else (
166-
_parse_since_days_param(since_days_raw)
167-
or DEFAULT_SEARCH_WINDOW_DAYS
168-
)
169-
),
170-
}
171-
return json_response(parse_warnings.attach_to(payload))
172-
173-
except sqlite3.OperationalError:
174-
_logger.exception("Search index unavailable")
175-
return _search_error(
176-
"Search index is temporarily unavailable",
177-
"search_index_unavailable",
178-
503,
179-
)
180-
except Exception:
181-
_logger.exception("Search failed")
182-
return _search_error("Search failed", "internal_error", 500)
183-
1+
"""
2+
API route for search — mirrors src/app/api/search/route.ts
3+
GET /api/search?q=...&type=all|chat|composer&all_history=1&workspace=<id>
4+
"""
5+
6+
from __future__ import annotations
7+
8+
import logging
9+
import os
10+
import re
11+
import sqlite3
12+
from typing import Any
13+
14+
from flask import Blueprint, Response, current_app, request
15+
16+
from api.flask_config import json_response
17+
18+
from models import ParseWarningCollector, SearchResult
19+
from services.search import (
20+
DEFAULT_SEARCH_WINDOW_DAYS,
21+
rank_results,
22+
resolve_search_since_ms,
23+
search_cli_sessions,
24+
search_global_storage,
25+
search_legacy_workspaces,
26+
)
27+
from utils.cli_chat_reader import list_cli_projects
28+
from utils.workspace_path import get_cli_chats_path, resolve_workspace_path
29+
30+
bp = Blueprint("search", __name__)
31+
_logger = logging.getLogger(__name__)
32+
33+
_MAX_SEARCH_SINCE_DAYS = 36_500 # ~100 years; avoids timedelta overflow on bad input
34+
_MAX_SEARCH_QUERY_LEN = 500
35+
_VALID_SEARCH_TYPES = frozenset({"all", "chat", "composer"})
36+
_SAFE_WORKSPACE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
37+
38+
39+
def _parse_since_days_param(raw: str | None) -> int | None:
40+
if raw is None or not str(raw).strip():
41+
return None
42+
try:
43+
days = int(raw)
44+
except ValueError:
45+
return None
46+
if days <= 0 or days > _MAX_SEARCH_SINCE_DAYS:
47+
return None
48+
return days
49+
50+
51+
def _search_error(
52+
message: str,
53+
code: str,
54+
status: int,
55+
) -> tuple[Response, int]:
56+
return json_response({"error": message, "code": code}, status)
57+
58+
59+
def _is_safe_workspace_folder_id(workspace_id: str) -> bool:
60+
"""Return whether *workspace_id* is a safe Cursor workspace folder name."""
61+
if not workspace_id or workspace_id in {".", ".."}:
62+
return False
63+
if (
64+
os.path.isabs(workspace_id)
65+
or ".." in workspace_id
66+
or "/" in workspace_id
67+
or "\\" in workspace_id
68+
):
69+
return False
70+
return _SAFE_WORKSPACE_ID_RE.fullmatch(workspace_id) is not None
71+
72+
73+
def _workspace_exists(workspace_id: str, workspace_path: str) -> bool:
74+
if workspace_id == "global":
75+
return True
76+
if workspace_id.startswith("cli:"):
77+
project_id = workspace_id[4:]
78+
if not _is_safe_workspace_folder_id(project_id):
79+
return False
80+
return any(
81+
cp.get("project_id") == project_id
82+
for cp in list_cli_projects(get_cli_chats_path())
83+
)
84+
if not _is_safe_workspace_folder_id(workspace_id):
85+
return False
86+
candidate = os.path.join(workspace_path, workspace_id)
87+
root = os.path.normpath(workspace_path)
88+
joined = os.path.normpath(candidate)
89+
if os.path.commonpath([root, joined]) != root:
90+
return False
91+
return os.path.isdir(joined)
92+
93+
94+
def _filter_results_by_workspace(
95+
results: list[SearchResult],
96+
workspace_id: str,
97+
) -> list[SearchResult]:
98+
return [r for r in results if r.get("workspaceId") == workspace_id]
99+
100+
101+
@bp.route("/api/search")
102+
def search() -> tuple[Response, int] | Response:
103+
"""Search chats, composers, and CLI sessions across Cursor storage.
104+
105+
Args:
106+
q: Search query string (required; 400 when empty).
107+
type: Filter scope — ``all`` (default), ``chat``, or ``composer``.
108+
workspace: Optional workspace folder hash; 404 when unknown.
109+
110+
Returns:
111+
JSON ``{"results": [...]}`` with optional ``warnings``. Structured
112+
``{"error", "code"}`` bodies for 400/404/503/500 failures.
113+
"""
114+
query = request.args.get("q", "").strip()
115+
if not query:
116+
return _search_error("No search query provided", "empty_query", 400)
117+
if len(query) > _MAX_SEARCH_QUERY_LEN:
118+
return _search_error("Search query is too long", "query_too_long", 400)
119+
120+
search_type = request.args.get("type", "all")
121+
if search_type not in _VALID_SEARCH_TYPES:
122+
return _search_error("Invalid search type", "invalid_type", 400)
123+
124+
since_days_raw = request.args.get("since_days")
125+
if (
126+
since_days_raw is not None
127+
and str(since_days_raw).strip()
128+
and _parse_since_days_param(since_days_raw) is None
129+
):
130+
return _search_error("Invalid since_days parameter", "invalid_since_days", 400)
131+
132+
workspace_filter = request.args.get("workspace", "").strip() or None
133+
134+
try:
135+
workspace_path = resolve_workspace_path()
136+
if workspace_filter and not _workspace_exists(workspace_filter, workspace_path):
137+
return _search_error("Workspace not found", "workspace_not_found", 404)
138+
139+
rules = current_app.config.get("EXCLUSION_RULES") or []
140+
all_history = request.args.get("all_history") in ("1", "true")
141+
since_ms = resolve_search_since_ms(
142+
all_history=all_history,
143+
since_days=_parse_since_days_param(since_days_raw),
144+
)
145+
146+
parse_warnings = ParseWarningCollector()
147+
query_lower = query.lower()
148+
149+
results: list[SearchResult] = []
150+
if search_type != "chat":
151+
results.extend(
152+
search_global_storage(
153+
workspace_path,
154+
query,
155+
query_lower,
156+
rules,
157+
parse_warnings,
158+
since_ms=since_ms,
159+
)
160+
)
161+
results.extend(
162+
search_legacy_workspaces(
163+
workspace_path,
164+
query,
165+
query_lower,
166+
search_type,
167+
rules,
168+
since_ms=since_ms,
169+
)
170+
)
171+
if search_type == "all":
172+
results.extend(
173+
search_cli_sessions(
174+
get_cli_chats_path(),
175+
query,
176+
query_lower,
177+
rules,
178+
parse_warnings,
179+
since_ms=since_ms,
180+
)
181+
)
182+
183+
ranked = rank_results(results)
184+
if workspace_filter:
185+
ranked = _filter_results_by_workspace(ranked, workspace_filter)
186+
187+
payload: dict[str, Any] = {
188+
"results": ranked,
189+
"allHistory": since_ms is None,
190+
"searchWindowDays": (
191+
None if since_ms is None else (
192+
_parse_since_days_param(since_days_raw)
193+
or DEFAULT_SEARCH_WINDOW_DAYS
194+
)
195+
),
196+
}
197+
return json_response(parse_warnings.attach_to(payload))
198+
199+
except sqlite3.OperationalError:
200+
_logger.exception("Search index unavailable")
201+
return _search_error(
202+
"Search index is temporarily unavailable",
203+
"search_index_unavailable",
204+
503,
205+
)
206+
except Exception:
207+
_logger.exception("Search failed")
208+
return _search_error("Search failed", "internal_error", 500)
209+

0 commit comments

Comments
 (0)