import html import math import re from urllib.parse import quote, urlencode from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request from fastapi.responses import HTMLResponse, RedirectResponse from ..auth import is_admin, is_developer, require_admin, require_developer, require_user from ..config import ( DEFAULT_GITEA_ORG, DELETE_APP_SCRIPT_NAME, DEPLOY_SCRIPT, GENERATE_COMPOSE_SCRIPT, get_appfactory_host, get_gitea_public_url, is_allowed_worker_script, read_env_value, ) from ..db.apps import ( create_app_ip_access_rule, create_app_variable, delete_app_ip_access_rule, delete_app_variable, get_app, get_app_ip_access_rule, get_app_ip_access_rules, get_app_template, get_app_deployments, get_app_templates, get_app_variables, get_apps, update_app_ip_access_rule, update_app_metadata, update_app_resources, upsert_created_app, update_app_variable, ) from ..db.audit import log_audit_event from ..db.health import get_latest_service_health, get_service_health, get_service_health_history from ..db.incidents import get_service_incidents from ..db.jobs import create_job, get_jobs, has_active_deploy_job from ..environment import AppEnvironmentError, apply_all_app_environments, apply_app_environment, validate_environment_key from ..repo_docs import RepoDocsError, is_markdown_path, list_markdown_files, read_markdown, render_markdown from ..routes.deployments import render_status_pill from ..routes.incidents import render_incident_history_rows from ..shell import run_command from ..templates.layout import page, render_result router = APIRouter() DEFAULT_PAGE_SIZE = 20 # Selectable method presets for IP access rules. WRITE = POST,PUT,PATCH,DELETE; ALL = every # common HTTP method including GET. The Caddy generator will later interpret these values. IP_ACCESS_METHOD_OPTIONS = ( "WRITE", "ALL", "GET", "POST", "PUT", "PATCH", "DELETE", "GET,POST", "POST,PATCH,DELETE", ) METADATA_FIELDS = ( "name", "description", "owner", "template", "runtime", "repository_url", "repository_name", "default_branch", "domain", "health_url", "container_port", "is_public", "is_enabled", ) # ID služby se používá jako Docker container/image name, Compose service name, Gitea repo, # adresář workspace a část URL/domény. Nejpřísnější je Docker image name (jen lowercase) # a DNS label (max 63 znaků, alfanumerické na začátku i konci). Proto povolujeme pouze # malá písmena a-z, číslice a pomlčky uvnitř, bez diakritiky, podtržítek a teček. APP_ID_MAX_LENGTH = 63 APP_ID_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$") def validate_new_app_id(app_id: str) -> str | None: """Vrátí chybovou hlášku, pokud ID služby nesplňuje omezení, jinak None.""" if not app_id: return "ID služby nesmí být prázdné." if app_id != app_id.lower(): return "ID služby musí být malými písmeny (lowercase)." if len(app_id) > APP_ID_MAX_LENGTH: return f"ID služby může mít nejvýše {APP_ID_MAX_LENGTH} znaků (limit DNS/Docker)." if not APP_ID_RE.match(app_id): return ( "Neplatné ID služby. Povolena jsou jen malá písmena a-z, číslice a pomlčky; " "musí začínat i končit písmenem nebo číslicí (např. gmail-service). " "Bez diakritiky, mezer, podtržítek a teček — ID se používá jako název " "Docker kontejneru, image a Gitea repozitáře." ) return None def clean_optional(value: str | None) -> str: return (value or "").strip() def parse_optional_int(value: str | None) -> int | None: value = clean_optional(value) if not value: return None try: return int(value) except ValueError: raise HTTPException(status_code=400, detail="Neplatne cislo") def bool_checked(value) -> str: return " checked" if value else "" def app_detail_url(app_id: str, anchor: str = "", message: str = "", error: str = "") -> str: params = {} if message: params["message"] = message if error: params["error"] = error url = f"/portal/apps/{quote(app_id, safe='')}" if params: url += f"?{urlencode(params)}" if anchor: url += f"#{anchor}" return url def redirect_app_detail(app_id: str, anchor: str = "", message: str = "", error: str = "") -> RedirectResponse: return RedirectResponse(url=app_detail_url(app_id, anchor=anchor, message=message, error=error), status_code=303) def apply_environment_message(app_id: str) -> str: result = apply_app_environment(app_id) if not result.get("changed"): return "Environment unchanged." if result.get("restarted"): return "Environment applied and container restarted." return "Environment file regenerated. Container was not running." def apply_all_environments_message() -> str: results = apply_all_app_environments() changed = sum(1 for result in results.values() if result.get("changed")) restarted = sum(1 for result in results.values() if result.get("restarted")) return f"Regenerated environments for {len(results)} apps. Changed: {changed}. Restarted: {restarted}." def ensure_variable_key_allowed(app_id: str, key: str, variable_id: int | None = None) -> None: try: validate_environment_key(key) except AppEnvironmentError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc for variable in get_app_variables(app_id): if variable_id is not None and int(variable.get("id")) == variable_id: continue if (variable.get("key") or "").strip() == key: raise HTTPException(status_code=400, detail="Variable key already exists") def render_template_options(templates: list[dict], selected_template: str, include_blank: bool = True) -> str: options = [''] if include_blank else [] selected_exists = not selected_template for template in templates: template_id = template.get("id", "") or "" name = template.get("name", "") or "" runtime = template.get("runtime", "") or "" description = template.get("description", "") or "" label_parts = [name] if runtime: label_parts.append(runtime) if description: label_parts.append(description) selected = " selected" if selected_template == template_id else "" if selected: selected_exists = True options.append( f'' ) if not selected_exists: options.append( f'' ) return "".join(options) def render_template_label(templates: list[dict], template_id: str) -> str: if not template_id: return "" for template in templates: if template.get("id") == template_id: return html.escape(template.get("name", "") or template_id) return "Neznámá šablona" def render_environment_usage_example(language: str, runtime: str, template: str) -> str: normalized = " ".join((language or "", runtime or "", template or "")).lower() title = "Použití v kódu" if "fastapi" in normalized or "python" in normalized: code = """import os value = os.environ["MY_VARIABLE"] optional_value = os.getenv("OPTIONAL_VARIABLE", "default")""" description = "Python / FastAPI čte Variables i Secrets ze systémového prostředí." elif "dotnet" in normalized or ".net" in normalized or "csharp" in normalized or "c#" in normalized: code = """var value = Environment.GetEnvironmentVariable("MY_VARIABLE"); var optionalValue = builder.Configuration["OPTIONAL_VARIABLE"] ?? "default";""" description = ".NET čte Variables i Secrets z environment variables, případně přes Configuration." elif "node" in normalized or "javascript" in normalized or "typescript" in normalized: code = """const value = process.env.MY_VARIABLE; const optionalValue = process.env.OPTIONAL_VARIABLE ?? "default";""" description = "Node.js čte Variables i Secrets přes process.env." elif "php" in normalized: code = """$value = getenv('MY_VARIABLE'); $optionalValue = getenv('OPTIONAL_VARIABLE') ?: 'default';""" description = "PHP čte Variables i Secrets ze systémového prostředí." elif "java" in normalized or "spring" in normalized: code = """String value = System.getenv("MY_VARIABLE"); String optionalValue = System.getenv().getOrDefault("OPTIONAL_VARIABLE", "default");""" description = "Java čte Variables i Secrets přes System.getenv." else: code = """MY_VARIABLE is available as an environment variable inside the container.""" description = "Variables i Secrets jsou dostupné jako environment variables v kontejneru." return f"""

