github
/
jumpserver
mirror of https://github.com/jumpserver/jumpserver
1
0
Fork 0
Code Issues Releases Wiki Activity

fix: Add type check for secure command execution

pull/14739/head
wangruidong 2024-12-23 11:18:24 +08:00 committed by Bryan
parent 7e6d2749ae
commit 35f88722af
1 changed files with 5 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
apps/ops/tasks.py
View File

@ -11,6 +11,7 @@ from django.conf import settings
from common.const.crontab import CRONTAB_AT_AM_TWO
from common.utils import get_logger, get_object_or_none, get_log_keep_day
from ops.celery import app
from ops.const import Types
from ops.serializers.job import JobExecutionSerializer
from orgs.utils import tmp_to_org, tmp_to_root_org
from .celery.decorator import (
@ -57,14 +58,13 @@ def _run_ops_job_execution(execution):
)
)
def run_ops_job(job_id):
if not settings.SECURITY_COMMAND_EXECUTION:
return
with tmp_to_root_org():
job = get_object_or_none(Job, id=job_id)
if not job:
logger.error("Did not get the execution: {}".format(job_id))
return
if not settings.SECURITY_COMMAND_EXECUTION and job.type != Types.upload_file:
return
with tmp_to_org(job.org):
execution = job.create_execution()
execution.creator = job.creator
@ -92,14 +92,13 @@ def job_execution_task_activity_callback(self, execution_id, *args, **kwargs):
)
)
def run_ops_job_execution(execution_id, **kwargs):
if not settings.SECURITY_COMMAND_EXECUTION:
return
with tmp_to_root_org():
execution = get_object_or_none(JobExecution, id=execution_id)
if not execution:
logger.error("Did not get the execution: {}".format(execution_id))
return
if not settings.SECURITY_COMMAND_EXECUTION and execution.job.type != Types.upload_file:
return
_run_ops_job_execution(execution)