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"""
"""
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_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"""
"""
# 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"""
""",
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.
{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.