diff --git a/src/api/handlers/mcp/routes/builds.py b/src/api/handlers/mcp/routes/builds.py index b06d01de..ac009a4e 100644 --- a/src/api/handlers/mcp/routes/builds.py +++ b/src/api/handlers/mcp/routes/builds.py @@ -1,19 +1,27 @@ """ MCP builds endpoints. -GET /api/v1/mcp/projects//builds -GET /api/v1/mcp/projects//builds/ -POST /api/v1/mcp/projects//trigger +GET /api/v1/mcp/projects//builds +GET /api/v1/mcp/projects//builds/ +POST /api/v1/mcp/projects//trigger +POST /api/v1/mcp/projects//builds//restart +DELETE /api/v1/mcp/projects//builds//abort """ +import json import uuid as _uuid from flask import g, abort from flask_restx import Resource from pyinfraboxutils.ibrestplus import api -from api.handlers.mcp.auth import mcp_auth_required, check_project_access_mcp, check_trigger_access_mcp +from api.handlers.mcp.auth import (mcp_auth_required, check_project_access_mcp, + check_trigger_access_mcp, get_mcp_user_id) from api.handlers.mcp.rate_limit import mcp_rate_limit from api.handlers.mcp.audit import audit_mcp +_ACCESS_DENIED = 'access to this project is not permitted for the current MCP token' +_TRIGGER_DENIED = 'this MCP token does not have trigger permission' +_TRIGGER_DENIED_REASON = 'trigger not allowed' + ns = api.namespace('MCP Builds', path='/api/v1/mcp/projects/', description='MCP build operations') @@ -28,7 +36,7 @@ def get(self, project_id): 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') + abort(403, _ACCESS_DENIED) try: rows = g.db.execute_many_dict(''' @@ -73,7 +81,7 @@ def get(self, project_id, build_id): details={'project_id': project_id, 'build_id': build_id}) if not check_project_access_mcp(project_id): audit_mcp('get_build', outcome='forbidden', details={'project_id': project_id}) - abort(403, 'access to this project is not permitted for the current MCP token') + abort(403, _ACCESS_DENIED) try: rows = g.db.execute_many_dict(''' @@ -116,11 +124,11 @@ def post(self, project_id): """Trigger a new build (requires allow_trigger on the MCP token).""" if not check_project_access_mcp(project_id): audit_mcp('trigger_build', outcome='forbidden', details={'project_id': project_id}) - abort(403, 'access to this project is not permitted for the current MCP token') + abort(403, _ACCESS_DENIED) if not check_trigger_access_mcp(): audit_mcp('trigger_build', outcome='forbidden', - details={'project_id': project_id, 'reason': 'trigger not allowed'}) + details={'project_id': project_id, 'reason': _TRIGGER_DENIED_REASON}) abort(403, 'this MCP token does not have trigger permission') project = g.db.execute_one_dict(''' @@ -152,6 +160,114 @@ def post(self, project_id): raise +@ns.route('/builds//restart') +class MCPBuildRestart(Resource): + @mcp_auth_required + @mcp_rate_limit('trigger_build') + def post(self, project_id, build_id): + """Restart a build (requires allow_trigger on the MCP token).""" + audit_mcp('restart_build', outcome='attempt', + details={'project_id': project_id, 'build_id': build_id}) + if not check_project_access_mcp(project_id): + audit_mcp('restart_build', outcome='forbidden', details={'project_id': project_id}) + abort(403, _ACCESS_DENIED) + if not check_trigger_access_mcp(): + audit_mcp('restart_build', outcome='forbidden', + details={'project_id': project_id, 'reason': _TRIGGER_DENIED_REASON}) + abort(403, _TRIGGER_DENIED) + + try: + build = g.db.execute_one_dict(''' + SELECT commit_id, build_number, source_upload_id + FROM build WHERE id = %s AND project_id = %s + ''', [build_id, project_id]) + if not build: + abort(404) + + result = g.db.execute_one_dict(''' + SELECT max(restart_counter) AS restart_counter + FROM build WHERE build_number = %s AND project_id = %s + ''', [build['build_number'], project_id]) + restart_counter = result['restart_counter'] + 1 + + new_result = g.db.execute_one_dict(''' + INSERT INTO build (commit_id, build_number, project_id, restart_counter, source_upload_id) + VALUES (%s, %s, %s, %s, %s) RETURNING id + ''', [build['commit_id'], build['build_number'], project_id, + restart_counter, build['source_upload_id']]) + new_build_id = new_result['id'] + + job = g.db.execute_one_dict(''' + SELECT repo, env_var, definition FROM job + WHERE project_id = %s AND name = 'Create Jobs' AND build_id = %s + ''', [project_id, build_id]) + + env_var = json.dumps(job['env_var']) if job and job['env_var'] else None + repo = json.dumps(job['repo']) if job and job['repo'] else None + definition = json.dumps(job['definition']) if job and job['definition'] else None + + job_id = str(_uuid.uuid4()) + msg = 'Build restarted by MCP token %s\n' % get_mcp_user_id() + g.db.execute(''' + INSERT INTO job (id, state, build_id, type, name, project_id, + dockerfile, repo, env_var, definition, cluster_name) + VALUES (%s, 'queued', %s, 'create_job_matrix', 'Create Jobs', + %s, '', %s, %s, %s, null); + INSERT INTO console (job_id, output) VALUES (%s, %s); + ''', [job_id, new_build_id, project_id, repo, env_var, definition, job_id, msg]) + g.db.commit() + + audit_mcp('restart_build', outcome='success', + details={'project_id': project_id, 'build_id': build_id, + 'new_build_id': new_build_id}) + return {'build_id': new_build_id, 'restart_counter': restart_counter}, 200 + except Exception as exc: + audit_mcp('restart_build', outcome='failure', + details={'project_id': project_id, 'build_id': build_id}, error=str(exc)) + raise + + +@ns.route('/builds//abort') +class MCPBuildAbort(Resource): + @mcp_auth_required + @mcp_rate_limit('trigger_build') + def delete(self, project_id, build_id): + """Abort all running jobs in a build (requires allow_trigger on the MCP token).""" + audit_mcp('abort_build', outcome='attempt', + details={'project_id': project_id, 'build_id': build_id}) + if not check_project_access_mcp(project_id): + audit_mcp('abort_build', outcome='forbidden', details={'project_id': project_id}) + abort(403, _ACCESS_DENIED) + if not check_trigger_access_mcp(): + audit_mcp('abort_build', outcome='forbidden', + details={'project_id': project_id, 'reason': _TRIGGER_DENIED_REASON}) + abort(403, _TRIGGER_DENIED) + + try: + user_id = get_mcp_user_id() + jobs = g.db.execute_many_dict(''' + SELECT id FROM job + WHERE build_id = %s AND project_id = %s + ''', [build_id, project_id]) + if not jobs: + abort(404) + + for j in jobs: + g.db.execute(''' + INSERT INTO abort(job_id, user_id) VALUES(%s, %s) + ''', [j['id'], user_id]) + g.db.commit() + + audit_mcp('abort_build', outcome='success', + details={'project_id': project_id, 'build_id': build_id, + 'jobs_aborted': len(jobs)}) + return {'message': 'Aborted all jobs', 'jobs_aborted': len(jobs)}, 200 + except Exception as exc: + audit_mcp('abort_build', outcome='failure', + details={'project_id': project_id, 'build_id': build_id}, error=str(exc)) + raise + + def _build_dict(r): return { 'id': r['id'], diff --git a/src/api/handlers/mcp/routes/jobs.py b/src/api/handlers/mcp/routes/jobs.py index 0b22fdc4..4194b4e4 100644 --- a/src/api/handlers/mcp/routes/jobs.py +++ b/src/api/handlers/mcp/routes/jobs.py @@ -1,21 +1,27 @@ """ MCP job endpoints. -GET /api/v1/mcp/projects//builds//jobs -GET /api/v1/mcp/projects//jobs/ -GET /api/v1/mcp/projects//jobs//log -GET /api/v1/mcp/projects//jobs//artifacts -GET /api/v1/mcp/projects//jobs//stats -GET /api/v1/mcp/projects//jobs//testruns -GET /api/v1/mcp/projects//jobs//manifest +GET /api/v1/mcp/projects//builds//jobs +GET /api/v1/mcp/projects//jobs/ +GET /api/v1/mcp/projects//jobs//log +GET /api/v1/mcp/projects//jobs//artifacts +GET /api/v1/mcp/projects//jobs//stats +GET /api/v1/mcp/projects//jobs//testruns +GET /api/v1/mcp/projects//jobs//manifest +POST /api/v1/mcp/projects//jobs//restart +POST /api/v1/mcp/projects//jobs//rerun +DELETE /api/v1/mcp/projects//jobs//abort """ import json import logging +import re +import uuid as _uuid from flask import g, abort from flask_restx import Resource from pyinfraboxutils.ibrestplus import api -from api.handlers.mcp.auth import mcp_auth_required, check_project_access_mcp +from api.handlers.mcp.auth import (mcp_auth_required, check_project_access_mcp, + check_trigger_access_mcp, get_mcp_user_id) from api.handlers.mcp.rate_limit import mcp_rate_limit from api.handlers.mcp.audit import audit_mcp @@ -302,3 +308,221 @@ def _compact_stats(series): return series step = len(series) // 100 return series[::step] + + +_RESTART_STATES = ('error', 'failure', 'finished', 'killed', 'unstable') +_RESTARTABLE_TYPES = ('run_project_container', 'run_docker_compose') +_TRIGGER_DENIED = 'this MCP token does not have trigger permission' +_TRIGGER_DENIED_REASON = 'trigger not allowed' + + +def _clone_jobs(project_id, job_ids_to_restart, build_id, actor_label): + """Clone the given job IDs (and remap their inter-dependencies). + + Returns the new job IDs on success. Raises on DB error. + """ + jobs = [] + for jid in job_ids_to_restart: + jobs += g.db.execute_many_dict(''' + SELECT id, build_id, type, dockerfile, name, project_id, dependencies, + repo, env_var, env_var_ref, build_arg, deployment, definition, cluster_name + FROM job WHERE id = %s + ''', [jid]) + + old_id_map = {} + for j in jobs: + g.db.execute('UPDATE job SET restarted = true WHERE id = %s', [j['id']]) + old_id_map[j['id']] = j + j['id'] = str(_uuid.uuid4()) + + # Rename: append .1 (or increment existing counter) + parts = j['name'].split('/') + last = parts[-1] + m = re.search(r'(.*?)\.\d+$', last) + if m: + suffix = '%s.%d' % (m.group(1), int(m.group(2)) + 1) + else: + suffix = last + '.1' + parts[-1] = suffix + j['name'] = '/'.join(parts) + + for j in jobs: + for dep in j.get('dependencies') or []: + if dep.get('job-id') in old_id_map: + dep['job'] = old_id_map[dep['job-id']]['name'] + dep['job-id'] = old_id_map[dep['job-id']]['id'] + + msg = 'Job restarted by %s\n' % actor_label + for j in jobs: + g.db.execute(''' + INSERT INTO job (state, id, build_id, type, dockerfile, name, project_id, + dependencies, repo, env_var, env_var_ref, build_arg, + deployment, definition, restarted, cluster_name) + VALUES ('queued', %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, false, %s); + INSERT INTO console (job_id, output) VALUES (%s, %s); + ''', [j['id'], j['build_id'], j['type'], j['dockerfile'], j['name'], + j['project_id'], json.dumps(j['dependencies']), json.dumps(j['repo']), + json.dumps(j['env_var']), json.dumps(j['env_var_ref']), + json.dumps(j['build_arg']), json.dumps(j['deployment']), + json.dumps(j['definition']), j['cluster_name'], j['id'], msg]) + + g.db.commit() + return [j['id'] for j in jobs] + + +@ns_job.route('/restart') +class MCPJobRestart(Resource): + @mcp_auth_required + @mcp_rate_limit('trigger_build') + def post(self, project_id, job_id): + """Restart a job and all its downstream dependents (requires allow_trigger).""" + audit_mcp('restart_job', outcome='attempt', + details={'project_id': project_id, 'job_id': job_id}) + if not check_project_access_mcp(project_id): + audit_mcp('restart_job', outcome='forbidden', details={'project_id': project_id}) + abort(403, _ACCESS_DENIED) + if not check_trigger_access_mcp(): + audit_mcp('restart_job', outcome='forbidden', + details={'project_id': project_id, 'reason': _TRIGGER_DENIED_REASON}) + abort(403, _TRIGGER_DENIED) + + try: + job = g.db.execute_one_dict(''' + SELECT state, type, build_id, restarted + FROM job WHERE id = %s AND project_id = %s + ''', [job_id, project_id]) + if not job: + abort(404) + + if job['type'] not in _RESTARTABLE_TYPES: + return {'message': 'job type cannot be restarted', 'status': 400}, 400 + if job['state'] not in _RESTART_STATES: + return {'message': 'job in state %s cannot be restarted' % job['state'], + 'status': 400}, 400 + if job['restarted']: + return {'message': 'job has already been restarted', 'status': 400}, 400 + + build_id = job['build_id'] + all_jobs = g.db.execute_many_dict(''' + SELECT id, state, dependencies, restarted FROM job + WHERE build_id = %s AND project_id = %s + ''', [build_id, project_id]) + + # Collect job + all downstream dependents + restart_set = [job_id] + while True: + found = False + for j in all_jobs: + if j['id'] in restart_set or j['restarted']: + continue + for dep in j.get('dependencies') or []: + if dep.get('job-id') in restart_set: + restart_set.append(j['id']) + found = True + break + if not found: + break + + for j in all_jobs: + if j['id'] not in restart_set: + continue + if j['state'] not in _RESTART_STATES and j['state'] not in ('skipped', 'queued'): + return {'message': 'some dependent jobs are still running', 'status': 400}, 400 + + new_ids = _clone_jobs(project_id, restart_set, build_id, + 'MCP token %s' % get_mcp_user_id()) + audit_mcp('restart_job', outcome='success', + details={'project_id': project_id, 'job_id': job_id, + 'jobs_restarted': len(new_ids)}) + return {'message': 'Successfully restarted job', 'jobs_restarted': len(new_ids)}, 200 + except Exception as exc: + audit_mcp('restart_job', outcome='failure', + details={'project_id': project_id, 'job_id': job_id}, error=str(exc)) + raise + + +@ns_job.route('/rerun') +class MCPJobRerun(Resource): + @mcp_auth_required + @mcp_rate_limit('trigger_build') + def post(self, project_id, job_id): + """Rerun a single job without restarting its downstream dependents (requires allow_trigger).""" + audit_mcp('rerun_job', outcome='attempt', + details={'project_id': project_id, 'job_id': job_id}) + if not check_project_access_mcp(project_id): + audit_mcp('rerun_job', outcome='forbidden', details={'project_id': project_id}) + abort(403, _ACCESS_DENIED) + if not check_trigger_access_mcp(): + audit_mcp('rerun_job', outcome='forbidden', + details={'project_id': project_id, 'reason': _TRIGGER_DENIED_REASON}) + abort(403, _TRIGGER_DENIED) + + try: + job = g.db.execute_one_dict(''' + SELECT state, type, build_id, restarted, dependencies, name + FROM job WHERE id = %s AND project_id = %s + ''', [job_id, project_id]) + if not job: + abort(404) + + if job['type'] not in _RESTARTABLE_TYPES: + return {'message': 'job type cannot be rerun', 'status': 400}, 400 + if job['state'] not in _RESTART_STATES: + return {'message': 'job in state %s cannot be rerun' % job['state'], + 'status': 400}, 400 + if job['restarted']: + return {'message': 'job has already been restarted', 'status': 400}, 400 + + # Ensure no parent is still running + for dep in job.get('dependencies') or []: + parent = g.db.execute_one_dict(''' + SELECT state FROM job WHERE id = %s AND project_id = %s + ''', [dep.get('job-id'), project_id]) + if parent and parent['state'] in ('queued', 'running'): + return {'message': 'parent job is still running', 'status': 400}, 400 + + new_ids = _clone_jobs(project_id, [job_id], job['build_id'], + 'MCP token %s' % get_mcp_user_id()) + audit_mcp('rerun_job', outcome='success', + details={'project_id': project_id, 'job_id': job_id}) + return {'message': 'Successfully rerun job', 'new_job_id': new_ids[0]}, 200 + except Exception as exc: + audit_mcp('rerun_job', outcome='failure', + details={'project_id': project_id, 'job_id': job_id}, error=str(exc)) + raise + + +@ns_job.route('/abort') +class MCPJobAbort(Resource): + @mcp_auth_required + @mcp_rate_limit('trigger_build') + def delete(self, project_id, job_id): + """Abort a running job (requires allow_trigger).""" + audit_mcp('abort_job', outcome='attempt', + details={'project_id': project_id, 'job_id': job_id}) + if not check_project_access_mcp(project_id): + audit_mcp('abort_job', outcome='forbidden', details={'project_id': project_id}) + abort(403, _ACCESS_DENIED) + if not check_trigger_access_mcp(): + audit_mcp('abort_job', outcome='forbidden', + details={'project_id': project_id, 'reason': _TRIGGER_DENIED_REASON}) + abort(403, _TRIGGER_DENIED) + + try: + job = g.db.execute_one_dict(_JOB_BY_PROJECT, [job_id, project_id]) + if not job: + abort(404) + + user_id = get_mcp_user_id() + g.db.execute(''' + INSERT INTO abort(job_id, user_id) VALUES(%s, %s) + ''', [job_id, user_id]) + g.db.commit() + + audit_mcp('abort_job', outcome='success', + details={'project_id': project_id, 'job_id': job_id}) + return {'message': 'Successfully aborted job'}, 200 + except Exception as exc: + audit_mcp('abort_job', outcome='failure', + details={'project_id': project_id, 'job_id': job_id}, error=str(exc)) + raise