mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-07 04:23:23 +09:00
Juan DenisandClaude Opus 4.8
d9b04077d4
Many per-app API route files lost their plan-29 gating (the `# Bucket:` header + require_app_member/admin_required decorators). Restored so the authz structural sweep + hardening matrix pass again, using the recovered rbac primitives (app_access_tier / require_app_member). - env_vars: can_access_app (read) / can_edit_app (write) across handlers - apps sleep/scale policy, container_status, snapshots, git deployments/webhooks (visibility-scoped), databases raw-docker routes (admin), status-page checks - wordpress: sealed reads + P0 destructive sealed-write + convergence reads + standalone host-ops admin-gated - bucket headers on the already-gated app_volumes/deploy/domains/private_urls/ python/security/templates test_authz_structural_sweep 2, test_api_authz_hardening 27, test_git_webhook_authz 5 — green; 430+ route regression tests still green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
500 lines
15 KiB
Python
500 lines
15 KiB
Python
# Bucket: PER-APP (plan 29 #9, #11). The per-app deployment read gates on app
|
|
# visibility (require_app_member); webhook reads are scoped to the linked app's
|
|
# visibility (an unlinked webhook stays panel-wide viewer-visible). Webhook
|
|
# mutations and the Gitea host lifecycle stay admin-only.
|
|
"""Git Server API endpoints for managing integrated Gitea instance and webhooks."""
|
|
|
|
from flask import Blueprint, request, jsonify
|
|
|
|
from ..middleware.rbac import (
|
|
admin_required, viewer_required, require_app_member,
|
|
get_current_user, app_access_tier,
|
|
)
|
|
from ..models import Application, GitWebhook
|
|
from ..services.git_service import GitService
|
|
from ..services.webhook_service import WebhookService
|
|
from ..services.gitea_api_service import GiteaAPIService
|
|
from ..services.git_deploy_service import GitDeployService
|
|
|
|
git_bp = Blueprint('git', __name__)
|
|
|
|
|
|
def _webhook_visible_to(user, webhook):
|
|
"""A webhook with no app linkage is a panel-wide viewer-visible resource; an
|
|
app-linked webhook is visible only to callers who can reach that app
|
|
(owner/admin/grant/workspace member — plan 29 #11)."""
|
|
if not getattr(webhook, 'app_id', None):
|
|
return True
|
|
app = Application.query.get(webhook.app_id)
|
|
if app is None:
|
|
return True
|
|
return app_access_tier(user, app) is not None
|
|
|
|
|
|
@git_bp.route('/status', methods=['GET'])
|
|
@viewer_required
|
|
def get_status():
|
|
"""Get Gitea installation status."""
|
|
result = GitService.get_gitea_status()
|
|
return jsonify(result), 200
|
|
|
|
|
|
@git_bp.route('/requirements', methods=['GET'])
|
|
@viewer_required
|
|
def get_requirements():
|
|
"""Get resource requirements for Gitea installation."""
|
|
result = GitService.get_gitea_resource_requirements()
|
|
return jsonify(result), 200
|
|
|
|
|
|
@git_bp.route('/install', methods=['POST'])
|
|
@admin_required
|
|
def install():
|
|
"""Install Gitea with PostgreSQL."""
|
|
data = request.get_json() or {}
|
|
|
|
result = GitService.install_gitea(
|
|
admin_user=data.get('adminUser', 'admin'),
|
|
admin_email=data.get('adminEmail'),
|
|
admin_password=data.get('adminPassword')
|
|
)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 201
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/uninstall', methods=['POST'])
|
|
@admin_required
|
|
def uninstall():
|
|
"""Uninstall Gitea and optionally remove data."""
|
|
data = request.get_json() or {}
|
|
|
|
result = GitService.uninstall_gitea(
|
|
remove_data=data.get('removeData', False)
|
|
)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/start', methods=['POST'])
|
|
@admin_required
|
|
def start():
|
|
"""Start Gitea server."""
|
|
result = GitService.start_gitea()
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/stop', methods=['POST'])
|
|
@admin_required
|
|
def stop():
|
|
"""Stop Gitea server."""
|
|
result = GitService.stop_gitea()
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/restart', methods=['POST'])
|
|
@admin_required
|
|
def restart():
|
|
"""Restart Gitea server."""
|
|
result = GitService.restart_gitea()
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
# ==================== WEBHOOK ENDPOINTS ====================
|
|
|
|
@git_bp.route('/webhooks', methods=['GET'])
|
|
@viewer_required
|
|
def list_webhooks():
|
|
"""List webhooks the caller can see (plan 29 #11): every unlinked webhook
|
|
plus any app-linked webhook whose app they can reach."""
|
|
user = get_current_user()
|
|
result = WebhookService.list_webhooks()
|
|
webhooks = result.get('webhooks', [])
|
|
visible = []
|
|
for w in webhooks:
|
|
wh = GitWebhook.query.get(w.get('id'))
|
|
if wh is None or _webhook_visible_to(user, wh):
|
|
visible.append(w)
|
|
result['webhooks'] = visible
|
|
if 'count' in result:
|
|
result['count'] = len(visible)
|
|
return jsonify(result), 200
|
|
|
|
|
|
@git_bp.route('/webhooks', methods=['POST'])
|
|
@admin_required
|
|
def create_webhook():
|
|
"""Create a new webhook."""
|
|
data = request.get_json() or {}
|
|
|
|
result = WebhookService.create_webhook(
|
|
name=data.get('name'),
|
|
source=data.get('source'),
|
|
source_repo_url=data.get('sourceRepoUrl'),
|
|
source_branch=data.get('sourceBranch', 'main'),
|
|
local_repo_name=data.get('localRepoName'),
|
|
sync_direction=data.get('syncDirection', 'pull'),
|
|
auto_sync=data.get('autoSync', True),
|
|
app_id=data.get('appId'),
|
|
deploy_on_push=data.get('deployOnPush', False),
|
|
pre_deploy_script=data.get('preDeployScript'),
|
|
post_deploy_script=data.get('postDeployScript'),
|
|
zero_downtime=data.get('zeroDowntime', False)
|
|
)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 201
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/webhooks/<int:webhook_id>', methods=['GET'])
|
|
@viewer_required
|
|
def get_webhook(webhook_id):
|
|
"""Get a specific webhook (scoped to the linked app's visibility, #11)."""
|
|
user = get_current_user()
|
|
wh = GitWebhook.query.get(webhook_id)
|
|
if wh is None or not _webhook_visible_to(user, wh):
|
|
return jsonify({'error': 'Not found'}), 404
|
|
|
|
result = WebhookService.get_webhook(webhook_id)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 404
|
|
|
|
|
|
@git_bp.route('/webhooks/<int:webhook_id>', methods=['PUT'])
|
|
@admin_required
|
|
def update_webhook(webhook_id):
|
|
"""Update a webhook."""
|
|
data = request.get_json() or {}
|
|
|
|
result = WebhookService.update_webhook(webhook_id, data)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/webhooks/<int:webhook_id>', methods=['DELETE'])
|
|
@admin_required
|
|
def delete_webhook(webhook_id):
|
|
"""Delete a webhook."""
|
|
result = WebhookService.delete_webhook(webhook_id)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/webhooks/<int:webhook_id>/toggle', methods=['POST'])
|
|
@admin_required
|
|
def toggle_webhook(webhook_id):
|
|
"""Toggle webhook active status."""
|
|
result = WebhookService.toggle_webhook(webhook_id)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/webhooks/<int:webhook_id>/logs', methods=['GET'])
|
|
@viewer_required
|
|
def get_webhook_logs(webhook_id):
|
|
"""Get logs for a specific webhook (scoped to the linked app's visibility)."""
|
|
user = get_current_user()
|
|
wh = GitWebhook.query.get(webhook_id)
|
|
if wh is None or not _webhook_visible_to(user, wh):
|
|
return jsonify({'error': 'Not found'}), 404
|
|
limit = request.args.get('limit', 50, type=int)
|
|
result = WebhookService.get_webhook_logs(webhook_id, limit=limit)
|
|
return jsonify(result), 200
|
|
|
|
|
|
@git_bp.route('/webhooks/<int:webhook_id>/test', methods=['POST'])
|
|
@admin_required
|
|
def test_webhook(webhook_id):
|
|
"""Test a webhook by triggering a manual sync."""
|
|
result = WebhookService.test_webhook(webhook_id)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
# Public endpoint - no auth required (verified by signature)
|
|
@git_bp.route('/webhooks/receive/<token>', methods=['POST'])
|
|
def receive_webhook(token):
|
|
"""Receive incoming webhook from GitHub/GitLab/Bitbucket."""
|
|
# Determine source from headers
|
|
if 'X-GitHub-Event' in request.headers:
|
|
source = 'github'
|
|
event_type = request.headers.get('X-GitHub-Event')
|
|
signature = request.headers.get('X-Hub-Signature-256')
|
|
delivery_id = request.headers.get('X-GitHub-Delivery')
|
|
elif 'X-Gitlab-Event' in request.headers:
|
|
source = 'gitlab'
|
|
event_type = request.headers.get('X-Gitlab-Event')
|
|
signature = request.headers.get('X-Gitlab-Token')
|
|
delivery_id = None
|
|
elif 'X-Event-Key' in request.headers:
|
|
source = 'bitbucket'
|
|
event_type = request.headers.get('X-Event-Key')
|
|
signature = request.headers.get('X-Hub-Signature')
|
|
delivery_id = request.headers.get('X-Request-Id')
|
|
else:
|
|
return jsonify({'error': 'Unknown webhook source'}), 400
|
|
|
|
result = WebhookService.handle_webhook(
|
|
token=token,
|
|
source=source,
|
|
event_type=event_type,
|
|
signature=signature,
|
|
delivery_id=delivery_id,
|
|
headers=dict(request.headers),
|
|
payload=request.get_data(),
|
|
payload_json=request.get_json(silent=True)
|
|
)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
# ==================== REPOSITORY ENDPOINTS ====================
|
|
|
|
@git_bp.route('/repos', methods=['GET'])
|
|
@viewer_required
|
|
def list_repositories():
|
|
"""List all repositories in Gitea."""
|
|
token = request.args.get('token')
|
|
limit = request.args.get('limit', 50, type=int)
|
|
|
|
result = GiteaAPIService.list_repositories(token=token, limit=limit)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>', methods=['GET'])
|
|
@viewer_required
|
|
def get_repository(owner, repo):
|
|
"""Get repository details."""
|
|
token = request.args.get('token')
|
|
|
|
result = GiteaAPIService.get_repository(owner, repo, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 404
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/stats', methods=['GET'])
|
|
@viewer_required
|
|
def get_repo_stats(owner, repo):
|
|
"""Get repository statistics."""
|
|
token = request.args.get('token')
|
|
|
|
result = GiteaAPIService.get_repo_stats(owner, repo, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/branches', methods=['GET'])
|
|
@viewer_required
|
|
def list_branches(owner, repo):
|
|
"""List repository branches."""
|
|
token = request.args.get('token')
|
|
|
|
result = GiteaAPIService.list_branches(owner, repo, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/branches/<branch>', methods=['GET'])
|
|
@viewer_required
|
|
def get_branch(owner, repo, branch):
|
|
"""Get branch details."""
|
|
token = request.args.get('token')
|
|
|
|
result = GiteaAPIService.get_branch(owner, repo, branch, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 404
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/commits', methods=['GET'])
|
|
@viewer_required
|
|
def list_commits(owner, repo):
|
|
"""List repository commits."""
|
|
token = request.args.get('token')
|
|
branch = request.args.get('branch')
|
|
page = request.args.get('page', 1, type=int)
|
|
limit = request.args.get('limit', 30, type=int)
|
|
|
|
result = GiteaAPIService.list_commits(
|
|
owner, repo,
|
|
branch=branch,
|
|
page=page,
|
|
limit=limit,
|
|
token=token
|
|
)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/commits/<sha>', methods=['GET'])
|
|
@viewer_required
|
|
def get_commit(owner, repo, sha):
|
|
"""Get commit details."""
|
|
token = request.args.get('token')
|
|
|
|
result = GiteaAPIService.get_commit(owner, repo, sha, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 404
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/contents', methods=['GET'])
|
|
@viewer_required
|
|
def list_files(owner, repo):
|
|
"""List files in repository directory."""
|
|
token = request.args.get('token')
|
|
ref = request.args.get('ref', 'main')
|
|
path = request.args.get('path', '')
|
|
|
|
result = GiteaAPIService.list_files(owner, repo, ref=ref, path=path, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/contents/<path:filepath>', methods=['GET'])
|
|
@viewer_required
|
|
def get_file_content(owner, repo, filepath):
|
|
"""Get file content."""
|
|
token = request.args.get('token')
|
|
ref = request.args.get('ref', 'main')
|
|
|
|
result = GiteaAPIService.get_file_content(owner, repo, filepath, ref=ref, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 404
|
|
|
|
|
|
@git_bp.route('/repos/<owner>/<repo>/readme', methods=['GET'])
|
|
@viewer_required
|
|
def get_readme(owner, repo):
|
|
"""Get repository README."""
|
|
token = request.args.get('token')
|
|
ref = request.args.get('ref')
|
|
|
|
result = GiteaAPIService.get_readme(owner, repo, ref=ref, token=token)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 404
|
|
|
|
|
|
@git_bp.route('/version', methods=['GET'])
|
|
@viewer_required
|
|
def get_gitea_version():
|
|
"""Get Gitea server version."""
|
|
result = GiteaAPIService.get_server_version()
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
# ==================== DEPLOYMENT ENDPOINTS ====================
|
|
|
|
@git_bp.route('/deployments/app/<int:app_id>', methods=['GET'])
|
|
@require_app_member()
|
|
def get_app_deployments(app_id):
|
|
"""Get deployment history for an application."""
|
|
limit = request.args.get('limit', 20, type=int)
|
|
result = GitDeployService.get_deployments(app_id, limit=limit)
|
|
return jsonify(result), 200
|
|
|
|
|
|
@git_bp.route('/deployments/<int:deployment_id>', methods=['GET'])
|
|
@viewer_required
|
|
def get_deployment(deployment_id):
|
|
"""Get a specific deployment with logs."""
|
|
include_logs = request.args.get('logs', 'false').lower() == 'true'
|
|
result = GitDeployService.get_deployment(deployment_id, include_logs=include_logs)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 404
|
|
|
|
|
|
@git_bp.route('/deployments/app/<int:app_id>/deploy', methods=['POST'])
|
|
@admin_required
|
|
def manual_deploy(app_id):
|
|
"""Trigger a manual deployment for an application."""
|
|
data = request.get_json() or {}
|
|
branch = data.get('branch')
|
|
|
|
result = GitDeployService.manual_deploy(app_id, branch=branch)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/deployments/app/<int:app_id>/rollback', methods=['POST'])
|
|
@admin_required
|
|
def rollback_deployment(app_id):
|
|
"""Rollback to a previous deployment version."""
|
|
data = request.get_json() or {}
|
|
target_version = data.get('targetVersion')
|
|
|
|
result = GitDeployService.rollback(app_id, target_version=target_version)
|
|
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
return jsonify(result), 400
|
|
|
|
|
|
@git_bp.route('/deployments/webhook/<int:webhook_id>', methods=['GET'])
|
|
@viewer_required
|
|
def get_webhook_deployments(webhook_id):
|
|
"""Get deployments triggered by a specific webhook."""
|
|
from app.models import GitDeployment
|
|
|
|
limit = request.args.get('limit', 20, type=int)
|
|
deployments = GitDeployment.query.filter_by(webhook_id=webhook_id)\
|
|
.order_by(GitDeployment.created_at.desc())\
|
|
.limit(limit).all()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'deployments': [d.to_dict() for d in deployments],
|
|
'count': len(deployments)
|
|
}), 200
|