A R R R Associates
263 lines
9.9 KiB
Python
263 lines
9.9 KiB
Python
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],
|
|
)
|