120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
"""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
|