fix: restrict script execution endpoints to analyst/admin roles

deploy, run, and run-deployed require analyst; undeploy requires admin.
Update test to use admin token for undeploy.
This commit is contained in:
ZdenekSrotyr 2026-04-09 16:31:42 +02:00
parent 449053bf8a
commit 2043594670
2 changed files with 13 additions and 11 deletions

View file

@ -13,7 +13,8 @@ from typing import Optional, List
import duckdb
from app.auth.dependencies import get_current_user, require_role, Role, _get_db
from app.auth.dependencies import get_current_user, require_role, _get_db
from src.rbac import Role
from src.repositories.notifications import ScriptRepository
router = APIRouter(prefix="/api/scripts", tags=["scripts"])
@ -53,7 +54,7 @@ async def list_scripts(
@router.post("/deploy", status_code=201)
async def deploy_script(
request: DeployScriptRequest,
user: dict = Depends(get_current_user),
user: dict = Depends(require_role(Role.ANALYST)),
conn: duckdb.DuckDBPyConnection = Depends(_get_db),
):
"""Deploy a Python script to be run on the server (optionally on schedule)."""
@ -75,7 +76,7 @@ async def deploy_script(
@router.post("/{script_id}/run")
async def run_deployed_script(
script_id: str,
user: dict = Depends(get_current_user),
user: dict = Depends(require_role(Role.ANALYST)),
conn: duckdb.DuckDBPyConnection = Depends(_get_db),
):
"""Run a deployed script by ID."""
@ -89,7 +90,7 @@ async def run_deployed_script(
@router.post("/run")
async def run_adhoc_script(
request: RunScriptRequest,
user: dict = Depends(get_current_user),
user: dict = Depends(require_role(Role.ANALYST)),
):
"""Run an ad-hoc Python script (not deployed)."""
if not request.source:
@ -100,7 +101,7 @@ async def run_adhoc_script(
@router.delete("/{script_id}", status_code=204)
async def undeploy_script(
script_id: str,
user: dict = Depends(get_current_user),
user: dict = Depends(require_role(Role.ADMIN)),
conn: duckdb.DuckDBPyConnection = Depends(_get_db),
):
repo = ScriptRepository(conn)

View file

@ -79,22 +79,23 @@ class TestScriptsAPI:
assert "disallowed" in detail or "Blocked" in detail
def test_deploy_run_undeploy(self, client):
c, _, analyst_token = client
headers = {"Authorization": f"Bearer {analyst_token}"}
c, admin_token, analyst_token = client
analyst_headers = {"Authorization": f"Bearer {analyst_token}"}
admin_headers = {"Authorization": f"Bearer {admin_token}"}
# Deploy
resp = c.post("/api/scripts/deploy", json={
"name": "calc", "source": "print(2+2)", "schedule": "0 8 * * MON",
}, headers=headers)
}, headers=analyst_headers)
script_id = resp.json()["id"]
# Run
resp = c.post(f"/api/scripts/{script_id}/run", headers=headers)
resp = c.post(f"/api/scripts/{script_id}/run", headers=analyst_headers)
assert resp.status_code == 200
assert "4" in resp.json()["stdout"]
# Undeploy
resp = c.delete(f"/api/scripts/{script_id}", headers=headers)
# Undeploy (requires admin)
resp = c.delete(f"/api/scripts/{script_id}", headers=admin_headers)
assert resp.status_code == 204