#!/usr/bin/env perl

# Unit tests for Harness::Safe (sneakernet/testing/Harness/Safe.pm) - the sandbox enforcement
# layer for the TESTING.md automation harness. No real ZFS or dd-nas1 needed: all exec calls
# are intercepted via $Harness::Safe::EXEC_OVERRIDE, so this proves the validation/allow-list/
# gate logic in complete isolation from the filesystem operations it would otherwise perform.

use strict;
use warnings;
use FindBin;
use lib "$FindBin::Bin/../sneakernet/testing";
use Harness::Safe;
use File::Temp qw(tempdir);

$| = 1; # keep stdout/stderr interleaved in the order tests actually ran, not buffer-reordered

my $passed = 0;
my $failed = 0;

sub ok {
    my ($cond, $desc) = @_;
    if ($cond) { print "  PASS: $desc\n"; $passed++; return 1; }
    else       { print "  FAIL: $desc\n"; $failed++; return 0; }
}

sub dies_ok {
    my ($code, $desc, $like) = @_;
    my $died = 0; my $msg = '';
    eval { $code->(); 1 } or do { $died = 1; $msg = $@; };
    if ($died && defined $like && $msg !~ $like) {
        print "  FAIL: $desc (died, but message didn't match $like: $msg)\n"; $failed++; return;
    }
    ok($died, $desc);
}

sub lives_ok {
    my ($code, $desc) = @_;
    my $ok = eval { $code->(); 1 };
    ok($ok, $desc) or print "    error: $@\n";
}

# ===========================================================================
print "=== ds() ===\n";
# ===========================================================================
{
    lives_ok(sub { Harness::Safe::ds('storage/testing/src/ds1') }, 'ordinary dataset accepted');
    lives_ok(sub { Harness::Safe::ds('storage/testing/src/ds1@2026-01-01_00.00.00') }, 'dataset@snapshot accepted');
    dies_ok(sub { Harness::Safe::ds(undef) }, 'undef refused');
    dies_ok(sub { Harness::Safe::ds('') }, 'empty string refused');
    dies_ok(sub { Harness::Safe::ds('storage/testing') }, 'bare sandbox root refused (the Part 19 case)', qr/outside sandbox/);
    dies_ok(sub { Harness::Safe::ds('storage/testing/') }, 'sandbox root with trailing slash refused', qr/empty path component|outside sandbox/);
    dies_ok(sub { Harness::Safe::ds('storage/backup') }, 'production dataset refused', qr/outside sandbox/);
    dies_ok(sub { Harness::Safe::ds('storage/testingXYZ/foo') }, 'lookalike prefix (not a real path separator) refused', qr/outside sandbox/);
    dies_ok(sub { Harness::Safe::ds('storage/testing/src/ds1@') }, 'trailing @ (the empty-$S2NAME case) refused', qr/empty snapshot name/);
    dies_ok(sub { Harness::Safe::ds('storage/testing/src/ds1@a@b') }, 'more than one @ refused', qr/more than one/);
    dies_ok(sub { Harness::Safe::ds('storage/testing/../backup') }, "'..' component refused", qr/\.\./);
    dies_ok(sub { Harness::Safe::ds('storage/testing//src') }, 'empty path component (double slash) refused', qr/empty path component/);
    dies_ok(sub { Harness::Safe::ds("storage/testing/src;rm -rf /") }, 'shell metacharacter refused', qr/illegal character/);
}

