#!/usr/bin/env python3 """Interactive Ubuntu disk provisioning and online growth. Preview by default. Python 3.8+; standard library only. See README.md for scope and dependencies. No command is run through a shell. Importing this module performs no I/O. """ import argparse import datetime from decimal import Decimal, InvalidOperation import fcntl import json import os from pathlib import Path import re import shlex import shutil import stat import subprocess import sys import tempfile import time class Stop(RuntimeError): """An unsupported state or failed safety check.""" PACKAGES = "python3 util-linux fdisk parted cloud-guest-utils e2fsprogs xfsprogs lvm2 udev" BASE_COMMANDS = ("lsblk", "findmnt", "wipefs", "blkid", "blockdev") LSBLK_COLUMNS = "NAME,KNAME,PKNAME,TYPE,SIZE,FSTYPE,MOUNTPOINTS,RO,RM,MODEL,SERIAL,WWN,UUID,MAJ:MIN,PTTYPE,PARTTYPE" def need(*commands): missing = [c for c in commands if not shutil.which(c)] if missing: raise Stop("Missing commands: %s\nInstall dependencies yourself with:\n" " sudo apt-get update\n sudo apt-get install %s" % (", ".join(missing), PACKAGES)) class Runner: def __init__(self, apply=False): self.apply = apply self.audit = None def record(self, text): if self.audit: with (self.audit / "operation.log").open("a") as stream: stream.write(text + "\n") stream.flush() os.fsync(stream.fileno()) def run(self, *args, change=False, input_text=None, allowed=(0,)): args = [str(a) for a in args] command = shlex.join(args) if change: print(" $ " + command) if not self.apply or self.audit is None: raise Stop("Internal guard: mutation requested without an approved apply operation.") self.record("$ " + command) if input_text is not None: self.record("stdin:\n" + input_text) result = subprocess.run(args, input=input_text, stdin=subprocess.DEVNULL if input_text is None else None, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=dict(os.environ, LC_ALL="C")) self.record("exit=%d\n%s%s" % (result.returncode, result.stdout, result.stderr)) if result.returncode not in allowed: raise Stop("Command failed (exit %d): %s\n%s%s" % (result.returncode, command, result.stdout, result.stderr)) if change: print((result.stdout + result.stderr).strip()) return result def data(self, *args): result = self.run(*args) try: return json.loads(result.stdout) except ValueError as exc: raise Stop("Invalid JSON from %s: %s" % (args[0], exc)) def start(self, snapshot): base = Path("/var/backups/ubuntu-storage") if base.is_symlink(): raise Stop("Backup directory must not be a symlink.") base.mkdir(mode=0o700, parents=True, exist_ok=True) info = base.stat() if info.st_uid != 0 or info.st_mode & 0o022: raise Stop("Backup directory must be root owned and not writable by group/others.") stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ-") self.audit = Path(tempfile.mkdtemp(prefix=stamp, dir=str(base))) self.save("before.json", json.dumps(snapshot, indent=2)) print("Operation log and metadata backups: " + str(self.audit)) def save(self, name, text): with (self.audit / name).open("x") as stream: stream.write(text) stream.flush() os.fsync(stream.fileno()) def real(path): return os.path.realpath(str(path)) def flatten(nodes): result = {} for node in nodes: node = dict(node) for field in ("ro", "rm"): value = node.get(field) if value in (True, 1, "1", "true"): node[field] = True elif value in (False, 0, "0", "false"): node[field] = False else: raise Stop("Cannot determine %s state for %s" % (field, node["name"])) node["children"] = list(node.get("children") or []) key = real(node["name"]) if key in result and result[key] != node: raise Stop("Ambiguous block-device topology: " + key) result[key] = node for key, child in flatten(node.get("children", [])).items(): if key in result and result[key] != child: raise Stop("Ambiguous block-device topology: " + key) result[key] = child return result def inventory(runner): return flatten(runner.data("lsblk", "--json", "--bytes", "--paths", "--tree", "--output", LSBLK_COLUMNS)["blockdevices"]) def mounts(runner): return runner.data("findmnt", "--json", "--list", "--output", "TARGET,SOURCE,FSTYPE,OPTIONS,MAJ:MIN,FSROOT")["filesystems"] def gib(value): return "%.2f GiB" % (int(value) / 1024 ** 3) def byte_count(value): try: count = Decimal(str(value)) if not count.is_finite() or count < 0 or count != count.to_integral_value(): raise Stop("Expected a non-negative integral byte count: " + str(value)) return int(count) except InvalidOperation: raise Stop("Invalid byte count: " + str(value)) def show_inventory(nodes): print("\n%-27s %-7s %-12s %-12s %s" % ("DEVICE", "TYPE", "SIZE", "FILESYSTEM", "MOUNTS / IDENTITY")) for path, node in nodes.items(): mounted = ",".join(str(p) for p in node.get("mountpoints", []) if p) identity = " ".join(str(node.get(k) or "").strip() for k in ("model", "serial", "wwn")) print("%-27s %-7s %-12s %-12s %s" % (path, node["type"], gib(node["size"]), node.get("fstype") or "-", mounted or identity)) def choose(nodes, prompt): path = real(input(prompt).strip()) if not path.startswith("/dev/") or path not in nodes: raise Stop("Select an existing block device from the inventory, using its full /dev path.") info = os.stat(path) if not stat.S_ISBLK(info.st_mode): raise Stop("Selection is not a block device.") actual = "%d:%d" % (os.major(info.st_rdev), os.minor(info.st_rdev)) if actual != nodes[path]["maj:min"]: raise Stop("Device identity changed; restart the script.") return path def fingerprint(node): return tuple(node.get(k) for k in ("name", "maj:min", "type", "size", "serial", "wwn", "uuid", "fstype", "ro", "pttype")) def sysnode(node): path = Path("/sys/dev/block") / node["maj:min"] if not path.exists(): raise Stop("Cannot inspect sysfs for " + node["name"]) return path.resolve() def relations(node, kind): return {p.name for p in (sysnode(node) / kind).iterdir()} def writable(node): if node.get("ro") or int(node["size"]) <= 0: raise Stop("Device is read-only or has no capacity: " + node["name"]) def signatures(runner, path): return runner.data("wipefs", "--no-act", "--json", path).get("signatures", []) def swap_devices(): result = set() for line in Path("/proc/swaps").read_text().splitlines()[1:]: path = re.sub(r"\\([0-7]{3})", lambda m: chr(int(m[1], 8)), line.split()[0]) info = os.stat(path) number = info.st_rdev if stat.S_ISBLK(info.st_mode) else info.st_dev result.add("%d:%d" % (os.major(number), os.minor(number))) return result def lvm_report(runner, command, section, fields, *args): need(command) data = runner.data(command, "--reportformat", "json", "--units", "b", "--nosuffix", "--options", fields, *args) if not any(section in report for report in data["report"]): raise Stop("Unexpected %s JSON report: missing section %s." % (command, section)) return [{k: str(v).strip() for k, v in row.items()} for report in data["report"] for row in report.get(section, [])] def decode_fstab(value): return re.sub(r"\\([0-7]{3})", lambda m: chr(int(m[1], 8)), value) def fstab_entries(text): for line in text.splitlines(): if line.strip() and not line.lstrip().startswith("#"): fields = line.split() if len(fields) < 4: raise Stop("Malformed existing /etc/fstab entry; correct it first.") yield decode_fstab(fields[0]), decode_fstab(fields[1]) def check_fstab(text, target, source=None, uuid=None): for old_source, old_target in fstab_entries(text): if real(old_target) == real(target): raise Stop("Mount point already appears in /etc/fstab: " + str(target)) if source and old_source.startswith("/dev/") and real(old_source) == real(source): raise Stop("Device already appears in /etc/fstab.") if uuid and old_source == "UUID=" + uuid: raise Stop("Filesystem UUID already appears in /etc/fstab.") def preflight_fstab(runner, target, source): path = Path("/etc/fstab") info = path.lstat() if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != 0 or info.st_mode & 0o022: raise Stop("/etc/fstab must be a regular, root-owned file with safe permissions.") check_fstab(path.read_text(), target, source) runner.run("findmnt", "--verify", "--tab-file", str(path)) def check_lvm_names(vg, lv): for name in (vg, lv): if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,62}", name): raise Stop("LVM names must start with a letter; use up to 63 letters, digits, or underscores.") reserved = ("_cdata", "_cmeta", "_corig", "_iorig", "_mimage", "_mlog", "_pmspare", "_rimage", "_rmeta", "_tdata", "_tmeta", "_vdata", "_vorigin", "_wcorig") if lv.startswith(("snapshot", "pvmove")) or any(word in lv for word in reserved): raise Stop("Logical volume name is reserved by LVM; choose another name.") if os.path.lexists("/dev/" + vg): raise Stop("Volume group name conflicts with an existing /dev entry.") def check_mount_dir(target, current_mounts): target = Path(target) if (not target.is_absolute() or str(target).startswith("//") or not re.fullmatch(r"/[A-Za-z0-9_./-]+", str(target))): raise Stop("Use an absolute mount path containing only letters, digits, /, ., _, and -.") if ".." in target.parts: raise Stop("Mount path must not contain '..'.") prohibited = ("/etc", "/dev", "/proc", "/sys", "/run", "/boot", "/usr", "/bin", "/sbin", "/lib", "/lib64") if str(target) in ("/", "/var", "/home", "/root", "/tmp", "/mnt", "/media") or any( str(target) == p or str(target).startswith(p + "/") for p in prohibited): raise Stop("Choose a dedicated data directory, for example /srv/data or /mnt/data.") for path in (target,) + tuple(target.parents): if path.is_symlink(): raise Stop("Mount path or an ancestor is a symlink: " + str(path)) if path.exists() and not path.is_dir(): raise Stop("Mount path has a non-directory component.") if path.exists(): info = path.stat() if info.st_uid != 0 or info.st_mode & 0o022: raise Stop("Existing mount path components must be root owned and not group/world writable.") for mount in current_mounts: other = mount["target"] if other == str(target) or other.startswith(str(target) + "/"): raise Stop("Mount point or a child directory is already mounted.") if target.exists() and any(target.iterdir()): raise Stop("Mount point must be empty; mounting would hide its existing files.") return target def check_blank(runner, path, nodes): node = nodes[path] writable(node) if node["type"] != "disk" or node.get("rm"): raise Stop("Provisioning requires a non-removable whole disk.") if node.get("children") or node.get("fstype") or node.get("pttype"): raise Stop("Disk contains partitions or a recognized format; provisioning is refused.") if relations(node, "holders") or relations(node, "slaves"): raise Stop("Disk participates in another storage layer.") if node["maj:min"] in swap_devices() or any(m["maj:min"] == node["maj:min"] for m in mounts(runner)): raise Stop("Disk is mounted or used by swap.") if signatures(runner, path): raise Stop("Disk has an existing signature. This script has no wipe option.") # Require LVM tools even for plain provisioning, so inactive PV membership is checked. pvs = lvm_report(runner, "pvs", "pv", "pv_name,vg_name,pv_uuid") if any(real(pv["pv_name"]) == path for pv in pvs): raise Stop("Disk is already an LVM physical volume.") for source, _ in fstab_entries(Path("/etc/fstab").read_text()): if source.startswith("/dev/") and real(source) == path: raise Stop("Disk is referenced by /etc/fstab.") def approve(runner, operation, path, plan): print("\nPlanned operation:") for line in plan: print(" " + line) if not runner.apply: print("\nPREVIEW COMPLETE: no storage changes made. Run again with --apply to execute.") return False print("\nConfirm the model/serial and backups independently. Signature checks cannot detect all raw data.") if operation == "PROVISION": print("This will OVERWRITE the selected disk with a new partition table and filesystem.") phrase = operation + " " + path if input("Type exactly '%s' to continue: " % phrase) != phrase: raise Stop("Confirmation did not match. No operation started.") return True def wait_for_partition(runner, disk): for _ in range(20): nodes = inventory(runner) children = nodes.get(disk, {}).get("children", []) if len(children) == 1 and children[0]["type"] == "part": part = real(children[0]["name"]) if os.path.exists(part): return part time.sleep(0.5) raise Stop("New partition did not appear. Stop and inspect lsblk/udev before continuing.") def fs_command(fstype, source, target): if fstype == "ext4": return ["resize2fs", source] if fstype == "xfs": return ["xfs_growfs", "-d", str(target)] raise Stop("Only ext4 and XFS are supported.") def add_fstab(runner, source, target, fstype, nofail, fstab_path=Path("/etc/fstab")): uuid = runner.run("blkid", "-s", "UUID", "-o", "value", source).stdout.strip() if not re.fullmatch(r"[A-Fa-f0-9-]+", uuid): raise Stop("Could not obtain an unambiguous filesystem UUID.") path = Path(fstab_path) fd = os.open(str(path), os.O_RDWR | os.O_NOFOLLOW) temporary = None try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) info = os.fstat(fd) if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != 0 or info.st_mode & 0o022: raise Stop("/etc/fstab must be a regular, root-owned file with safe permissions.") with os.fdopen(os.dup(fd), "r") as stream: old = stream.read() check_fstab(old, target, source, uuid) options = "defaults,nofail,x-systemd.device-timeout=10s" if nofail else "defaults" entry = "UUID=%s %s %s %s 0 %d\n" % (uuid, target, fstype, options, 2 if fstype == "ext4" else 0) new = old.rstrip("\n") + "\n\n# Added by ubuntu-storage\n" + entry runner.save("fstab.before", old) staged, temporary = tempfile.mkstemp(prefix=".fstab-ubuntu-storage-", dir=str(path.parent)) with os.fdopen(staged, "w") as stream: os.fchmod(stream.fileno(), stat.S_IMODE(info.st_mode)) os.fchown(stream.fileno(), info.st_uid, info.st_gid) stream.write(new) stream.flush() os.fsync(stream.fileno()) runner.run("findmnt", "--verify", "--tab-file", temporary) latest = path.lstat() if (latest.st_dev, latest.st_ino) != (info.st_dev, info.st_ino) or path.read_text() != old: raise Stop("/etc/fstab changed concurrently; the staged entry was not installed.") runner.record("Install fstab entry: " + entry.strip()) os.replace(temporary, str(path)) temporary = None parent_fd = os.open(str(path.parent), os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(parent_fd) finally: os.close(parent_fd) finally: if temporary: os.unlink(temporary) os.close(fd) if Path("/run/systemd/system").exists(): runner.run("systemctl", "daemon-reload", change=True) def provision(runner): need(*BASE_COMMANDS, "pvs", "sfdisk", "partprobe", "udevadm", "mount") nodes = inventory(runner) show_inventory(nodes) path = choose(nodes, "\nWhole disk to provision (for example /dev/sdb): ") check_blank(runner, path, nodes) fstype = input("Filesystem [ext4/xfs] (ext4): ").strip().lower() or "ext4" if fstype not in ("ext4", "xfs"): raise Stop("Choose ext4 or xfs.") need("mkfs." + fstype) layout = input("Layout [plain/lvm] (plain): ").strip().lower() or "plain" if layout not in ("plain", "lvm"): raise Stop("Choose plain or lvm.") vg = lv = None if layout == "lvm": need("pvcreate", "vgcreate", "lvcreate", "vgs", "vgcfgbackup") vg = input("New volume group name (data_vg): ").strip() or "data_vg" lv = input("New logical volume name (data): ").strip() or "data" check_lvm_names(vg, lv) if any(v["vg_name"] == vg for v in lvm_report(runner, "vgs", "vg", "vg_name")): raise Stop("Volume group already exists; select a new name.") target = check_mount_dir(input("Mount directory (/srv/data): ").strip() or "/srv/data", mounts(runner)) persist = ask_yes("Add a UUID entry to /etc/fstab for mounting after reboot?", True) nofail = ask_yes("Allow boot to continue if this data disk is absent (nofail)?", True) if persist else False if persist: preflight_fstab(runner, target, path) if Path("/run/systemd/system").exists(): need("systemctl") partition_type = "E6D6D379-F507-44C2-A23C-238F2A3DF928" if vg else "0FC63DAF-8483-4772-8E79-3D69D8477DE4" table = "label: gpt\n\nstart=1MiB, type=%s\n" % partition_type future = "/dev/%s/%s" % (vg, lv) if vg else "" plan = ["Device: %s (%s), model=%s serial=%s WWN=%s" % (path, gib(nodes[path]["size"]), nodes[path].get("model"), nodes[path].get("serial"), nodes[path].get("wwn")), "sfdisk --wipe never --wipe-partitions never %s: GPT, one partition from 1 MiB to end" % path] if vg: plan += ["pvcreate ", "vgcreate %s " % vg, "lvcreate --extents 100%%FREE --name %s %s" % (lv, vg)] plan += ["mkfs.%s %s" % (fstype, future), "mount %s %s" % (future, target), "Mount directory stays root:root; set application ownership separately."] if persist: plan += ["Back up, validate and atomically update /etc/fstab using the new UUID (%s)." % ("nofail" if nofail else "required at boot")] if not approve(runner, "PROVISION", path, plan): return fresh = inventory(runner) if path not in fresh or fingerprint(fresh[path]) != fingerprint(nodes[path]): raise Stop("Disk changed after selection; restart and inspect it.") check_blank(runner, path, fresh) check_mount_dir(target, mounts(runner)) if persist: preflight_fstab(runner, target, path) if vg and any(v["vg_name"] == vg for v in lvm_report(runner, "vgs", "vg", "vg_name")): raise Stop("Volume group name was claimed concurrently.") if vg: check_lvm_names(vg, lv) runner.start(fresh) runner.run("sfdisk", "--wipe", "never", "--wipe-partitions", "never", path, change=True, input_text=table) runner.run("partprobe", path, change=True) runner.run("udevadm", "settle", "--timeout=15", change=True) part = wait_for_partition(runner, path) verify_partition(runner, part, inventory(runner)) if signatures(runner, part): raise Stop("A signature was discovered inside the new partition; refusing to format it.") source = part if vg: runner.run("pvcreate", part, change=True) runner.run("vgcreate", vg, part, change=True) runner.run("lvcreate", "--extents", "100%FREE", "--name", lv, vg, change=True) runner.run("udevadm", "settle", "--timeout=15", change=True) source = "/dev/%s/%s" % (vg, lv) runner.run("vgcfgbackup", "--file", runner.audit / "vg.after.conf", vg, change=True) # No force flag: never suppress a filesystem tool's existing-data checks. runner.run("mkfs." + fstype, source, change=True) check_mount_dir(target, mounts(runner)) target.mkdir(mode=0o755, parents=True, exist_ok=True) runner.record("Create mount directory: " + str(target)) runner.run("mount", "-t", fstype, source, str(target), change=True) if persist: add_fstab(runner, source, target, fstype, nofail) verify_mounted(runner, real(source), fstype) print("\nProvisioning complete. " + str(target) + " is mounted.") if not persist: print("This mount is temporary; no /etc/fstab entry was added.") print(runner.run("df", "-hT", str(target)).stdout) def ask_yes(prompt, default=False): value = input(prompt + (" [Y/n]: " if default else " [y/N]: ")).strip().lower() if not value: return default if value not in ("y", "yes", "n", "no"): raise Stop("Please answer yes or no.") return value in ("y", "yes") def verify_mounted(runner, path, fstype): nodes = inventory(runner) if path not in nodes: raise Stop("Filesystem device disappeared.") candidates = [m for m in mounts(runner) if m["maj:min"] == nodes[path]["maj:min"] and m["fstype"] == fstype and m.get("fsroot") == "/" and "rw" in m["options"].split(",")] if not candidates: raise Stop("Expansion requires the full filesystem to be mounted read-write; bind-only or offline mounts are unsupported.") candidates.sort(key=lambda m: (len(m["target"]), m["target"])) return candidates[0]["target"] def partition_info(runner, path, nodes): node = nodes[path] if node["type"] != "part": return None parent = node.get("pkname") disk = real(parent if str(parent).startswith("/") else "/dev/" + str(parent)) if (disk not in nodes or nodes[disk]["type"] != "disk" or relations(nodes[disk], "slaves") or relations(nodes[disk], "holders")): raise Stop("Partition must sit directly on an ordinary disk.") writable(nodes[disk]) number = int((sysnode(node) / "partition").read_text().strip()) table = runner.data("sfdisk", "--json", disk)["partitiontable"] if table.get("label") not in ("gpt", "dos") or table.get("unit") != "sectors": raise Stop("Only GPT or primary MBR partitions are supported.") if table["label"] == "dos" and (number > 4 or any( re.sub(r"^0x", "", str(p["type"]).lower()) in ("5", "f", "85", "05", "0f") for p in table["partitions"])): raise Stop("MBR extended/logical partitions require manual handling.") selected = [p for p in table["partitions"] if real(p["node"]) == path] if len(selected) != 1: raise Stop("Cannot unambiguously identify the on-disk partition.") entry = selected[0] if any(int(p["start"]) > int(entry["start"]) for p in table["partitions"]): raise Stop("Only the last partition by physical position can be expanded by this script.") sector = int(table.get("sectorsize") or runner.run("blockdev", "--getss", disk).stdout) return {"disk": disk, "number": number, "start": int(entry["start"]), "size": int(entry["size"]), "sector": sector, "table": table} def verify_partition(runner, path, nodes): part = partition_info(runner, path, nodes) if part: kernel_bytes = int(runner.run("blockdev", "--getsize64", path).stdout) kernel_start = int((sysnode(nodes[path]) / "start").read_text().strip()) * 512 if kernel_bytes != part["size"] * part["sector"] or kernel_start != part["start"] * part["sector"]: raise Stop("Kernel partition size/start differs from its on-disk table. Reboot or refresh it safely, then rerun. No higher layer was resized.") return part def get_vg(runner, name): rows = lvm_report(runner, "vgs", "vg", "vg_name,vg_uuid,pv_count,vg_attr,vg_extent_size,vg_free_count") matches = [v for v in rows if v["vg_name"] == name] if len(matches) != 1: raise Stop("Volume group is missing or ambiguous.") vg = matches[0] if int(vg["pv_count"]) != 1 or not re.fullmatch(r"w[z-]--n-", vg["vg_attr"]): raise Stop("Only complete, local, writable volume groups containing one PV are supported.") return vg def expansion_layout(runner, path, nodes): node = nodes[path] writable(node) fstype = node.get("fstype") if fstype not in ("ext4", "xfs"): raise Stop("Select the ext4/XFS filesystem device (the LV for LVM), not the underlying PV/disk.") probe = runner.run("blkid", "-p", "-s", "TYPE", "-o", "value", path).stdout.strip() if probe != fstype: raise Stop("Filesystem probe disagrees with inventory.") target = verify_mounted(runner, path, fstype) if relations(node, "holders"): raise Stop("Filesystem device has another storage layer above it.") base, lvm = path, None if node["type"] == "lvm": rows = lvm_report(runner, "lvs", "seg", "lv_path,lv_uuid,vg_name,lv_attr,segtype,origin,pool_lv", "--segments") selected = [row for row in rows if real(row["lv_path"]) == path] if not selected or any(row["segtype"] != "linear" or row["origin"] or row["pool_lv"] or not row["lv_attr"].startswith("-w") or row["lv_attr"][4:5] != "a" for row in selected): raise Stop("Only active, ordinary linear LVs are supported; snapshots, thin, RAID, and cached LVs are excluded.") vg = get_vg(runner, selected[0]["vg_name"]) pvs = [p for p in lvm_report(runner, "pvs", "pv", "pv_name,pv_uuid,vg_name,pv_size,pe_start,pv_attr") if p["vg_name"] == vg["vg_name"]] if len(pvs) != 1 or "m" in pvs[0]["pv_attr"]: raise Stop("Missing or multiple physical volumes are unsupported.") base = real(pvs[0]["pv_name"]) if base not in nodes or nodes[base]["type"] not in ("disk", "part"): raise Stop("LVM PV must sit directly on a disk or partition; encryption/RAID/multipath are unsupported.") expected = {Path(nodes[real(row["lv_path"])]["kname"]).name for row in rows if row["vg_name"] == vg["vg_name"] and real(row["lv_path"]) in nodes} if not relations(nodes[base], "holders").issubset(expected): raise Stop("PV has an unrecognized holder.") lvm = {"vg": vg, "pv": pvs[0], "lv_uuid": selected[0]["lv_uuid"]} elif node["type"] not in ("disk", "part") or node.get("children"): raise Stop("Unsupported filesystem device topology.") writable(nodes[base]) if relations(nodes[base], "slaves"): raise Stop("Underlying storage layers are unsupported.") if nodes[base]["type"] == "disk" and (nodes[base].get("pttype") or any( n["type"] == "part" for n in nodes[base].get("children", []))): raise Stop("Whole-device storage conflicts with a partition table.") part = verify_partition(runner, base, nodes) if lvm and byte_count(lvm["pv"]["pv_size"]) + byte_count(lvm["pv"]["pe_start"]) > int(nodes[base]["size"]): raise Stop("PV is larger than its block device. Shrinking and recovery are unsupported.") return {"path": path, "fstype": fstype, "target": target, "base": base, "part": part, "lvm": lvm} def growth_possible(runner, part): result = runner.run("growpart", "--dry-run", part["disk"], str(part["number"]), allowed=(0, 1)) if result.returncode == 1 and "NOCHANGE:" not in result.stdout + result.stderr: raise Stop("growpart returned 1 without a recognized NOCHANGE result. Inspect it manually.") return result.returncode == 0 def extend(runner): need(*BASE_COMMANDS, "sfdisk") nodes = inventory(runner) show_inventory(nodes) print("\nFirst enlarge the disk in your hypervisor/cloud provider. Confirm lsblk shows its new size.") print("For LVM, select the filesystem LV (for example /dev/mapper/ubuntu--vg-ubuntu--lv).") path = choose(nodes, "Filesystem device to expand: ") layout = expansion_layout(runner, path, nodes) base, part, lvm = layout["base"], layout["part"], layout["lvm"] fs_args = fs_command(layout["fstype"], path, layout["target"]) need(fs_args[0]) can_grow = False if part: need("growpart", "udevadm") can_grow = growth_possible(runner, part) allocation = "new" if lvm: need("pvresize", "lvextend", "vgcfgbackup") print("VG currently has %s free." % gib(int(lvm["vg"]["vg_free_count"]) * byte_count(lvm["vg"]["vg_extent_size"]))) allocation = input("LV allocation [new/all/none] (new): ").strip().lower() or "new" if allocation not in ("new", "all", "none"): raise Stop("Choose new (new PV extents only), all (all VG free extents), or none (filesystem catch-up).") plan = ["Online growth of %s mounted at %s (%s)" % (path, layout["target"], layout["fstype"])] if part: plan += ["Back up the partition table for " + part["disk"], ("growpart %s %s" % (part["disk"], part["number"])) if can_grow else "Partition has no usable adjacent free space; continue checking higher layers.", "Verify the partition start is unchanged and kernel size matches the on-disk table."] if lvm: plan += ["Back up LVM metadata for " + lvm["vg"]["vg_name"], "pvresize " + base] if allocation == "new": plan += ["lvextend --extents + " + path, "Existing VG free space is left available. If PV is already enlarged, choose 'all' to use it."] elif allocation == "all": plan += ["lvextend --extents + " + path, "This allocates existing AND newly added VG free space to this LV."] else: plan += ["Leave LV size unchanged; update the PV and grow the filesystem to its current LV size."] plan += [shlex.join(fs_args), "No shrinking, unmounting, or automatic rollback."] if not approve(runner, "EXPAND", path, plan): return fresh = inventory(runner) paths = {path, base} | ({part["disk"]} if part else set()) if any(p not in fresh or fingerprint(nodes[p]) != fingerprint(fresh[p]) for p in paths): raise Stop("Storage changed after selection; restart and inspect it.") current = expansion_layout(runner, path, fresh) if current != layout: raise Stop("Mount, partition, or LVM metadata changed after selection; restart.") runner.start(fresh) if part: runner.save("partition-table.before.sfdisk", runner.run("sfdisk", "--dump", part["disk"]).stdout) if lvm: runner.run("vgcfgbackup", "--file", runner.audit / "vg.before.conf", lvm["vg"]["vg_name"], change=True) if part: if growth_possible(runner, part): result = runner.run("growpart", part["disk"], str(part["number"]), change=True, allowed=(0, 1)) if result.returncode == 1 and "NOCHANGE:" not in result.stdout + result.stderr: raise Stop("Unrecognized growpart result; stopped before resizing higher layers.") runner.run("udevadm", "settle", "--timeout=15", change=True) after = verify_partition(runner, base, inventory(runner)) if after["start"] != part["start"] or after["size"] < part["size"]: raise Stop("Unexpected partition boundary change; stopped before resizing higher layers.") if lvm: before_free = int(lvm["vg"]["vg_free_count"]) runner.run("pvresize", base, change=True) vg = get_vg(runner, lvm["vg"]["vg_name"]) if vg["vg_uuid"] != lvm["vg"]["vg_uuid"]: raise Stop("Volume group identity changed.") free = int(vg["vg_free_count"]) count = free if allocation == "all" else max(0, free - before_free) if allocation == "new" else 0 if count: runner.run("lvextend", "--extents", "+%d" % count, path, change=True) else: print("No LV extents allocated. Continuing with filesystem growth (safe to rerun).") runner.run("vgcfgbackup", "--file", runner.audit / "vg.after.conf", vg["vg_name"], change=True) if verify_mounted(runner, path, layout["fstype"]) != layout["target"]: raise Stop("Mount changed; stopped before filesystem growth.") runner.run(*fs_args, change=True) runner.save("after.json", json.dumps(inventory(runner), indent=2)) print("\nGrowth checks and commands completed. Compare the resulting capacity below:") print(runner.run("df", "-hT", layout["target"]).stdout) def main(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--apply", action="store_true", help="enable changes; exact typed confirmation is still required") parser.add_argument("--inventory", action="store_true", help="display block devices and exit (read-only)") args = parser.parse_args(argv) runner = Runner(args.apply) lock_fd = None try: if sys.platform != "linux": raise Stop("Run this script on the Ubuntu server. This host is not Linux.") release = Path("/etc/os-release").read_text() if not re.search(r'^ID=[\"\']?ubuntu[\"\']?$', release, re.MULTILINE): raise Stop("This script targets Ubuntu. Other distributions have not been validated.") os.environ["PATH"] = "/usr/sbin:/usr/bin:/sbin:/bin" need(*BASE_COMMANDS) if args.inventory: show_inventory(inventory(runner)) return 0 if os.geteuid() != 0: raise Stop("Run with sudo, including preview, so safety probes can read all device metadata.") if not sys.stdin.isatty(): raise Stop("An interactive terminal is required. Do not pipe answers into this script.") if Path("/.dockerenv").exists() or Path("/run/.containerenv").exists() or os.environ.get("container"): raise Stop("Run on the server/VM host, not inside a container.") if os.readlink("/proc/self/ns/mnt") != os.readlink("/proc/1/ns/mnt"): raise Stop("Run in the host mount namespace, where all system mounts are visible.") # A directory flock prevents overlapping instances without writing a lock file in preview. lock_fd = os.open("/run/lock", os.O_RDONLY | os.O_DIRECTORY) fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) os.umask(0o077) print("Ubuntu storage assistant — " + ("APPLY mode" if args.apply else "PREVIEW mode")) print("Use verified backups and a maintenance window. Do not change disks concurrently in another tool.") print("1) Provision a new data disk\n2) Expand a mounted filesystem\n3) Show inventory\nq) Quit") action = input("Choose an action: ").strip().lower() if action == "1": provision(runner) elif action == "2": extend(runner) elif action == "3": show_inventory(inventory(runner)) elif action not in ("q", "quit"): raise Stop("Unknown action.") return 0 except (Stop, OSError, ValueError, KeyError) as exc: print("\nSTOPPED: " + str(exc), file=sys.stderr) if runner.audit: print("Some steps may already have completed. Nothing was rolled back.\n" "Inspect lsblk, findmnt, and the log before retrying: " + str(runner.audit), file=sys.stderr) return 1 except (KeyboardInterrupt, EOFError): print("\nCancelled. Completed changes, if any, were not rolled back.", file=sys.stderr) if runner.audit: print("Inspect the log before retrying: " + str(runner.audit), file=sys.stderr) return 130 finally: if lock_fd is not None: os.close(lock_fd) if __name__ == "__main__": sys.exit(main())