|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
|
|
from devplacepy.database import get_setting
|
|
from devplacepy.services.containers import store
|
|
from devplacepy.stealth import stealth_async_client
|
|
|
|
from . import tunnels
|
|
|
|
API_PATH = "/api/v1"
|
|
ISSUE_TIMEOUT_SECONDS = 180.0
|
|
|
|
|
|
class CertError(Exception):
|
|
pass
|
|
|
|
|
|
def _base_url() -> str:
|
|
return (get_setting("workspace_molohttp_base_url", "") or "").rstrip("/")
|
|
|
|
|
|
def _auth_headers() -> dict[str, str]:
|
|
api_key = get_setting("workspace_molohttp_api_key", "") or ""
|
|
if api_key:
|
|
return {"x-api-key": api_key}
|
|
return {}
|
|
|
|
|
|
def _basic_auth() -> tuple[str, str] | None:
|
|
username = get_setting("workspace_molohttp_username", "") or ""
|
|
password = get_setting("workspace_molohttp_password", "") or ""
|
|
if username and password:
|
|
return (username, password)
|
|
return None
|
|
|
|
|
|
def configured() -> bool:
|
|
return bool(_base_url()) and bool(_auth_headers() or _basic_auth())
|
|
|
|
|
|
async def issue(hostname: str) -> None:
|
|
if not hostname:
|
|
raise CertError("no hostname")
|
|
base = _base_url()
|
|
if not base:
|
|
raise CertError("workspace_molohttp_base_url is not configured")
|
|
headers = _auth_headers()
|
|
auth = _basic_auth()
|
|
if not headers and not auth:
|
|
raise CertError("no molohttp credentials configured")
|
|
payload: dict[str, object] = {"domains": [hostname]}
|
|
email = get_setting("workspace_acme_email", "") or ""
|
|
if email:
|
|
payload["email"] = email
|
|
async with stealth_async_client(timeout=ISSUE_TIMEOUT_SECONDS) as client:
|
|
response = await client.post(
|
|
f"{base}{API_PATH}/certs/issue",
|
|
json=payload,
|
|
headers=headers,
|
|
auth=auth,
|
|
)
|
|
if response.status_code >= 400:
|
|
raise CertError(
|
|
f"molohttp cert issue failed ({response.status_code}): "
|
|
f"{response.text[:200]}"
|
|
)
|
|
|
|
|
|
async def certify(row: dict, log: Callable[[str], None] | None = None) -> bool:
|
|
hostname = row.get("hostname", "")
|
|
tunnels.update(
|
|
row["uid"], {"status": tunnels.STATUS_PROVISIONING, "last_error": ""}
|
|
)
|
|
try:
|
|
await issue(hostname)
|
|
except Exception as error:
|
|
tunnels.update(
|
|
row["uid"],
|
|
{"status": tunnels.STATUS_FAILED, "last_error": str(error)[:500]},
|
|
)
|
|
if log:
|
|
log(f"tunnel certificate for {hostname} failed: {error}")
|
|
return False
|
|
tunnels.update(row["uid"], {"status": tunnels.STATUS_ACTIVE, "last_error": ""})
|
|
if log:
|
|
log(f"tunnel certificate issued for {hostname}")
|
|
announce(row, hostname)
|
|
return True
|
|
|
|
|
|
def announce(row: dict, hostname: str) -> None:
|
|
from devplacepy.utils import create_notification
|
|
|
|
from . import provision
|
|
|
|
instance = store.get_instance(row.get("instance_uid", ""))
|
|
if not instance:
|
|
return
|
|
owner_uid = instance.get("workspace_owner_uid", "")
|
|
if not owner_uid:
|
|
return
|
|
url = f"https://{hostname}"
|
|
if provision.is_editor_tunnel(instance, row):
|
|
password = instance.get("editor_password") or ""
|
|
message = f"Your VS Code workspace is live at {url} (password: {password})."
|
|
else:
|
|
message = f"Tunnel {row.get('label') or hostname} is live at {url}."
|
|
create_notification(
|
|
owner_uid,
|
|
"workspace",
|
|
message,
|
|
instance["uid"],
|
|
f"/projects/{instance.get('project_uid', '')}/workspace",
|
|
)
|