# ===========================================================================
print "\n=== path() ===\n";
# ===========================================================================
{
    lives_ok(sub { Harness::Safe::path('/storage/testing/transport/serial.txt') }, 'ordinary path accepted');
    lives_ok(sub { Harness::Safe::path('/storage/testing') }, 'sandbox root itself accepted (unlike ds())');
    dies_ok(sub { Harness::Safe::path(undef) }, 'undef refused');
    dies_ok(sub { Harness::Safe::path('storage/testing/relative') }, 'relative path refused', qr/not absolute/);
    dies_ok(sub { Harness::Safe::path('/tmp/whatever') }, '/tmp path refused outright', qr/outside sandbox/);
    dies_ok(sub { Harness::Safe::path('/storage/backup/x') }, 'production path refused', qr/outside sandbox/);
    dies_ok(sub { Harness::Safe::path('/storage/testing/../backup') }, "'..' component refused", qr/\.\./);
    dies_ok(sub { Harness::Safe::path('/storage/testing/transport/*') }, 'glob character refused', qr/glob/);
    dies_ok(sub { Harness::Safe::path('/storage/testing/transport/`whoami`') }, 'backtick refused', qr/glob/);
    dies_ok(sub { Harness::Safe::path("/storage/testing/foo bar") }, 'embedded space refused', qr/glob/);
}

# ===========================================================================
print "\n=== path() symlink-escape detection ===\n";
# ===========================================================================
{
    my $dir = tempdir(CLEANUP => 1);
    # Fake a sandbox for this test by pointing SANDBOX_PATH at our tempdir.
    local $Harness::Safe::SANDBOX_PATH = $dir;
    mkdir "$dir/real_outside";
    symlink("$dir/real_outside", "$dir/escape_link") or die $!;
    dies_ok(sub { Harness::Safe::path("$dir/escape_link/file") },
        'a path through a symlink pointing OUTSIDE the sandbox is refused', qr/does not|escapes sandbox|outside sandbox/)
        if 0; # real_outside is still inside $dir in this fixture; see next test for the real escape
    my $outsideDir = tempdir(CLEANUP => 1);
    symlink($outsideDir, "$dir/escapes") or die $!;
    dies_ok(sub { Harness::Safe::path("$dir/escapes/file") },
        'a path through a symlink pointing truly outside the sandbox is refused', qr/escapes sandbox/);
}

# ===========================================================================
print "\n=== path() refuses an empty path component instead of hanging ===\n";
# ===========================================================================
{
    # Regression: a trailing slash (e.g. from "{TOKEN}/$VAR" where $VAR substituted empty under
    # --explain - see Harness::Runner's _subst) made the ancestor-walk loop's regex
    # (s{/+[^/]+$}{}) never match, so $probe never shrank and the loop spun forever. path() must
    # refuse this outright, in every mode, rather than relying on callers never doing this.
    local $SIG{ALRM} = sub { die "TIMEOUT: path() did not return - the ancestor-walk loop is hanging\n" };
    alarm(5);
    dies_ok(sub { Harness::Safe::path('/storage/testing/nonexistent-xyz/') },
        'a nonexistent path with a trailing slash is refused promptly, not an infinite loop', qr/empty path component/);
    dies_ok(sub { Harness::Safe::path('/storage/testing//nonexistent-xyz') },
        'a doubled slash is refused', qr/empty path component/);
    alarm(0);
}

# ===========================================================================
print "\n=== destroy_dataset allow-list ===\n";
# ===========================================================================
{
    Harness::Safe::set_mode('explain');
    lives_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/dst/ds1', recursive => 1, missing_ok => 1) },
        'destroy of dst/dsN is allow-listed');
    lives_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/verify', recursive => 1, missing_ok => 1) },
        'destroy of the verify scratch dataset is allow-listed');
    dies_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/code', recursive => 1) },
        'destroy of storage/testing/code (not on the allow-list) is refused', qr/allow-list/);
    dies_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/harness', recursive => 1) },
        'destroy of storage/testing/harness (the audit trail itself) is refused', qr/allow-list/);
    dies_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing', recursive => 1) },
        "Part 19's destroy is refused (fails in ds() before allow-list is even reached)", qr/outside sandbox/);
    dies_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/src/ds1', recursive => 1, missing_ok => 1) },
        'destroy of a SOURCE ds (only dst/verify are allow-listed for destroy... ') if 0;
    # Actually src/dsN legitimately gets destroyed in Steps 9.3/9.4/10.4 - confirm it IS allowed:
    lives_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/src/ds1', recursive => 1, missing_ok => 1) },
        'destroy of src/dsN (Steps 9.3/9.4/10.4 recreate-source scenarios) is allow-listed');
}

