"""Reading and updating patchbay.conf (shared with patchbayd, see config.c).""" import os import tempfile DEFAULTS = { "role": "client", "database": "/etc/patchbay/patchbay.db", "rundir": "/run/patchbay", "webbind": "0.0.0.0", "webport": "8443", "tlscert": "/etc/patchbay/tls.crt", "tlskey": "/etc/patchbay/tls.key", "sessionhours": "12", "sysopuser": "admin", "sysopemail": "", "sysoppassword": "", "mailhost": "", "mailport": "587", "mailsecurity": "starttls", "mailuser": "", "mailpassword": "", "mailfrom": "", } # Canonical spelling used when a key has to be appended to the file. CANONICAL = { "sysoppassword": "SysopPassword", "sysopemail": "SysopEmail", "mailhost": "MailHost", "mailport": "MailPort", "mailsecurity": "MailSecurity", "mailuser": "MailUser", "mailpassword": "MailPassword", "mailfrom": "MailFrom", } def _parse_line(line): s = line.strip() if not s or s[0] in "#;[" or "=" not in s: return None key, val = s.split("=", 1) val = val.strip() if len(val) >= 2 and val[0] == '"' and val[-1] == '"': val = val[1:-1] return key.strip().lower(), val class Config: """Keys are case-insensitive and stored lower-case.""" def __init__(self, path): self.path = path self.values = dict(DEFAULTS) with open(path, encoding="utf-8") as f: for line in f: kv = _parse_line(line) if kv: self.values[kv[0]] = kv[1] def get(self, key, default=None): return self.values.get(key.lower(), default) def getint(self, key, default=0): try: return int(self.get(key, default)) except (TypeError, ValueError): return default def update(self, changes): """Rewrites the given keys in place, keeping comments and order.""" changes = {k.lower(): v for k, v in changes.items()} for v in changes.values(): if "\n" in str(v) or "\r" in str(v): raise ValueError("newline in config value") with open(self.path, encoding="utf-8") as f: lines = f.readlines() done = set() out = [] for line in lines: kv = _parse_line(line) if kv and kv[0] in changes: if kv[0] in done: continue # drop duplicates of a rewritten key key_text = line.split("=", 1)[0].strip() out.append(f"{key_text} = {self._quote(changes[kv[0]])}\n") done.add(kv[0]) else: out.append(line) missing = [k for k in changes if k not in done] if missing: if out and not out[-1].endswith("\n"): out[-1] += "\n" for k in missing: out.append(f"{CANONICAL.get(k, k)} = {self._quote(changes[k])}\n") st = os.stat(self.path) fd, tmp = tempfile.mkstemp(dir=os.path.dirname(self.path) or ".", prefix=".patchbay.conf.") try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.writelines(out) os.chmod(tmp, st.st_mode & 0o777) try: os.chown(tmp, st.st_uid, st.st_gid) except PermissionError: pass os.replace(tmp, self.path) except BaseException: if os.path.exists(tmp): os.unlink(tmp) raise self.values.update({k: str(v) for k, v in changes.items()}) @staticmethod def _quote(v): v = str(v) return f'"{v}"' if v != v.strip() else v