#!/usr/bin/env python3 """Linux POSIX ACL metadata probe. Creates only private temporary fixtures. SPDX-License-Identifier: MIT """ import argparse import datetime import hashlib import json import os from pathlib import Path import platform import stat import subprocess import tempfile PAYLOAD = b"Hesela ACL recovery control\n" * 16 UID_A, UID_B = 12345, 12346 def run(*args): env = {**os.environ, "LC_ALL": "C"} env.pop("POSIXLY_CORRECT", None) return subprocess.run([str(a) for a in args], check=True, capture_output=True, text=True, env=env).stdout.strip() def parse_acl(text): entries = {} for line in text.splitlines(): line = line.split("#", 1)[0].strip() if not line: continue parts = line.split(":") if len(parts) != 3 or parts[0] not in ("user", "group", "mask", "other"): raise ValueError(f"Unsupported access ACL entry: {line}") kind, qualifier, permissions = parts if kind in ("mask", "other") and qualifier: raise ValueError(f"Unexpected ACL qualifier: {line}") if qualifier: if not qualifier.isascii() or not qualifier.isdecimal(): raise ValueError(f"Expected numeric ACL qualifier: {line}") qualifier = str(int(qualifier)) if len(permissions) != 3 or any(c not in (expected, "-") for c, expected in zip(permissions, "rwx")): raise ValueError(f"Invalid permissions: {line}") key = f"{kind}:{qualifier}" if key in entries: raise ValueError(f"Duplicate ACL entry: {key}") entries[key] = sum(bit for bit, c in zip((4, 2, 1), permissions) if c != "-") if not all(key in entries for key in ("user:", "group:", "other:")): raise ValueError("Incomplete ACL") if any(k.startswith(("user:", "group:")) and k.split(":")[1] for k in entries): if "mask:" not in entries: raise ValueError("Named entries require a mask") return entries def snapshot(path): raw = run("getfacl", "--numeric", "--omit-header", "--access", path) entries = parse_acl(raw) mask = entries.get("mask:", 7) named = entries.get(f"user:{UID_A}") data = path.read_bytes() if data != PAYLOAD: raise AssertionError("Payload changed") return { "mode": f"{stat.S_IMODE(path.stat().st_mode):04o}", "acl": raw, "entries": entries, "namedUserEntryEffective": None if named is None else named & mask, "owningGroupEntryEffective": entries["group:"] & mask, "sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data), } def set_acl(path, mask="rw-"): run("setfacl", "--set", f"u::rw-,u:{UID_A}:rw-,g::---,m::{mask},o::---", path) def fixture(root, name, mask="r--"): path = root / name path.write_bytes(PAYLOAD) set_acl(path, mask) return path def create(path, mode): old = os.umask(0o077) try: fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode) with os.fdopen(fd, "wb") as output: output.write(PAYLOAD) finally: os.umask(old) def observe(directory): if os.getuid() in (UID_A, UID_B): raise ValueError("Fixture UID must differ from file owner") rows = [] with tempfile.TemporaryDirectory(prefix="hesela-acl-", dir=directory) as temp: root = Path(temp) run("setfacl", "-b", root) run("setfacl", "-k", root) source = fixture(root, "source", "rw-") original = snapshot(source) for flag in ("-a", "-aX", "-aA"): destination = root / ("copy" + flag) run("rsync", flag, "--", source, destination) rows.append({"case": "rsync" + flag, "before": original, "after": snapshot(destination)}) changed = fixture(root, "chmod") before = snapshot(changed) os.chmod(changed, 0o660) rows.append({"case": "chmod-group-write", "before": before, "after": snapshot(changed)}) for no_mask in (False, True): path = fixture(root, f"modify-{no_mask}") before = snapshot(path) flags = ["-n"] if no_mask else [] run("setfacl", *flags, "-m", f"u:{UID_B}:r--", path) rows.append({"case": "setfacl-no-mask" if no_mask else "setfacl-auto-mask", "before": before, "after": snapshot(path)}) for acl_last in (False, True): path = fixture(root, f"restore-{acl_last}") before = snapshot(path) if acl_last: os.chmod(path, 0o600) set_acl(path, "r--") else: set_acl(path, "r--") os.chmod(path, 0o600) rows.append({"case": "restore-acl-last" if acl_last else "restore-mode-last", "before": before, "after": snapshot(path)}) plain = root / "plain" plain.mkdir() inherited = root / "inherited" inherited.mkdir() run("setfacl", "-d", "--set", f"u::rwx,u:{UID_A}:rwx,g::---,m::rwx,o::---", inherited) for name, parent, mode in (("create-plain-0666", plain, 0o666), ("create-default-0666", inherited, 0o666), ("create-default-0600", inherited, 0o600)): path = parent / name create(path, mode) rows.append({"case": name, "requestedMode": f"{mode:04o}", "umask": "0077", "parentDefaultAcl": run("getfacl", "-n", "-c", "-d", parent), "after": snapshot(path)}) return rows def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--directory", action="append", required=True) parser.add_argument("--repetitions", type=int, default=3) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() if not 1 <= args.repetitions <= 100: parser.error("--repetitions must be 1..100") evidence = { "schemaVersion": "1.0", "kind": "controlled-metadata-observation", "generatedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "license": "CC0-1.0", "probeLicense": "MIT", "probeSha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "environment": {"kernel": platform.release(), "python": platform.python_version(), "uid": os.getuid(), "gid": os.getgid(), "setfacl": run("setfacl", "--version"), "getfacl": run("getfacl", "--version"), "rsync": run("rsync", "--version").splitlines()[0]}, "fixtureUids": [UID_A, UID_B], "rightsInterpretation": "Entry AND mask only; not an impersonated access check.", "runs": [], } for directory in args.directory: mount = run("findmnt", "-n", "-o", "FSTYPE", "-T", directory) evidence["runs"].append({"filesystem": mount, "repetitions": [ observe(directory) for _ in range(args.repetitions)]}) args.output.write_text(json.dumps(evidence, indent=2) + "\n") print(json.dumps({"runs": len(evidence["runs"]), "cases": sum(len(rows) for r in evidence["runs"] for rows in r["repetitions"])})) if __name__ == "__main__": main()