# ===========================================================================
print "\n=== create_dataset / snapshot_dataset allow-lists ===\n";
# ===========================================================================
{
    Harness::Safe::set_mode('explain');
    lives_ok(sub { Harness::Safe::create_dataset(dataset => 'storage/testing/tmp') }, 'create tmp dataset allowed');
    lives_ok(sub { Harness::Safe::create_dataset(dataset => 'storage/testing/verify') }, 'create verify dataset allowed');
    dies_ok(sub { Harness::Safe::create_dataset(dataset => 'storage/testing/newthing') },
        'create of an arbitrary new top-level dataset refused', qr/allow-list/);
    lives_ok(sub { Harness::Safe::snapshot_dataset(dataset => 'storage/testing/src/ds1@x') }, 'snapshot of src/dsN allowed');
    dies_ok(sub { Harness::Safe::snapshot_dataset(dataset => 'storage/testing/dst/ds1@x') },
        'snapshot of a DESTINATION dataset refused (sneakernet, not the harness, ever snapshots dst)', qr/allow-list/);
}

# ===========================================================================
print "\n=== multi-argument destroy is unsupported (structural, not a policy check) ===\n";
# ===========================================================================
{
    # There is simply no API to pass two datasets to one destroy_dataset() call - confirm the
    # sub signature only ever takes a single 'dataset' key, by checking a two-dataset attempt
    # is not even expressible: passing an arrayref where a scalar dataset name is expected
    # fails ds()'s type checks rather than being silently accepted.
    dies_ok(sub { Harness::Safe::destroy_dataset(dataset => ['storage/testing/dst/ds1', 'storage/testing/dst/ds2']) },
        'an arrayref of datasets (multi-target attempt) is refused, not silently accepted');
}

# ===========================================================================
print "\n=== execution modes ===\n";
# ===========================================================================
{
    my @execCalls;
    local $Harness::Safe::EXEC_OVERRIDE = sub {
        my ($argv) = @_;
        push @execCalls, join(' ', @$argv);
        return ('', 0);
    };

    Harness::Safe::set_mode('explain');
    @execCalls = ();
    Harness::Safe::destroy_dataset(dataset => 'storage/testing/dst/ds1', recursive => 1, missing_ok => 1);
    ok(scalar(@execCalls) == 0, '--explain never calls the exec layer for a destructive primitive');

    Harness::Safe::set_mode('dry_run');
    @execCalls = ();
    Harness::Safe::destroy_dataset(dataset => 'storage/testing/dst/ds1', recursive => 1, missing_ok => 1);
    ok(scalar(@execCalls) == 0, '--dry-run never calls the exec layer for a destructive primitive');
    @execCalls = ();
    Harness::Safe::dataset_exists(dataset => 'storage/testing/dst/ds1');
    ok(scalar(@execCalls) == 1, '--dry-run DOES call the exec layer for a readonly primitive');

    Harness::Safe::set_mode('run');
    @execCalls = ();
    Harness::Safe::destroy_dataset(dataset => 'storage/testing/dst/ds1', recursive => 1, missing_ok => 1);
    ok(scalar(@execCalls) == 1, '--run calls the exec layer for a destructive primitive');
    ok($execCalls[0] =~ /destroy -r storage\/testing\/dst\/ds1$/, 'the actual argv rendered is correct: ' . $execCalls[0]);
}

