97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
"""UCI profile snapshots under /mmc/mk8/profiles/<name>/{pineapd,wireless,network}"""
|
|
import os, re, time
|
|
|
|
PROFILES_DIR = '/mmc/mk8/profiles'
|
|
CONFIGS = ('pineapd', 'wireless', 'network')
|
|
NAME_RE = re.compile(r'^[A-Za-z0-9._-]{1,64}$')
|
|
|
|
|
|
def run_cmd(args, timeout=20, input_data=None):
|
|
"""Lazy import avoids a circular import with server.py; tests monkeypatch."""
|
|
from server import device_run
|
|
return device_run(args, timeout=timeout, input_data=input_data)
|
|
|
|
|
|
def _path(name):
|
|
"""Resolve a profile name to its directory. HTTP-supplied names are never
|
|
trusted: reject anything but [A-Za-z0-9._-]{1,64} and explicitly refuse
|
|
'.'/'..' so traversal can never escape PROFILES_DIR."""
|
|
if not isinstance(name, str) or not NAME_RE.fullmatch(name) \
|
|
or name in ('.', '..'):
|
|
raise ValueError('invalid profile name')
|
|
return os.path.join(PROFILES_DIR, name)
|
|
|
|
|
|
def snapshot(name):
|
|
dest = _path(name)
|
|
try:
|
|
os.makedirs(dest, exist_ok=True)
|
|
wrote = False
|
|
for cfg in CONFIGS:
|
|
rc, out, err = run_cmd(['uci', 'export', cfg])
|
|
if rc != 0 or not (out or '').strip():
|
|
continue
|
|
with open(os.path.join(dest, cfg + '.uci'), 'w') as f:
|
|
f.write(out)
|
|
wrote = True
|
|
return wrote
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def auto_name(op):
|
|
return 'pre-%s-%d' % (op, int(time.time()))
|
|
|
|
|
|
def list_profiles():
|
|
try:
|
|
out = []
|
|
for d in os.listdir(PROFILES_DIR):
|
|
try:
|
|
if os.path.isdir(_path(d)):
|
|
out.append(d)
|
|
except ValueError:
|
|
continue
|
|
return sorted(out)
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def delete(name):
|
|
import shutil
|
|
shutil.rmtree(_path(name), ignore_errors=True)
|
|
|
|
|
|
def restore(name):
|
|
"""Restore configs then commit once per config. Caller runs wifi reload
|
|
/ service restart as appropriate for the operation."""
|
|
src = _path(name)
|
|
restored = []
|
|
if not os.path.isdir(src):
|
|
return {'ok': False, 'restored': [], 'error': 'profile not found'}
|
|
for cfg in CONFIGS:
|
|
fpath = os.path.join(src, cfg + '.uci')
|
|
if not os.path.isfile(fpath):
|
|
continue
|
|
with open(fpath) as f:
|
|
text = f.read()
|
|
rc, _, err = run_cmd(['uci', 'import', cfg], input_data=text)
|
|
if rc != 0:
|
|
return {'ok': False, 'restored': restored,
|
|
'error': 'import failed'}
|
|
crc, _, cerr = run_cmd(['uci', 'commit', cfg])
|
|
if crc != 0:
|
|
return {'ok': False, 'restored': restored,
|
|
'error': 'commit failed: %s' % (cerr or cfg)}
|
|
restored.append(cfg)
|
|
return {'ok': True, 'restored': restored}
|
|
|
|
|
|
LASTKNOWN_GOOD = 'lastknown-good'
|
|
|
|
|
|
def promote_lastknown_good():
|
|
"""Replace the lastknown-good profile with the live config."""
|
|
delete(LASTKNOWN_GOOD)
|
|
return snapshot(LASTKNOWN_GOOD)
|