Hotovo. Do detailu plánovaného skriptu jsem přidal sekci Obsah skriptu.

Implementováno v app/routes/scheduled_scripts.py:
bezpečné načtení pouze z /opt/appfactory/workspace/appfactory-tools/maintenance/{script_name},
validace script_name: .sh, bez /, \, .., neprázdné,
zobrazení obsahu v textarea,
pokud soubor neexistuje, zobrazí se hláška a výchozí obsah:bash

#!/usr/bin/env bash
set -euo pipefail

echo "TODO"

admin může obsah uložit přes POST /portal/scheduled-scripts/{id}/script,
neadmin vidí read-only editor,
obsah nesmí být prázdný ani větší než 100 KB,
pokud chybí shebang #!/usr/bin/env bash, automaticky se doplní,
po uložení se nastaví chmod 755,
audit event scheduled_script.file_updated s scheduled_script_id a script_name.
Tlačítko Spustit nyní zůstává na detailu dostupné.
This commit is contained in:
JiriUhlir
2026-06-03 12:33:37 +02:00
parent f7a8007af8
commit 64f8fe806d
+150
View File
@@ -1,4 +1,6 @@
import html import html
import os
from pathlib import Path
from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.responses import HTMLResponse, RedirectResponse
@@ -18,6 +20,13 @@ from app.routes.deployments import render_status_pill
from app.templates.layout import page from app.templates.layout import page
router = APIRouter() router = APIRouter()
MAINTENANCE_DIR = Path("/opt/appfactory/workspace/appfactory-tools/maintenance")
MAX_SCRIPT_BYTES = 100 * 1024
DEFAULT_SCRIPT_CONTENT = """#!/usr/bin/env bash
set -euo pipefail
echo "TODO"
"""
SCHEDULE_TYPES = ("hourly", "daily", "weekly", "monthly") SCHEDULE_TYPES = ("hourly", "daily", "weekly", "monthly")
SCHEDULE_LABELS = { SCHEDULE_LABELS = {
"hourly": "každou hodinu", "hourly": "každou hodinu",
@@ -31,6 +40,10 @@ def clean_optional(value: str | None) -> str:
return (value or "").strip() return (value or "").strip()
def is_admin(user) -> bool:
return (user.get("role") or "").lower() == "admin"
def render_bool(value) -> str: def render_bool(value) -> str:
return "Ano" if value else "Ne" return "Ano" if value else "Ne"
@@ -71,6 +84,64 @@ def validate_script_name(value: str) -> str:
return script_name return script_name
def script_path(script_name: str) -> Path:
safe_name = validate_script_name(script_name)
path = (MAINTENANCE_DIR / safe_name).resolve()
base = MAINTENANCE_DIR.resolve()
try:
path.relative_to(base)
except ValueError:
raise HTTPException(status_code=400, detail="Neplatný název skriptu")
return path
def read_script_file(script_name: str) -> dict:
path = script_path(script_name)
if not path.exists():
return {
"exists": False,
"content": DEFAULT_SCRIPT_CONTENT,
"error": None,
"warning": "Soubor zatím neexistuje. Můžete ho vytvořit z výchozího obsahu.",
}
if not path.is_file():
return {"exists": False, "content": "", "error": "Cesta není soubor.", "warning": None}
try:
if path.stat().st_size > MAX_SCRIPT_BYTES:
return {"exists": True, "content": "", "error": "Soubor je větší než 100 KB.", "warning": None}
content = path.read_text(encoding="utf-8")
except Exception:
return {"exists": True, "content": "", "error": "Soubor se nepodařilo načíst.", "warning": None}
warning = None
if not content.startswith("#!/usr/bin/env bash"):
warning = "První řádek by měl být #!/usr/bin/env bash."
return {"exists": True, "content": content, "error": None, "warning": warning}
def normalize_script_content(content: str) -> str:
value = content.replace("\r\n", "\n").replace("\r", "\n")
if not value.strip():
raise HTTPException(status_code=400, detail="Obsah skriptu nesmí být prázdný")
if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
raise HTTPException(status_code=400, detail="Obsah skriptu je větší než 100 KB")
if not value.startswith("#!/usr/bin/env bash"):
value = "#!/usr/bin/env bash\n" + value.lstrip("\n")
if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
raise HTTPException(status_code=400, detail="Obsah skriptu je větší než 100 KB")
return value
def save_script_file(script_name: str, content: str) -> None:
path = script_path(script_name)
value = normalize_script_content(content)
try:
path.write_text(value, encoding="utf-8", newline="\n")
os.chmod(path, 0o755)
except Exception:
raise HTTPException(status_code=500, detail="Soubor se nepodařilo uložit")
def validate_schedule_type(value: str) -> str: def validate_schedule_type(value: str) -> str:
schedule_type = clean_optional(value) schedule_type = clean_optional(value)
if schedule_type not in SCHEDULE_TYPES: if schedule_type not in SCHEDULE_TYPES:
@@ -158,6 +229,56 @@ def render_form(script: dict | None, action: str, include_script_name: bool) ->
""" """
def render_script_content_section(script: dict, user: dict) -> str:
script_name = script.get("script_name", "") or ""
script_name_html = html.escape(script_name)
try:
script_file = read_script_file(script_name)
except HTTPException:
script_file = {
"exists": False,
"content": "",
"error": "Název skriptu není bezpečný.",
"warning": None,
}
content = html.escape(script_file.get("content", "") or "")
warning = ""
if script_file.get("warning"):
warning = f'<p class="alert">{script_file["warning"]}</p>'
error = ""
if script_file.get("error"):
error = f'<p class="alert alert-danger">{script_file["error"]}</p>'
if not is_admin(user):
readonly_hint = '<p class="muted">Obsah skriptu je dostupn&yacute; pouze pro &ccaron;ten&iacute;. Ukl&aacute;dat m&uring;&zcaron;e jen administr&aacute;tor.</p>'
return f"""
<div class="card">
<h2>Obsah skriptu</h2>
<p><strong>Soubor:</strong> {script_name_html}</p>
{warning}
{error}
{readonly_hint}
<textarea rows="18" readonly>{content}</textarea>
</div>
"""
return f"""
<div class="card">
<h2>Obsah skriptu</h2>
<p><strong>Soubor:</strong> {script_name_html}</p>
{warning}
{error}
<form method="post" action="/portal/scheduled-scripts/{html.escape(str(script.get("id")))}/script" class="metadata-form">
<label>Obsah souboru</label>
<textarea name="content" rows="22" spellcheck="false">{content}</textarea>
<div class="form-actions">
<button type="submit">Ulo&zcaron;it skript</button>
</div>
</form>
</div>
"""
@router.get("/scheduled-scripts", response_class=HTMLResponse) @router.get("/scheduled-scripts", response_class=HTMLResponse)
def scheduled_scripts_page(request: Request, user=Depends(require_user)): def scheduled_scripts_page(request: Request, user=Depends(require_user)):
scripts = get_scheduled_scripts() scripts = get_scheduled_scripts()
@@ -303,6 +424,7 @@ def scheduled_script_detail(script_id: int, request: Request, user=Depends(requi
<button type="submit" class="danger">Smazat</button> <button type="submit" class="danger">Smazat</button>
</form> </form>
""" """
script_content_section = render_script_content_section(script, user)
return page( return page(
html.escape(script.get("name", "") or "Pl&aacute;novan&yacute; skript"), html.escape(script.get("name", "") or "Pl&aacute;novan&yacute; skript"),
@@ -342,6 +464,8 @@ def scheduled_script_detail(script_id: int, request: Request, user=Depends(requi
<tr><th>Posledn&iacute; &uacute;loha</th><td>{last_job}</td></tr> <tr><th>Posledn&iacute; &uacute;loha</th><td>{last_job}</td></tr>
</table> </table>
</div> </div>
{script_content_section}
""", """,
user=user, user=user,
) )
@@ -420,6 +544,32 @@ def run_scheduled_script_now(script_id: int, user=Depends(require_user)):
return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303) return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303)
@router.post("/scheduled-scripts/{script_id}/script")
def update_scheduled_script_file(
script_id: int,
content: str = Form(...),
user=Depends(require_user),
):
if not is_admin(user):
raise HTTPException(status_code=403, detail="Skript m&uring;&zcaron;e upravit pouze administr&aacute;tor")
script = get_scheduled_script(script_id)
if not script:
raise HTTPException(status_code=404, detail="Scheduled script not found")
script_name = validate_script_name(script.get("script_name", "") or "")
save_script_file(script_name, content)
log_audit_event(
user,
action="scheduled_script.file_updated",
target_type="scheduled_script",
target_id=script_id,
metadata={
"scheduled_script_id": script_id,
"script_name": script_name,
},
)
return RedirectResponse(url=f"/portal/scheduled-scripts/{script_id}", status_code=303)
@router.post("/scheduled-scripts/{script_id}/toggle") @router.post("/scheduled-scripts/{script_id}/toggle")
def toggle_scheduled_script(script_id: int, user=Depends(require_user)): def toggle_scheduled_script(script_id: int, user=Depends(require_user)):
script = get_scheduled_script(script_id) script = get_scheduled_script(script_id)