{html.escape(title)}

{html.escape(description)}

{html.escape(code)}
""" def diff_metadata(before: dict, after: dict) -> dict: changes = {} for field in METADATA_FIELDS: old_value = before.get(field) new_value = after.get(field) if field in ("is_public", "is_enabled"): changed = bool(old_value) != bool(new_value) else: changed = str(old_value or "") != str(new_value or "") if changed: changes[field] = {"old": old_value, "new": new_value} return changes def render_health_status(status: str | None) -> str: value = status or "" normalized = value.lower() labels = { "healthy": "zdravá", "unhealthy": "nezdravá", "unreachable": "nedostupná", } class_name = "pill pill-muted" if normalized == "healthy": class_name = "pill pill-success" elif normalized == "unhealthy": class_name = "pill pill-warning" elif normalized == "unreachable": class_name = "pill pill-danger" return f'{html.escape(labels.get(normalized, value or "neznámá"))}' def render_health_dot(status: str | None, checked_at: str | None) -> str: normalized = (status or "").lower() labels = { "healthy": "zdravá", "unhealthy": "nezdravá", "unreachable": "nedostupná", } class_name = "health-dot health-dot-muted" if normalized == "healthy": class_name = "health-dot health-dot-success" elif normalized == "unhealthy": class_name = "health-dot health-dot-warning" elif normalized == "unreachable": class_name = "health-dot health-dot-danger" label = labels.get(normalized, "zdraví neznámé") title = label if checked_at: title = f"{label}, poslední kontrola: {checked_at}" return f'' def render_section(title: str, body: str, section_id: str = "", is_open: bool = False) -> str: """Sbalitelná karta detailu služby (defaultně zavřená, ať detail nezabírá tolik místa). Title je hotové HTML (obvykle ikona + text), body celý obsah sekce. Odkaz s kotvou na zavřenou sekci ji otevře až portal.js (openHashSection) - proto má sekce vlastní id. """ if not body.strip(): return "" id_attr = f' id="{section_id}"' if section_id else "" open_attr = " open" if is_open else "" return f"""

{title}

