diff --git a/src/api/handlers/mcp/pagination.py b/src/api/handlers/mcp/pagination.py new file mode 100644 index 00000000..3ea0085a --- /dev/null +++ b/src/api/handlers/mcp/pagination.py @@ -0,0 +1,61 @@ +""" +Pagination helpers for MCP list endpoints. + +Every MCP list endpoint returns a bounded page by default so a single call can +never overflow the client/model context or hang on very large result sets. + +Usage in a handler: + + from api.handlers.mcp.pagination import parse_pagination, page + + limit, offset = parse_pagination() # reads ?limit=&offset=, validates + ...run query with LIMIT %s OFFSET %s and a COUNT(*)... + return page(items, total, limit, offset) + +Callers that need the *entire* set page through by increasing offset until +offset + len(items) >= total. +""" +from flask import request, abort + +DEFAULT_LIMIT = 50 +MAX_LIMIT = 500 + + +def parse_pagination(): + """Read and validate limit/offset from the query string. + + Returns (limit, offset). Missing params fall back to defaults, so an + endpoint always applies a bound even when the caller passes nothing. + Invalid values abort with 400. + """ + limit = _parse_int('limit', DEFAULT_LIMIT) + offset = _parse_int('offset', 0) + + if limit < 1: + abort(400, 'limit must be >= 1') + if limit > MAX_LIMIT: + limit = MAX_LIMIT + if offset < 0: + abort(400, 'offset must be >= 0') + + return limit, offset + + +def _parse_int(name, default): + raw = request.args.get(name) + if raw is None or raw == '': + return default + try: + return int(raw) + except (TypeError, ValueError): + abort(400, '%s must be an integer' % name) + + +def page(items, total, limit, offset): + """Wrap a page of items in the standard pagination envelope.""" + return { + 'items': items, + 'total': total, + 'limit': limit, + 'offset': offset, + } diff --git a/src/api/handlers/mcp/routes/builds.py b/src/api/handlers/mcp/routes/builds.py index af6a4ed9..3e3ee2d0 100644 --- a/src/api/handlers/mcp/routes/builds.py +++ b/src/api/handlers/mcp/routes/builds.py @@ -21,6 +21,7 @@ ) from api.handlers.mcp.rate_limit import mcp_rate_limit from api.handlers.mcp.audit import audit_mcp +from api.handlers.mcp.pagination import parse_pagination, page ns = api.namespace('MCP Builds', path='/api/v1/mcp/projects/', @@ -32,13 +33,17 @@ class MCPBuilds(Resource): @mcp_auth_required @mcp_rate_limit('list_builds') def get(self, project_id): - """List builds for a project.""" + """List builds for a project. Paginated (default limit 50).""" audit_mcp('list_builds', outcome='attempt', details={'project_id': project_id}) if not check_project_access_mcp(project_id): audit_mcp('list_builds', outcome='forbidden', details={'project_id': project_id}) abort(403, 'access to this project is not permitted for the current MCP token') + limit, offset = parse_pagination() + try: + total = g.db.execute_one_dict( + 'SELECT count(*) AS c FROM build WHERE project_id = %s', [project_id])['c'] rows = g.db.execute_many_dict(''' SELECT b.id, b.build_number, b.restart_counter, b.project_id, b.commit_id, c.branch, @@ -59,12 +64,12 @@ def get(self, project_id): WHERE b.project_id = %s GROUP BY b.id, b.build_number, b.restart_counter, b.project_id, b.commit_id, c.branch ORDER BY b.build_number DESC, b.restart_counter DESC - LIMIT 50 - ''', [project_id]) - result = [_build_dict(r) for r in rows] + LIMIT %s OFFSET %s + ''', [project_id, limit, offset]) + items = [_build_dict(r) for r in rows] audit_mcp('list_builds', outcome='success', - details={'project_id': project_id, 'count': len(result)}) - return result + details={'project_id': project_id, 'count': len(items), 'total': total}) + return page(items, total, limit, offset) except Exception as exc: # Clear any aborted/pending transaction so the failure audit # (which shares g.db) can write, and no stray write is committed. diff --git a/src/api/handlers/mcp/routes/jobs.py b/src/api/handlers/mcp/routes/jobs.py index eb641008..594e8ef3 100644 --- a/src/api/handlers/mcp/routes/jobs.py +++ b/src/api/handlers/mcp/routes/jobs.py @@ -16,7 +16,7 @@ import re import uuid as _uuid -from flask import g, abort +from flask import g, request, abort from flask_restx import Resource from pyinfraboxutils.ibrestplus import api @@ -28,12 +28,17 @@ ) from api.handlers.mcp.rate_limit import mcp_rate_limit from api.handlers.mcp.audit import audit_mcp +from api.handlers.mcp.pagination import parse_pagination, page logger = logging.getLogger('mcp_jobs') _ACCESS_DENIED = 'access to this project is not permitted for the current MCP token' _JOB_BY_PROJECT = 'SELECT id FROM job WHERE id = %s AND project_id = %s' +# Log byte caps: bound get_job_log so one call can't overflow the context. +DEFAULT_LOG_BYTES = 1024 * 1024 # 1 MB tail by default +MAX_LOG_BYTES = 5 * 1024 * 1024 # 5 MB hard ceiling per request + ns_build_jobs = api.namespace('MCP Build Jobs', path='/api/v1/mcp/projects//builds/', description='MCP job list') @@ -48,25 +53,46 @@ class MCPJobList(Resource): @mcp_auth_required @mcp_rate_limit('list_jobs') def get(self, project_id, build_id): - """List jobs for a build.""" + """List jobs for a build. + + Paginated (default limit 50). Optional ?state= filters by job state + (e.g. failure). Returns {items, total, limit, offset}. + """ audit_mcp('list_jobs', outcome='attempt', details={'project_id': project_id, 'build_id': build_id}) if not check_project_access_mcp(project_id): audit_mcp('list_jobs', outcome='forbidden', details={'project_id': project_id}) abort(403, _ACCESS_DENIED) + limit, offset = parse_pagination() + state = request.args.get('state') or None + + # Build the WHERE clause; state is an optional, exact-match filter. + # Compare against state::text so an unknown value (not in the job_state + # enum) simply matches nothing instead of raising a 22P02 invalid-enum + # error (which would surface as a 500). + where = 'j.build_id = %s AND j.project_id = %s' + params = [build_id, project_id] + if state: + where += ' AND j.state::text = %s' + params.append(state) + try: + total = g.db.execute_one_dict( + 'SELECT count(*) AS c FROM job j WHERE ' + where, params)['c'] rows = g.db.execute_many_dict(''' SELECT j.id, j.name, j.state, j.build_id, j.project_id, j.start_date, j.end_date, j.message FROM job j - WHERE j.build_id = %s AND j.project_id = %s + WHERE ''' + where + ''' ORDER BY j.name - ''', [build_id, project_id]) - result = [_job_dict(r) for r in rows] + LIMIT %s OFFSET %s + ''', params + [limit, offset]) + items = [_job_dict(r) for r in rows] audit_mcp('list_jobs', outcome='success', - details={'project_id': project_id, 'build_id': build_id, 'count': len(result)}) - return result + details={'project_id': project_id, 'build_id': build_id, + 'count': len(items), 'total': total}) + return page(items, total, limit, offset) except Exception as exc: # Clear any aborted/pending transaction so the failure audit # (which shares g.db) can write, and no stray write is committed. @@ -120,7 +146,15 @@ class MCPJobLog(Resource): @mcp_auth_required @mcp_rate_limit('get_job_log') def get(self, project_id, job_id): - """Get console log for a job.""" + """Get console log for a job. + + Bounded by default: returns at most the last DEFAULT_LOG_BYTES (1 MB) of + the log so a single call cannot overflow the client/model context. + Query params: + - max_bytes: cap on returned bytes (default 1 MB, max 5 MB) + - offset/length: read an explicit byte range instead of the tail + Returns {log, total_bytes, offset, length, truncated}. + """ audit_mcp('get_job_log', outcome='attempt', details={'project_id': project_id, 'job_id': job_id}) if not check_project_access_mcp(project_id): @@ -148,10 +182,13 @@ def get(self, project_id, job_id): ''', [job_id]) log = ''.join(r['output'] for r in rows) + sliced = _slice_log(log) audit_mcp('get_job_log', outcome='success', details={'project_id': project_id, 'job_id': job_id, - 'bytes': len(log)}) - return log, 200, {'Content-Type': 'text/plain; charset=utf-8'} + 'total_bytes': sliced['total_bytes'], + 'returned_bytes': sliced['length'], + 'truncated': sliced['truncated']}) + return sliced except Exception as exc: # Clear any aborted/pending transaction so the failure audit # (which shares g.db) can write, and no stray write is committed. @@ -190,12 +227,18 @@ def get(self, project_id, job_id): ''', [job_id, project_id]) archive = (row or {}).get('archive') or [] - result = [{'filename': a.get('filename'), - 'filesize': a.get('size') or a.get('filesize')} - for a in archive] + all_items = [{'filename': a.get('filename'), + 'filesize': a.get('size') or a.get('filesize')} + for a in archive] + # archive is a single jsonb[] column (already in memory), so paginate + # by slicing the list rather than in SQL. + limit, offset = parse_pagination() + total = len(all_items) + items = all_items[offset:offset + limit] audit_mcp('list_job_artifacts', outcome='success', - details={'project_id': project_id, 'job_id': job_id, 'count': len(result)}) - return result + details={'project_id': project_id, 'job_id': job_id, + 'count': len(items), 'total': total}) + return page(items, total, limit, offset) except Exception as exc: # Clear any aborted/pending transaction so the failure audit # (which shares g.db) can write, and no stray write is committed. @@ -266,15 +309,23 @@ def get(self, project_id, job_id): abort(403, _ACCESS_DENIED) try: + limit, offset = parse_pagination() + total = g.db.execute_one_dict(''' + SELECT count(*) AS c FROM test_run tr + WHERE tr.job_id = %s AND tr.project_id = %s + ''', [job_id, project_id])['c'] rows = g.db.execute_many_dict(''' SELECT tr.state, tr.name, tr.suite, tr.duration, tr.message, tr.stack, to_char(tr.timestamp, 'YYYY-MM-DD HH24:MI:SS') AS timestamp FROM test_run tr WHERE tr.job_id = %s AND tr.project_id = %s - ''', [job_id, project_id]) + ORDER BY tr.suite, tr.name + LIMIT %s OFFSET %s + ''', [job_id, project_id, limit, offset]) audit_mcp('get_job_testruns', outcome='success', - details={'project_id': project_id, 'job_id': job_id, 'count': len(rows)}) - return rows + details={'project_id': project_id, 'job_id': job_id, + 'count': len(rows), 'total': total}) + return page(rows, total, limit, offset) except Exception as exc: # Clear any aborted/pending transaction so the failure audit # (which shares g.db) can write, and no stray write is committed. @@ -363,6 +414,59 @@ def _job_dict(r): } +def _slice_log(log): + """Bound a job log to a byte window. + + Default (no params): return the LAST DEFAULT_LOG_BYTES of the log — the tail + is where failures/stack traces live, and it's what's useful when truncating. + Query params: + - max_bytes: cap the returned size (clamped to [1, MAX_LOG_BYTES]) + - offset/length: read an explicit byte range from the start instead of the tail + Returns {log, total_bytes, offset, length, truncated}. + """ + data = (log or '').encode('utf-8') + total = len(data) + + raw_offset = request.args.get('offset') + raw_length = request.args.get('length') + raw_max = request.args.get('max_bytes') + + if raw_offset is not None or raw_length is not None: + # Explicit range read from the start of the log. + offset = _pos_int('offset', raw_offset, 0) + length = _pos_int('length', raw_length, DEFAULT_LOG_BYTES) + length = min(length, MAX_LOG_BYTES) + chunk = data[offset:offset + length] + else: + # Default: tail of the log, capped by max_bytes. + max_bytes = _pos_int('max_bytes', raw_max, DEFAULT_LOG_BYTES) + max_bytes = min(max(max_bytes, 1), MAX_LOG_BYTES) + offset = max(0, total - max_bytes) + chunk = data[offset:] + length = max_bytes + + text = chunk.decode('utf-8', errors='replace') + return { + 'log': text, + 'total_bytes': total, + 'offset': offset, + 'length': len(chunk), + 'truncated': offset > 0 or (offset + len(chunk)) < total, + } + + +def _pos_int(name, raw, default): + if raw is None or raw == '': + return default + try: + v = int(raw) + except (TypeError, ValueError): + abort(400, '%s must be an integer' % name) + if v < 0: + abort(400, '%s must be >= 0' % name) + return v + + def _compact_stats(series): """Downsample a stats time series to at most 100 points.""" if not isinstance(series, list) or len(series) <= 100: diff --git a/src/api/handlers/mcp/routes/projects.py b/src/api/handlers/mcp/routes/projects.py index 6f9ca611..af3f4725 100644 --- a/src/api/handlers/mcp/routes/projects.py +++ b/src/api/handlers/mcp/routes/projects.py @@ -10,6 +10,7 @@ from api.handlers.mcp.auth import mcp_auth_required, check_project_access_mcp, get_mcp_user_id from api.handlers.mcp.rate_limit import mcp_rate_limit from api.handlers.mcp.audit import audit_mcp +from api.handlers.mcp.pagination import parse_pagination, page ns = api.namespace('MCP Projects', path='/api/v1/mcp', @@ -21,44 +22,43 @@ class MCPProjects(Resource): @mcp_auth_required @mcp_rate_limit('list_projects') def get(self): - """List projects accessible to the current MCP token or session user.""" + """List projects accessible to the current MCP token or session user. + + Paginated (default limit 50). Returns {items, total, limit, offset}. + """ audit_mcp('list_projects', outcome='attempt') try: user_id = get_mcp_user_id() enabled = getattr(g, 'mcp_enabled_projects', None) - if enabled is not None: - # MCP token path - if not enabled: - # empty dict = all projects the user is a collaborator on - rows = g.db.execute_many_dict(''' - SELECT p.id, p.name, p.type, p.public - FROM project p - INNER JOIN collaborator co ON co.project_id = p.id AND co.user_id = %s - ORDER BY p.name - ''', [user_id]) - else: - project_ids = list(enabled.keys()) - rows = g.db.execute_many_dict(''' - SELECT p.id, p.name, p.type, p.public - FROM project p - INNER JOIN collaborator co ON co.project_id = p.id AND co.user_id = %s - WHERE p.id = ANY(%s::uuid[]) - ORDER BY p.name - ''', [user_id, project_ids]) - else: - # Session path: return all projects the user is a collaborator on - rows = g.db.execute_many_dict(''' - SELECT p.id, p.name, p.type, p.public - FROM project p - INNER JOIN collaborator co ON co.project_id = p.id AND co.user_id = %s - ORDER BY p.name - ''', [user_id]) + # Build a shared WHERE clause across the three access paths: + # - MCP token with an explicit project scope -> filter by those ids + # - MCP token with empty scope, or session user -> all collaborations + where = 'co.user_id = %s' + params = [user_id] + if enabled: + where += ' AND p.id = ANY(%s::uuid[])' + params.append(list(enabled.keys())) - result = [{'id': r['id'], 'name': r['name'], 'type': r['type'], 'public': r['public']} - for r in rows] - audit_mcp('list_projects', outcome='success', details={'count': len(result)}) - return result + limit, offset = parse_pagination() + total = g.db.execute_one_dict(''' + SELECT count(*) AS c + FROM project p + INNER JOIN collaborator co ON co.project_id = p.id AND ''' + where, + params)['c'] + rows = g.db.execute_many_dict(''' + SELECT p.id, p.name, p.type, p.public + FROM project p + INNER JOIN collaborator co ON co.project_id = p.id AND ''' + where + ''' + ORDER BY p.name + LIMIT %s OFFSET %s + ''', params + [limit, offset]) + + items = [{'id': r['id'], 'name': r['name'], 'type': r['type'], 'public': r['public']} + for r in rows] + audit_mcp('list_projects', outcome='success', + details={'count': len(items), 'total': total}) + return page(items, total, limit, offset) except Exception as exc: # Clear any aborted/pending transaction so the failure audit # (which shares g.db) can write, and no stray write is committed.