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 ..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''
@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 ""
)
detail_tabs = (
"""
"""
if can_manage
else ""
)
variables_card = (
f"""
Proměnné
{environment_usage_example}
Key
Value
Secret
Akce
{variable_rows}
"""
if can_manage
else ""
)
ip_access_card = (
f"""
Security / IP Access
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.