247 lines
11 KiB
Python
Executable File
247 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Convert an existing Z-Push installation's configuration into ActivePush settings JSON.
|
|
|
|
Usage:
|
|
python3 zpush-import.py /usr/share/z-push # source layout (config.php, backend/*/config.php)
|
|
python3 zpush-import.py /etc/z-push # Debian package layout (*.conf.php)
|
|
python3 zpush-import.py /var/www/z-push > zpush.json
|
|
active-push config apply zpush.json
|
|
|
|
Only settings are converted. Device sync states are not portable: devices notice the new
|
|
server (unknown sync key / policy key) and resynchronise automatically.
|
|
"""
|
|
import configparser
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
|
|
DEFINE = re.compile(r"""^\s*define\s*\(\s*['"]([A-Z0-9_]+)['"]\s*,\s*(.+?)\)\s*;""", re.S)
|
|
|
|
|
|
def php_value(raw):
|
|
raw = raw.strip()
|
|
if raw.startswith(("'", '"')) and raw.endswith(raw[0]):
|
|
return raw[1:-1].replace("\\'", "'").replace('\\"', '"')
|
|
low = raw.lower()
|
|
if low in ("true", "false"):
|
|
return low == "true"
|
|
if re.fullmatch(r"-?\d+", raw):
|
|
return int(raw)
|
|
if low.startswith("array(") or raw.startswith("["):
|
|
inner = raw[raw.index("(") + 1: raw.rindex(")")] if low.startswith("array(") else raw[1:-1]
|
|
out = {}
|
|
for k, v in re.findall(r"""['"]([^'"]+)['"]\s*=>\s*('[^']*'|"[^"]*"|[^,\)]+)""", inner):
|
|
out[k] = php_value(v)
|
|
return out
|
|
if low.startswith("serialize("):
|
|
return None
|
|
return {"__const__": raw}
|
|
|
|
|
|
def read_defines(paths):
|
|
defs = {}
|
|
for p in paths:
|
|
text = p.read_text(errors="replace")
|
|
# strip block comments and line comments
|
|
text = re.sub(r"/\*.*?\*/", "", text, flags=re.S)
|
|
lines = [l for l in text.splitlines() if not l.strip().startswith(("//", "#"))]
|
|
joined = "\n".join(lines)
|
|
for m in re.finditer(r"define\s*\(.*?\)\s*;", joined, re.S):
|
|
d = DEFINE.match(m.group(0))
|
|
if d:
|
|
defs[d.group(1)] = php_value(d.group(2))
|
|
return defs
|
|
|
|
|
|
def find_configs(root):
|
|
root = pathlib.Path(root)
|
|
files = []
|
|
for pat in ["config.php", "*.conf.php", "backend/*/config.php", "autodiscover/config.php", "src/config.php", "src/backend/*/config.php", "src/autodiscover/config.php"]:
|
|
files += [f for f in root.glob(pat) if f.is_file()]
|
|
policies = [f for f in [root / "policies.ini", root / "src/policies.ini"] if f.is_file()]
|
|
return sorted(set(files)), policies
|
|
|
|
|
|
FILTER = {"SYNC_FILTERTYPE_ALL": 0, "SYNC_FILTERTYPE_1DAY": 1, "SYNC_FILTERTYPE_3DAYS": 2, "SYNC_FILTERTYPE_1WEEK": 3,
|
|
"SYNC_FILTERTYPE_2WEEKS": 4, "SYNC_FILTERTYPE_1MONTH": 5, "SYNC_FILTERTYPE_3MONTHS": 6, "SYNC_FILTERTYPE_6MONTHS": 7}
|
|
|
|
POLICY_KEYS = {
|
|
"devpwenabled": bool, "alphanumpwreq": bool, "devencenabled": bool, "attenabled": bool, "mindevpwlenngth": int,
|
|
"maxinacttimedevlock": int, "maxdevpwfailedattempts": int, "maxattsize": int, "allowsimpledevpw": bool,
|
|
"devpwexpiration": int, "devpwhistory": int, "allowstoragecard": bool, "allowcam": bool, "reqdevenc": bool,
|
|
"allowunsignedapps": bool, "mindevcomplexchars": int, "allowwifi": bool, "allowtextmessaging": bool,
|
|
"allowpopimapemail": bool, "allowbluetooth": int, "allowirda": bool, "reqmansyncroam": bool, "maxcalagefilter": int,
|
|
"allowhtmlemail": bool, "maxemailagefilter": int, "maxemailbodytruncsize": int, "maxemailhtmlbodytruncsize": int,
|
|
"reqsignedsmimemessages": bool, "reqencsmimemessages": bool, "reqsignedsmimealgorithm": int,
|
|
"reqencsmimealgorithm": int, "allowsmimeencalgneg": int, "allowsmimesoftcerts": bool, "allowbrowser": bool,
|
|
"allowconsumeremail": bool, "allowinternetsharing": bool, "allowremotedesktop": bool, "allowdesktopsync": bool,
|
|
"allowunsignedinstallationpackages": bool, "pwrecoveryenabled": bool,
|
|
}
|
|
|
|
|
|
def parse_policies(path):
|
|
cp = configparser.ConfigParser(interpolation=None, strict=False)
|
|
cp.read_string(path.read_text(errors="replace"))
|
|
out = []
|
|
for sec in cp.sections():
|
|
p = {"name": sec}
|
|
for k, typ in POLICY_KEYS.items():
|
|
if k in cp[sec]:
|
|
v = cp[sec][k].strip().strip("'\"")
|
|
if v == "":
|
|
continue
|
|
try:
|
|
p[k] = (int(v) != 0) if typ is bool else int(v)
|
|
except ValueError:
|
|
pass
|
|
out.append(p)
|
|
return out
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
print(__doc__, file=sys.stderr)
|
|
sys.exit(2)
|
|
files, pol_files = find_configs(sys.argv[1])
|
|
if not files:
|
|
sys.exit(f"no Z-Push configuration files found in {sys.argv[1]}")
|
|
d = read_defines(files)
|
|
notes = [f"read {f}" for f in files]
|
|
s = {"general": {}, "imap": {}, "smtp": {}, "caldav": {}, "carddav": {}, "sync": {}, "ping": {}, "provisioning": {}, "autodiscover": {}}
|
|
|
|
if isinstance(d.get("TIMEZONE"), str) and d["TIMEZONE"]:
|
|
s["general"]["default_timezone"] = d["TIMEZONE"]
|
|
if "USE_FULLEMAIL_FOR_LOGIN" in d:
|
|
s["general"]["login_format"] = "as_given" if d["USE_FULLEMAIL_FOR_LOGIN"] else "local_part"
|
|
|
|
# IMAP
|
|
if "IMAP_SERVER" in d:
|
|
s["imap"]["host"] = d["IMAP_SERVER"]
|
|
if "IMAP_PORT" in d:
|
|
s["imap"]["port"] = int(d["IMAP_PORT"])
|
|
opts = str(d.get("IMAP_OPTIONS", "")).lower()
|
|
if opts:
|
|
if "/ssl" in opts:
|
|
s["imap"]["security"] = "tls"
|
|
elif "/tls" in opts:
|
|
s["imap"]["security"] = "start_tls"
|
|
elif "/notls" in opts:
|
|
s["imap"]["security"] = "none"
|
|
s["imap"]["verify_tls"] = "novalidate-cert" not in opts
|
|
for zk, ak in [("IMAP_FOLDER_PREFIX", "folder_prefix"), ("IMAP_FOLDER_INBOX", "folder_inbox"), ("IMAP_FOLDER_SENT", "folder_sent"),
|
|
("IMAP_FOLDER_DRAFT", "folder_drafts"), ("IMAP_FOLDER_TRASH", "folder_trash"), ("IMAP_FOLDER_SPAM", "folder_spam"),
|
|
("IMAP_FOLDER_ARCHIVE", "folder_archive")]:
|
|
if isinstance(d.get(zk), str):
|
|
s["imap"][ak] = d[zk]
|
|
if "IMAP_AUTOSEEN_ON_DELETE" in d:
|
|
s["imap"]["autoseen_on_delete"] = bool(d["IMAP_AUTOSEEN_ON_DELETE"])
|
|
if "IMAP_INLINE_FORWARD" in d:
|
|
s["imap"]["inline_forward"] = bool(d["IMAP_INLINE_FORWARD"])
|
|
if isinstance(d.get("IMAP_EXCLUDED_FOLDERS"), str) and d["IMAP_EXCLUDED_FOLDERS"]:
|
|
s["imap"]["excluded_folders"] = [x for x in d["IMAP_EXCLUDED_FOLDERS"].split("|") if x]
|
|
df = d.get("IMAP_DEFAULTFROM")
|
|
if isinstance(df, str):
|
|
if df.startswith("@"):
|
|
s["general"]["email_template"] = "{local}" + df
|
|
elif df == "domain":
|
|
s["general"]["email_template"] = "{local}@{domain}"
|
|
elif df in ("", "username"):
|
|
s["general"]["email_template"] = "{login}"
|
|
elif df in ("sql", "ldap"):
|
|
notes.append("IMAP_DEFAULTFROM sql/ldap is not supported; using {login} - adjust email_template")
|
|
|
|
# SMTP
|
|
method = str(d.get("IMAP_SMTP_METHOD", "")).lower()
|
|
if method in ("mail", "sendmail"):
|
|
s["smtp"]["method"] = "sendmail"
|
|
notes.append("SMTP method 'mail'/'sendmail' -> sendmail (consider SMTP submission on port 587 instead)")
|
|
elif method == "smtp":
|
|
s["smtp"]["method"] = "smtp"
|
|
params = d.get("IMAP_SMTP_PARAMS") if isinstance(d.get("IMAP_SMTP_PARAMS"), dict) else {}
|
|
host = str(params.get("host", "localhost"))
|
|
if host.startswith("ssl://") or host.startswith("tls://"):
|
|
s["smtp"]["security"] = "tls"
|
|
host = host.split("://", 1)[1]
|
|
s["smtp"]["host"] = host
|
|
s["smtp"]["port"] = int(params.get("port", 25))
|
|
if s["smtp"]["port"] == 587:
|
|
s["smtp"]["security"] = "start_tls"
|
|
if params.get("auth"):
|
|
if params.get("username") == "imap_username":
|
|
s["smtp"]["auth"] = "user_credentials"
|
|
else:
|
|
s["smtp"]["auth"] = "static"
|
|
s["smtp"]["username"] = str(params.get("username", ""))
|
|
s["smtp"]["password"] = str(params.get("password", ""))
|
|
else:
|
|
s["smtp"]["auth"] = "none"
|
|
if params.get("verify_peer") is False:
|
|
s["smtp"]["verify_tls"] = False
|
|
|
|
# CalDAV / CardDAV
|
|
def dav(prefix, key):
|
|
server = d.get(f"{prefix}_SERVER")
|
|
if not isinstance(server, str) or server in ("", "caldavserver.domain.com"):
|
|
return
|
|
proto = d.get(f"{prefix}_PROTOCOL", "https")
|
|
port = d.get(f"{prefix}_PORT", 443 if proto == "https" else 80)
|
|
default_port = (proto == "https" and str(port) == "443") or (proto == "http" and str(port) == "80")
|
|
s[key]["enabled"] = True
|
|
s[key]["base_url"] = f"{proto}://{server}" + ("" if default_port else f":{port}")
|
|
if isinstance(d.get(f"{prefix}_PATH"), str):
|
|
s[key]["path"] = d[f"{prefix}_PATH"]
|
|
if f"{prefix}_PATH_INCLUDE" in d and isinstance(d[f"{prefix}_PATH_INCLUDE"], str):
|
|
s[key]["include_regex"] = d[f"{prefix}_PATH_INCLUDE"].strip("/")
|
|
if f"{prefix}_PATH_EXCLUDE" in d and isinstance(d[f"{prefix}_PATH_EXCLUDE"], str):
|
|
s[key]["exclude_regex"] = d[f"{prefix}_PATH_EXCLUDE"].strip("/")
|
|
ssl = d.get("DAV_SSL_VERIFY")
|
|
if isinstance(ssl, dict) and "NONE" in ssl.get("__const__", ""):
|
|
s[key]["verify_tls"] = False
|
|
dav("CALDAV", "caldav")
|
|
dav("CARDDAV", "carddav")
|
|
if isinstance(d.get("CARDDAV_GAL_PATH"), str):
|
|
s["carddav"]["gal_path"] = d["CARDDAV_GAL_PATH"]
|
|
if isinstance(d.get("CARDDAV_GAL_MIN_LENGTH"), int):
|
|
s["carddav"]["gal_min_length"] = d["CARDDAV_GAL_MIN_LENGTH"]
|
|
if isinstance(d.get("CARDDAV_DEFAULT_PATH"), str):
|
|
s["carddav"]["default_collection"] = d["CARDDAV_DEFAULT_PATH"].rstrip("/").rsplit("/", 1)[-1]
|
|
|
|
# sync / ping / provisioning
|
|
if isinstance(d.get("SYNC_MAX_ITEMS"), int):
|
|
s["sync"]["max_items"] = d["SYNC_MAX_ITEMS"]
|
|
ft = d.get("SYNC_FILTERTIME_MAX")
|
|
if isinstance(ft, dict) and ft.get("__const__") in FILTER:
|
|
s["sync"]["filter_time_max"] = FILTER[ft["__const__"]]
|
|
sc = d.get("SYNC_CONFLICT_DEFAULT")
|
|
if isinstance(sc, dict):
|
|
s["sync"]["server_wins"] = "OVERWRITE_PIM" in sc["__const__"]
|
|
for zk, sec, ak in [("PING_INTERVAL", "ping", "check_interval_secs"), ("PING_LOWER_BOUND_LIFETIME", "ping", "lifetime_min"),
|
|
("PING_HIGHER_BOUND_LIFETIME", "ping", "lifetime_max"), ("RETRY_AFTER_DELAY", "sync", "retry_after_secs"),
|
|
("SEARCH_WAIT", "sync", "search_wait_secs"), ("SEARCH_MAXRESULTS", "sync", "search_max_results")]:
|
|
if isinstance(d.get(zk), int) and not isinstance(d.get(zk), bool):
|
|
s[sec][ak] = d[zk]
|
|
if isinstance(d.get("SYNC_CONTACTS_MAXPICTURESIZE"), int):
|
|
s["carddav"]["max_picture_kb"] = max(1, d["SYNC_CONTACTS_MAXPICTURESIZE"] // 1024)
|
|
if "PROVISIONING" in d:
|
|
s["provisioning"]["enabled"] = bool(d["PROVISIONING"])
|
|
if "LOOSE_PROVISIONING" in d:
|
|
s["provisioning"]["loose"] = bool(d["LOOSE_PROVISIONING"])
|
|
if pol_files:
|
|
pols = parse_policies(pol_files[0])
|
|
if pols:
|
|
s["provisioning"]["policies"] = pols
|
|
s["provisioning"]["default_policy"] = "default" if any(p["name"] == "default" for p in pols) else pols[0]["name"]
|
|
notes.append(f"imported {len(pols)} polic{'y' if len(pols) == 1 else 'ies'} from {pol_files[0]}")
|
|
if isinstance(d.get("ZPUSH_HOST"), str) and d["ZPUSH_HOST"]:
|
|
s["autodiscover"]["server_url"] = f"https://{d['ZPUSH_HOST']}/Microsoft-Server-ActiveSync"
|
|
|
|
s = {k: v for k, v in s.items() if v}
|
|
for n in notes:
|
|
print(f"# {n}", file=sys.stderr)
|
|
print(json.dumps(s, indent=2, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|