diff --git a/app/routes/scheduled_scripts.py b/app/routes/scheduled_scripts.py index c7e7bfd..3026a69 100644 --- a/app/routes/scheduled_scripts.py +++ b/app/routes/scheduled_scripts.py @@ -1,4 +1,6 @@ import html +import os +from pathlib import Path from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import HTMLResponse, RedirectResponse @@ -18,6 +20,13 @@ from app.routes.deployments import render_status_pill from app.templates.layout import page 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_LABELS = { "hourly": "každou hodinu", @@ -31,6 +40,10 @@ def clean_optional(value: str | None) -> str: return (value or "").strip() +def is_admin(user) -> bool: + return (user.get("role") or "").lower() == "admin" + + def render_bool(value) -> str: return "Ano" if value else "Ne" @@ -71,6 +84,64 @@ def validate_script_name(value: str) -> str: 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: schedule_type = clean_optional(value) 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'
{script_file["warning"]}
' + error = "" + if script_file.get("error"): + error = f'{script_file["error"]}
' + + if not is_admin(user): + readonly_hint = 'Obsah skriptu je dostupný pouze pro čtení. Ukládat může jen administrátor.
' + return f""" +Soubor: {script_name_html}
+ {warning} + {error} + {readonly_hint} + +Soubor: {script_name_html}
+ {warning} + {error} + +