1656 lines
57 KiB
Python
1656 lines
57 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Stalwart v0.16 migration helper.
|
|
|
|
Two modes:
|
|
|
|
dump — pull all settings and principals from a Stalwart server via the
|
|
management API into two JSON files.
|
|
|
|
convert — read those two JSON files and emit:
|
|
* config.json — plain DataStore object (Stalwart's main config)
|
|
* export.json — array of `update`/`create` ops for everything
|
|
else, in load order.
|
|
|
|
Usage:
|
|
python migrate_v016.py dump --url https://mail.example.com \
|
|
--username admin --password s3cret \
|
|
--settings settings.json --principals principals.json
|
|
|
|
python migrate_v016.py convert \
|
|
--settings settings.json --principals principals.json \
|
|
--config config.json --output export.json
|
|
"""
|
|
|
|
# SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.parse
|
|
from collections import defaultdict
|
|
from typing import Any
|
|
|
|
import requests
|
|
import urllib3
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
ALL_PRINCIPAL_TYPES = [
|
|
"individual",
|
|
"group",
|
|
"resource",
|
|
"location",
|
|
"list",
|
|
"other",
|
|
"domain",
|
|
"tenant",
|
|
"role",
|
|
"apiKey",
|
|
"oauthClient",
|
|
]
|
|
|
|
PAGE_SIZE = 200
|
|
REQUEST_TIMEOUT = 60
|
|
|
|
class ApiError(RuntimeError):
|
|
pass
|
|
|
|
class ConvertError(RuntimeError):
|
|
pass
|
|
|
|
class StalwartClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
*,
|
|
token: str | None = None,
|
|
username: str | None = None,
|
|
password: str | None = None,
|
|
verify: bool = False,
|
|
):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.session = requests.Session()
|
|
self.session.verify = verify
|
|
self.session.headers.update({"Accept": "application/json"})
|
|
if token:
|
|
self.session.headers["Authorization"] = f"Bearer {token}"
|
|
elif username is not None and password is not None:
|
|
self.session.auth = (username, password)
|
|
else:
|
|
raise ValueError("need either a token or username/password")
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
json_body: Any = None,
|
|
) -> Any:
|
|
resp = self.session.request(
|
|
method,
|
|
self.base_url + path,
|
|
params=params,
|
|
json=json_body,
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
if resp.status_code == 401:
|
|
raise ApiError(f"401 Unauthorized for {method} {path}")
|
|
if resp.status_code == 403:
|
|
raise ApiError(f"403 Forbidden for {method} {path}")
|
|
if resp.status_code == 404:
|
|
raise ApiError(f"404 Not Found for {method} {path}")
|
|
if not resp.ok:
|
|
raise ApiError(
|
|
f"{resp.status_code} {resp.reason} for {method} {path}: {resp.text[:500]}"
|
|
)
|
|
try:
|
|
payload = resp.json()
|
|
except ValueError as exc:
|
|
raise ApiError(f"Non-JSON response from {path}: {exc}")
|
|
if isinstance(payload, dict) and "error" in payload and "data" not in payload:
|
|
raise ApiError(f"Server error on {path}: {payload}")
|
|
if isinstance(payload, dict) and "data" in payload:
|
|
return payload["data"]
|
|
return payload
|
|
|
|
def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
|
|
return self._request("GET", path, params=params)
|
|
|
|
def dump_all_settings(self) -> dict[str, str]:
|
|
|
|
merged: dict[str, str] = {}
|
|
page = 1
|
|
last_progress_page = 0
|
|
while True:
|
|
data = self.get(
|
|
"/api/settings/list",
|
|
params={
|
|
"prefix": "",
|
|
"page": str(page),
|
|
"limit": str(PAGE_SIZE),
|
|
},
|
|
)
|
|
items = data.get("items", {}) or {}
|
|
total = int(data.get("total", len(items)) or 0)
|
|
|
|
before = len(merged)
|
|
merged.update(items)
|
|
gained = len(merged) - before
|
|
|
|
if not items:
|
|
break
|
|
if len(merged) >= total:
|
|
break
|
|
if gained == 0:
|
|
raise ApiError(
|
|
f"Settings pagination made no progress on page {page} "
|
|
f"(have {len(merged)}/{total}). Server may not support "
|
|
"paging /api/settings/list with page/limit."
|
|
)
|
|
last_progress_page = page
|
|
page += 1
|
|
|
|
if page - last_progress_page > 5:
|
|
raise ApiError(
|
|
f"Settings pagination stalled at page {page} "
|
|
f"(have {len(merged)}/{total})."
|
|
)
|
|
return merged
|
|
|
|
def list_principal_names(self) -> list[tuple[str, str]]:
|
|
out: list[tuple[str, str]] = []
|
|
seen: set[tuple[str, str]] = set()
|
|
for principal_type in ALL_PRINCIPAL_TYPES:
|
|
try:
|
|
self._list_principals_of_type(principal_type, out, seen)
|
|
except ApiError as exc:
|
|
print(
|
|
f" WARN skipping principal type {principal_type!r}: {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
return out
|
|
|
|
def _list_principals_of_type(
|
|
self,
|
|
principal_type: str,
|
|
out: list[tuple[str, str]],
|
|
seen: set[tuple[str, str]],
|
|
) -> None:
|
|
before = len(out)
|
|
page = 1
|
|
while True:
|
|
data = self.get(
|
|
"/api/principal",
|
|
params={
|
|
"page": str(page),
|
|
"limit": str(PAGE_SIZE),
|
|
"types": principal_type,
|
|
},
|
|
)
|
|
items = data.get("items", []) or []
|
|
total = int(data.get("total", 0) or 0)
|
|
for p in items:
|
|
typ = p.get("type") or principal_type
|
|
name = _principal_name(p)
|
|
if not name:
|
|
continue
|
|
key = (typ, name)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(key)
|
|
if not items:
|
|
break
|
|
if (len(out) - before) >= total:
|
|
break
|
|
page += 1
|
|
|
|
def get_principal(self, name: str) -> dict[str, Any]:
|
|
quoted = urllib.parse.quote(name, safe="")
|
|
return self.get(f"/api/principal/{quoted}")
|
|
|
|
def _principal_name(p: dict[str, Any]) -> str:
|
|
v = p.get("name")
|
|
if isinstance(v, str):
|
|
return v
|
|
if isinstance(v, dict):
|
|
if isinstance(v.get("string"), str):
|
|
return v["string"]
|
|
sl = v.get("stringList")
|
|
if isinstance(sl, list) and sl:
|
|
return sl[0]
|
|
if isinstance(v, list) and v:
|
|
return v[0]
|
|
return ""
|
|
|
|
def cmd_dump(args: argparse.Namespace) -> int:
|
|
if args.token:
|
|
client = StalwartClient(args.url, token=args.token, verify=False)
|
|
elif args.username and args.password:
|
|
client = StalwartClient(
|
|
args.url,
|
|
username=args.username,
|
|
password=args.password,
|
|
verify=False,
|
|
)
|
|
else:
|
|
print("error: either --token or --username/--password is required",
|
|
file=sys.stderr)
|
|
return 2
|
|
|
|
print("Fetching settings...", file=sys.stderr)
|
|
settings = client.dump_all_settings()
|
|
with open(args.settings, "w", encoding="utf-8") as f:
|
|
json.dump(settings, f, indent=2, sort_keys=True, ensure_ascii=False)
|
|
print(f" wrote {len(settings)} settings keys to {args.settings}",
|
|
file=sys.stderr)
|
|
|
|
print("Listing principals...", file=sys.stderr)
|
|
names = client.list_principal_names()
|
|
print(f" found {len(names)} principals across all types", file=sys.stderr)
|
|
|
|
principals: list[dict[str, Any]] = []
|
|
for i, (typ, name) in enumerate(names, 1):
|
|
try:
|
|
full = client.get_principal(name)
|
|
except ApiError as exc:
|
|
print(f" [{i}/{len(names)}] WARN failed to fetch {typ} {name!r}: {exc}",
|
|
file=sys.stderr)
|
|
continue
|
|
principals.append(full)
|
|
if i % 50 == 0 or i == len(names):
|
|
print(f" [{i}/{len(names)}] fetched {typ} {name}", file=sys.stderr)
|
|
|
|
missing_id = [p for p in principals if p.get("id") is None]
|
|
if missing_id:
|
|
print(
|
|
f"warning: {len(missing_id)} principal(s) returned no 'id' field",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
with open(args.principals, "w", encoding="utf-8") as f:
|
|
json.dump(principals, f, indent=2, ensure_ascii=False)
|
|
print(f" wrote {len(principals)} principals to {args.principals}",
|
|
file=sys.stderr)
|
|
|
|
return 0
|
|
|
|
_DURATION_UNITS_MS = {
|
|
"ms": 1,
|
|
"s": 1000,
|
|
"m": 60_000,
|
|
"h": 3_600_000,
|
|
"d": 86_400_000,
|
|
"w": 604_800_000,
|
|
}
|
|
|
|
_SIZE_UNITS_BYTES = {
|
|
"": 1,
|
|
"b": 1,
|
|
"kb": 1024,
|
|
"mb": 1024 * 1024,
|
|
"gb": 1024 * 1024 * 1024,
|
|
}
|
|
|
|
def parse_duration_ms(s: str | None) -> int | None:
|
|
|
|
if s is None:
|
|
return None
|
|
s = str(s).strip().lower()
|
|
if not s:
|
|
return None
|
|
m = re.fullmatch(r"(-?\d+)(ms|s|m|h|d|w)?", s)
|
|
if not m:
|
|
raise ConvertError(f"Could not parse duration {s!r}")
|
|
n = int(m.group(1))
|
|
unit = m.group(2) or "ms"
|
|
return n * _DURATION_UNITS_MS[unit]
|
|
|
|
def parse_size_bytes(s: str | None) -> int | None:
|
|
|
|
if s is None:
|
|
return None
|
|
s = str(s).strip().lower()
|
|
if not s:
|
|
return None
|
|
m = re.fullmatch(r"(-?\d+)\s*(b|kb|mb|gb)?", s)
|
|
if not m:
|
|
raise ConvertError(f"Could not parse size {s!r}")
|
|
return int(m.group(1)) * _SIZE_UNITS_BYTES[m.group(2) or ""]
|
|
|
|
def parse_int(s: str | None) -> int | None:
|
|
if s is None:
|
|
return None
|
|
s = str(s).strip()
|
|
if not s:
|
|
return None
|
|
try:
|
|
return int(s)
|
|
except ValueError as exc:
|
|
raise ConvertError(f"Could not parse integer {s!r}") from exc
|
|
|
|
def parse_bool(s: str | None) -> bool | None:
|
|
if s is None:
|
|
return None
|
|
v = str(s).strip().lower()
|
|
if v in ("true", "1", "yes", "on"):
|
|
return True
|
|
if v in ("false", "0", "no", "off", ""):
|
|
return False
|
|
raise ConvertError(f"Could not parse bool {s!r}")
|
|
|
|
def pv_string(v: Any) -> str:
|
|
if v is None:
|
|
return ""
|
|
if isinstance(v, str):
|
|
return v
|
|
if isinstance(v, (int, float)):
|
|
return str(v)
|
|
if isinstance(v, dict):
|
|
if isinstance(v.get("string"), str):
|
|
return v["string"]
|
|
sl = v.get("stringList")
|
|
if isinstance(sl, list) and sl:
|
|
return str(sl[0])
|
|
iv = v.get("integer")
|
|
if isinstance(iv, int):
|
|
return str(iv)
|
|
if isinstance(v, list) and v:
|
|
return pv_string(v[0])
|
|
return ""
|
|
|
|
def pv_int(v: Any) -> int | None:
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, bool):
|
|
return int(v)
|
|
if isinstance(v, int):
|
|
return v
|
|
if isinstance(v, str):
|
|
try:
|
|
return int(v)
|
|
except ValueError:
|
|
return None
|
|
if isinstance(v, dict):
|
|
iv = v.get("integer")
|
|
if isinstance(iv, int):
|
|
return iv
|
|
return None
|
|
|
|
def pv_list(v: Any) -> list:
|
|
|
|
if v is None:
|
|
return []
|
|
if isinstance(v, list):
|
|
return list(v)
|
|
if isinstance(v, dict):
|
|
sl = v.get("stringList")
|
|
if isinstance(sl, list):
|
|
return list(sl)
|
|
il = v.get("integerList")
|
|
if isinstance(il, list):
|
|
return list(il)
|
|
if "string" in v and isinstance(v["string"], str):
|
|
return [v["string"]]
|
|
if "integer" in v and isinstance(v["integer"], int):
|
|
return [v["integer"]]
|
|
return []
|
|
|
|
return [v]
|
|
|
|
def split_email(addr: str) -> tuple[str, str] | None:
|
|
|
|
if "@" not in addr:
|
|
return None
|
|
local, _, domain = addr.rpartition("@")
|
|
domain = domain.strip().lower()
|
|
if not domain:
|
|
return None
|
|
return (local, domain)
|
|
|
|
def group_settings_by_prefix(settings: dict[str, str], prefix: str) -> dict[str, dict[str, str]]:
|
|
|
|
raise NotImplementedError
|
|
|
|
def build_sub_trees(
|
|
settings: dict[str, str],
|
|
prefix: str,
|
|
discriminator: str,
|
|
) -> dict[str, dict[str, str]]:
|
|
|
|
record_ids: list[str] = []
|
|
disc_suffix = "." + discriminator
|
|
prefix_dot = prefix + "."
|
|
for k in settings:
|
|
if k.startswith(prefix_dot) and k.endswith(disc_suffix):
|
|
rid = k[len(prefix_dot):-len(disc_suffix)]
|
|
if rid:
|
|
record_ids.append(rid)
|
|
|
|
record_ids.sort(key=lambda s: (-len(s), s))
|
|
|
|
trees: dict[str, dict[str, str]] = {rid: {} for rid in record_ids}
|
|
claimed: set[str] = set()
|
|
for rid in record_ids:
|
|
head = prefix_dot + rid + "."
|
|
for k, v in settings.items():
|
|
if k in claimed:
|
|
continue
|
|
if k.startswith(head):
|
|
sub = k[len(head):]
|
|
trees[rid][sub] = v
|
|
claimed.add(k)
|
|
return trees
|
|
|
|
def collect_array(sub: dict[str, str], field: str) -> list[str]:
|
|
|
|
items: list[tuple[int, str]] = []
|
|
head = field + "."
|
|
for k, v in sub.items():
|
|
if k.startswith(head):
|
|
tail = k[len(head):]
|
|
if tail.isdigit():
|
|
items.append((int(tail), v))
|
|
items.sort()
|
|
return [v for _, v in items]
|
|
|
|
_PEM_RE = re.compile(
|
|
r"-----BEGIN (?P<kind>[A-Z][A-Z0-9 ]*?)-----"
|
|
r"[\s\S]+?"
|
|
r"-----END (?P=kind)-----",
|
|
re.MULTILINE,
|
|
)
|
|
|
|
|
|
def split_pem_bundle(blob: str) -> tuple[str, str]:
|
|
certs: list[str] = []
|
|
key: str | None = None
|
|
for m in _PEM_RE.finditer(blob):
|
|
kind = m.group("kind")
|
|
block = m.group(0) + "\n"
|
|
if "PRIVATE KEY" in kind:
|
|
if key is None:
|
|
key = block
|
|
elif kind == "CERTIFICATE":
|
|
certs.append(block)
|
|
return ("".join(certs), key or "")
|
|
|
|
|
|
def _make_certificate_object(cert_pem: str, key_pem: str) -> dict[str, Any]:
|
|
return {
|
|
"certificate": {"@type": "Text", "value": cert_pem},
|
|
"privateKey": {"@type": "Text", "secret": key_pem},
|
|
}
|
|
|
|
|
|
def is_app_password(secret: str) -> bool:
|
|
return secret.startswith("$app$")
|
|
|
|
def is_otpauth(secret: str) -> bool:
|
|
return secret.startswith("otpauth://")
|
|
|
|
def secret_key_optional(value: str | None) -> dict[str, Any]:
|
|
|
|
if value is None or value == "":
|
|
return {"@type": "None"}
|
|
return {"@type": "Value", "secret": value}
|
|
|
|
def secret_key(value: str | None) -> dict[str, Any]:
|
|
|
|
if value is None or value == "":
|
|
return {"@type": "None"}
|
|
return {"@type": "Value", "secret": value}
|
|
|
|
def secret_text(value: str | None) -> dict[str, Any]:
|
|
|
|
if value is None or value == "":
|
|
return {"@type": "None"}
|
|
return {"@type": "Text", "secret": value}
|
|
|
|
_REDIS_PROTOCOL_MAP = {
|
|
"resp2": "resp2",
|
|
"resp3": "resp3",
|
|
}
|
|
|
|
_S3_REGION_MAP = {
|
|
"us-east-1": "UsEast1", "us-east-2": "UsEast2",
|
|
"us-west-1": "UsWest1", "us-west-2": "UsWest2",
|
|
"ca-central-1": "CaCentral1",
|
|
"af-south-1": "AfSouth1",
|
|
"ap-east-1": "ApEast1", "ap-south-1": "ApSouth1",
|
|
"ap-northeast-1": "ApNortheast1", "ap-northeast-2": "ApNortheast2",
|
|
"ap-northeast-3": "ApNortheast3",
|
|
"ap-southeast-1": "ApSoutheast1", "ap-southeast-2": "ApSoutheast2",
|
|
"cn-north-1": "CnNorth1", "cn-northwest-1": "CnNorthwest1",
|
|
"eu-north-1": "EuNorth1",
|
|
"eu-central-1": "EuCentral1", "eu-central-2": "EuCentral2",
|
|
"eu-west-1": "EuWest1", "eu-west-2": "EuWest2", "eu-west-3": "EuWest3",
|
|
"il-central-1": "IlCentral1",
|
|
"me-south-1": "MeSouth1",
|
|
"sa-east-1": "SaEast1",
|
|
"do-nyc3": "DoNyc3", "do-ams3": "DoAms3",
|
|
"do-sgp1": "DoSgp1", "do-fra1": "DoFra1",
|
|
"yandex": "Yandex",
|
|
"wa-us-east-1": "WaUsEast1", "wa-us-east-2": "WaUsEast2",
|
|
"wa-us-central-1": "WaUsCentral1", "wa-us-west-1": "WaUsWest1",
|
|
"wa-ca-central-1": "WaCaCentral1",
|
|
"wa-eu-central-1": "WaEuCentral1", "wa-eu-central-2": "WaEuCentral2",
|
|
"wa-eu-west-1": "WaEuWest1", "wa-eu-west-2": "WaEuWest2",
|
|
"wa-ap-northeast-1": "WaApNortheast1", "wa-ap-northeast-2": "WaApNortheast2",
|
|
"wa-ap-southeast-1": "WaApSoutheast1", "wa-ap-southeast-2": "WaApSoutheast2",
|
|
}
|
|
|
|
class Converter:
|
|
def __init__(
|
|
self,
|
|
principals: list[dict[str, Any]],
|
|
settings: dict[str, str],
|
|
):
|
|
self.principals = principals
|
|
self.settings = settings
|
|
|
|
self.by_name: dict[str, dict[str, Any]] = {}
|
|
self.by_id: dict[int, dict[str, Any]] = {}
|
|
for p in principals:
|
|
n = pv_string(p.get("name"))
|
|
if n:
|
|
self.by_name[n] = p
|
|
pid = pv_int(p.get("id"))
|
|
if pid is not None:
|
|
self.by_id[pid] = p
|
|
|
|
self.tenant_name_to_cid: dict[str, str] = {}
|
|
self.domain_name_to_cid: dict[str, str] = {}
|
|
self.domain_cid_to_name: dict[str, str] = {}
|
|
self.domain_cid_to_tenant_cid: dict[str, str] = {}
|
|
self.default_domain_cid: str | None = None
|
|
self._create_counter = 0
|
|
|
|
def _next_create_cid(self) -> str:
|
|
cid = f"create-{self._create_counter}"
|
|
self._create_counter += 1
|
|
return cid
|
|
|
|
@staticmethod
|
|
def _account_cid(old_id: int) -> str:
|
|
return f"restore-{old_id}"
|
|
|
|
def run(self) -> dict[str, Any]:
|
|
tenants = self._build_tenants()
|
|
domains = self._build_domains()
|
|
self._pick_default_domain()
|
|
accounts = self._build_accounts()
|
|
mailing_lists = self._build_mailing_lists()
|
|
dkim_signatures = self._build_dkim_signatures()
|
|
certificates = self._build_certificates()
|
|
|
|
self._check_duplicate_emails(accounts, mailing_lists)
|
|
|
|
data_store = self._build_data_store()
|
|
blob_store = self._build_blob_store()
|
|
in_memory_store = self._build_in_memory_store()
|
|
search_store = self._build_search_store()
|
|
enterprise = self._build_enterprise()
|
|
system_settings = self._build_system_settings()
|
|
|
|
out: dict[str, Any] = {}
|
|
if system_settings is not None:
|
|
out["SystemSettings"] = system_settings
|
|
if enterprise is not None:
|
|
out["Enterprise"] = enterprise
|
|
if data_store is not None:
|
|
out["DataStore"] = data_store
|
|
if blob_store is not None:
|
|
out["BlobStore"] = blob_store
|
|
if in_memory_store is not None:
|
|
out["InMemoryStore"] = in_memory_store
|
|
if search_store is not None:
|
|
out["SearchStore"] = search_store
|
|
if tenants:
|
|
out["Tenant"] = tenants
|
|
if domains:
|
|
out["Domain"] = domains
|
|
if accounts:
|
|
out["Account"] = accounts
|
|
if mailing_lists:
|
|
out["MailingList"] = mailing_lists
|
|
if dkim_signatures:
|
|
out["DkimSignature"] = dkim_signatures
|
|
if certificates:
|
|
out["Certificate"] = certificates
|
|
return out
|
|
|
|
def _build_tenants(self) -> dict[str, dict[str, Any]]:
|
|
tenants = [p for p in self.principals if p.get("type") == "tenant"]
|
|
tenants.sort(key=lambda p: pv_string(p.get("name")))
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for p in tenants:
|
|
name = pv_string(p.get("name"))
|
|
if not name:
|
|
continue
|
|
cid = self._next_create_cid()
|
|
self.tenant_name_to_cid[name] = cid
|
|
obj: dict[str, Any] = {"name": name}
|
|
logo = pv_string(p.get("picture"))
|
|
if logo:
|
|
obj["logo"] = logo
|
|
quotas: dict[str, int] = {}
|
|
q = pv_int(p.get("quota"))
|
|
if q:
|
|
quotas["maxDiskQuota"] = q
|
|
obj["quotas"] = quotas
|
|
out[cid] = obj
|
|
return out
|
|
|
|
def _collect_domain_names(self) -> set[str]:
|
|
names: set[str] = set()
|
|
|
|
def add(addr: str) -> None:
|
|
parts = split_email(addr)
|
|
if parts is None:
|
|
return
|
|
_, dom = parts
|
|
if dom:
|
|
names.add(dom)
|
|
|
|
for p in self.principals:
|
|
t = p.get("type")
|
|
if t == "domain":
|
|
n = pv_string(p.get("name")).strip().lower()
|
|
if n:
|
|
names.add(n)
|
|
elif t in ("individual", "group", "list"):
|
|
|
|
nm = pv_string(p.get("name"))
|
|
if "@" in nm:
|
|
add(nm)
|
|
for addr in pv_list(p.get("emails")):
|
|
if isinstance(addr, str):
|
|
add(addr)
|
|
|
|
sigs = build_sub_trees(self.settings, "signature", "algorithm")
|
|
for _, sub in sigs.items():
|
|
d = sub.get("domain", "").strip().lower()
|
|
if d:
|
|
names.add(d)
|
|
|
|
return names
|
|
|
|
def _build_domains(self) -> dict[str, dict[str, Any]]:
|
|
declared: dict[str, dict[str, Any]] = {}
|
|
for p in self.principals:
|
|
if p.get("type") == "domain":
|
|
n = pv_string(p.get("name")).strip().lower()
|
|
if n:
|
|
declared[n] = p
|
|
|
|
names = sorted(self._collect_domain_names())
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for dname in names:
|
|
cid = self._next_create_cid()
|
|
self.domain_name_to_cid[dname] = cid
|
|
self.domain_cid_to_name[cid] = dname
|
|
obj: dict[str, Any] = {"name": dname}
|
|
p = declared.get(dname)
|
|
if p is not None:
|
|
desc = pv_string(p.get("description"))
|
|
if desc:
|
|
obj["description"] = desc
|
|
logo = pv_string(p.get("picture"))
|
|
if logo:
|
|
obj["logo"] = logo
|
|
tname = pv_string(p.get("tenant"))
|
|
if tname and tname in self.tenant_name_to_cid:
|
|
t_cid = self.tenant_name_to_cid[tname]
|
|
obj["memberTenantId"] = "#" + t_cid
|
|
self.domain_cid_to_tenant_cid[cid] = t_cid
|
|
out[cid] = obj
|
|
return out
|
|
|
|
def _pick_default_domain(self) -> None:
|
|
if not self.domain_name_to_cid:
|
|
self.default_domain_cid = None
|
|
return
|
|
if len(self.domain_name_to_cid) == 1:
|
|
self.default_domain_cid = next(iter(self.domain_name_to_cid.values()))
|
|
return
|
|
|
|
counts: dict[str, int] = defaultdict(int)
|
|
for p in self.principals:
|
|
t = p.get("type")
|
|
if t not in ("individual", "group", "list"):
|
|
continue
|
|
d = self._infer_primary_domain(p)
|
|
if d is not None:
|
|
counts[d] += 1
|
|
if not counts:
|
|
first = sorted(self.domain_name_to_cid.keys())[0]
|
|
self.default_domain_cid = self.domain_name_to_cid[first]
|
|
return
|
|
|
|
best = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[0][0]
|
|
self.default_domain_cid = self.domain_name_to_cid[best]
|
|
|
|
def _infer_primary_domain(self, p: dict[str, Any]) -> str | None:
|
|
|
|
nm = pv_string(p.get("name"))
|
|
if "@" in nm:
|
|
parts = split_email(nm)
|
|
if parts and parts[1]:
|
|
return parts[1]
|
|
|
|
for addr in pv_list(p.get("emails")):
|
|
if not isinstance(addr, str):
|
|
continue
|
|
parts = split_email(addr)
|
|
if parts is None:
|
|
continue
|
|
local, dom = parts
|
|
if local and local == nm and dom:
|
|
return dom
|
|
return None
|
|
|
|
def _resolve_name_and_domain(self, p: dict[str, Any]) -> tuple[str, str]:
|
|
nm = pv_string(p.get("name"))
|
|
|
|
if "@" in nm:
|
|
parts = split_email(nm)
|
|
if parts is None or not parts[1]:
|
|
raise ConvertError(f"principal name {nm!r} is malformed")
|
|
local, dom = parts
|
|
if dom not in self.domain_name_to_cid:
|
|
raise ConvertError(f"domain {dom!r} missing from domain index")
|
|
return (local, self.domain_name_to_cid[dom])
|
|
|
|
dom = self._infer_primary_domain(p)
|
|
if dom:
|
|
if dom not in self.domain_name_to_cid:
|
|
raise ConvertError(f"domain {dom!r} missing from domain index")
|
|
return (nm, self.domain_name_to_cid[dom])
|
|
|
|
if self.default_domain_cid is None:
|
|
raise ConvertError(
|
|
f"principal {nm!r} has no domain and no default domain is set"
|
|
)
|
|
return (nm, self.default_domain_cid)
|
|
|
|
def _build_aliases(
|
|
self,
|
|
p: dict[str, Any],
|
|
primary_name: str,
|
|
primary_domain_cid: str,
|
|
) -> dict[str, dict[str, Any]]:
|
|
aliases: dict[str, dict[str, Any]] = {}
|
|
idx = 0
|
|
seen: set[tuple[str, str]] = set()
|
|
for addr in pv_list(p.get("emails")):
|
|
if not isinstance(addr, str):
|
|
continue
|
|
parts = split_email(addr)
|
|
if parts is None:
|
|
continue
|
|
local, dom = parts
|
|
if local == "":
|
|
continue
|
|
if dom not in self.domain_name_to_cid:
|
|
continue
|
|
d_cid = self.domain_name_to_cid[dom]
|
|
if local == primary_name and d_cid == primary_domain_cid:
|
|
continue
|
|
key = (local, d_cid)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
aliases[str(idx)] = {"name": local, "domainId": "#" + d_cid}
|
|
idx += 1
|
|
return aliases
|
|
|
|
def _build_accounts(self) -> dict[str, dict[str, Any]]:
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for p in self.principals:
|
|
t = p.get("type")
|
|
if t == "individual":
|
|
cid, obj = self._build_user(p)
|
|
elif t == "group":
|
|
cid, obj = self._build_group(p)
|
|
else:
|
|
continue
|
|
if cid in out:
|
|
raise ConvertError(f"duplicate account client-id {cid!r}")
|
|
out[cid] = obj
|
|
return out
|
|
|
|
def _build_user(self, p: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
|
local, dom_cid = self._resolve_name_and_domain(p)
|
|
uid = pv_int(p.get("id"))
|
|
if uid is None:
|
|
raise ConvertError(f"user {pv_string(p.get('name'))!r} has no id")
|
|
body: dict[str, Any] = {
|
|
"@type": "User",
|
|
"name": local,
|
|
"domainId": "#" + dom_cid,
|
|
"aliases": self._build_aliases(p, local, dom_cid),
|
|
"credentials": self._build_credentials(p),
|
|
"memberGroupIds": self._build_member_group_ids(p),
|
|
"quotas": self._build_account_quotas(p),
|
|
}
|
|
desc = pv_string(p.get("description"))
|
|
if desc:
|
|
body["description"] = desc
|
|
tname = pv_string(p.get("tenant"))
|
|
if tname and tname in self.tenant_name_to_cid:
|
|
body["memberTenantId"] = "#" + self.tenant_name_to_cid[tname]
|
|
return (self._account_cid(uid), body)
|
|
|
|
def _build_group(self, p: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
|
local, dom_cid = self._resolve_name_and_domain(p)
|
|
gid = pv_int(p.get("id"))
|
|
if gid is None:
|
|
raise ConvertError(f"group {pv_string(p.get('name'))!r} has no id")
|
|
body: dict[str, Any] = {
|
|
"@type": "Group",
|
|
"name": local,
|
|
"domainId": "#" + dom_cid,
|
|
"aliases": self._build_aliases(p, local, dom_cid),
|
|
"quotas": self._build_account_quotas(p),
|
|
}
|
|
desc = pv_string(p.get("description"))
|
|
if desc:
|
|
body["description"] = desc
|
|
tname = pv_string(p.get("tenant"))
|
|
if tname and tname in self.tenant_name_to_cid:
|
|
body["memberTenantId"] = "#" + self.tenant_name_to_cid[tname]
|
|
return (self._account_cid(gid), body)
|
|
|
|
def _build_account_quotas(self, p: dict[str, Any]) -> dict[str, int]:
|
|
quotas: dict[str, int] = {}
|
|
q = pv_int(p.get("quota"))
|
|
if q:
|
|
quotas["maxDiskQuota"] = q
|
|
return quotas
|
|
|
|
def _build_credentials(self, p: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
|
|
secrets = [s for s in pv_list(p.get("secrets")) if isinstance(s, str)]
|
|
password = next(
|
|
(s for s in secrets if not is_app_password(s) and not is_otpauth(s)),
|
|
None,
|
|
)
|
|
otp = next((s for s in secrets if is_otpauth(s)), None)
|
|
if password is None:
|
|
return {}
|
|
cred: dict[str, Any] = {
|
|
"@type": "Password",
|
|
"secret": password,
|
|
}
|
|
if otp is not None:
|
|
cred["otpAuth"] = otp
|
|
return {"0": cred}
|
|
|
|
def _build_member_group_ids(self, p: dict[str, Any]) -> dict[str, bool]:
|
|
out: dict[str, bool] = {}
|
|
for ref in pv_list(p.get("memberOf")):
|
|
target = self._resolve_principal_ref(ref)
|
|
if target is None:
|
|
continue
|
|
if target.get("type") != "group":
|
|
continue
|
|
gid = pv_int(target.get("id"))
|
|
if gid is None:
|
|
continue
|
|
out["#" + self._account_cid(gid)] = True
|
|
return out
|
|
|
|
def _resolve_principal_ref(self, ref: Any) -> dict[str, Any] | None:
|
|
if isinstance(ref, str):
|
|
return self.by_name.get(ref)
|
|
if isinstance(ref, int):
|
|
return self.by_id.get(ref)
|
|
if isinstance(ref, dict):
|
|
if "string" in ref:
|
|
return self.by_name.get(str(ref["string"]))
|
|
if "integer" in ref:
|
|
return self.by_id.get(int(ref["integer"]))
|
|
return None
|
|
|
|
def _build_mailing_lists(self) -> dict[str, dict[str, Any]]:
|
|
lists = [p for p in self.principals if p.get("type") == "list"]
|
|
lists.sort(key=lambda p: (pv_int(p.get("id")) or 0))
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for p in lists:
|
|
try:
|
|
local, dom_cid = self._resolve_name_and_domain(p)
|
|
except ConvertError:
|
|
print(
|
|
f"warning: skipping mailing list {pv_string(p.get('name'))!r} "
|
|
f"(cannot resolve its domain)",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
body: dict[str, Any] = {
|
|
"name": local,
|
|
"domainId": "#" + dom_cid,
|
|
"aliases": self._build_aliases(p, local, dom_cid),
|
|
"recipients": self._build_recipients(p),
|
|
}
|
|
desc = pv_string(p.get("description"))
|
|
if desc:
|
|
body["description"] = desc
|
|
tname = pv_string(p.get("tenant"))
|
|
if tname and tname in self.tenant_name_to_cid:
|
|
body["memberTenantId"] = "#" + self.tenant_name_to_cid[tname]
|
|
out[self._next_create_cid()] = body
|
|
return out
|
|
|
|
def _build_recipients(self, p: dict[str, Any]) -> dict[str, bool]:
|
|
out: dict[str, bool] = {}
|
|
for ref in pv_list(p.get("members")):
|
|
target = self._resolve_principal_ref(ref)
|
|
if target is None:
|
|
continue
|
|
if target.get("type") not in ("individual", "group"):
|
|
continue
|
|
try:
|
|
local, dom_cid = self._resolve_name_and_domain(target)
|
|
except ConvertError:
|
|
continue
|
|
dname = self.domain_cid_to_name.get(dom_cid)
|
|
if dname is None:
|
|
continue
|
|
out[f"{local}@{dname}"] = True
|
|
|
|
for addr in pv_list(p.get("externalMembers")):
|
|
if isinstance(addr, str) and "@" in addr:
|
|
out[addr] = True
|
|
return out
|
|
|
|
def _build_dkim_signatures(self) -> dict[str, dict[str, Any]]:
|
|
sigs = build_sub_trees(self.settings, "signature", "algorithm")
|
|
ids = sorted(sigs.keys())
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for sid in ids:
|
|
sub = sigs[sid]
|
|
algo = sub.get("algorithm", "").strip().lower()
|
|
if algo == "rsa-sha1":
|
|
continue
|
|
if algo == "ed25519-sha256":
|
|
tag = "Dkim1Ed25519Sha256"
|
|
elif algo == "rsa-sha256":
|
|
tag = "Dkim1RsaSha256"
|
|
else:
|
|
print(f"warning: skipping DKIM signature {sid!r}: "
|
|
f"unknown algorithm {algo!r}", file=sys.stderr)
|
|
continue
|
|
selector = sub.get("selector", "").strip()
|
|
if not selector:
|
|
print(f"warning: skipping DKIM signature {sid!r}: no selector",
|
|
file=sys.stderr)
|
|
continue
|
|
domain = sub.get("domain", "").strip().lower()
|
|
if domain not in self.domain_name_to_cid:
|
|
print(f"warning: skipping DKIM signature {sid!r}: "
|
|
f"unknown domain {domain!r}", file=sys.stderr)
|
|
continue
|
|
dom_cid = self.domain_name_to_cid[domain]
|
|
canon = sub.get("canonicalization", "relaxed/relaxed").strip().lower()
|
|
if not canon:
|
|
canon = "relaxed/relaxed"
|
|
body: dict[str, Any] = {
|
|
"@type": tag,
|
|
"canonicalization": canon,
|
|
"domainId": "#" + dom_cid,
|
|
"privateKey": secret_text(sub.get("private-key")),
|
|
"selector": selector,
|
|
}
|
|
t_cid = self.domain_cid_to_tenant_cid.get(dom_cid)
|
|
if t_cid is not None:
|
|
body["memberTenantId"] = "#" + t_cid
|
|
out[self._next_create_cid()] = body
|
|
return out
|
|
|
|
def _stores(self) -> dict[str, dict[str, str]]:
|
|
return build_sub_trees(self.settings, "store", "type")
|
|
|
|
def _referenced_store_id(self, key: str) -> str | None:
|
|
v = self.settings.get(key)
|
|
if v is None:
|
|
return None
|
|
v = v.strip()
|
|
return v or None
|
|
|
|
def _build_data_store(self) -> dict[str, Any] | None:
|
|
sid = self._referenced_store_id("storage.data")
|
|
if sid is None:
|
|
return None
|
|
stores = self._stores()
|
|
if sid not in stores:
|
|
raise ConvertError(
|
|
f"storage.data = {sid!r} but no store.{sid}.type is defined"
|
|
)
|
|
sub = stores[sid]
|
|
stype = sub.get("type", "").strip().lower()
|
|
if stype == "rocksdb":
|
|
return self._build_rocksdb(sub)
|
|
if stype == "sqlite":
|
|
return self._build_sqlite(sub)
|
|
if stype == "foundationdb":
|
|
return self._build_foundationdb(sub)
|
|
if stype == "postgresql":
|
|
return self._build_postgresql(sub)
|
|
if stype == "mysql":
|
|
return self._build_mysql(sub)
|
|
raise ConvertError(
|
|
f"storage.data points at store {sid!r} of unsupported type {stype!r} "
|
|
f"(DataStore requires rocksdb/sqlite/foundationdb/postgresql/mysql)"
|
|
)
|
|
|
|
def _build_blob_store(self) -> dict[str, Any] | None:
|
|
sid = self._referenced_store_id("storage.blob")
|
|
if sid is None:
|
|
return {"@type": "Default"}
|
|
data_sid = self._referenced_store_id("storage.data")
|
|
if sid == data_sid:
|
|
return {"@type": "Default"}
|
|
stores = self._stores()
|
|
if sid not in stores:
|
|
raise ConvertError(
|
|
f"storage.blob = {sid!r} but no store.{sid}.type is defined"
|
|
)
|
|
sub = stores[sid]
|
|
stype = sub.get("type", "").strip().lower()
|
|
if stype == "s3":
|
|
return self._build_s3(sub)
|
|
if stype == "azure":
|
|
return self._build_azure(sub)
|
|
if stype == "fs":
|
|
return self._build_fs(sub)
|
|
if stype == "foundationdb":
|
|
return self._build_foundationdb(sub, for_blob=True)
|
|
if stype == "postgresql":
|
|
return self._build_postgresql(sub, for_blob=True)
|
|
if stype == "mysql":
|
|
return self._build_mysql(sub, for_blob=True)
|
|
|
|
return {"@type": "Default"}
|
|
|
|
def _build_in_memory_store(self) -> dict[str, Any] | None:
|
|
sid = self._referenced_store_id("storage.lookup")
|
|
if sid is None:
|
|
return {"@type": "Default"}
|
|
data_sid = self._referenced_store_id("storage.data")
|
|
if sid == data_sid:
|
|
return {"@type": "Default"}
|
|
stores = self._stores()
|
|
if sid not in stores:
|
|
raise ConvertError(
|
|
f"storage.lookup = {sid!r} but no store.{sid}.type is defined"
|
|
)
|
|
sub = stores[sid]
|
|
stype = sub.get("type", "").strip().lower()
|
|
if stype == "redis":
|
|
redis_type = sub.get("redis-type", "single").strip().lower()
|
|
if redis_type == "cluster":
|
|
return self._build_redis_cluster(sub)
|
|
return self._build_redis_single(sub)
|
|
return {"@type": "Default"}
|
|
|
|
def _build_search_store(self) -> dict[str, Any] | None:
|
|
sid = self._referenced_store_id("storage.fts")
|
|
if sid is None:
|
|
return {"@type": "Default"}
|
|
data_sid = self._referenced_store_id("storage.data")
|
|
if sid == data_sid:
|
|
return {"@type": "Default"}
|
|
stores = self._stores()
|
|
if sid not in stores:
|
|
raise ConvertError(
|
|
f"storage.fts = {sid!r} but no store.{sid}.type is defined"
|
|
)
|
|
sub = stores[sid]
|
|
stype = sub.get("type", "").strip().lower()
|
|
if stype == "elasticsearch":
|
|
return self._build_elasticsearch(sub)
|
|
if stype == "meilisearch":
|
|
return self._build_meilisearch(sub)
|
|
if stype == "foundationdb":
|
|
return self._build_foundationdb(sub, for_search=True)
|
|
if stype == "postgresql":
|
|
return self._build_postgresql(sub, for_search=True)
|
|
if stype == "mysql":
|
|
return self._build_mysql(sub, for_search=True)
|
|
return {"@type": "Default"}
|
|
|
|
def _build_rocksdb(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
path = sub.get("path", "").strip()
|
|
if not path:
|
|
raise ConvertError("rocksdb store missing required 'path'")
|
|
body: dict[str, Any] = {"@type": "RocksDb", "path": path}
|
|
bs = parse_size_bytes(sub.get("settings.min-blob-size"))
|
|
if bs is not None:
|
|
body["blobSize"] = bs
|
|
wb = parse_size_bytes(sub.get("settings.write-buffer-size"))
|
|
if wb is not None:
|
|
body["bufferSize"] = wb
|
|
pw = parse_int(sub.get("pool.workers"))
|
|
if pw is not None:
|
|
body["poolWorkers"] = pw
|
|
return body
|
|
|
|
def _build_sqlite(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
path = sub.get("path", "").strip()
|
|
if not path:
|
|
raise ConvertError("sqlite store missing required 'path'")
|
|
body: dict[str, Any] = {"@type": "Sqlite", "path": path}
|
|
pmc = parse_int(sub.get("pool.max-connections"))
|
|
if pmc is not None:
|
|
body["poolMaxConnections"] = pmc
|
|
pw = parse_int(sub.get("pool.workers"))
|
|
if pw is not None:
|
|
body["poolWorkers"] = pw
|
|
return body
|
|
|
|
def _build_foundationdb(
|
|
self,
|
|
sub: dict[str, str],
|
|
*,
|
|
for_blob: bool = False,
|
|
for_search: bool = False,
|
|
) -> dict[str, Any]:
|
|
body: dict[str, Any] = {"@type": "FoundationDb"}
|
|
cf = sub.get("cluster-file", "").strip()
|
|
if cf:
|
|
body["clusterFile"] = cf
|
|
dc = sub.get("ids.datacenter", "").strip()
|
|
if dc:
|
|
body["datacenterId"] = dc
|
|
mid = sub.get("ids.machine", "").strip()
|
|
if mid:
|
|
body["machineId"] = mid
|
|
trd = parse_duration_ms(sub.get("transaction.max-retry-delay"))
|
|
if trd is not None:
|
|
body["transactionRetryDelay"] = trd
|
|
trl = parse_int(sub.get("transaction.retry-limit"))
|
|
if trl is not None:
|
|
body["transactionRetryLimit"] = trl
|
|
tt = parse_duration_ms(sub.get("transaction.timeout"))
|
|
if tt is not None:
|
|
body["transactionTimeout"] = tt
|
|
return body
|
|
|
|
def _build_sql_common(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
out: dict[str, Any] = {}
|
|
host = sub.get("host", "").strip()
|
|
if not host:
|
|
raise ConvertError("SQL store missing required 'host'")
|
|
out["host"] = host
|
|
db = sub.get("database", "").strip()
|
|
if not db:
|
|
raise ConvertError("SQL store missing required 'database'")
|
|
out["database"] = db
|
|
port = parse_int(sub.get("port"))
|
|
if port is not None:
|
|
out["port"] = port
|
|
user = sub.get("user", "").strip()
|
|
if user:
|
|
out["authUsername"] = user
|
|
out["authSecret"] = secret_key_optional(sub.get("password"))
|
|
tls_enable = parse_bool(sub.get("tls.enable"))
|
|
if tls_enable is not None:
|
|
out["useTls"] = tls_enable
|
|
tls_invalid = parse_bool(sub.get("tls.allow-invalid-certs"))
|
|
if tls_invalid is not None:
|
|
out["allowInvalidCerts"] = tls_invalid
|
|
tout = parse_duration_ms(sub.get("timeout"))
|
|
if tout is not None:
|
|
out["timeout"] = tout
|
|
pmc = parse_int(sub.get("pool.max-connections"))
|
|
if pmc is not None:
|
|
out["poolMaxConnections"] = pmc
|
|
return out
|
|
|
|
def _build_postgresql(
|
|
self,
|
|
sub: dict[str, str],
|
|
*,
|
|
for_blob: bool = False,
|
|
for_search: bool = False,
|
|
) -> dict[str, Any]:
|
|
body = {"@type": "PostgreSql"}
|
|
body.update(self._build_sql_common(sub))
|
|
return body
|
|
|
|
def _build_mysql(
|
|
self,
|
|
sub: dict[str, str],
|
|
*,
|
|
for_blob: bool = False,
|
|
for_search: bool = False,
|
|
) -> dict[str, Any]:
|
|
body = {"@type": "MySql"}
|
|
body.update(self._build_sql_common(sub))
|
|
map_ = parse_size_bytes(sub.get("max-allowed-packet"))
|
|
if map_ is not None:
|
|
body["maxAllowedPacket"] = map_
|
|
pmin = parse_int(sub.get("pool.min-connections"))
|
|
if pmin is not None:
|
|
body["poolMinConnections"] = pmin
|
|
return body
|
|
|
|
def _build_s3(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
body: dict[str, Any] = {"@type": "S3"}
|
|
bucket = sub.get("bucket", "").strip()
|
|
if not bucket:
|
|
raise ConvertError("s3 store missing required 'bucket'")
|
|
body["bucket"] = bucket
|
|
ak = sub.get("access-key", "").strip()
|
|
if ak:
|
|
body["accessKey"] = ak
|
|
body["secretKey"] = secret_key_optional(sub.get("secret-key"))
|
|
body["securityToken"] = secret_key_optional(sub.get("security-token"))
|
|
profile = sub.get("profile", "").strip()
|
|
if profile:
|
|
body["profile"] = profile
|
|
kp = sub.get("key-prefix", "").strip()
|
|
if kp:
|
|
body["keyPrefix"] = kp
|
|
mr = parse_int(sub.get("max-retries"))
|
|
if mr is not None:
|
|
body["maxRetries"] = mr
|
|
to = parse_duration_ms(sub.get("timeout"))
|
|
if to is not None:
|
|
body["timeout"] = to
|
|
region_raw = sub.get("region", "").strip().lower()
|
|
endpoint = sub.get("endpoint", "").strip()
|
|
if endpoint:
|
|
body["region"] = {
|
|
"@type": "Custom",
|
|
"customEndpoint": endpoint,
|
|
"customRegion": region_raw or "custom",
|
|
}
|
|
elif region_raw in _S3_REGION_MAP:
|
|
body["region"] = {"@type": _S3_REGION_MAP[region_raw]}
|
|
elif region_raw:
|
|
|
|
body["region"] = {
|
|
"@type": "Custom",
|
|
"customEndpoint": "",
|
|
"customRegion": region_raw,
|
|
}
|
|
return body
|
|
|
|
def _build_azure(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
body: dict[str, Any] = {"@type": "Azure"}
|
|
sa = sub.get("storage-account", "").strip()
|
|
if not sa:
|
|
raise ConvertError("azure store missing required 'storage-account'")
|
|
body["storageAccount"] = sa
|
|
cont = sub.get("container", "").strip()
|
|
if not cont:
|
|
raise ConvertError("azure store missing required 'container'")
|
|
body["container"] = cont
|
|
body["accessKey"] = secret_key_optional(sub.get("azure-access-key"))
|
|
body["sasToken"] = secret_key_optional(sub.get("sas-token"))
|
|
kp = sub.get("key-prefix", "").strip()
|
|
if kp:
|
|
body["keyPrefix"] = kp
|
|
mr = parse_int(sub.get("max-retries"))
|
|
if mr is not None:
|
|
body["maxRetries"] = mr
|
|
to = parse_duration_ms(sub.get("timeout"))
|
|
if to is not None:
|
|
body["timeout"] = to
|
|
return body
|
|
|
|
def _build_fs(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
path = sub.get("path", "").strip()
|
|
if not path:
|
|
raise ConvertError("fs store missing required 'path'")
|
|
body: dict[str, Any] = {"@type": "FileSystem", "path": path}
|
|
depth = parse_int(sub.get("depth"))
|
|
if depth is not None:
|
|
body["depth"] = depth
|
|
return body
|
|
|
|
def _build_redis_single(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
urls = collect_array(sub, "urls")
|
|
if not urls:
|
|
raise ConvertError("redis store missing required 'urls'")
|
|
body: dict[str, Any] = {"@type": "Redis", "url": urls[0]}
|
|
to = parse_duration_ms(sub.get("timeout"))
|
|
if to is not None:
|
|
body["timeout"] = to
|
|
return body
|
|
|
|
def _build_redis_cluster(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
urls = collect_array(sub, "urls")
|
|
if not urls:
|
|
raise ConvertError("redis-cluster store missing required 'urls'")
|
|
body: dict[str, Any] = {
|
|
"@type": "RedisCluster",
|
|
"urls": {u: True for u in urls},
|
|
}
|
|
body["authSecret"] = secret_key_optional(sub.get("password"))
|
|
user = sub.get("user", "").strip()
|
|
if user:
|
|
body["authUsername"] = user
|
|
mr = parse_int(sub.get("retry.total"))
|
|
if mr is not None:
|
|
body["maxRetries"] = mr
|
|
mxw = parse_duration_ms(sub.get("retry.max-wait"))
|
|
if mxw is not None:
|
|
body["maxRetryWait"] = mxw
|
|
mnw = parse_duration_ms(sub.get("retry.min-wait"))
|
|
if mnw is not None:
|
|
body["minRetryWait"] = mnw
|
|
rfr = parse_bool(sub.get("read-from-replicas"))
|
|
if rfr is not None:
|
|
body["readFromReplicas"] = rfr
|
|
pv = sub.get("protocol-version", "").strip().lower()
|
|
if pv in _REDIS_PROTOCOL_MAP:
|
|
body["protocolVersion"] = _REDIS_PROTOCOL_MAP[pv]
|
|
to = parse_duration_ms(sub.get("timeout"))
|
|
if to is not None:
|
|
body["timeout"] = to
|
|
return body
|
|
|
|
def _build_http_auth(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
token = sub.get("auth.token", "").strip()
|
|
if token:
|
|
return {"@type": "Bearer", "bearerToken": secret_key(token)}
|
|
username = sub.get("auth.username", "").strip()
|
|
secret = sub.get("auth.secret", "")
|
|
if username:
|
|
return {"@type": "Basic", "username": username,
|
|
"secret": secret_key(secret)}
|
|
return {"@type": "Unauthenticated"}
|
|
|
|
def _build_elasticsearch(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
url = sub.get("url", "").strip()
|
|
if not url:
|
|
raise ConvertError("elasticsearch store missing required 'url'")
|
|
body: dict[str, Any] = {
|
|
"@type": "ElasticSearch",
|
|
"url": url,
|
|
"httpAuth": self._build_http_auth(sub),
|
|
}
|
|
aic = parse_bool(sub.get("tls.allow-invalid-certs"))
|
|
if aic is not None:
|
|
body["allowInvalidCerts"] = aic
|
|
nr = parse_int(sub.get("index.replicas"))
|
|
if nr is not None:
|
|
body["numReplicas"] = nr
|
|
ns = parse_int(sub.get("index.shards"))
|
|
if ns is not None:
|
|
body["numShards"] = ns
|
|
return body
|
|
|
|
def _build_meilisearch(self, sub: dict[str, str]) -> dict[str, Any]:
|
|
url = sub.get("url", "").strip()
|
|
if not url:
|
|
raise ConvertError("meilisearch store missing required 'url'")
|
|
body: dict[str, Any] = {
|
|
"@type": "Meilisearch",
|
|
"url": url,
|
|
"httpAuth": self._build_http_auth(sub),
|
|
}
|
|
aic = parse_bool(sub.get("tls.allow-invalid-certs"))
|
|
if aic is not None:
|
|
body["allowInvalidCerts"] = aic
|
|
pi = parse_duration_ms(sub.get("task.poll-interval"))
|
|
if pi is not None:
|
|
body["pollInterval"] = pi
|
|
return body
|
|
|
|
def _build_enterprise(self) -> dict[str, Any] | None:
|
|
lk = self.settings.get("enterprise.license-key", "").strip()
|
|
ak = self.settings.get("enterprise.api-key", "").strip()
|
|
lu = self.settings.get("enterprise.logo-url", "").strip()
|
|
if not (lk or ak or lu):
|
|
return None
|
|
body: dict[str, Any] = {
|
|
"licenseKey": secret_key_optional(lk),
|
|
"apiKey": secret_key_optional(ak),
|
|
}
|
|
if lu:
|
|
body["logoUrl"] = lu
|
|
return body
|
|
|
|
def _build_system_settings(self) -> dict[str, Any] | None:
|
|
if self.default_domain_cid is None:
|
|
return None
|
|
hostname = self.settings.get("server.hostname", "").strip()
|
|
body: dict[str, Any] = {
|
|
"defaultDomainId": "#" + self.default_domain_cid,
|
|
"defaultHostname": hostname,
|
|
}
|
|
return body
|
|
|
|
def _build_certificates(self) -> dict[str, dict[str, Any]]:
|
|
out: dict[str, dict[str, Any]] = {}
|
|
|
|
for sid, sub in sorted(
|
|
build_sub_trees(self.settings, "certificate", "cert").items()
|
|
):
|
|
cert = sub.get("cert", "").strip()
|
|
key = sub.get("private-key", "").strip()
|
|
if not cert or not key:
|
|
print(
|
|
f"warning: skipping certificate.{sid}: "
|
|
"missing cert or private-key",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
if "-----BEGIN " not in cert or "-----BEGIN " not in key:
|
|
print(
|
|
f"warning: skipping certificate.{sid}: value is not PEM "
|
|
"(likely a file or env placeholder)",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
out[self._next_create_cid()] = _make_certificate_object(cert, key)
|
|
|
|
for sid, sub in sorted(
|
|
build_sub_trees(self.settings, "acme", "cert").items()
|
|
):
|
|
blob_b64 = sub.get("cert", "").strip()
|
|
if not blob_b64:
|
|
continue
|
|
padded = blob_b64 + "=" * (-len(blob_b64) % 4)
|
|
blob = None
|
|
for decoder in (base64.b64decode, base64.urlsafe_b64decode):
|
|
try:
|
|
blob = decoder(padded).decode("latin-1")
|
|
break
|
|
except Exception as exc:
|
|
last_err = exc
|
|
if blob is None:
|
|
print(
|
|
f"warning: skipping acme.{sid}: base64 decode failed: {last_err}",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
cert_pem, key_pem = split_pem_bundle(blob)
|
|
if not cert_pem or not key_pem:
|
|
print(
|
|
f"warning: skipping acme.{sid}: decoded bundle lacks "
|
|
"cert or private key",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
out[self._next_create_cid()] = _make_certificate_object(
|
|
cert_pem, key_pem
|
|
)
|
|
|
|
return out
|
|
|
|
def _check_duplicate_emails(
|
|
self,
|
|
accounts: dict[str, dict[str, Any]],
|
|
mailing_lists: dict[str, dict[str, Any]],
|
|
) -> None:
|
|
owners: dict[tuple[str, str], str] = {}
|
|
|
|
def claim(local: str, domain_ref: str, owner: str) -> None:
|
|
d_cid = domain_ref[1:] if domain_ref.startswith("#") else domain_ref
|
|
key = (local, d_cid)
|
|
if key in owners:
|
|
dname = self.domain_cid_to_name.get(d_cid, d_cid)
|
|
raise ConvertError(
|
|
f"duplicate email address {local}@{dname!s} — "
|
|
f"claimed by both {owners[key]} and {owner}"
|
|
)
|
|
owners[key] = owner
|
|
|
|
for cid, obj in accounts.items():
|
|
kind = obj.get("@type", "Account")
|
|
claim(obj["name"], obj["domainId"], f"{kind} {cid} ({obj['name']})")
|
|
for alias in obj.get("aliases", {}).values():
|
|
claim(alias["name"], alias["domainId"],
|
|
f"alias of {kind} {cid} ({obj['name']})")
|
|
|
|
for cid, obj in mailing_lists.items():
|
|
claim(obj["name"], obj["domainId"],
|
|
f"MailingList {cid} ({obj['name']})")
|
|
for alias in obj.get("aliases", {}).values():
|
|
claim(alias["name"], alias["domainId"],
|
|
f"alias of MailingList {cid} ({obj['name']})")
|
|
|
|
SINGLETON_ORDER = [
|
|
"SystemSettings",
|
|
"Enterprise",
|
|
"BlobStore",
|
|
"InMemoryStore",
|
|
"SearchStore",
|
|
]
|
|
|
|
COLLECTION_ORDER = [
|
|
"Tenant",
|
|
"Domain",
|
|
"Account",
|
|
"MailingList",
|
|
"DkimSignature",
|
|
"Certificate",
|
|
]
|
|
|
|
def build_export_ops(result: dict[str, Any]) -> list[dict[str, Any]]:
|
|
ops: list[dict[str, Any]] = []
|
|
for name in COLLECTION_ORDER:
|
|
if name not in result:
|
|
continue
|
|
records: dict[str, dict[str, Any]] = result[name]
|
|
if not records:
|
|
continue
|
|
if name == "Account":
|
|
groups = {c: r for c, r in records.items() if r.get("@type") == "Group"}
|
|
users = {c: r for c, r in records.items() if r.get("@type") == "User"}
|
|
if groups:
|
|
ops.append({"@type": "create", "object": name, "value": groups})
|
|
if users:
|
|
ops.append({"@type": "create", "object": name, "value": users})
|
|
else:
|
|
ops.append({"@type": "create", "object": name, "value": records})
|
|
for name in SINGLETON_ORDER:
|
|
if name in result:
|
|
ops.append({
|
|
"@type": "update",
|
|
"object": name,
|
|
"value": result[name],
|
|
})
|
|
return ops
|
|
|
|
def cmd_convert(args: argparse.Namespace) -> int:
|
|
with open(args.settings, "r", encoding="utf-8") as f:
|
|
settings = json.load(f)
|
|
if not isinstance(settings, dict):
|
|
print(f"error: {args.settings} is not a JSON object", file=sys.stderr)
|
|
return 2
|
|
settings = {str(k): ("" if v is None else str(v)) for k, v in settings.items()}
|
|
|
|
with open(args.principals, "r", encoding="utf-8") as f:
|
|
principals = json.load(f)
|
|
if not isinstance(principals, list):
|
|
print(f"error: {args.principals} is not a JSON array", file=sys.stderr)
|
|
return 2
|
|
|
|
conv = Converter(principals, settings)
|
|
result = conv.run()
|
|
|
|
data_store = result.pop("DataStore", None)
|
|
if data_store is None:
|
|
raise ConvertError(
|
|
"DataStore could not be built (storage.data missing or invalid)"
|
|
)
|
|
with open(args.config, "w", encoding="utf-8") as f:
|
|
json.dump(data_store, f, indent=2, ensure_ascii=False)
|
|
print(f"wrote {args.config} (DataStore: @type={data_store.get('@type')!r})",
|
|
file=sys.stderr)
|
|
|
|
ops = build_export_ops(result)
|
|
with open(args.output, "w", encoding="utf-8") as f:
|
|
json.dump(ops, f, indent=2, ensure_ascii=False)
|
|
print(f"wrote {args.output} ({len(ops)} ops)", file=sys.stderr)
|
|
for op in ops:
|
|
kind = op["@type"]
|
|
name = op["object"]
|
|
if kind == "update":
|
|
print(f" update {name}", file=sys.stderr)
|
|
else:
|
|
print(f" create {name}: {len(op['value'])} records",
|
|
file=sys.stderr)
|
|
return 0
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(
|
|
description="Dump / migrate a Stalwart server via its management API.",
|
|
)
|
|
sub = p.add_subparsers(dest="command", required=True)
|
|
|
|
d = sub.add_parser("dump", help="Dump settings and principals to JSON files.")
|
|
d.add_argument("--url", required=True,
|
|
help="Base URL of the Stalwart server, e.g. https://mail.example.com")
|
|
d.add_argument("--token", help="Bearer token.")
|
|
d.add_argument("--username", help="Admin username for HTTP Basic auth.")
|
|
d.add_argument("--password", help="Admin password for HTTP Basic auth.")
|
|
d.add_argument("--settings", default="settings.json",
|
|
help="Output file for settings (default: settings.json).")
|
|
d.add_argument("--principals", default="principals.json",
|
|
help="Output file for principals (default: principals.json).")
|
|
d.set_defaults(func=cmd_dump)
|
|
|
|
c = sub.add_parser("convert",
|
|
help="Convert dumped JSON into the new JMAP-object format.")
|
|
c.add_argument("--settings", default="settings.json",
|
|
help="Input settings JSON (default: settings.json).")
|
|
c.add_argument("--principals", default="principals.json",
|
|
help="Input principals JSON (default: principals.json).")
|
|
c.add_argument("--config", default="config.json",
|
|
help="Output file for the DataStore object "
|
|
"(default: config.json).")
|
|
c.add_argument("--output", default="export.json",
|
|
help="Output file for the operations array "
|
|
"(default: export.json).")
|
|
c.set_defaults(func=cmd_convert)
|
|
|
|
return p
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
try:
|
|
return args.func(args)
|
|
except (ApiError, ConvertError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
except KeyboardInterrupt:
|
|
return 130
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|