{body}
""" @router.get("/") def portal_home(user=Depends(require_user)): return RedirectResponse(url="/portal/apps", status_code=303) @router.get("/apps", response_class=HTMLResponse) def apps_page( request: Request, q: str = Query(""), status: str = Query(""), message: str = Query(""), error: str = Query(""), page_number: int = Query(1, alias="page", ge=1), user=Depends(require_user), ): apps = get_apps() latest_health = get_latest_service_health() query = q.strip() selected_status = status.strip() if query: apps = [ item for item in apps if query.lower() in (item.get("id", "") or "").lower() or query.lower() in (item.get("name", "") or "").lower() ] if selected_status: apps = [item for item in apps if (item.get("status", "") or "") == selected_status] sort = request.query_params.get("sort", "").strip() if sort == "health": order = {"unreachable": 0, "unhealthy": 1, "healthy": 2} apps = sorted( apps, key=lambda item: ( order.get((latest_health.get(item.get("id", "")) or {}).get("status"), 3), item.get("id", ""), ), ) status_values = sorted({item.get("status", "") for item in get_apps() if item.get("status")}) total_apps = len(apps) total_pages = max(1, math.ceil(total_apps / DEFAULT_PAGE_SIZE)) if page_number > total_pages: page_number = total_pages offset = (page_number - 1) * DEFAULT_PAGE_SIZE apps = apps[offset : offset + DEFAULT_PAGE_SIZE] gitea_url = get_gitea_public_url() gitea_org = read_env_value("GITEA_ORG", DEFAULT_GITEA_ORG) host = get_appfactory_host(request.url.hostname or "") # Role-based visibility: a viewer only sees which services run and their documentation. # Developers also get Git/clone and operational actions; only admins see the delete button. can_manage = is_developer(user) can_delete = is_admin(user) rows = "" for index, item in enumerate(apps, start=1): app_id = html.escape(item.get("id", "")) app_url_id = quote(item.get("id", ""), safe="") app_name = html.escape(item.get("name", "") or "") # Ve sloupci Služba zobrazujeme "Název (id)"; když název chybí nebo je shodný s ID, # zůstane jen ID bez zdvojení. service_label = f"{app_name} ({app_id})" if app_name and app_name != app_id else app_id resource_modal_id = f"resources-{page_number}-{index}" status = html.escape(item.get("status", "")) health = latest_health.get(item.get("id", "")) or {} health_checked_at = html.escape(health.get("checked_at", "") or "") health_dot = render_health_dot(health.get("status"), health.get("checked_at")) docs = html.escape(item.get("docs", f"/apps/{app_id}/docs")) memory = html.escape(item.get("memory", "") or "") cpus = html.escape(item.get("cpus", "") or "") memory_label = memory or "výchozí" cpus_label = cpus or "výchozí" http_clone = html.escape(f"git clone {gitea_url}/{gitea_org}/{app_id}.git") if gitea_url else "" ssh_clone = html.escape(f"git clone ssh://git@{host}:2222/{gitea_org}/{app_id}.git") if host else "" # Resources: developers/admins can edit; viewers see only the read-only summary. if can_manage: resource_edit = f"""

Paměť je limit RAM. CPU určuje maximální podíl výpočetního výkonu. Příklad: 0,50 = polovina jádra, 1,00 = celé jádro.

""" else: resource_edit = "" # Git clone commands are an operational concern — visible to developers/admins only. if can_manage and (http_clone or ssh_clone): git_cell = f"""
Příkazy pro klonování
""" else: git_cell = '' # Deploy actions for developers/admins; delete is admin-only. manage_icons = ( f"""
""" if can_manage else "" ) delete_icon = ( f"""
""" if can_delete else "" ) rows += f"""
{health_dot}
{service_label}
/apps/{app_id}
{status} Swagger
Paměť: {memory_label} CPU: {cpus_label} {resource_edit} {git_cell}
{manage_icons} {delete_icon}
""" if not rows: rows = 'Zatím nejsou nasazené žádné služby.' status_options = [''] for value in status_values: selected = " selected" if selected_status == value else "" escaped_value = html.escape(value) status_options.append(f'') first_item = offset + 1 if total_apps else 0 last_item = min(offset + len(apps), total_apps) def page_url(page: int) -> str: params = {"page": page} if query: params["q"] = query if selected_status: params["status"] = selected_status if sort: params["sort"] = sort return f"/portal/apps?{urlencode(params)}" previous_link = ( f' Předchozí' if page_number > 1 else "" ) next_link = ( f'Další ' if page_number < total_pages else "" ) pagination = "" if total_pages > 1: pagination = f""" """ notice = "" if message: notice = f'

{html.escape(message)}

' if error: notice = f'

{html.escape(error)}

' # Creating services and regenerating .env files are write actions — developers/admins only. manage_toolbar = ( """

Nová služba

""" if can_manage else "" ) return page( "Služby", f"""

Nasazené služby