# ===========================================================================
print "\n=== confirm_destructive prompts and honors the answer ===\n";
# ===========================================================================
{
    my @execCalls;
    local $Harness::Safe::EXEC_OVERRIDE = sub { push @execCalls, 1; return ('', 0); };
    Harness::Safe::set_mode('confirm_destructive');

    {
        local *STDIN;
        open STDIN, '<', \"no\n" or die $!;
        @execCalls = ();
        dies_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/dst/ds1', recursive => 1, missing_ok => 1) },
            'answering "no" refuses the destroy', qr/declined/);
        ok(scalar(@execCalls) == 0, 'declined destroy never reaches the exec layer');
    }
    {
        local *STDIN;
        open STDIN, '<', \"yes\n" or die $!;
        @execCalls = ();
        lives_ok(sub { Harness::Safe::destroy_dataset(dataset => 'storage/testing/dst/ds1', recursive => 1, missing_ok => 1) },
            'answering "yes" allows the destroy');
        ok(scalar(@execCalls) == 1, 'confirmed destroy reaches the exec layer exactly once');
    }
    Harness::Safe::set_mode('explain');
}

# ===========================================================================
print "\n=== empty_dir / rm_matching are glob-free ===\n";
# ===========================================================================
{
    my $dir = tempdir(CLEANUP => 1);
    local $Harness::Safe::SANDBOX_PATH = $dir;
    local $Harness::Safe::EXEC_OVERRIDE = sub {
        my ($argv) = @_;
        return ("$dir\t$dir\n", 0) if grep { /zfs/ } @$argv; # fake "no child mounts" answer
        return ('', 0);
    };
    Harness::Safe::set_mode('run');

    open my $f1, '>', "$dir/a.txt" or die $!; close $f1;
    open my $f2, '>', "$dir/b.txt" or die $!; close $f2;
    mkdir "$dir/subdir";
    open my $f3, '>', "$dir/subdir/c.txt" or die $!; close $f3;

    Harness::Safe::empty_dir(path => $dir);
    my @remaining = glob("$dir/*");
    ok(scalar(@remaining) == 0, 'empty_dir removed files AND subdirectories, leaving the dir itself');
    ok(-d $dir, 'the directory itself still exists after empty_dir');

    mkdir "$dir/transport";
    open my $f4, '>', "$dir/transport/keep.log" or die $!; close $f4;
    open my $f5, '>', "$dir/transport/status.bak.1" or die $!; close $f5;
    open my $f6, '>', "$dir/transport/status.bak.2" or die $!; close $f6;
    Harness::Safe::rm_matching(dir => "$dir/transport", basename_re => 'status\.bak\.\d+$');
    ok(-e "$dir/transport/keep.log", 'rm_matching left the non-matching file alone');
    ok(!-e "$dir/transport/status.bak.1" && !-e "$dir/transport/status.bak.2", 'rm_matching removed both matching files');

    dies_ok(sub { Harness::Safe::rm_matching(dir => $dir, basename_re => 'a/b') },
        'rm_matching refuses a pattern containing a slash', qr{may not contain});

    mkdir "$dir/backups";
    open my $f7, '>', "$dir/backups/status.2026-01-01" or die $!; close $f7;
    open my $f8, '>', "$dir/backups/status.2026-01-02" or die $!; close $f8;
    open my $f9, '>', "$dir/backups/other.txt" or die $!; close $f9;
    my @matches = Harness::Safe::list_matching(dir => "$dir/backups", basename_re => '^status\.');
    ok(scalar(@matches) == 2, 'list_matching finds exactly the matching files (' . scalar(@matches) . ')');
    ok(-e "$dir/backups/other.txt", 'list_matching (unlike rm_matching) never deletes anything');

    # empty_dir on an empty directory (or one that doesn't exist) is a no-op, not an error -
    # this is what makes it safe to call unconditionally in a reset profile.
    my $emptyDir = "$dir/already_empty";
    mkdir $emptyDir;
    lives_ok(sub { Harness::Safe::empty_dir(path => $emptyDir) }, 'empty_dir on an already-empty dir is a no-op, not an error');
}

