mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-08 21:14:47 +09:00
Juan DenisandTyler Boykin
9fb0a6cab5
An authenticated user of any role could inject arguments into git subprocesses running as the backend user (root under systemd): - repo_url was passed unvalidated to git ls-remote/clone, enabling the ext:: transport (host command execution on git < 2.52 defaults) and file:// local-repository disclosure (all git versions). - app_path on POST /deploy/clone was passed without a '--' terminator, enabling --upload-pack=<cmd> option injection independent of git version and protocol.*.allow policy. - branch flowed into 'git fetch origin <branch>' as an unterminated refspec. Fix (new app/utils/git_security.py applied at every affected sink): - repo_url scheme allowlist (https/http/git/ssh + scp-like SSH); rejects ext::/file://, local paths, leading '-', whitespace. - '-c protocol.ext.allow=never -c protocol.file.allow=never' pinned on every remote-touching git call, plus a restrictive GIT_ALLOW_PROTOCOL in the subprocess environment. - '--' positional terminator before URL/path/refspec arguments — co-load-bearing with protocol pinning, which alone does not stop option injection such as --upload-pack=<cmd>. - branch/ref validation (leading '-', ref-format metacharacters) for clone --branch, fetch refspecs and stored deploy configs. - POST /deploy/clone now confines app_path to the managed apps root, mirroring how /apps/from-repository derives it server-side. Adds 31 regression tests; bumps version to 1.7.86. Reported-by: CaptBoykin Co-authored-by: Tyler Boykin <25166954+CaptBoykin@users.noreply.github.com>
305 lines
10 KiB
Python
305 lines
10 KiB
Python
# Bucket: PER-APP (plan 29 #9). Deploy config reads/writes gate on the shared
|
|
# app-access seam (can_access_app / can_edit_app); the public inbound webhook is
|
|
# token/signature-verified (ALLOWLISTed — no per-app user session).
|
|
import os
|
|
|
|
from flask import Blueprint, request, jsonify
|
|
from flask_jwt_extended import jwt_required, get_jwt_identity
|
|
from app import paths
|
|
from app.middleware.rbac import admin_required, app_access_tier
|
|
from app.models import User, Application
|
|
from app.services.git_service import GitService
|
|
from app.services.resource_grant_service import ResourceGrantService
|
|
|
|
deploy_bp = Blueprint('deploy', __name__)
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/config', methods=['GET'])
|
|
@jwt_required()
|
|
def get_deploy_config(app_id):
|
|
"""Get deployment configuration for an app."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app = Application.query.get(app_id)
|
|
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
|
|
if not ResourceGrantService.can_access_app(user, app):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
config = GitService.get_app_config(app_id)
|
|
if not config:
|
|
return jsonify({'configured': False}), 200
|
|
|
|
return jsonify({
|
|
'configured': True,
|
|
'config': config
|
|
}), 200
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/config', methods=['POST'])
|
|
@jwt_required()
|
|
@admin_required
|
|
def configure_deployment(app_id):
|
|
"""Configure Git deployment for an app."""
|
|
app = Application.query.get(app_id)
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return jsonify({'error': 'No data provided'}), 400
|
|
|
|
if 'repo_url' not in data:
|
|
return jsonify({'error': 'repo_url is required'}), 400
|
|
|
|
result = GitService.configure_deployment(
|
|
app_id=app_id,
|
|
app_path=app.root_path,
|
|
repo_url=data['repo_url'],
|
|
branch=data.get('branch', 'main'),
|
|
auto_deploy=data.get('auto_deploy', True),
|
|
pre_deploy_script=data.get('pre_deploy_script'),
|
|
post_deploy_script=data.get('post_deploy_script')
|
|
)
|
|
|
|
return jsonify(result), 201 if result['success'] else 400
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/config', methods=['DELETE'])
|
|
@jwt_required()
|
|
@admin_required
|
|
def remove_deployment(app_id):
|
|
"""Remove deployment configuration."""
|
|
result = GitService.remove_deployment(app_id)
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/deploy', methods=['POST'])
|
|
@jwt_required()
|
|
def trigger_deploy(app_id):
|
|
"""Trigger a deployment."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app = Application.query.get(app_id)
|
|
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
|
|
if not ResourceGrantService.can_edit_app(user, app):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
data = request.get_json() or {}
|
|
result = GitService.deploy(app_id, force=data.get('force', False))
|
|
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/pull', methods=['POST'])
|
|
@jwt_required()
|
|
def pull_changes(app_id):
|
|
"""Pull latest changes without running deploy scripts."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app = Application.query.get(app_id)
|
|
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
|
|
if not ResourceGrantService.can_edit_app(user, app):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
data = request.get_json() or {}
|
|
result = GitService.pull_changes(app.root_path, data.get('branch'))
|
|
|
|
return jsonify(result), 200 if result['success'] else 400
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/git-status', methods=['GET'])
|
|
@jwt_required()
|
|
def get_git_status(app_id):
|
|
"""Get Git status for an app."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app = Application.query.get(app_id)
|
|
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
|
|
if not ResourceGrantService.can_access_app(user, app):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
status = GitService.get_git_status(app.root_path)
|
|
return jsonify(status), 200
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/commit', methods=['GET'])
|
|
@jwt_required()
|
|
def get_commit_info(app_id):
|
|
"""Get current commit info for an app."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app = Application.query.get(app_id)
|
|
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
|
|
if not ResourceGrantService.can_access_app(user, app):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
commit_info = GitService.get_commit_info(app.root_path)
|
|
if commit_info:
|
|
return jsonify({'commit': commit_info}), 200
|
|
return jsonify({'error': 'Not a Git repository'}), 404
|
|
|
|
|
|
@deploy_bp.route('/history', methods=['GET'])
|
|
@jwt_required()
|
|
def get_deployment_history():
|
|
"""Get deployment history. App-scoped queries require access to that app;
|
|
global queries (no app_id) are restricted to panel admins."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app_id = request.args.get('app_id', type=int)
|
|
limit = request.args.get('limit', 50, type=int)
|
|
|
|
if app_id:
|
|
app = Application.query.get(app_id)
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
if not user or app_access_tier(user, app) is None:
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
elif not user or not user.is_admin:
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
history = GitService.get_deployment_history(app_id, limit)
|
|
return jsonify({'deployments': history}), 200
|
|
|
|
|
|
@deploy_bp.route('/clone', methods=['POST'])
|
|
@jwt_required()
|
|
@admin_required
|
|
def clone_repository():
|
|
"""Clone a Git repository to a new location."""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return jsonify({'error': 'No data provided'}), 400
|
|
|
|
required = ['repo_url', 'app_path']
|
|
missing = [f for f in required if f not in data]
|
|
if missing:
|
|
return jsonify({'error': f'Missing required fields: {", ".join(missing)}'}), 400
|
|
|
|
# GHSA-8vx6-432p-h62q: app_path is attacker-controlled here (unlike
|
|
# /apps/from-repository, which derives it server-side). Confine clones to
|
|
# the managed apps root, mirroring _assert_managed_app_path.
|
|
base_dir = os.path.abspath(paths.APPS_DIR)
|
|
app_path = os.path.abspath(data['app_path'])
|
|
if app_path == base_dir or not app_path.startswith(base_dir + os.sep):
|
|
return jsonify({'error': 'app_path must be inside the managed apps directory'}), 400
|
|
|
|
result = GitService.clone_repository(
|
|
app_path=app_path,
|
|
repo_url=data['repo_url'],
|
|
branch=data.get('branch', 'main')
|
|
)
|
|
|
|
return jsonify(result), 201 if result['success'] else 400
|
|
|
|
|
|
@deploy_bp.route('/apps/<int:app_id>/branches', methods=['GET'])
|
|
@jwt_required()
|
|
def get_branches(app_id):
|
|
"""Get list of remote branches for an app's repository."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app = Application.query.get(app_id)
|
|
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
|
|
if not ResourceGrantService.can_access_app(user, app):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
result = GitService.get_remote_branches(app.root_path)
|
|
return jsonify(result), 200 if result.get('success') else 400
|
|
|
|
|
|
@deploy_bp.route('/branches', methods=['POST'])
|
|
@jwt_required()
|
|
def get_branches_from_url():
|
|
"""Get list of branches from a repository URL (before cloning)."""
|
|
data = request.get_json()
|
|
if not data or 'repo_url' not in data:
|
|
return jsonify({'error': 'repo_url is required'}), 400
|
|
|
|
result = GitService.get_remote_branches_from_url(data['repo_url'])
|
|
return jsonify(result), 200 if result.get('success') else 400
|
|
|
|
|
|
@deploy_bp.route('/webhook-logs', methods=['GET'])
|
|
@jwt_required()
|
|
def get_webhook_logs():
|
|
"""Get webhook logs for debugging. App-scoped queries require access to that
|
|
app; global queries (no app_id) are restricted to panel admins."""
|
|
current_user_id = get_jwt_identity()
|
|
user = User.query.get(current_user_id)
|
|
app_id = request.args.get('app_id', type=int)
|
|
limit = request.args.get('limit', 50, type=int)
|
|
|
|
if app_id:
|
|
app = Application.query.get(app_id)
|
|
if not app:
|
|
return jsonify({'error': 'Application not found'}), 404
|
|
if not user or app_access_tier(user, app) is None:
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
elif not user or not user.is_admin:
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
logs = GitService.get_webhook_logs(app_id, limit)
|
|
return jsonify({'logs': logs}), 200
|
|
|
|
|
|
# Webhook endpoint (no auth required)
|
|
@deploy_bp.route('/webhook/<int:app_id>/<token>', methods=['POST'])
|
|
def webhook(app_id, token):
|
|
"""Handle incoming webhook from Git provider."""
|
|
# Detect provider based on headers
|
|
provider = 'github' # default
|
|
signature = None
|
|
|
|
if request.headers.get('X-Gitlab-Token'):
|
|
provider = 'gitlab'
|
|
signature = request.headers.get('X-Gitlab-Token')
|
|
elif request.headers.get('X-Hub-Signature-256'):
|
|
# Could be GitHub or Bitbucket
|
|
if request.headers.get('X-Bitbucket-Type'):
|
|
provider = 'bitbucket'
|
|
else:
|
|
provider = 'github'
|
|
signature = request.headers.get('X-Hub-Signature-256')
|
|
elif request.headers.get('X-Hub-Signature'):
|
|
# Bitbucket uses X-Hub-Signature
|
|
provider = 'bitbucket'
|
|
signature = request.headers.get('X-Hub-Signature')
|
|
|
|
# Log webhook for debugging
|
|
GitService.log_webhook(app_id, provider, request.headers.to_wsgi_list(), request.data)
|
|
|
|
# Verify webhook
|
|
if not GitService.verify_webhook(app_id, token, signature, request.data, provider):
|
|
return jsonify({'error': 'Invalid webhook'}), 403
|
|
|
|
# Parse payload
|
|
try:
|
|
payload = request.get_json()
|
|
except Exception:
|
|
return jsonify({'error': 'Invalid payload'}), 400
|
|
|
|
# Handle webhook
|
|
result = GitService.handle_webhook(app_id, payload)
|
|
|
|
return jsonify(result), 200 if result.get('success') else 400
|