{notice} {manage_toolbar}
Reset
{pagination} {rows}
Služba Status Dokumentace Prostředky Git Akce
{pagination}
""", user=user, ) @router.post("/apps/environment/apply-all") def apply_all_app_environments_action(user=Depends(require_developer)): try: message = apply_all_environments_message() except AppEnvironmentError as exc: return RedirectResponse(url="/portal/apps?error=" + quote(f"Environment apply failed: {exc}"), status_code=303) log_audit_event( user, action="app.environment.applied_all", target_type="app", metadata={"result": message}, ) return RedirectResponse(url="/portal/apps?message=" + quote(message), status_code=303) @router.get("/apps/{app_id}", response_class=HTMLResponse) def app_detail(app_id: str, request: Request, message: str = "", error: str = "", user=Depends(require_user)): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") # Viewers may open the detail (services + docs are read-only here); editing the service, # its variables or triggering a redeploy is reserved for developers/admins. can_manage = is_developer(user) log_audit_event( user, action="service.health.view", target_type="service", target_id=app_id, ) escaped_app_id = html.escape(app.get("id", "")) app_url_id = quote(app.get("id", ""), safe="") name = html.escape(app.get("name", "") or "") language_raw = app.get("language", "") or "" runtime_raw = app.get("runtime", "") or "" template_raw = app.get("template", "") or "" language = html.escape(language_raw) version = html.escape(app.get("version", "") or "") status = html.escape(app.get("status", "") or "") memory = html.escape(app.get("memory", "") or "") cpus = html.escape(app.get("cpus", "") or "") updated_at = html.escape(app.get("updated_at", "") or "") description = html.escape(app.get("description", "") or "") owner = html.escape(app.get("owner", "") or "") runtime = html.escape(runtime_raw) repository_url = html.escape(app.get("repository_url", "") or "") repository_name = html.escape(app.get("repository_name", "") or "") default_branch = html.escape(app.get("default_branch", "") or "") domain = html.escape(app.get("domain", "") or "") health_url = html.escape(app.get("health_url", "") or "") container_port = html.escape(str(app.get("container_port") or "")) is_public = bool(app.get("is_public")) is_enabled = bool(app.get("is_enabled")) templates = get_app_templates() template_options = render_template_options(templates, app.get("template", "") or "") template_value = render_template_label(templates, app.get("template", "") or "") environment_usage_example = render_environment_usage_example(language_raw, runtime_raw, template_raw) variables = get_app_variables(app.get("id", "")) ip_access_rules = get_app_ip_access_rules(app.get("id", "")) incidents = get_service_incidents(app.get("id", ""), limit=20) current_health = get_service_health(app.get("id", "")) health_history = get_service_health_history(app.get("id", ""), limit=50) rows = "" for deployment in get_app_deployments(app.get("id", ""), limit=10): deployment_id = html.escape(str(deployment.get("id", ""))) started_at = html.escape(deployment.get("started_at", "") or "") triggered_by = html.escape( deployment.get("triggered_by_display_name") or deployment.get("triggered_by_username") or "" ) rows += f""" #{deployment_id} {render_status_pill(deployment.get("status"))} {started_at} {triggered_by} """ if not rows: rows = 'Zatím nejsou evidovaná žádná nasazení této služby.' job_rows = "" for job in get_jobs(limit=100, target=app.get("id", "")): if job.get("target_type") != "app" or job.get("target_id") != app.get("id", ""): continue job_id = html.escape(str(job.get("id", ""))) job_rows += f""" #{job_id} {render_status_pill(job.get("status"))} {html.escape(job.get("type", "") or "")} {html.escape(job.get("created_at", "") or "")} """ if not job_rows: job_rows = 'Zatím nejsou evidované žádné úlohy této služby.' health_status = render_health_status(current_health.get("status") if current_health else None) health_http_status = html.escape(str(current_health.get("http_status") or "")) if current_health else "" health_response_time = html.escape(str(current_health.get("response_time_ms") or "")) if current_health else "" health_checked_at = html.escape(current_health.get("checked_at", "") or "") if current_health else "" health_rows = "" for item in health_history: error_text = item.get("error_text", "") or "" error_preview = error_text if len(error_text) <= 140 else f"{error_text[:137]}..." health_rows += f""" {html.escape(item.get("checked_at", "") or "")} {render_health_status(item.get("status"))} {html.escape(str(item.get("http_status") or ""))} {html.escape(str(item.get("response_time_ms") or ""))} {html.escape(error_preview)} """ if not health_rows: health_rows = 'Zatím nejsou evidované žádné kontroly zdraví.' variable_rows = "" for variable in variables: variable_id = html.escape(str(variable.get("id", ""))) variable_key = html.escape(variable.get("key", "") or "") variable_value_raw = variable.get("value", "") or "" variable_is_secret = bool(variable.get("is_secret")) variable_value = "---" if variable_is_secret else html.escape(variable_value_raw) value_input = "" if variable_is_secret else html.escape(variable_value_raw) secret_checked = bool_checked(variable_is_secret) secret_hint = ' placeholder="---"' if variable_is_secret else "" variable_rows += f""" {variable_key} {variable_value} {"Ano" if variable_is_secret else "Ne"}
""" if not variable_rows: variable_rows = 'Zatím nejsou evidované žádné proměnné.' def render_method_options(selected: str) -> str: selected_upper = (selected or "").upper() options = list(IP_ACCESS_METHOD_OPTIONS) if selected_upper and selected_upper not in options: options.append(selected_upper) out = "" for opt in options: is_sel = " selected" if opt == selected_upper else "" out += f'' return out # IP access rules — inputs in each column are associated with the per-row update form via the # HTML5 form="..." attribute, so the whole row is editable inline while staying valid markup. ip_access_rule_rows = "" for rule in ip_access_rules: rid = html.escape(str(rule.get("id", ""))) form_id = f"iprule-{rid}" rule_ip = html.escape(rule.get("ip_cidr", "") or "", quote=True) rule_desc = html.escape(rule.get("description", "") or "", quote=True) rule_enabled = bool(rule.get("is_enabled", 1)) row_class = "" if rule_enabled else " rule-disabled" enabled_checked = bool_checked(rule_enabled) ip_label = html.escape(rule.get("ip_cidr", "") or "") # Pravidla navázaná na katalog IP adres drží IP/CIDR i popis podle katalogu, aby se obě # místa nerozešla – tady jsou proto jen ke čtení a mění se v centrální správě IP Access. address_id = rule.get("address_id") if address_id: address_label = html.escape(rule.get("address_label") or rule.get("description") or "", quote=True) source_cell = ( f'{address_label}' ) ip_cell = f'' description_cell = f'' else: source_cell = 'ručně' ip_cell = f'' description_cell = f'' ip_access_rule_rows += f""" {ip_cell} {description_cell} {source_cell}
""" if not ip_access_rule_rows: ip_access_rule_rows = 'Zatím nejsou evidovaná žádná IP access pravidla.' notice = "" if message: notice = f'

{html.escape(message)}

' if error: notice = f'

{html.escape(error)}

' # Editable sections are developer/admin only. Viewers keep the read-only Souhrn / Zdrav\u00ed / # Historie cards below, which already present the service metadata without write access. redeploy_block = ( f"""
""" if can_manage else "" ) # Odkazy jen otevřou a najedou na sekci - všechny sekce jsou defaultně sbalené. manage_tabs = ( """ Metadata Proměnné Security / IP Access """ if can_manage else "" ) detail_tabs = f"""
Dokumentace {manage_tabs} Historie
""" # Seznam *.md souborů se tahá z Gitea, proto se načítá až po rozbalení sekce (portal.js) - # nedostupná Gitea tak nikdy nezdrží vykreslení detailu. docs_card = render_section( ' Dokumentace', f"""