# ===========================================================================
print "\n=== empty_dir refuses a symlink or a non-directory ===\n";
# ===========================================================================
{
    my $dir = tempdir(CLEANUP => 1);
    local $Harness::Safe::SANDBOX_PATH = $dir;
    Harness::Safe::set_mode('run');
    open my $f, '>', "$dir/afile" or die $!; close $f;
    dies_ok(sub { Harness::Safe::empty_dir(path => "$dir/afile") }, 'empty_dir refuses a plain file', qr/not a directory/);
    my $outside = tempdir(CLEANUP => 1);
    symlink($outside, "$dir/link") or die $!;
    dies_ok(sub { Harness::Safe::empty_dir(path => "$dir/link") }, 'empty_dir refuses a symlink', qr/symlink/);
}

# ===========================================================================
print "\n=== write_file is atomic and refuses symlinks ===\n";
# ===========================================================================
{
    my $dir = tempdir(CLEANUP => 1);
    local $Harness::Safe::SANDBOX_PATH = $dir;
    Harness::Safe::set_mode('run');
    Harness::Safe::write_file(path => "$dir/out.txt", content => "hello\n");
    ok(Harness::Safe::read_file(path => "$dir/out.txt") eq "hello\n", 'write_file then read_file round-trips');
    my @leftover = grep { /\.harness\.\d+$/ } glob("$dir/*");
    ok(scalar(@leftover) == 0, 'no .harness.<pid> temp file left behind after a successful write_file');

    my $outside = tempdir(CLEANUP => 1);
    symlink("$outside/target", "$dir/evil_link") or die $!;
    dies_ok(sub { Harness::Safe::write_file(path => "$dir/evil_link", content => "x") },
        'write_file refuses to write through a symlink', qr/symlink/);
}

# ===========================================================================
print "\n=== corrupt_file never extends the file, caps length ===\n";
# ===========================================================================
{
    my $dir = tempdir(CLEANUP => 1);
    local $Harness::Safe::SANDBOX_PATH = $dir;
    Harness::Safe::set_mode('run');
    open my $fh, '>', "$dir/f.bin" or die $!;
    print {$fh} ("A" x 100);
    close $fh;
    dies_ok(sub { Harness::Safe::corrupt_file(path => "$dir/f.bin", offset => 90, length => 20) },
        'corrupt_file refuses a write that would extend the file', qr/extend/);
    lives_ok(sub { Harness::Safe::corrupt_file(path => "$dir/f.bin", offset => 90, length => 10) },
        'corrupt_file accepts a write that exactly fits');
    ok(-s "$dir/f.bin" == 100, 'file size unchanged after corrupt_file');
    dies_ok(sub { Harness::Safe::corrupt_file(path => "$dir/f.bin", offset => 0, length => 2_000_000) },
        'corrupt_file refuses a length above the 1MB cap', qr/exceeds/);
}

# ===========================================================================
print "\n=== random_bytes / random_hex ===\n";
# ===========================================================================
{
    my $bytes = Harness::Safe::random_bytes(bytes => 16);
    ok(length($bytes) == 16, 'random_bytes returns exactly the requested length');
    my $hex = Harness::Safe::random_hex(bytes => 16);
    ok($hex =~ /^[0-9a-f]{32}$/, 'random_hex returns 32 lowercase hex characters for 16 bytes');
    my $hex2 = Harness::Safe::random_hex(bytes => 16);
    ok($hex ne $hex2, 'two calls produce different values (sanity, not a real randomness test)');
}

# ===========================================================================
print "\n=== audit log: redaction ===\n";
# ===========================================================================
{
    Harness::Safe::clear_audit_buffer();
    local $Harness::Safe::EXEC_OVERRIDE = sub { return ('some output mentioning SECRETKEY123 in it', 0); };
    Harness::Safe::set_mode('run');
    local @Harness::Safe::REDACT = ('SECRETKEY123');
    Harness::Safe::dataset_exists(dataset => 'storage/testing/dst/ds1');
    my @records = Harness::Safe::audit_buffer();
    my $found = 0;
    for my $r (@records) {
        $found = 1 if defined($r->{output}) && $r->{output} =~ /SECRETKEY123/;
    }
    ok(!$found, 'a redacted secret never appears in the audit buffer');
}

