mirror of
https://github.com/usmannasir/cyberpanel.git
synced 2025-11-06 13:25:51 +01:00
344 lines
12 KiB
Python
344 lines
12 KiB
Python
import json
|
|
import docker
|
|
from django.http import HttpResponse
|
|
from .models import DockerSites
|
|
from loginSystem.models import Administrator
|
|
from plogical.acl import ACLManager
|
|
from django.shortcuts import redirect
|
|
from loginSystem.views import loadLoginPage
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from plogical.CyberCPLogFileWriter import CyberCPLogFileWriter as logging
|
|
import requests
|
|
import re
|
|
|
|
def require_login(view_func):
|
|
def wrapper(request, *args, **kwargs):
|
|
try:
|
|
userID = request.session['userID']
|
|
return view_func(request, *args, **kwargs)
|
|
except KeyError:
|
|
return redirect(loadLoginPage)
|
|
return wrapper
|
|
|
|
class DockerManager:
|
|
def __init__(self):
|
|
self.client = docker.from_env()
|
|
|
|
def get_container(self, container_id):
|
|
try:
|
|
return self.client.containers.get(container_id)
|
|
except docker.errors.NotFound:
|
|
return None
|
|
except Exception as e:
|
|
logging.writeToFile(f"Error getting container {container_id}: {str(e)}")
|
|
return None
|
|
|
|
def get_n8n_version(self, container):
|
|
try:
|
|
# Execute npm list command in the container to get n8n version
|
|
exec_result = container.exec_run(
|
|
cmd="npm list n8n --json",
|
|
workdir="/usr/local/lib/node_modules/n8n"
|
|
)
|
|
if exec_result.exit_code == 0:
|
|
npm_output = json.loads(exec_result.output.decode())
|
|
# Extract version from npm output
|
|
if 'dependencies' in npm_output and 'n8n' in npm_output['dependencies']:
|
|
return npm_output['dependencies']['n8n']['version']
|
|
return None
|
|
except Exception as e:
|
|
logging.writeToFile(f"Error getting n8n version: {str(e)}")
|
|
return None
|
|
|
|
def get_latest_n8n_version(self):
|
|
try:
|
|
# Fetch latest version from npm registry
|
|
response = requests.get('https://registry.npmjs.org/n8n/latest')
|
|
if response.status_code == 200:
|
|
return response.json()['version']
|
|
return None
|
|
except Exception as e:
|
|
logging.writeToFile(f"Error fetching latest n8n version: {str(e)}")
|
|
return None
|
|
|
|
@csrf_exempt
|
|
@require_login
|
|
def startContainer(request):
|
|
try:
|
|
if request.method == 'POST':
|
|
userID = request.session['userID']
|
|
currentACL = ACLManager.loadedACL(userID)
|
|
admin = Administrator.objects.get(pk=userID)
|
|
|
|
data = json.loads(request.body)
|
|
container_id = data.get('container_id')
|
|
site_name = data.get('name')
|
|
|
|
# Verify Docker site ownership
|
|
try:
|
|
docker_site = DockerSites.objects.get(SiteName=site_name)
|
|
if currentACL['admin'] != 1 and docker_site.admin != admin and docker_site.admin.owner != admin.pk:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Not authorized to access this container'
|
|
}))
|
|
except DockerSites.DoesNotExist:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Docker site not found'
|
|
}))
|
|
|
|
docker_manager = DockerManager()
|
|
container = docker_manager.get_container(container_id)
|
|
|
|
if not container:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Container not found'
|
|
}))
|
|
|
|
container.start()
|
|
return HttpResponse(json.dumps({'status': 1}))
|
|
|
|
return HttpResponse('Not allowed')
|
|
except Exception as e:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': str(e)
|
|
}))
|
|
|
|
@csrf_exempt
|
|
@require_login
|
|
def stopContainer(request):
|
|
try:
|
|
if request.method == 'POST':
|
|
userID = request.session['userID']
|
|
currentACL = ACLManager.loadedACL(userID)
|
|
admin = Administrator.objects.get(pk=userID)
|
|
|
|
data = json.loads(request.body)
|
|
container_id = data.get('container_id')
|
|
site_name = data.get('name')
|
|
|
|
# Verify Docker site ownership
|
|
try:
|
|
docker_site = DockerSites.objects.get(SiteName=site_name)
|
|
if currentACL['admin'] != 1 and docker_site.admin != admin and docker_site.admin.owner != admin.pk:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Not authorized to access this container'
|
|
}))
|
|
except DockerSites.DoesNotExist:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Docker site not found'
|
|
}))
|
|
|
|
docker_manager = DockerManager()
|
|
container = docker_manager.get_container(container_id)
|
|
|
|
if not container:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Container not found'
|
|
}))
|
|
|
|
container.stop()
|
|
return HttpResponse(json.dumps({'status': 1}))
|
|
|
|
return HttpResponse('Not allowed')
|
|
except Exception as e:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': str(e)
|
|
}))
|
|
|
|
@csrf_exempt
|
|
@require_login
|
|
def restartContainer(request):
|
|
try:
|
|
if request.method == 'POST':
|
|
userID = request.session['userID']
|
|
currentACL = ACLManager.loadedACL(userID)
|
|
admin = Administrator.objects.get(pk=userID)
|
|
|
|
data = json.loads(request.body)
|
|
container_id = data.get('container_id')
|
|
site_name = data.get('name')
|
|
|
|
# Verify Docker site ownership
|
|
try:
|
|
docker_site = DockerSites.objects.get(SiteName=site_name)
|
|
if currentACL['admin'] != 1 and docker_site.admin != admin and docker_site.admin.owner != admin.pk:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Not authorized to access this container'
|
|
}))
|
|
except DockerSites.DoesNotExist:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Docker site not found'
|
|
}))
|
|
|
|
docker_manager = DockerManager()
|
|
container = docker_manager.get_container(container_id)
|
|
|
|
if not container:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Container not found'
|
|
}))
|
|
|
|
container.restart()
|
|
return HttpResponse(json.dumps({'status': 1}))
|
|
|
|
return HttpResponse('Not allowed')
|
|
except Exception as e:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': str(e)
|
|
}))
|
|
|
|
@csrf_exempt
|
|
@require_login
|
|
def check_n8n_version(request):
|
|
try:
|
|
if request.method == 'POST':
|
|
userID = request.session['userID']
|
|
currentACL = ACLManager.loadedACL(userID)
|
|
admin = Administrator.objects.get(pk=userID)
|
|
|
|
data = json.loads(request.body)
|
|
container_id = data.get('container_id')
|
|
site_name = data.get('name')
|
|
|
|
# Verify Docker site ownership
|
|
try:
|
|
docker_site = DockerSites.objects.get(SiteName=site_name)
|
|
if currentACL['admin'] != 1 and docker_site.admin != admin and docker_site.admin.owner != admin.pk:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Not authorized to access this container'
|
|
}))
|
|
except DockerSites.DoesNotExist:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Docker site not found'
|
|
}))
|
|
|
|
docker_manager = DockerManager()
|
|
container = docker_manager.get_container(container_id)
|
|
|
|
if not container:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Container not found'
|
|
}))
|
|
|
|
# Get current version
|
|
current_version = docker_manager.get_n8n_version(container)
|
|
if not current_version:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Could not determine current n8n version'
|
|
}))
|
|
|
|
# Get latest version
|
|
latest_version = docker_manager.get_latest_n8n_version()
|
|
if not latest_version:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Could not fetch latest n8n version'
|
|
}))
|
|
|
|
# Compare versions
|
|
update_available = current_version != latest_version
|
|
|
|
return HttpResponse(json.dumps({
|
|
'status': 1,
|
|
'current_version': current_version,
|
|
'latest_version': latest_version,
|
|
'update_available': update_available
|
|
}))
|
|
|
|
return HttpResponse('Not allowed')
|
|
except Exception as e:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': str(e)
|
|
}))
|
|
|
|
@csrf_exempt
|
|
@require_login
|
|
def update_n8n(request):
|
|
try:
|
|
if request.method == 'POST':
|
|
userID = request.session['userID']
|
|
currentACL = ACLManager.loadedACL(userID)
|
|
admin = Administrator.objects.get(pk=userID)
|
|
|
|
data = json.loads(request.body)
|
|
container_id = data.get('container_id')
|
|
site_name = data.get('name')
|
|
|
|
# Verify Docker site ownership
|
|
try:
|
|
docker_site = DockerSites.objects.get(SiteName=site_name)
|
|
if currentACL['admin'] != 1 and docker_site.admin != admin and docker_site.admin.owner != admin.pk:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Not authorized to access this container'
|
|
}))
|
|
except DockerSites.DoesNotExist:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Docker site not found'
|
|
}))
|
|
|
|
docker_manager = DockerManager()
|
|
container = docker_manager.get_container(container_id)
|
|
|
|
if not container:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': 'Container not found'
|
|
}))
|
|
|
|
# Update n8n
|
|
try:
|
|
# Run npm update in the container
|
|
exec_result = container.exec_run(
|
|
cmd="npm install -g n8n@latest",
|
|
workdir="/usr/local/lib/node_modules/n8n"
|
|
)
|
|
|
|
if exec_result.exit_code != 0:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': f'Update failed: {exec_result.output.decode()}'
|
|
}))
|
|
|
|
# Get new version after update
|
|
new_version = docker_manager.get_n8n_version(container)
|
|
|
|
# Restart the container to apply changes
|
|
container.restart()
|
|
|
|
return HttpResponse(json.dumps({
|
|
'status': 1,
|
|
'message': 'n8n updated successfully',
|
|
'new_version': new_version
|
|
}))
|
|
|
|
except Exception as e:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': f'Update failed: {str(e)}'
|
|
}))
|
|
|
|
return HttpResponse('Not allowed')
|
|
except Exception as e:
|
|
return HttpResponse(json.dumps({
|
|
'status': 0,
|
|
'error_message': str(e)
|
|
})) |