Swagger / OpenAPI Interaktivní dokumentace API běžící služby.

Soubory *.md v repozitáři

Seznam souborů se načte po rozbalení sekce.

""", section_id="dokumentace", ) metadata_card = ( render_section( ' Metadata', f"""
""", section_id="metadata", ) if can_manage else "" ) variables_card = ( render_section( ' Proměnné', f"""
{environment_usage_example} {variable_rows}
Key Value Secret Akce

Přidat proměnnou

""", section_id="promenne", ) if can_manage else "" ) ip_access_card = ( render_section( ' Security / IP Access', f"""

Pravidla určují, z jakých IP/CIDR adres lze volat dané HTTP metody. WRITE = POST, PUT, PATCH, DELETE; ALL = všechny běžné metody včetně GET. Pravidla se zatím pouze ukládají (Caddy je začne vynucovat později).

Hromadné přiřazení IP adres Pojmenované adresy z katalogu (sloupec Zdroj) se spravují centrálně — tam jednou adresou obsloužíte i deset služeb.

{ip_access_rule_rows}
IP/CIDR Methods Description Zdroj Enabled Actions

Přidat pravidlo

""", section_id="ip-access", ) if can_manage else "" ) summary_card = render_section( ' Souhrn', f"""
ID{escaped_app_id}
N\u00e1zev{name}
Popis{description}
Vlastn\u00edk{owner}
Template{template_value}
Runtime{runtime}
Repository URL{repository_url}
Repository Name{repository_name}
Default Branch{default_branch}
Domain{domain}
Health URL{health_url}
Container Port{container_port}
Ve\u0159ejn\u00e1 slu\u017eba{"Ano" if is_public else "Ne"}
Aktivn\u00ed slu\u017eba{"Ano" if is_enabled else "Ne"}
Jazyk{language}
Verze{version}
Status{status}
Pam\u011b\u0165{memory}
CPU{cpus}
Upraveno{updated_at}
LogyZobrazit \u00falohy a logy
""", section_id="souhrn", ) health_card = render_section( ' Zdrav\u00ed slu\u017eby', f"""
Aktu\u00e1ln\u00ed status{health_status}
HTTP status{health_http_status}
Odezva{health_response_time}
Posledn\u00ed kontrola{health_checked_at}
""", section_id="zdravi", ) health_history_card = render_section( ' Historie kontrol', f""" {health_rows}
\u010cas Status HTTP status Odezva Chyba
""", section_id="historie", ) incidents_card = render_section( ' Incidenty slu\u017eby', f""" {render_incident_history_rows(incidents, include_service=False)}
N\u00e1zev Za\u010d\u00e1tek Konec Trv\u00e1n\u00ed Stav
""", section_id="incidenty", ) deployments_card = render_section( ' Historie nasazen\u00ed', f""" {rows}
ID Status \u010cas Spustil
""", section_id="nasazeni", ) jobs_card = render_section( ' Historie \u00faloh', f""" {job_rows}
ID Status Typ Vytvo\u0159eno
""", section_id="ulohy", ) return page( "Detail slu\u017eby", f"""

{escaped_app_id}

{name}

{notice}

← Zpět na služby Nasazení služby Úlohy služby

{redeploy_block}
{detail_tabs} {docs_card} {metadata_card} {variables_card} {ip_access_card} {summary_card} {health_card} {health_history_card} {incidents_card} {deployments_card} {jobs_card} """, user=user, ) def render_doc_size(size: int) -> str: if size < 1024: return f"{size} B" return f"{round(size / 1024)} kB" @router.get("/apps/{app_id}/docs-files", response_class=HTMLResponse) def app_docs_files(app_id: str, user=Depends(require_user)): """HTML fragment se seznamem *.md souborů služby (načítá ho portal.js po rozbalení sekce).""" app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") repo = (app.get("repository_name") or app_id).strip() branch = (app.get("default_branch") or "").strip() try: files = list_markdown_files(repo, branch) except RepoDocsError as exc: return HTMLResponse(f'

Dokumentaci se nepodařilo načíst: {html.escape(str(exc))}

') if not files: return HTMLResponse('

V repozitáři služby nejsou žádné soubory *.md.

') items = "" for item in files: doc_path = item["path"] doc_query = urlencode({"path": doc_path}) items += f"""
{html.escape(doc_path)} {render_doc_size(item["size"])}

Obsah se načte po rozbalení.