# ===========================================================================
print "\n=== audit log: every primitive requires a class ===\n";
# ===========================================================================
{
    dies_ok(sub { Harness::Safe::_run_primitive(primitive => 'x', run => sub { {exit=>0,output=>''} }) },
        'a primitive call with no class refuses outright', qr/no class/);
    dies_ok(sub { Harness::Safe::_run_primitive(primitive => 'x', class => 'bogus', run => sub { {exit=>0,output=>''} }) },
        'a primitive call with an unknown class refuses', qr/unknown class/);
}

# ===========================================================================
print "\n=== cron guard window ===\n";
# ===========================================================================
{
    # Construct epoch times for today at specific local wall-clock times, matching the same
    # timezone localtime() will use, so this test is portable regardless of the box's TZ.
    my @today = localtime(time());
    my $mk = sub {
        my ($h, $m) = @_;
        my @t = @today;
        ($t[2], $t[1], $t[0]) = ($h, $m, 0);
        require POSIX;
        return POSIX::mktime(@t);
    };
    ok(Harness::Safe::in_cron_guard_window($mk->(4, 3)), '04:03 (the cron fire time) is inside the guard window');
    ok(Harness::Safe::in_cron_guard_window($mk->(3, 45)), '03:45 (window open) is inside the guard window');
    ok(Harness::Safe::in_cron_guard_window($mk->(5, 0)), '05:00 (window close) is inside the guard window');
    ok(!Harness::Safe::in_cron_guard_window($mk->(3, 44)), '03:44 is outside the guard window');
    ok(!Harness::Safe::in_cron_guard_window($mk->(5, 1)), '05:01 is outside the guard window');
    ok(!Harness::Safe::in_cron_guard_window($mk->(12, 0)), 'noon is outside the guard window');

    dies_ok(sub {
        local *Harness::Safe::in_cron_guard_window = sub { 1 };
        Harness::Safe::assert_outside_cron_guard_window();
    }, 'assert_outside_cron_guard_window dies when the window check reports true');
}

