c2b0c3d35e
A fresh server gets boot-tools/blancco from the USB, which can be an older Blancco (7.15.1) than our preferences (7.20). The version guard refused to patch its config.img and the exception failed the whole share apply, so the first-boot playbook failed. refresh_live_prefs() now returns a warning for config.img and still writes the served preferences.xml (UEFI path).
318 lines
14 KiB
Python
318 lines
14 KiB
Python
"""Tools > Share accounts: the pxe-upload, blancco and clonezilla Samba
|
|
passwords, and every file each one goes into.
|
|
|
|
Run from the webapp folder:
|
|
PYTHONPATH=.:<deps> python -m unittest tests.test_share_accounts -v
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
from unittest import mock
|
|
|
|
_T = tempfile.mkdtemp(prefix="pxe-shares-test-")
|
|
for _k, _sub in {"SAMBA_SHARE": "samba", "CLONEZILLA_SHARE": "cz", "BLANCCO_REPORTS": "br",
|
|
"ENROLLMENT_SHARE": "enr", "WEB_ROOT": "www", "IMAGING_DIR": "img", "UPDATES_DIR": "upd",
|
|
"TFTP_ROOT": "tftp", "BIOS_DIR": "bios", "UPLOAD_DIR": "upl", "SAMBA_LOG_DIR": "smblog",
|
|
"STARTNET_BACKUPS_DIR": "snb"}.items():
|
|
os.environ.setdefault(_k, os.path.join(_T, _sub))
|
|
os.makedirs(os.environ[_k], exist_ok=True)
|
|
for _k in ("AUDIT_LOG", "HARDWARE_STATE", "BLANCCO_STATE", "DNSMASQ_LEASES", "APACHE_ACCESS_LOG",
|
|
"DNSMASQ_SYSLOG", "NIC_STATUS_FILE", "NIC_CONF", "DNSMASQ_NIC_CONF", "DNSMASQ_CONF", "CLONEZILLA_STATE"):
|
|
os.environ.setdefault(_k, os.path.join(_T, _k.lower()))
|
|
|
|
WEBAPP = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
REPO = os.path.dirname(WEBAPP)
|
|
sys.path.insert(0, WEBAPP)
|
|
os.chdir(WEBAPP)
|
|
|
|
import config # noqa: E402
|
|
import app as A # noqa: E402
|
|
from services import blancco_prefs as BP # noqa: E402
|
|
from services import blancco_update as BU # noqa: E402
|
|
from services import share_accounts as SA # noqa: E402
|
|
from services import wim # noqa: E402
|
|
|
|
PW = "Site-Pass_2026"
|
|
|
|
|
|
def newc(files):
|
|
"""A newc cpio archive of {name: bytes} (what Blancco's config.img is)."""
|
|
out = b""
|
|
def entry(name, data, mode):
|
|
nb = name.encode() + b"\0"
|
|
hdr = b"070701" + b"".join(b"%08X" % v for v in (
|
|
1, mode, 0, 0, 1, 0, len(data), 0, 0, 0, 0, len(nb), 0))
|
|
blob = hdr + nb
|
|
blob += b"\0" * (-len(blob) % 4)
|
|
blob += data + b"\0" * (-len(data) % 4)
|
|
return blob
|
|
for name, data in files.items():
|
|
out += entry(name, data, 0o100644)
|
|
return out + entry("TRAILER!!!", b"", 0)
|
|
|
|
|
|
def cpio_file(data, name):
|
|
return BU.cpio_member(data, name)
|
|
|
|
|
|
def startnet_text():
|
|
with open(os.path.join(REPO, "playbook", "startnet.cmd"), encoding="utf-8") as f:
|
|
return f.read()
|
|
|
|
|
|
def menu_text():
|
|
with open(os.path.join(REPO, "playbook", "ipxe", "GetPxeScript.aspx"), encoding="utf-8") as f:
|
|
return f.read().replace("__CLONEZILLA_SMB_PASSWORD__", "clonezilla-backup")
|
|
|
|
|
|
class Fixture(unittest.TestCase):
|
|
def setUp(self):
|
|
for d in (config.UPDATES_DIR, config.WEB_ROOT, config.IMAGING_DIR, wim.BACKUPS_DIR):
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
os.makedirs(d)
|
|
# boot.wim with the repo startnet.cmd
|
|
src = tempfile.mkdtemp(dir=_T)
|
|
os.makedirs(os.path.join(src, "Windows", "System32"))
|
|
with open(os.path.join(src, "Windows", "System32", "startnet.cmd"), "w", newline="\r\n") as f:
|
|
f.write(startnet_text().replace("\r\n", "\n"))
|
|
os.makedirs(os.path.dirname(config.BOOT_WIM))
|
|
subprocess.run(["wimcapture", src, config.BOOT_WIM, "--compress=none"], check=True,
|
|
capture_output=True)
|
|
# iPXE menu as the playbook renders it
|
|
os.makedirs(os.path.dirname(BU.menu_path()))
|
|
with open(BU.menu_path(), "w", newline="") as f:
|
|
f.write(menu_text())
|
|
# Blancco: repo preferences, served copy, and a live config.img
|
|
os.makedirs(BU.blancco_root(), exist_ok=True)
|
|
with open(os.path.join(REPO, "playbook", "blancco-preferences.xml"), "rb") as f:
|
|
repo = f.read()
|
|
with open(BP.repo_prefs_path(), "wb") as f:
|
|
f.write(repo)
|
|
BP.render()
|
|
with open(BP.served_prefs_path(), "rb") as f:
|
|
served = f.read()
|
|
prefs_in_img = served + b"\0" * 3000 + b"<!-- " + b"x" * 200 + b" -->\n"
|
|
self.cfg = newc({"report.key": b"KEY", "preferences.xml": prefs_in_img, "config.xml": b"<c/>"})
|
|
with open(os.path.join(BU.blancco_root(), "config.img"), "wb") as f:
|
|
f.write(self.cfg)
|
|
self.smb = mock.patch.object(SA, "_smbpasswd").start()
|
|
self.addCleanup(mock.patch.stopall)
|
|
|
|
def startnet(self):
|
|
return wim.extract_startnet(config.BOOT_WIM)
|
|
|
|
def menu(self):
|
|
with open(BU.menu_path()) as f:
|
|
return f.read()
|
|
|
|
def prefs_share_pw(self, data):
|
|
m = re.search(rb'<network_share>\s*<username encrypted="false">blancco</username>\s*'
|
|
rb'<password encrypted="false">([^<]*)</password>', data)
|
|
return m.group(1).decode()
|
|
|
|
|
|
class ShareAccounts(Fixture):
|
|
def test_defaults_and_public_hides_passwords(self):
|
|
self.assertEqual(SA.load(), SA.DEFAULTS)
|
|
pub = SA.public()
|
|
self.assertTrue(all(r["default"] for r in pub))
|
|
self.assertNotIn("pxe", json.dumps(pub).replace("pxe-upload", ""))
|
|
|
|
def test_rules(self):
|
|
for bad in ("short", "has space 123", "semi;colon12", "pct%PATH%xx", "a" * 65, "amp&ersand12"):
|
|
with self.assertRaises(ValueError, msg=bad):
|
|
SA.check("pxe-upload", bad)
|
|
with self.assertRaisesRegex(ValueError, "shipped default"):
|
|
SA.check("clonezilla", "clonezilla-backup")
|
|
with self.assertRaises(ValueError):
|
|
SA.check("root", PW)
|
|
SA.check("blancco", "Ok.pass_~-123")
|
|
|
|
def test_pxe_upload_goes_into_samba_and_startnet(self):
|
|
done = SA.set_password("pxe-upload", PW)
|
|
self.smb.assert_called_once_with("pxe-upload", PW)
|
|
self.assertIn("boot.wim: startnet.cmd updated", done)
|
|
sn = self.startnet()
|
|
self.assertRegex(sn, r"(?m)^set SHAREPASS=%s\r$" % re.escape(PW))
|
|
self.assertIn("set SHAREUSER=pxe-upload", sn)
|
|
self.assertEqual(sn.count("\r\n"), sn.count("\n")) # still CRLF
|
|
# Only that line changed.
|
|
a = startnet_text().replace("\r\n", "\n").splitlines()
|
|
b = sn.replace("\r\n", "\n").splitlines()
|
|
self.assertEqual([i for i, (x, y) in enumerate(zip(a, b)) if x != y],
|
|
[a.index("set SHAREPASS=pxe")])
|
|
# The prior startnet was snapshotted.
|
|
self.assertTrue(any(s["note"] == "SHARE_PASSWORD" for s in wim.list_snapshots()))
|
|
# Stored privately.
|
|
st = os.stat(SA.state_path())
|
|
self.assertEqual(stat.S_IMODE(st.st_mode), 0o600)
|
|
self.assertEqual(SA.load()["pxe-upload"], PW)
|
|
self.assertFalse(SA.public()[0]["default"])
|
|
|
|
def test_clonezilla_goes_into_the_menu_without_a_backup_left(self):
|
|
before = set(os.listdir(os.path.dirname(BU.menu_path())))
|
|
SA.set_password("clonezilla", PW)
|
|
m = self.menu()
|
|
self.assertEqual(m.count(f"smb3://clonezilla:{PW}@"), 2)
|
|
self.assertNotIn("clonezilla-backup", m)
|
|
self.assertEqual(set(os.listdir(os.path.dirname(BU.menu_path()))), before)
|
|
self.assertEqual(m, menu_text().replace("clonezilla:clonezilla-backup@", f"clonezilla:{PW}@"))
|
|
|
|
def test_blancco_goes_into_served_prefs_and_live_config_img(self):
|
|
SA.set_password("blancco", PW)
|
|
with open(BP.served_prefs_path(), "rb") as f:
|
|
served = f.read()
|
|
self.assertEqual(self.prefs_share_pw(served), PW)
|
|
with open(os.path.join(BU.blancco_root(), "config.img"), "rb") as f:
|
|
cfg = f.read()
|
|
self.assertEqual(len(cfg), len(self.cfg))
|
|
self.assertEqual(self.prefs_share_pw(cpio_file(cfg, "preferences.xml")), PW)
|
|
self.assertEqual(cpio_file(cfg, "report.key"), b"KEY")
|
|
# A release staged later gets it too (apply_prefs runs blancco_prefs.apply).
|
|
with open(BP.repo_prefs_path(), "rb") as f:
|
|
self.assertEqual(self.prefs_share_pw(BP.apply(f.read())), PW)
|
|
|
|
def test_older_live_release_warns_instead_of_failing(self):
|
|
# A config.img from an older Blancco than our preferences (the repo
|
|
# payload can lag): the version guard refuses it. The served
|
|
# preferences (UEFI path) still change; the change is not refused.
|
|
old = cpio_file(self.cfg, "preferences.xml").replace(
|
|
b"<major>20</major>", b"<major>15</major>", 1)
|
|
self.assertIn(b"<major>15</major>", old)
|
|
cfg = newc({"report.key": b"KEY", "preferences.xml": old, "config.xml": b"<c/>"})
|
|
with open(os.path.join(BU.blancco_root(), "config.img"), "wb") as f:
|
|
f.write(cfg)
|
|
done = SA.set_password("blancco", PW)
|
|
self.assertTrue(any("WARNING" in d and "config.img" in d for d in done), done)
|
|
with open(BP.served_prefs_path(), "rb") as f:
|
|
self.assertEqual(self.prefs_share_pw(f.read()), PW)
|
|
with open(os.path.join(BU.blancco_root(), "config.img"), "rb") as f:
|
|
self.assertEqual(f.read(), cfg) # left as it was
|
|
|
|
def test_apply_is_idempotent(self):
|
|
SA.set_password("pxe-upload", PW)
|
|
SA.set_password("clonezilla", PW + "x")
|
|
SA.set_password("blancco", PW + "y")
|
|
done = SA.apply()
|
|
self.assertIn("boot.wim: unchanged", done)
|
|
self.assertIn("iPXE menu: unchanged", done)
|
|
self.assertIn("Blancco preferences: unchanged", done)
|
|
|
|
def test_refused_while_a_pc_images(self):
|
|
now = datetime.now().astimezone()
|
|
with open(os.path.join(config.IMAGING_DIR, "ABC123.json"), "w") as f:
|
|
json.dump({"serial": "ABC123", "hostname": "BAY-7", "status": "in_progress",
|
|
"stage_index": 3, "last_updated": now.isoformat()}, f)
|
|
with open(os.path.join(config.IMAGING_DIR, "OLD1.json"), "w") as f:
|
|
json.dump({"serial": "OLD1", "status": "in_progress", "stage_index": 3,
|
|
"last_updated": (now - timedelta(hours=5)).isoformat()}, f)
|
|
self.assertEqual(SA.busy(), ["BAY-7"])
|
|
with self.assertRaisesRegex(RuntimeError, "BAY-7"):
|
|
SA.set_password("pxe-upload", PW)
|
|
self.assertEqual(SA.load()["pxe-upload"], "pxe") # nothing stored
|
|
self.smb.assert_not_called()
|
|
SA.set_password("pxe-upload", PW, force=True)
|
|
self.assertEqual(SA.load()["pxe-upload"], PW)
|
|
|
|
def test_seed_only_for_accounts_not_held(self):
|
|
SA.set_password("blancco", PW)
|
|
env = {"SHARE_SEED_PXE_UPLOAD": "Seed-Upload-1", "SHARE_SEED_BLANCCO": "Seed-Blancco-1",
|
|
"SHARE_SEED_CLONEZILLA": "clonezilla-backup"}
|
|
with mock.patch.dict(os.environ, env):
|
|
self.assertEqual(SA.seed_from_env(), ["pxe-upload"])
|
|
cur = SA.load()
|
|
self.assertEqual((cur["pxe-upload"], cur["blancco"], cur["clonezilla"]),
|
|
("Seed-Upload-1", PW, "clonezilla-backup"))
|
|
with mock.patch.dict(os.environ, {"SHARE_SEED_CLONEZILLA": "bad pw"}):
|
|
with self.assertRaises(ValueError):
|
|
SA.seed_from_env()
|
|
|
|
def test_cli_apply_seed(self):
|
|
with mock.patch.dict(os.environ, {"SHARE_SEED_CLONEZILLA": "Seed-Clone-1"}):
|
|
self.assertEqual(SA.main(["apply", "--seed"]), 0)
|
|
self.assertIn("smb3://clonezilla:Seed-Clone-1@", self.menu())
|
|
self.assertEqual(self.smb.call_count, 3)
|
|
|
|
def test_failure_after_store_is_reported_as_such(self):
|
|
self.smb.side_effect = RuntimeError("smbpasswd pxe-upload failed: no such user")
|
|
with self.assertRaises(SA.ApplyFailed):
|
|
SA.set_password("pxe-upload", PW)
|
|
|
|
def test_no_server_files_yet(self):
|
|
os.remove(config.BOOT_WIM)
|
|
os.remove(BU.menu_path())
|
|
os.remove(os.path.join(BU.blancco_root(), "config.img"))
|
|
done = SA.apply()
|
|
self.assertTrue(any("skipped" in d for d in done))
|
|
|
|
|
|
class StartnetPinning(Fixture):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.c = A.app.test_client()
|
|
html = self.c.get("/startnet").get_data(as_text=True)
|
|
self.tok = re.search(r'name="csrf-token" content="([^"]+)"', html).group(1)
|
|
|
|
def test_editor_save_cannot_bring_back_an_old_password(self):
|
|
SA.set_password("pxe-upload", PW)
|
|
old = startnet_text() # carries SHAREPASS=pxe, like an old snapshot
|
|
r = self.c.post("/startnet/save", data={"content": old, "_csrf_token": self.tok})
|
|
self.assertIn(r.status_code, (200, 302))
|
|
self.assertRegex(self.startnet(), r"(?m)^set SHAREPASS=%s\r$" % re.escape(PW))
|
|
|
|
def test_new_menu_block_uses_the_variables(self):
|
|
blk = wim._synth_block({"target": "x", "target_image": "gea-x", "label": "X"}, "172.16.9.1")
|
|
self.assertIn(r"net use Z: \\172.16.9.1\winpeapps\gea-x /user:%SHAREUSER% %SHAREPASS% /persistent:no", blk)
|
|
|
|
|
|
class Page(Fixture):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.c = A.app.test_client()
|
|
|
|
def tok(self):
|
|
html = self.c.get("/tools/share-accounts/").get_data(as_text=True)
|
|
return re.search(r'name="csrf-token" content="([^"]+)"', html).group(1)
|
|
|
|
def test_page_and_save(self):
|
|
r = self.c.get("/tools/share-accounts/")
|
|
self.assertEqual(r.status_code, 200)
|
|
html = r.get_data(as_text=True)
|
|
for a in SA.ACCOUNTS:
|
|
self.assertIn(a, html)
|
|
self.assertIn("Default password", html)
|
|
tok = self.tok()
|
|
r = self.c.post("/tools/share-accounts/clonezilla",
|
|
data={"password": PW, "confirm": PW, "_csrf_token": tok}, follow_redirects=True)
|
|
html = r.get_data(as_text=True)
|
|
self.assertIn("clonezilla password changed", html)
|
|
self.assertNotIn(PW, html)
|
|
self.assertIn(f"clonezilla:{PW}@", self.menu())
|
|
|
|
def test_mismatch_and_bad_password(self):
|
|
tok = self.tok()
|
|
r = self.c.post("/tools/share-accounts/blancco",
|
|
data={"password": PW, "confirm": PW + "z", "_csrf_token": tok}, follow_redirects=True)
|
|
self.assertIn("differ", r.get_data(as_text=True))
|
|
r = self.c.post("/tools/share-accounts/blancco",
|
|
data={"password": "bad", "confirm": "bad", "_csrf_token": tok}, follow_redirects=True)
|
|
self.assertIn("not changed", r.get_data(as_text=True))
|
|
self.assertEqual(SA.load()["blancco"], "blancco")
|
|
self.smb.assert_not_called()
|
|
|
|
def test_csrf_required(self):
|
|
r = self.c.post("/tools/share-accounts/blancco", data={"password": PW, "confirm": PW})
|
|
self.assertIn(r.status_code, (400, 403))
|
|
self.assertEqual(SA.load()["blancco"], "blancco")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|