""" return HTMLResponse(items) @router.get("/apps/{app_id}/docs-file", response_class=HTMLResponse) def app_docs_file(app_id: str, path: str = Query(""), user=Depends(require_user)): """HTML fragment s obsahem jednoho *.md souboru z repozitáře služby.""" app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") if not is_markdown_path(path): raise HTTPException(status_code=400, detail="Neplatná cesta k souboru dokumentace") repo = (app.get("repository_name") or app_id).strip() branch = (app.get("default_branch") or "").strip() try: text = read_markdown(repo, path, branch) except RepoDocsError as exc: return HTMLResponse(f'

Soubor se nepodařilo načíst: {html.escape(str(exc))}

') return HTMLResponse(render_markdown(text, repo)) @router.post("/apps/{app_id}/metadata") def save_app_metadata( app_id: str, name: str = Form(...), description: str = Form(""), owner: str = Form(""), template: str = Form(""), runtime: str = Form(""), repository_url: str = Form(""), repository_name: str = Form(""), default_branch: str = Form(""), domain: str = Form(""), health_url: str = Form(""), container_port: str = Form(""), is_public: str | None = Form(None), is_enabled: str | None = Form(None), user=Depends(require_developer), ): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") metadata = { "name": clean_optional(name), "description": clean_optional(description), "owner": clean_optional(owner), "template": clean_optional(template), "runtime": clean_optional(runtime), "repository_url": clean_optional(repository_url), "repository_name": clean_optional(repository_name), "default_branch": clean_optional(default_branch), "domain": clean_optional(domain), "health_url": clean_optional(health_url), "container_port": parse_optional_int(container_port), "is_public": bool(is_public), "is_enabled": bool(is_enabled), } changes = diff_metadata(app, metadata) update_app_metadata(app_id, metadata) if changes: log_audit_event( user, action="app.metadata.updated", target_type="app", target_id=app_id, metadata={"changes": changes}, ) return RedirectResponse(url=f"/portal/apps/{quote(app_id, safe='')}#metadata", status_code=303) @router.post("/apps/{app_id}/variables") def add_app_variable( app_id: str, key: str = Form(...), value: str = Form(""), is_secret: str | None = Form(None), user=Depends(require_developer), ): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") variable_key = clean_optional(key) if not variable_key: raise HTTPException(status_code=400, detail="Key is required") ensure_variable_key_allowed(app_id, variable_key) create_app_variable(app_id, variable_key, value, bool(is_secret)) log_audit_event( user, action="app.variable.created", target_type="app", target_id=app_id, metadata={"key": variable_key, "is_secret": bool(is_secret)}, ) try: message = apply_environment_message(app_id) except AppEnvironmentError as exc: return redirect_app_detail(app_id, anchor="promenne", error=f"Variable saved, but environment apply failed: {exc}") return redirect_app_detail(app_id, anchor="promenne", message=message) @router.post("/apps/{app_id}/variables/{variable_id}/update") def save_app_variable( app_id: str, variable_id: int, key: str = Form(...), value: str = Form(""), is_secret: str | None = Form(None), user=Depends(require_developer), ): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") existing = next((item for item in get_app_variables(app_id) if int(item.get("id")) == variable_id), None) if not existing: raise HTTPException(status_code=404, detail="Variable not found") variable_key = clean_optional(key) if not variable_key: raise HTTPException(status_code=400, detail="Key is required") ensure_variable_key_allowed(app_id, variable_key, variable_id=variable_id) stored_value = None if existing.get("is_secret") and bool(is_secret) and value == "" else value update_app_variable(variable_id, app_id, variable_key, stored_value, bool(is_secret)) log_audit_event( user, action="app.variable.updated", target_type="app", target_id=app_id, metadata={"key": variable_key, "is_secret": bool(is_secret)}, ) try: message = apply_environment_message(app_id) except AppEnvironmentError as exc: return redirect_app_detail(app_id, anchor="promenne", error=f"Variable saved, but environment apply failed: {exc}") return redirect_app_detail(app_id, anchor="promenne", message=message) @router.post("/apps/{app_id}/variables/{variable_id}/delete") def remove_app_variable(app_id: str, variable_id: int, user=Depends(require_developer)): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") existing = next((item for item in get_app_variables(app_id) if int(item.get("id")) == variable_id), None) if not existing: raise HTTPException(status_code=404, detail="Variable not found") delete_app_variable(variable_id, app_id) log_audit_event( user, action="app.variable.deleted", target_type="app", target_id=app_id, metadata={"key": existing.get("key"), "is_secret": bool(existing.get("is_secret"))}, ) try: message = apply_environment_message(app_id) except AppEnvironmentError as exc: return redirect_app_detail(app_id, anchor="promenne", error=f"Variable deleted, but environment apply failed: {exc}") return redirect_app_detail(app_id, anchor="promenne", message=message) @router.post("/apps/{app_id}/environment/apply") def apply_app_environment_action(app_id: str, user=Depends(require_developer)): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") try: message = apply_environment_message(app_id) except AppEnvironmentError as exc: return redirect_app_detail(app_id, anchor="promenne", error=f"Environment apply failed: {exc}") log_audit_event( user, action="app.environment.applied", target_type="app", target_id=app_id, metadata={"result": message}, ) return redirect_app_detail(app_id, anchor="promenne", message=message) @router.post("/apps/{app_id}/ip-access-rules") def add_app_ip_access_rule( app_id: str, ip_cidr: str = Form(...), methods: str = Form("WRITE"), description: str = Form(""), is_enabled: str | None = Form(None), user=Depends(require_developer), ): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") rule_ip = clean_optional(ip_cidr) rule_methods = clean_optional(methods).upper() if not rule_ip: return redirect_app_detail(app_id, anchor="ip-access", error="IP/CIDR nesmí být prázdné.") if not rule_methods: return redirect_app_detail(app_id, anchor="ip-access", error="Methods nesmí být prázdné.") create_app_ip_access_rule(app_id, rule_ip, rule_methods, description.strip(), bool(is_enabled)) log_audit_event( user, action="app.ip_access_rule.created", target_type="app", target_id=app_id, metadata={"ip_cidr": rule_ip, "methods": rule_methods, "is_enabled": bool(is_enabled)}, ) return redirect_app_detail(app_id, anchor="ip-access", message="IP access pravidlo přidáno.") @router.post("/apps/{app_id}/ip-access-rules/{rule_id}/update") def save_app_ip_access_rule( app_id: str, rule_id: int, ip_cidr: str = Form(...), methods: str = Form("WRITE"), description: str = Form(""), is_enabled: str | None = Form(None), user=Depends(require_developer), ): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") existing = get_app_ip_access_rule(rule_id, app_id) if not existing: return redirect_app_detail(app_id, anchor="ip-access", error="IP access pravidlo nebylo nalezeno.") rule_ip = clean_optional(ip_cidr) rule_methods = clean_optional(methods).upper() rule_description = description.strip() if not rule_methods: return redirect_app_detail(app_id, anchor="ip-access", error="Methods nesmí být prázdné.") # Pravidla z katalogu IP adres si IP/CIDR ani popis nedrží sama – ty patří katalogu, jinak by # se obě místa rozešla. Editovatelné odtud zůstávají jen metody a zapnutí/vypnutí. if existing.get("address_id"): rule_ip = existing.get("ip_cidr") or "" rule_description = existing.get("description") or "" elif not rule_ip: return redirect_app_detail(app_id, anchor="ip-access", error="IP/CIDR nesmí být prázdné.") update_app_ip_access_rule(rule_id, app_id, rule_ip, rule_methods, rule_description, bool(is_enabled)) log_audit_event( user, action="app.ip_access_rule.updated", target_type="app", target_id=app_id, metadata={"rule_id": rule_id, "ip_cidr": rule_ip, "methods": rule_methods, "is_enabled": bool(is_enabled)}, ) return redirect_app_detail(app_id, anchor="ip-access", message="IP access pravidlo upraveno.") @router.post("/apps/{app_id}/ip-access-rules/{rule_id}/delete") def remove_app_ip_access_rule(app_id: str, rule_id: int, user=Depends(require_developer)): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") existing = get_app_ip_access_rule(rule_id, app_id) if not existing: return redirect_app_detail(app_id, anchor="ip-access", error="IP access pravidlo nebylo nalezeno.") delete_app_ip_access_rule(rule_id, app_id) log_audit_event( user, action="app.ip_access_rule.deleted", target_type="app", target_id=app_id, metadata={"rule_id": rule_id, "ip_cidr": existing.get("ip_cidr"), "methods": existing.get("methods")}, ) return redirect_app_detail(app_id, anchor="ip-access", message="IP access pravidlo smazáno.") @router.post("/apps/{app_id}/redeploy") def redeploy_app(app_id: str, user=Depends(require_developer)): app = get_app(app_id) if not app: raise HTTPException(status_code=404, detail="App not found") try: apply_environment_message(app_id) except AppEnvironmentError as exc: return redirect_app_detail(app_id, error=f"Environment apply failed: {exc}") active_job = has_active_deploy_job("app", app_id) if active_job: return RedirectResponse(url=f"/portal/jobs/{active_job['id']}", status_code=303) job_id = create_job( job_type="deploy_app", target_type="app", target_id=app_id, payload={}, user=user, source="portal", ) log_audit_event( user, action="app.redeploy.queued", target_type="app", target_id=app_id, metadata={"job_id": job_id}, ) return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303) @router.get("/new-app", response_class=HTMLResponse) def new_app_form(request: Request, user=Depends(require_developer)): template_options = render_template_options(get_app_templates(create_enabled=True), "", include_blank=False) return page( "Nová služba", f"""