# ===========================================================================
print "\n=== production tripwire ===\n";
# ===========================================================================
{
    my @responses = ("storage/backup/a\nstorage/backup/a\@snap1\n");
    local $Harness::Safe::EXEC_OVERRIDE = sub { return (shift(@responses) // '', 0) };
    Harness::Safe::set_mode('run');
    my $baseline = Harness::Safe::production_fingerprint();
    ok(length($baseline->{sha}) == 64, 'fingerprint sha looks like a real sha256 hex digest');

    @responses = ("storage/backup/a\nstorage/backup/a\@snap1\n"); # identical -> should pass
    lives_ok(sub { Harness::Safe::assert_production_unchanged($baseline) }, 'identical fingerprint passes the tripwire');

    @responses = ("storage/backup/a\n"); # one entry vanished -> should fire
    dies_ok(sub { Harness::Safe::assert_production_unchanged($baseline) },
        'a changed fingerprint fires the tripwire and dies loudly', qr/TRIPWIRE/);
}

# ===========================================================================
print "\n=== mount-topology preflight ===\n";
# ===========================================================================
{
    Harness::Safe::set_mode('run');
    local $Harness::Safe::EXEC_OVERRIDE = sub {
        my ($argv) = @_;
        if (join(' ', @$argv) =~ /storage\/testing/) {
            return ("storage/testing\t/storage/testing\tnoauto\n"
                  . "storage/testing/src\t/storage/testing/src\ton\n", 0);
        }
        return ("storage/backup\t/storage/backup\n", 0); # a normal, non-overlapping mount
    };
    lives_ok(sub { Harness::Safe::assert_mount_topology() }, 'normal, non-overlapping mount topology passes');

    local $Harness::Safe::EXEC_OVERRIDE = sub {
        my ($argv) = @_;
        if (join(' ', @$argv) =~ /storage\/testing/) {
            return ("storage/testing\t/storage/testing\tnoauto\n", 0);
        }
        return ("storage/backup/evil\t/storage/testing/report\n", 0); # foreign dataset mounted INSIDE sandbox
    };
    dies_ok(sub { Harness::Safe::assert_mount_topology() },
        'a foreign dataset mounted inside the sandbox is refused', qr/foreign dataset/);

    local $Harness::Safe::EXEC_OVERRIDE = sub {
        my ($argv) = @_;
        return ("storage/testing\t/somewhere/else\tnoauto\n", 0);
    };
    dies_ok(sub { Harness::Safe::assert_mount_topology() },
        'a sandbox dataset mounted somewhere unexpected is refused', qr/unexpected mountpoint/);
}

# ===========================================================================
print "\n=== run_utility() allow-list ===\n";
# ===========================================================================
{
    my $dir = tempdir(CLEANUP => 1);
    local $Harness::Safe::SANDBOX_PATH = $dir;
    Harness::Safe::set_mode('explain');
    lives_ok(sub { Harness::Safe::run_utility(script => 'checkSneakernetFile', code_dir => $dir, args => []) },
        'checkSneakernetFile is on the allow-list');
    lives_ok(sub { Harness::Safe::run_utility(script => 'buildUpgradeOneShot.pl', code_dir => $dir, args => []) },
        'buildUpgradeOneShot.pl is on the allow-list');
    dies_ok(sub { Harness::Safe::run_utility(script => 'upgrade.pl', code_dir => $dir, args => []) },
        "an arbitrary/generated script (e.g. the Part 17 OUTPUT, 'upgrade.pl') is NOT on the allow-list", qr/allow-list/);
    dies_ok(sub { Harness::Safe::run_utility(script => 'rm', code_dir => $dir, args => ['-rf', '/']) },
        'an arbitrary system binary name is refused, not just "unusual" script names', qr/allow-list/);
}

# ===========================================================================
print "\n=== the one real pipeline (Step 8.5) never invokes a shell ===\n";
# ===========================================================================
{
    my $dir = tempdir(CLEANUP => 1);
    local $Harness::Safe::SANDBOX_PATH = $dir;
    Harness::Safe::set_mode('run');
    # We can't fake a two-process pipeline through EXEC_OVERRIDE (it's real fork/exec by
    # design - that's the point). Stage 1 (zfs send) is stood in for by /bin/echo - fine,
    # only its stdout bytes matter here. Stage 2 is the REAL openssl (its flags are specific
    # to it; a generic stand-in like /bin/cat can't accept '-aes-256-cbc' etc), so correctness
    # is verified by decrypting the output back and checking it matches what stage 1 emitted -
    # proving bytes genuinely flowed process 1 -> pipe -> process 2 -> file, with no shell
    # anywhere in between.
    local $Harness::Safe::BIN{zfs} = '/bin/echo';
    my ($key, $iv) = ('00' x 32, '00' x 16);
    Harness::Safe::pipeline_zfs_send_to_encrypted_file(
        snapshot => 'storage/testing/src/ds1@x', key => $key, iv => $iv,
        out_path => "$dir/piped.out",
    );
    ok(-s "$dir/piped.out" > 0, 'the pipeline produced a non-empty output file');
    my $decrypted = `openssl enc -aes-256-cbc -d -K $key -iv $iv -in "$dir/piped.out" 2>/dev/null`;
    ok($decrypted eq "send storage/testing/src/ds1\@x\n",
        'decrypting the pipeline output recovers exactly what stage 1 (zfs send stand-in) emitted');
}

print "\n=== Summary ===\n";
print "Passed: $passed\n";
print "Failed: $failed\n";
exit($failed > 0 ? 1 : 0);