admin/ServerKit
1
0 Fork 0
mirror of https://github.com/jhd3197/ServerKit.git synced 2026-11-09 05:24:49 +09:00
Code Issues Packages Projects Releases Wiki Activity
Files
b006f2e4dabbbf5f4a796090bcfa415ce6337d21
ServerKit/backend/app/api/git.py
T
Juan DenisandClaude Opus 5 5f8a3ef204 feat(recycle-bin): applications are soft-deletable, and a delete stops keeping data hostage 
Plan 70. Domain went through this and TEN read paths were wrong, four with
effects outside the panel. Application has 211 call sites across 68 files and
drives more -- container lifecycles, nginx, backups, registry pulls, the OS
crontab -- so it was audited in full first (docs/plans/70_...md). 180 sites are
now query_active(); the 35 left unfiltered are each commented in place.

**The delete / purge split is what makes the tombstone honest.** delete_app used
to run compose_down(volumes=True) and rmtree(). Keeping a tombstone AND doing
that would put a Restore button next to a row whose data the delete had already
destroyed -- the one promise a recycle bin cannot break. So:

  DELETE   containers down (VOLUMES KEPT), vhost removed, firewall closed,
           cron suspended, row tombstoned. All reversible.
  RESTORE  vhost rewritten, firewall reopened, cron resumed. Containers left
           STOPPED -- "put it back" must not silently start processes or open
           ports, the same reasoning that stops domain restore re-issuing certs.
  PURGE    volumes removed, uploaded source deleted, cron detached.

register() gained on_purge for that, wired into BOTH purge paths. And
purge_expired had NO scheduler -- fine when a tombstone was just a row, not fine
when it holds data volumes and a source tree, so a daily builtin job now runs
it. Without that, "kept until purge" would have meant kept forever and delete
would have quietly stopped reclaiming disk.

A side effect worth naming: the old _remove_data_flag protection for database
engines is now the DEFAULT for every app, and the flag is carried into
?purge=true so an explicit skip-the-bin delete still honours it.

**Cron was the blocker with no mechanical answer.** before_delete never fires on
a soft delete, and clear_application is LOSSY -- it nulls the association
without recording it, so nothing could put it back. suspend_for_application
instead disables only the jobs that were CURRENTLY ENABLED and stamps each
suspended_by_app; resume re-enables exactly those. That flag is the whole point:
without it a restore switches back on jobs the user deliberately turned off.
clear_application survives as purge-only, where lossiness is correct.

**The indirect reach (audit 3.1) is the class that grep cannot find.**
ContainerSleepPolicy, BackupPolicy, DeploymentJob and WafPolicy are enumerated
on a SCHEDULE and resolve application_id afterwards, so they never appear in a
search for Application.query. Left alone, a tombstoned app's sleep policy is
swept forever and wake_app restarts the containers the delete tore down. Those
loops test Application.deleted_ids().

Deliberately NOT scoped: reservation checks (port, name, directory) -- a
tombstone must keep holding those or a new app takes them and the restore is
refused; api/git.py's webhook VISIBILITY gate, which falls through to
"panel-wide visible" so hiding a tombstone would WIDEN access; and cron
attribution, which only labels a row.

Verified: 15 new tests, the migration replayed on a fresh DB (a tombstone frees
its private_slug, two live rows still collide), and a full delete -> restore
round trip. The 20 remaining full-suite failures are the known
serverkit-wordpress-not-installed class and are identical on a clean tree.

Two traps caught by testing, both recorded in the plan: a bad import made the
ENTIRE purge cleanup silently no-op while purge() still reported success
(_run_purge_hook swallows by design, so assert on the warning, not just ok), and
private_slug had to lose unique=True at the MODEL level or a models-built
database burns the slug regardless of the migration.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-12 20:26:04 -04:00

519 lines
16 KiB
Python
Raw Blame History

# 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
# Deliberately unfiltered: this gate falls through to "panel-wide visible"
# when the app can't be resolved, so hiding tombstones here would WIDEN
# access — deleting an app would publish its webhook to every viewer.
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."""
from app.models import GitDeployment
deployment = GitDeployment.query.get(deployment_id)
if deployment is None:
return jsonify({'success': False, 'error': 'Deployment not found'}), 404
app = Application.query_active().filter_by(id=deployment.app_id).first()
if app is None or app_access_tier(get_current_user(), app) is None:
return jsonify({'success': False, 'error': 'Access denied'}), 403
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
webhook = GitWebhook.query.get(webhook_id)
if webhook is None:
return jsonify({'success': False, 'error': 'Webhook not found'}), 404
if not _webhook_visible_to(get_current_user(), webhook):
return jsonify({'success': False, 'error': 'Access denied'}), 403
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
Reference in new issue View Git Blame Copy Permalink