Vytvořit novou službu

Vytvoří Gitea repozitář, webhook, lokální workspace, první commit a nasadí službu.



Jen malá písmena, číslice a pomlčky (např. gmail-service). ID se použije jako název Docker kontejneru, Gitea repozitáře a v URL — později ho nelze změnit.



← Zpět

""", user=user, ) @router.post("/new-app", response_class=HTMLResponse) def create_app( app_id: str = Form(...), app_name: str = Form(...), template: str = Form(...), description: str = Form(""), owner: str = Form(""), memory: str = Form(""), cpus: str = Form(""), user=Depends(require_developer), ): # Normalizace + validace ID: ID musí být vždy lowercase a splňovat omezení Dockeru, # DNS a Gitea (viz validate_new_app_id). Bez toho by create script vytvořil kontejner # nebo repozitář, který později nejde nasadit či adresovat. app_id = clean_optional(app_id).lower() app_name = clean_optional(app_name) id_error = validate_new_app_id(app_id) if id_error: return HTMLResponse(id_error, status_code=400) if not app_name: return HTMLResponse("Název služby nesmí být prázdný.", status_code=400) if get_app(app_id): return HTMLResponse(f"Služba s ID '{app_id}' už existuje.", status_code=400) selected_template = get_app_template(template, create_enabled=True) if not selected_template: return HTMLResponse("Nepodporovaná šablona", status_code=400) create_script = clean_optional(selected_template.get("create_script")) if not create_script: return HTMLResponse("Šablona nemá nastavený create script", status_code=400) create_result = run_command([create_script, app_id, app_name]) if create_result.returncode != 0: log_audit_event( user, action="create_app", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "app_name": app_name, "template": selected_template.get("id"), "status": "FAILED", "create_returncode": create_result.returncode, }, ) return render_result( title="Vytvoření služby: FAILED", back_url="/portal/apps", sections=[ ("Výstup vytvoření", create_result.stdout), ("Chyba vytvoření", create_result.stderr), ], user=user, ) gitea_url = get_gitea_public_url() gitea_org = read_env_value("GITEA_ORG", DEFAULT_GITEA_ORG) repository_url = f"{gitea_url}/{gitea_org}/{app_id}.git" if gitea_url else None owner_value = clean_optional(owner) or user.get("display_name") or user.get("username") or None try: upsert_created_app( app_id, { "name": app_name, "template": selected_template.get("id"), "runtime": selected_template.get("runtime"), "language": selected_template.get("language"), "version": "1.0.0", "status": "created", "memory": clean_optional(memory), "cpus": clean_optional(cpus), "description": clean_optional(description), "owner": owner_value, "repository_name": app_id, "repository_url": repository_url, "default_branch": "main", "health_url": selected_template.get("default_health_path"), "container_port": selected_template.get("default_port"), "is_public": True, "is_enabled": True, }, ) except Exception as exc: log_audit_event( user, action="create_app", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "app_name": app_name, "template": selected_template.get("id"), "status": "FAILED", "create_returncode": create_result.returncode, "registration_error": str(exc), }, ) return render_result( title="Registrace služby: FAILED", back_url="/portal/apps", sections=[ ("Výstup vytvoření", create_result.stdout), ("Chyba vytvoření", create_result.stderr), ("Chyba registrace", str(exc)), ], user=user, ) try: apply_environment_message(app_id) except AppEnvironmentError as exc: log_audit_event( user, action="app.environment.apply_failed", target_type="app", target_id=app_id, metadata={"error": str(exc)}, ) return render_result( title="GenerovánĂ­ prostĹ™edĂ­: FAILED", back_url="/portal/apps", sections=[ ("VĂ˝stup vytvoĹ™enĂ­", create_result.stdout), ("Chyba vytvoĹ™enĂ­", create_result.stderr), ("Chyba prostĹ™edĂ­", str(exc)), ], user=user, ) job_id = create_job( job_type="deploy_app", target_type="app", target_id=app_id, payload={"template": selected_template.get("id")}, user=user, source="portal", ) log_audit_event( user, action="create_app", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "app_name": app_name, "template": selected_template.get("id"), "status": "OK", "create_returncode": create_result.returncode, "job_id": job_id, }, ) return render_result( title="Vytvoření služby: OK", back_url="/portal/apps", sections=[ ("Výstup vytvoření", create_result.stdout), ("Chyba vytvoření", create_result.stderr), ("Nasazení", f"Deploy job #{job_id} byl zařazen do fronty."), ], extra_link=f"/portal/jobs/{job_id}", extra_label="Otevřít deploy job", user=user, ) @router.post("/delete-app", response_class=HTMLResponse) def delete_app(app_id: str = Form(...), user=Depends(require_admin)): # Portál NEMAŽE workspace/container/image/gitea repo přímo – běží bez práv k workspace, # takže rm padá na "Permission denied". Místo toho jen zařadí run_script job a mazání # provede Worker přes delete-app.sh s právy serveru. Veškerá runtime/Docker logika patří # do Workeru a shell skriptů, ne do portálu. # # KONTRAKT NÁZVU AKCE (proč dříve padalo "Invalid script name"): # - do jobu se ukládá HOLÝ název skriptu: script_name = "delete-app.sh" # (stejný formát jako ostatní run_script joby, viz preflight-check.sh / bootstrap-v2.sh # a portálová validace v scheduled_scripts.validate_script_name); # - Worker přijímá jen holý název, adresář scripts/ si doplní sám, a cokoli s cestou # ("scripts/delete-app.sh", "../...", absolutní cesta) odmítne jako "Invalid script name". # Job 221 padal právě proto, že se posílalo "scripts/delete-app.sh" (název s cestou). # Stejný allowlist/validaci děláme i tady, ať portál nikdy nezaloží job, který Worker zahodí. if not is_allowed_worker_script(DELETE_APP_SCRIPT_NAME): return render_result( title="Smazání služby: CHYBA", back_url="/portal/apps", sections=[ ( "Chyba", f"Neplatný název skriptu pro Worker job: '{DELETE_APP_SCRIPT_NAME}'. " "Musí to být holý povolený název bez cesty (např. 'delete-app.sh'), jinak " "Worker job odmítne hláškou 'Invalid script name'. Job nebyl vytvořen.", ), ], user=user, ) payload = { "script_name": DELETE_APP_SCRIPT_NAME, "args": [app_id], "arguments": [app_id], } job_id = create_job( job_type="run_script", target_type="app", target_id=app_id, payload=payload, user=user, source="portal", ) log_audit_event( user, action="delete_app", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "status": "QUEUED", "script_name": DELETE_APP_SCRIPT_NAME, "job_id": job_id, }, ) # Worker zpracuje job asynchronně; pošleme uživatele rovnou na detail jobu, ať vidí # živé logy mazání a výsledek. return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303) @router.post("/update-resources", response_class=HTMLResponse) def update_resources( app_id: str = Form(...), memory: str = Form(""), cpus: str = Form(""), user=Depends(require_developer), ): memory = memory.strip() cpus = cpus.strip() update_app_resources(app_id, memory, cpus) catalog_result = run_command(["/tools/generate-catalog.sh"]) compose_result = run_command([GENERATE_COMPOSE_SCRIPT]) deploy_result = run_command([DEPLOY_SCRIPT, app_id]) status = ( "OK" if catalog_result.returncode == 0 and compose_result.returncode == 0 and deploy_result.returncode == 0 else "FAILED" ) log_audit_event( user, action="update_resources", target_type="app", target_id=app_id, metadata={ "app_id": app_id, "memory": memory, "cpus": cpus, "status": status, "catalog_returncode": catalog_result.returncode, "compose_returncode": compose_result.returncode, "deploy_returncode": deploy_result.returncode, }, ) return render_result( title=f"Úprava prostředků: {status}", back_url="/portal/apps", sections=[ ("Výstup katalogu", catalog_result.stdout), ("Chyba katalogu", catalog_result.stderr), ("Výstup compose", compose_result.stdout), ("Chyba compose", compose_result.stderr), ("Výstup nasazení", deploy_result.stdout), ("Chyba nasazení", deploy_result.stderr), ], user=user, )