from __future__ import annotations import json import os import ssl from dataclasses import dataclass from datetime import datetime, timezone from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import urljoin from urllib.request import Request, urlopen from sqlalchemy.orm import Session from app.modules.domain_management.models import DomainMapping from app.modules.domain_management.services import normalize_domain @dataclass(frozen=True) class CoolifyConfig: api_base_url: str api_token: str application_uuid: str verify_tls: bool = True @property def is_configured(self) -> bool: return bool(self.api_base_url and self.api_token and self.application_uuid) @dataclass(frozen=True) class CoolifyDomainSyncResult: ok: bool status: str message: str domain_url: str | None = None existing_domains: tuple[str, ...] = () updated_domains: tuple[str, ...] = () api_status_code: int | None = None api_response: str | None = None def get_coolify_config() -> CoolifyConfig: """Read Coolify API configuration from environment. Required environment variables: COOLIFY_API_BASE_URL Example: https://coolify.example.com/api/v1 COOLIFY_API_TOKEN Bearer token created in Coolify COOLIFY_ERP_APPLICATION_UUID Coolify ERP application UUID Optional: COOLIFY_VERIFY_TLS=false Only for private/self-signed Coolify API endpoints """ base = (os.getenv("COOLIFY_API_BASE_URL") or os.getenv("COOLIFY_API_URL") or "").strip().rstrip("/") token = (os.getenv("COOLIFY_API_TOKEN") or "").strip() app_uuid = (os.getenv("COOLIFY_ERP_APPLICATION_UUID") or os.getenv("COOLIFY_APPLICATION_UUID") or "").strip() verify_tls_raw = (os.getenv("COOLIFY_VERIFY_TLS") or "true").strip().lower() verify_tls = verify_tls_raw not in {"0", "false", "no", "off"} return CoolifyConfig(api_base_url=base, api_token=token, application_uuid=app_uuid, verify_tls=verify_tls) def is_coolify_configured() -> bool: return get_coolify_config().is_configured def _normalise_api_base_url(value: str) -> str: base = (value or "").strip().rstrip("/") if not base: return "" if not base.endswith("/api/v1"): base = base + "/api/v1" return base + "/" def _request_json(config: CoolifyConfig, method: str, endpoint: str, payload: dict[str, Any] | None = None) -> tuple[int, dict[str, Any], str]: base = _normalise_api_base_url(config.api_base_url) url = urljoin(base, endpoint.lstrip("/")) body: bytes | None = None headers = { "Authorization": f"Bearer {config.api_token}", "Accept": "application/json", } if payload is not None: body = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" req = Request(url=url, data=body, method=method.upper(), headers=headers) context = None if config.verify_tls else ssl._create_unverified_context() # noqa: SLF001 - explicit opt-out via env only try: with urlopen(req, timeout=25, context=context) as response: # nosec B310 - endpoint is admin-configured raw = response.read().decode("utf-8", errors="replace") try: data = json.loads(raw) if raw else {} except Exception: data = {"raw": raw} return int(response.status), data, raw except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") try: data = json.loads(raw) if raw else {} except Exception: data = {"raw": raw} return int(exc.code), data, raw except URLError as exc: raise RuntimeError(f"Coolify API connection failed: {exc.reason}") from exc except TimeoutError as exc: raise RuntimeError("Coolify API request timed out.") from exc def _split_domain_urls(value: str | None) -> tuple[str, ...]: if not value: return () seen: set[str] = set() out: list[str] = [] for part in value.split(","): item = part.strip().rstrip("/") if not item: continue key = item.lower() if key not in seen: seen.add(key) out.append(item) return tuple(out) def _domain_to_coolify_url(domain_name: str) -> str: domain = normalize_domain(domain_name) if not domain: return "" return f"https://{domain}" def get_coolify_application_domains() -> CoolifyDomainSyncResult: config = get_coolify_config() if not config.is_configured: return CoolifyDomainSyncResult( ok=False, status="not_configured", message="Coolify API is not configured. Set COOLIFY_API_BASE_URL, COOLIFY_API_TOKEN and COOLIFY_ERP_APPLICATION_UUID.", ) try: status_code, data, raw = _request_json(config, "GET", f"applications/{config.application_uuid}") except Exception as exc: return CoolifyDomainSyncResult(ok=False, status="api_error", message=str(exc)) if status_code < 200 or status_code >= 300: return CoolifyDomainSyncResult( ok=False, status="api_error", message=f"Coolify returned HTTP {status_code} while reading the application.", api_status_code=status_code, api_response=raw[:1000], ) domains = _split_domain_urls(str(data.get("fqdn") or data.get("domains") or "")) return CoolifyDomainSyncResult( ok=True, status="read_ok", message="Coolify application domains were read successfully.", existing_domains=domains, updated_domains=domains, api_status_code=status_code, api_response=raw[:1000], ) def sync_domain_to_coolify(db: Session, mapping: DomainMapping, *, user_id: int | None = None) -> CoolifyDomainSyncResult: """Add the mapping domain to the configured Coolify ERP application domains. This function deliberately does not create/update ERP domain mappings. It only updates the Coolify application domain list and records proxy/SSL status on the existing mapping. """ config = get_coolify_config() domain_url = _domain_to_coolify_url(mapping.domain_name) if not domain_url: return CoolifyDomainSyncResult(ok=False, status="invalid_domain", message="Domain name is empty or invalid.") if not config.is_configured: return CoolifyDomainSyncResult( ok=False, status="not_configured", message="Coolify API is not configured. Set COOLIFY_API_BASE_URL, COOLIFY_API_TOKEN and COOLIFY_ERP_APPLICATION_UUID.", domain_url=domain_url, ) read_result = get_coolify_application_domains() if not read_result.ok: return CoolifyDomainSyncResult( ok=False, status=read_result.status, message=read_result.message, domain_url=domain_url, api_status_code=read_result.api_status_code, api_response=read_result.api_response, ) existing = list(read_result.updated_domains) existing_keys = {d.lower().rstrip("/") for d in existing} if domain_url.lower() in existing_keys: mapping.ssl_mode = "coolify" mapping.ssl_provider = "coolify" mapping.ssl_status = mapping.ssl_status or "pending_ssl" mapping.ssl_last_error = None mapping.updated_by_user_id = user_id mapping.updated_at_utc = datetime.now(timezone.utc) db.commit() return CoolifyDomainSyncResult( ok=True, status="already_present", message="Domain is already present in Coolify application domains.", domain_url=domain_url, existing_domains=tuple(existing), updated_domains=tuple(existing), ) updated = existing + [domain_url] domains_csv = ",".join(updated) # Current Coolify application API writes domain URLs through the request field named "domains". # Some older builds used/accepted fqdn, so we fallback only if the first attempt is rejected. patch_attempts = ({"domains": domains_csv}, {"fqdn": domains_csv}) last_status = None last_raw = "" for payload in patch_attempts: try: status_code, _data, raw = _request_json(config, "PATCH", f"applications/{config.application_uuid}", payload) except Exception as exc: return CoolifyDomainSyncResult(ok=False, status="api_error", message=str(exc), domain_url=domain_url) last_status = status_code last_raw = raw if 200 <= status_code < 300: mapping.ssl_mode = "coolify" mapping.ssl_provider = "coolify" mapping.ssl_status = "pending_ssl" mapping.ssl_last_error = None mapping.updated_by_user_id = user_id mapping.updated_at_utc = datetime.now(timezone.utc) db.commit() return CoolifyDomainSyncResult( ok=True, status="synced", message="Domain was added to Coolify application domains. Redeploy/restart the ERP app if your Coolify version does not refresh labels immediately.", domain_url=domain_url, existing_domains=tuple(existing), updated_domains=tuple(updated), api_status_code=status_code, api_response=raw[:1000], ) mapping.ssl_mode = "coolify" mapping.ssl_provider = "coolify" mapping.ssl_status = "failed" mapping.ssl_last_error = f"Coolify API returned HTTP {last_status}: {last_raw[:500]}" mapping.updated_by_user_id = user_id mapping.updated_at_utc = datetime.now(timezone.utc) db.commit() return CoolifyDomainSyncResult( ok=False, status="api_error", message=f"Coolify did not accept the domain update. HTTP {last_status}.", domain_url=domain_url, existing_domains=tuple(existing), updated_domains=tuple(updated), api_status_code=last_status, api_response=last_raw[:1000], )