|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
from devplacepy.database import get_setting
|
|
from devplacepy.stealth import stealth_async_client
|
|
|
|
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]}"
|
|
)
|