#! /usr/bin/env perl

# Simplified BSD License (FreeBSD License)
#
# Copyright (c) 2026, Daily Data Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice, this
#    list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

# Harness::Safe - the sandbox enforcement layer for the sneakernet TESTING.md automation
# harness. See sneakernet/Documentation/TESTING_automation.md section 6 for the full design
# rationale; this module implements that section.
#
# THE CENTRAL RULE: this is the only module in the harness permitted to fork, exec, unlink,
# rename, or open a file for writing. Every other harness module calls INTO here with already-
# typed, already-validated arguments; nothing outside this file ever holds a shell command
# string, because there is no shell command string anywhere in this harness's process tree.
# Every side effect is an exec() of an argv LIST (never a string passed to /bin/sh), or a
# direct Perl filesystem call.
#
# Two constructors gate every argument that reaches an exec or a filesystem call:
#   ds($name)   - a ZFS dataset (or dataset@snapshot) name, required to be under storage/testing/
#   path($p)    - an absolute filesystem path, required to be under /storage/testing/
# Both die (via _refuse, which logs a REFUSED audit record first) rather than return false, so
# a caller cannot forget to check a return value.

package Harness::Safe;

use strict;
use warnings;
use Carp qw(croak);
use Cwd qw(abs_path getcwd);
use File::Basename qw(dirname basename);
use File::Path qw(make_path);
use Fcntl qw(:flock O_WRONLY O_RDONLY O_CREAT O_EXCL SEEK_SET);
use POSIX qw(strftime WNOHANG);
use JSON::PP qw(encode_json);
use Digest::SHA qw(sha256_hex);
use Time::HiRes qw(time);

our $VERSION = '0.1.0';

our $SANDBOX_DS   = 'storage/testing';
our $SANDBOX_PATH = '/storage/testing';

# Absolute paths to the binaries this harness is willing to exec, resolved once. Real argv[0]
# always comes from this table, NEVER from $ENV{PATH} search - a hostile or merely different
# PATH cannot substitute a different program.
our %BIN = (
    zfs        => '/sbin/zfs',
    openssl    => '/usr/bin/openssl',
    xz         => '/usr/local/bin/xz',
    perl       => '/usr/local/bin/perl',
    svn        => '/usr/local/bin/svn',
    svnversion => '/usr/local/bin/svnversion',
    mount      => '/sbin/mount',
    sendmail   => '/usr/sbin/sendmail',
);

# ---------------------------------------------------------------------------
# Execution mode. One of: explain | dry_run | confirm_destructive | run.
# See TESTING_automation.md 6.8. Set via Harness::Safe::set_mode('explain') before use.
# ---------------------------------------------------------------------------
our $MODE = 'explain';
our %VALID_MODES = map { $_ => 1 } qw(explain dry_run confirm_destructive run);

sub set_mode {
    my ($mode) = @_;
    croak "Harness::Safe: unknown mode '$mode'" unless $VALID_MODES{$mode};
    $MODE = $mode;
}

# Injection seam for tests: if set, ALL exec calls go through this coderef instead of really
# forking. Signature: $coderef->(\@argv) -> ($combinedOutput, $exitStatus). Production code
# never sets this; testLibrary/test_Harness_Safe.pl does, so the whole primitive/allow-list/
# gate surface is testable without real ZFS or a real dd-nas1.
our $EXEC_OVERRIDE;

# Where the audit JSONL is appended. Set by the runner at startup via set_audit_log(). If
# unset, audit records are only kept in memory (_AUDIT_BUFFER) - used by tests and --explain.
our $AUDIT_LOG_PATH;
my @AUDIT_BUFFER;
my $CURRENT_STEP = '-';
my $SEQ = 0;

sub set_audit_log { my ($path) = @_; $AUDIT_LOG_PATH = $path; }
sub set_current_step { my ($id) = @_; $CURRENT_STEP = $id; }
sub audit_buffer { return @AUDIT_BUFFER; }
sub clear_audit_buffer { @AUDIT_BUFFER = (); $SEQ = 0; }

# Values that must never reach the audit log or the console verbatim - e.g. the transport
# encryption key, passed in argv to openssl by Steps 7.2c/8.5. Callers push onto this before
# an operation that will expose a secret in argv, and Harness::Safe redacts every occurrence.
our @REDACT;

sub _redact {
    my ($text) = @_;
    return $text unless defined $text;
    for my $secret (grep { defined && length } @REDACT) {
        my $q = quotemeta($secret);
        $text =~ s/$q/[REDACTED]/g;
    }
    return $text;
}

sub _audit {
    my (%rec) = @_;
    $rec{ts}   = strftime('%Y-%m-%dT%H:%M:%S', gmtime(int($rec{_epoch} // time))) . 'Z';
    delete $rec{_epoch};
    $rec{step} = $CURRENT_STEP;
    $rec{seq}  = ++$SEQ;
    $rec{mode} = $MODE;
    if (exists $rec{args}) {
        my %a = %{ $rec{args} };
        $a{$_} = _redact($a{$_}) for keys %a;
        $rec{args} = \%a;
    }
    $rec{output} = _redact($rec{output}) if exists $rec{output};
    push @AUDIT_BUFFER, \%rec;
    if (defined $AUDIT_LOG_PATH) {
        open my $fh, '>>', $AUDIT_LOG_PATH or croak "Harness::Safe: cannot append audit log '$AUDIT_LOG_PATH': $!";
        flock($fh, LOCK_EX) or croak "Harness::Safe: cannot lock audit log: $!";
        print {$fh} encode_json(\%rec) . "\n";
        close $fh;
    }
    return \%rec;
}

sub _refuse {
    my ($msg) = @_;
    _audit(class => 'n/a', primitive => 'n/a', verdict => 'REFUSED', reason => $msg);
    die "Harness::Safe: REFUSED: $msg\n";
}

# ---------------------------------------------------------------------------
# Type constructors - see TESTING_automation.md 6.2
# ---------------------------------------------------------------------------

sub ds {
    my ($name) = @_;
    _refuse('dataset is undef') unless defined $name;
    _refuse('dataset is empty string') unless length $name;
    _refuse("dataset has illegal character: '$name'") if $name =~ m{[^A-Za-z0-9_.:@/-]};
    _refuse("dataset has '..' path component: '$name'") if $name =~ m{(^|/)\.\.?(/|@|$)};
    _refuse("dataset has an empty path component: '$name'")
        if $name =~ m{//} || $name =~ m{/\@} || $name =~ m{^/} || $name =~ m{/$};
    _refuse("dataset has an empty snapshot name: '$name'") if $name =~ m{\@$};
    _refuse("dataset has more than one '\@': '$name'") if (() = $name =~ /\@/g) > 1;
    # The trailing slash here is deliberate: it makes the bare string "storage/testing" itself
    # structurally impossible to pass as a dataset argument (see TESTING_automation.md 6.2).
    _refuse("dataset outside sandbox: '$name'") unless $name =~ m{^\Q$SANDBOX_DS\E/};
    return $name;
}

sub path {
    my ($p) = @_;
    _refuse('path is undef') unless defined $p;
    _refuse('path is empty string') unless length $p;
    _refuse("path is not absolute: '$p'") unless $p =~ m{^/};
    _refuse("path contains a glob/shell metacharacter: '$p'") if $p =~ m{[*?\[\]{}|;&<>`\$\s]};
    _refuse("path has a '..' component: '$p'") if $p =~ m{(^|/)\.\.(/|$)};
    _refuse("path outside sandbox: '$p'") unless $p =~ m{^\Q$SANDBOX_PATH\E(?:/|$)};
    # An empty path component (a doubled slash, or a trailing slash - e.g. from "{TOKEN}/$VAR"
    # where $VAR substituted empty) is refused structurally, in every mode: past this point
    # nothing else in path() can distinguish it from a real directory reference, and the
    # ancestor-walk loop below never terminates on a string ending in '/' (its regex requires a
    # non-slash character before the anchor, so a trailing slash can never be stripped and
    # $probe never shrinks). This check is intentionally NOT relaxed under --explain - unlike
    # _subst's empty-value guard, path() is a general safety primitive called directly by other
    # code (including tests exercising its symlink-escape detection) and must behave identically
    # regardless of $MODE.
    _refuse("path has an empty path component: '$p'") if $p =~ m{//} || $p =~ m{/$};

    # Symlink escape check: resolve the deepest existing ancestor and re-check containment.
    my $probe = $p;
    $probe =~ s{/+[^/]+$}{} while length($probe) > length($SANDBOX_PATH) && !-e $probe;
    if (-e $probe) {
        my $real = abs_path($probe);
        _refuse("cannot resolve '$probe'") unless defined $real;
        _refuse("path escapes sandbox via symlink: '$p' -> '$real'")
            unless $real eq $SANDBOX_PATH || $real =~ m{^\Q$SANDBOX_PATH\E/};
    }
    return $p;
}

# ---------------------------------------------------------------------------
# Per-primitive allow-lists - TESTING_automation.md 6.3
# ---------------------------------------------------------------------------

our @DESTROY_ALLOW = (
    qr{^\Q$SANDBOX_DS\E/(?:src|dst)/ds[0-9]+(?:\@[A-Za-z0-9_.:-]+)?$},
    qr{^\Q$SANDBOX_DS\E/verify(?:/[A-Za-z0-9_.-]+)?$},
);

our @CREATE_ALLOW = (
    qr{^\Q$SANDBOX_DS\E/(?:src|dst)$},              # the parent containers (Part 0.3 only)
    qr{^\Q$SANDBOX_DS\E/(?:src|dst)/ds[0-9]+$},
    qr{^\Q$SANDBOX_DS\E/(?:verify|tmp|harness)$},
);

our @SNAPSHOT_ALLOW = (
    qr{^\Q$SANDBOX_DS\E/src/ds[0-9]+\@[A-Za-z0-9_.:-]+$},
    qr{^\Q$SANDBOX_DS\E/code\@harness_[A-Za-z0-9_]+$},
);

sub _check_allow {
    my ($name, $listRef, $what) = @_;
    _refuse("$what target not on allow-list: '$name'")
        unless grep { $name =~ $_ } @$listRef;
}

# ---------------------------------------------------------------------------
# Low-level exec: no shell, ever. Runs @$argv directly via fork+exec (PROGRAM form, which
# bypasses the shell even for a single-element list), with stderr merged into stdout - the
# harness always wants "what would a human seeing '2>&1 | tee' have seen".
# ---------------------------------------------------------------------------

sub _exec_argv {
    my ($argv) = @_;
    if ($EXEC_OVERRIDE) {
        return $EXEC_OVERRIDE->($argv);
    }
    pipe(my $rd, my $wr) or croak "Harness::Safe: pipe: $!";
    my $pid = fork();
    croak "Harness::Safe: fork: $!" unless defined $pid;
    if ($pid == 0) {
        close $rd;
        open(STDOUT, '>&', $wr) or POSIX::_exit(126);
        open(STDERR, '>&', $wr) or POSIX::_exit(126);
        close $wr;
        { exec { $argv->[0] } @$argv; }
        POSIX::_exit(127); # exec failed
    }
    close $wr;
    local $/;
    my $output = <$rd>;
    close $rd;
    waitpid($pid, 0);
    return ($output // '', $?);
}

# Two-stage pipeline (argv1 | argv2 > $outPath), used for exactly one thing in the whole
# catalogue: Step 8.5's "zfs send SNAP | openssl enc ... > file". No shell is invoked; the
# pipe is a real POSIX pipe between two directly-exec'd children.
sub _exec_pipeline {
    my ($argv1, $argv2, $outPath) = @_;
    pipe(my $midRd, my $midWr) or croak "Harness::Safe: pipe: $!";
    sysopen(my $outFh, $outPath, O_WRONLY | O_CREAT | O_EXCL, 0644)
        or croak "Harness::Safe: cannot create '$outPath': $!";

    my $pid1 = fork();
    croak "Harness::Safe: fork: $!" unless defined $pid1;
    if ($pid1 == 0) {
        close $midRd;
        open(STDOUT, '>&', $midWr) or POSIX::_exit(126);
        close $midWr;
        { exec { $argv1->[0] } @$argv1; }
        POSIX::_exit(127);
    }

    my $pid2 = fork();
    croak "Harness::Safe: fork: $!" unless defined $pid2;
    if ($pid2 == 0) {
        close $midWr;
        open(STDIN, '<&', $midRd) or POSIX::_exit(126);
        open(STDOUT, '>&', $outFh) or POSIX::_exit(126);
        close $midRd;
        { exec { $argv2->[0] } @$argv2; }
        POSIX::_exit(127);
    }

    close $midRd;
    close $midWr;
    close $outFh;
    waitpid($pid1, 0);
    my $status1 = $?;
    waitpid($pid2, 0);
    my $status2 = $?;
    return ($status1, $status2);
}

# ---------------------------------------------------------------------------
# The dispatcher every primitive goes through. Handles mode gating (explain/dry_run/
# confirm_destructive/run), the class requirement, and audit logging uniformly.
# ---------------------------------------------------------------------------

sub _run_primitive {
    my (%p) = @_;
    # class is mandatory - a primitive call with no class refuses outright.
    _refuse("primitive '$p{primitive}' called with no class") unless defined $p{class};
    _refuse("primitive '$p{primitive}' has unknown class '$p{class}'")
        unless $p{class} =~ /^(?:readonly|mutating|destructive)$/;

    my $wouldRunDesc = $p{describe} ? $p{describe}->() : $p{primitive};

    if ($MODE eq 'explain') {
        _audit(class => $p{class}, primitive => $p{primitive}, args => $p{args} // {},
               verdict => 'EXPLAIN', would_run => $wouldRunDesc);
        return $p{explain_result} // ($p{class} eq 'readonly' ? [] : 1);
    }

    if ($MODE eq 'dry_run' && $p{class} ne 'readonly') {
        _audit(class => $p{class}, primitive => $p{primitive}, args => $p{args} // {},
               verdict => 'SIMULATED', would_run => $wouldRunDesc);
        return $p{explain_result} // 1;
    }

    if ($MODE eq 'confirm_destructive' && $p{class} eq 'destructive') {
        print STDERR "\n>>> ABOUT TO RUN (destructive): $wouldRunDesc\n>>> proceed? [y/N] ";
        my $answer = <STDIN> // '';
        chomp $answer;
        unless ($answer =~ /^y(es)?$/i) {
            _audit(class => $p{class}, primitive => $p{primitive}, args => $p{args} // {},
                   verdict => 'DECLINED_BY_OPERATOR', would_run => $wouldRunDesc);
            die "Harness::Safe: operator declined: $wouldRunDesc\n";
        }
    }

    my $result = $p{run}->();
    _audit(class => $p{class}, primitive => $p{primitive}, args => $p{args} // {},
           verdict => 'ALLOWED', ran => $wouldRunDesc,
           exit => $result->{exit}, output => $result->{output});
    return $result;
}

# ---------------------------------------------------------------------------
# ZFS primitives
# ---------------------------------------------------------------------------

sub destroy_dataset {
    my (%a) = @_;
    my $d = ds($a{dataset});
    _check_allow($d, \@DESTROY_ALLOW, 'destroy');
    _refuse("recursive destroy of a snapshot is not meaningful: '$d'") if $a{recursive} && $d =~ /\@/;
    my @argv = ($BIN{zfs}, 'destroy', ($a{recursive} ? ('-r') : ()), $d);
    return _run_primitive(
        primitive => 'destroy_dataset', class => 'destructive', args => { dataset => $d },
        describe  => sub { join(' ', @argv) },
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            if ($status != 0 && !$a{missing_ok}) {
                _refuse("destroy_dataset('$d') failed: $out");
            }
            return { exit => $status >> 8, output => $out };
        },
    );
}

sub create_dataset {
    my (%a) = @_;
    my $d = ds($a{dataset});
    _check_allow($d, \@CREATE_ALLOW, 'create');
    my @argv = ($BIN{zfs}, 'create', '-p', $d);
    return _run_primitive(
        primitive => 'create_dataset', class => 'mutating', args => { dataset => $d },
        describe  => sub { join(' ', @argv) },
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            _refuse("create_dataset('$d') failed: $out") if $status != 0;
            return { exit => 0, output => $out };
        },
    );
}

sub snapshot_dataset {
    my (%a) = @_;
    my $d = ds($a{dataset});
    _check_allow($d, \@SNAPSHOT_ALLOW, 'snapshot');
    my @argv = ($BIN{zfs}, 'snapshot', $d);
    return _run_primitive(
        primitive => 'snapshot_dataset', class => 'mutating', args => { dataset => $d },
        describe  => sub { join(' ', @argv) },
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            _refuse("snapshot_dataset('$d') failed: $out") if $status != 0;
            return { exit => 0, output => $out };
        },
    );
}

sub dataset_exists {
    my (%a) = @_;
    my $d = ds($a{dataset});
    my @argv = ($BIN{zfs}, 'list', '-H', '-o', 'name', $d);
    my $result = _run_primitive(
        primitive => 'dataset_exists', class => 'readonly', args => { dataset => $d },
        describe  => sub { join(' ', @argv) },
        explain_result => 0,
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            return { exit => 0, output => ($status == 0 ? 1 : 0) };
        },
    );
    return ref($result) eq 'HASH' ? $result->{output} : $result;
}

sub zfs_list_snapshots {
    my (%a) = @_;
    my $d = ds($a{dataset});
    my @argv = ($BIN{zfs}, 'list', '-H', '-o', 'name', '-t', 'snap', '-r', $d);
    my $result = _run_primitive(
        primitive => 'zfs_list_snapshots', class => 'readonly', args => { dataset => $d },
        describe  => sub { join(' ', @argv) },
        explain_result => [],
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            my @lines = grep { length } split /\n/, ($out // '');
            return { exit => 0, output => \@lines };
        },
    );
    my $listRef = ref($result) eq 'HASH' ? $result->{output} : $result;
    return @{ $listRef // [] };
}

sub zfs_list_all_snapshots_under {
    my (%a) = @_;
    my $d = ds($a{dataset});
    my @argv = ($BIN{zfs}, 'list', '-H', '-o', 'name', '-t', 'all', '-r', $d);
    my $result = _run_primitive(
        primitive => 'zfs_list_all_snapshots_under', class => 'readonly', args => { dataset => $d },
        describe  => sub { join(' ', @argv) },
        explain_result => [],
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            my @lines = grep { length } split /\n/, ($out // '');
            return { exit => 0, output => \@lines };
        },
    );
    my $listRef = ref($result) eq 'HASH' ? $result->{output} : $result;
    return @{ $listRef // [] };
}

# Read-only inventory used by the mount-topology preflight and the production tripwire -
# not gated by dataset() being under the sandbox, since it deliberately inspects OUTSIDE it
# (storage/backup) and the pool root. This is the one place that is intentional and reviewed;
# it never writes anything.
sub zfs_list_raw {
    my (%a) = @_;
    my @argv = ($BIN{zfs}, 'list', '-H', @{ $a{opts} // [] }, $a{target});
    my $result = _run_primitive(
        primitive => 'zfs_list_raw', class => 'readonly', args => { target => $a{target} },
        describe  => sub { join(' ', @argv) },
        explain_result => [],
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            my @lines = grep { length } split /\n/, ($out // '');
            return { exit => 0, output => \@lines };
        },
    );
    my $listRef = ref($result) eq 'HASH' ? $result->{output} : $result;
    return @{ $listRef // [] };
}

# ---------------------------------------------------------------------------
# Filesystem primitives - no globs, ever.
# ---------------------------------------------------------------------------

sub _has_child_mount_beneath {
    my ($dir) = @_;
    my @mounts = zfs_list_raw(target => $SANDBOX_DS, opts => ['-o', 'name,mountpoint', '-r']);
    for my $line (@mounts) {
        my ($name, $mp) = split /\t/, $line;
        next unless defined $mp && length $mp;
        next if $mp eq $dir;
        return 1 if index($mp, "$dir/") == 0;
    }
    return 0;
}

sub empty_dir {
    my (%a) = @_;
    my $dir = path($a{path});
    return _run_primitive(
        primitive => 'empty_dir', class => 'destructive', args => { path => $dir },
        describe  => sub { "empty_dir($dir)" },
        run       => sub {
            _refuse("empty_dir: '$dir' does not exist") unless -e $dir;
            _refuse("empty_dir: '$dir' is a symlink") if -l $dir;
            _refuse("empty_dir: '$dir' is not a directory") unless -d $dir;
            _refuse("empty_dir: a child dataset is mounted beneath '$dir'")
                if _has_child_mount_beneath($dir);
            opendir(my $dh, $dir) or _refuse("empty_dir: cannot opendir '$dir': $!");
            my @entries = grep { $_ ne '.' && $_ ne '..' } readdir($dh);
            closedir $dh;
            for my $e (@entries) {
                my $full = "$dir/$e";
                if (-l $full) { unlink($full) or _refuse("empty_dir: unlink symlink '$full': $!"); next; }
                if (-d $full) { File::Path::remove_tree($full, { safe => 1 }) or _refuse("empty_dir: remove_tree '$full' failed"); next; }
                unlink($full) or _refuse("empty_dir: unlink '$full': $!");
            }
            return { exit => 0, output => scalar(@entries) . ' entries removed' };
        },
    );
}

## Step 8.3's before/after transport-size comparison (the compression shrinkage check). Sums
## regular-file sizes directly inside a directory - non-recursive, since transport/datasets
## never has subdirectories.
sub dir_size_bytes {
    my (%a) = @_;
    my $dir = path($a{path});
    my $result = _run_primitive(
        primitive => 'dir_size_bytes', class => 'readonly', args => { path => $dir },
        describe  => sub { "dir_size_bytes($dir)" },
        explain_result => 0,
        run       => sub {
            return { exit => 0, output => 0 } unless -d $dir;
            opendir(my $dh, $dir) or _refuse("dir_size_bytes: cannot opendir '$dir': $!");
            my @entries = grep { $_ ne '.' && $_ ne '..' } readdir($dh);
            closedir $dh;
            my $total = 0;
            for my $e (@entries) {
                my $full = "$dir/$e";
                $total += -s $full if -f $full && !-l $full;
            }
            return { exit => 0, output => $total };
        },
    );
    return ref($result) eq 'HASH' ? $result->{output} : $result;
}

sub rm_matching {
    my (%a) = @_;
    my $dir = path($a{dir});
    my $reSrc = $a{basename_re};
    _refuse("rm_matching: pattern may not contain '/': $reSrc") if $reSrc =~ m{/};
    my $re = qr/$reSrc/;
    return _run_primitive(
        primitive => 'rm_matching', class => 'destructive', args => { dir => $dir, pattern => $reSrc },
        describe  => sub { "rm_matching($dir, $reSrc)" },
        run       => sub {
            return { exit => 0, output => '0 entries (dir absent)' } unless -d $dir;
            opendir(my $dh, $dir) or _refuse("rm_matching: cannot opendir '$dir': $!");
            my @hits = grep { $_ ne '.' && $_ ne '..' && $_ =~ $re } readdir($dh);
            closedir $dh;
            for my $f (@hits) {
                my $full = "$dir/$f";
                next if -d $full; # never descend - basename-matched files only
                unlink($full) or _refuse("rm_matching: unlink '$full': $!");
            }
            return { exit => 0, output => scalar(@hits) . ' entries removed' };
        },
    );
}

sub list_matching {
    my (%a) = @_;
    my $dir = path($a{dir});
    my $reSrc = $a{basename_re};
    _refuse("list_matching: pattern may not contain '/': $reSrc") if $reSrc =~ m{/};
    my $re = qr/$reSrc/;
    my $result = _run_primitive(
        primitive => 'list_matching', class => 'readonly', args => { dir => $dir, pattern => $reSrc },
        describe  => sub { "list_matching($dir, $reSrc)" },
        explain_result => [],
        run       => sub {
            return { exit => 0, output => [] } unless -d $dir;
            opendir(my $dh, $dir) or _refuse("list_matching: cannot opendir '$dir': $!");
            my @hits = sort grep { $_ ne '.' && $_ ne '..' && $_ =~ $re } readdir($dh);
            closedir $dh;
            return { exit => 0, output => \@hits };
        },
    );
    my $listRef = ref($result) eq 'HASH' ? $result->{output} : $result;
    return @{ $listRef // [] };
}

sub write_file {
    my (%a) = @_;
    my $p = path($a{path});
    return _run_primitive(
        primitive => 'write_file', class => 'mutating', args => { path => $p },
        describe  => sub { "write_file($p, " . length($a{content} // '') . " bytes)" },
        run       => sub {
            _refuse("write_file: refusing to write through a symlink: '$p'") if -l $p;
            my $tmp = "$p.harness.$$";
            sysopen(my $fh, $tmp, O_WRONLY | O_CREAT | O_EXCL, $a{mode} // 0644)
                or _refuse("write_file: cannot create '$tmp': $!");
            print {$fh} ($a{content} // '') or _refuse("write_file: write to '$tmp' failed: $!");
            close $fh;
            rename($tmp, $p) or do { unlink $tmp; _refuse("write_file: rename '$tmp' -> '$p' failed: $!"); };
            return { exit => 0, output => "wrote $p" };
        },
    );
}

# Part 14/15 need to stage a copy of a real cleanup script (helloWorld, updateConfigKeys) into
# oneshot/ - plain read+write rather than pulling in File::Copy, matching write_file's own style.
sub copy_file {
    my (%a) = @_; # from, to
    my $from = path($a{from});
    my $to = path($a{to});
    return _run_primitive(
        primitive => 'copy_file', class => 'mutating', args => { from => $from, to => $to },
        describe  => sub { "copy_file($from -> $to)" },
        run       => sub {
            _refuse("copy_file: source '$from' does not exist") unless -f $from;
            _refuse("copy_file: source '$from' is a symlink") if -l $from;
            _refuse("copy_file: refusing to write through a symlink: '$to'") if -l $to;
            open my $ifh, '<', $from or _refuse("copy_file: cannot open '$from': $!");
            local $/;
            my $content = <$ifh>;
            close $ifh;
            my $tmp = "$to.harness.$$";
            my $mode = (stat($from))[2] // 0644;
            sysopen(my $ofh, $tmp, O_WRONLY | O_CREAT | O_EXCL, $mode & 07777)
                or _refuse("copy_file: cannot create '$tmp': $!");
            print {$ofh} $content or _refuse("copy_file: write to '$tmp' failed: $!");
            close $ofh;
            rename($tmp, $to) or do { unlink $tmp; _refuse("copy_file: rename '$tmp' -> '$to' failed: $!"); };
            return { exit => 0, output => "copied $from -> $to" };
        },
    );
}

# Part 15 needs to customize a copied one-shot script's hardcoded 'my @updates = (...)' array
# (TESTING.md Step 15.1) - a narrowly-scoped structural edit rather than a general find/replace
# primitive, matching this harness's preference for purpose-built operations over broad power.
sub patch_oneshot_updates {
    my (%a) = @_; # path, entries (ARRAYREF of plain 'key=value' strings)
    my $p = path($a{path});
    my @entries = @{ $a{entries} // [] };
    _refuse("patch_oneshot_updates: no entries given") unless @entries;
    for my $e (@entries) {
        _refuse("patch_oneshot_updates: entry contains a single quote or newline: '$e'") if $e =~ /['\n]/;
    }
    return _run_primitive(
        primitive => 'patch_oneshot_updates', class => 'mutating', args => { path => $p, entries => \@entries },
        describe  => sub { "patch_oneshot_updates($p, " . scalar(@entries) . " entries)" },
        run       => sub {
            _refuse("patch_oneshot_updates: '$p' does not exist") unless -f $p;
            open my $fh, '<', $p or _refuse("patch_oneshot_updates: cannot open '$p': $!");
            local $/;
            my $text = <$fh>;
            close $fh;
            my $block = "my \@updates = (\n" . join('', map { "    '$_',\n" } @entries) . ");";
            my $count = ( $text =~ s/my\s+\@updates\s*=\s*\([^)]*\)\s*;/$block/s );
            _refuse("patch_oneshot_updates: 'my \@updates = (...)' block not found in '$p'") unless $count;
            my $tmp = "$p.harness.$$";
            sysopen(my $ofh, $tmp, O_WRONLY | O_CREAT | O_EXCL, 0755)
                or _refuse("patch_oneshot_updates: cannot create '$tmp': $!");
            print {$ofh} $text or _refuse("patch_oneshot_updates: write to '$tmp' failed: $!");
            close $ofh;
            rename($tmp, $p) or do { unlink $tmp; _refuse("patch_oneshot_updates: rename '$tmp' -> '$p' failed: $!"); };
            return { exit => 0, output => "patched \@updates in $p" };
        },
    );
}

# Step 8.1/8.3 need a large, highly-compressible file (TESTING.md: 50,000 repeated lines) to
# demonstrate xz shrinkage - built here rather than as a literal multi-megabyte string in the
# JSON catalogue.
sub write_repeated_line {
    my (%a) = @_; # path, line, count
    my $p = path($a{path});
    my $count = $a{count};
    _refuse("write_repeated_line: count must be a positive integer") unless $count =~ /^[1-9]\d*$/;
    _refuse("write_repeated_line: count $count exceeds 1,000,000 cap") if $count > 1_000_000;
    return _run_primitive(
        primitive => 'write_repeated_line', class => 'mutating', args => { path => $p, count => $count },
        describe  => sub { "write_repeated_line($p, " . length($a{line} // '') . "-byte line x $count)" },
        run       => sub {
            _refuse("write_repeated_line: refusing to write through a symlink: '$p'") if -l $p;
            my $tmp = "$p.harness.$$";
            sysopen(my $fh, $tmp, O_WRONLY | O_CREAT | O_EXCL, 0644)
                or _refuse("write_repeated_line: cannot create '$tmp': $!");
            print {$fh} (($a{line} // '') . "\n") x $count
                or _refuse("write_repeated_line: write to '$tmp' failed: $!");
            close $fh;
            rename($tmp, $p) or do { unlink $tmp; _refuse("write_repeated_line: rename '$tmp' -> '$p' failed: $!"); };
            return { exit => 0, output => "wrote $p ($count lines)" };
        },
    );
}

## Read the sole file in a directory matching a basename regex. Used where a filename is
## discovered at runtime (e.g. Step 15.2's timestamped sneakernet.conf.yaml.bak.<timestamp>) -
## avoids ever needing to interpolate a captured runtime value into a path() string, which under
## --explain (where captures from readonly primitives are empty placeholders) can produce a
## degenerate path shape that path() must legitimately refuse.
sub read_file_matching {
    my (%a) = @_; # dir, basename_re
    my $dir = path($a{dir});
    my $reSrc = $a{basename_re};
    _refuse("read_file_matching: pattern may not contain '/': $reSrc") if $reSrc =~ m{/};
    my $re = qr/$reSrc/;
    my $result = _run_primitive(
        primitive => 'read_file_matching', class => 'readonly', args => { dir => $dir, pattern => $reSrc },
        describe  => sub { "read_file_matching($dir, $reSrc)" },
        explain_result => '',
        run       => sub {
            _refuse("read_file_matching: '$dir' does not exist") unless -d $dir;
            opendir(my $dh, $dir) or _refuse("read_file_matching: cannot opendir '$dir': $!");
            my @hits = sort grep { $_ ne '.' && $_ ne '..' && $_ =~ $re } readdir($dh);
            closedir $dh;
            _refuse("read_file_matching: expected exactly 1 match in '$dir' for /$reSrc/, got " . scalar(@hits))
                unless @hits == 1;
            my $full = "$dir/$hits[0]";
            open my $fh, '<', $full or _refuse("read_file_matching: cannot open '$full': $!");
            local $/;
            my $text = <$fh>;
            close $fh;
            return { exit => 0, output => $text // '' };
        },
    );
    return ref($result) eq 'HASH' ? $result->{output} : $result;
}

sub read_file {
    my (%a) = @_;
    my $p = path($a{path});
    my $result = _run_primitive(
        primitive => 'read_file', class => 'readonly', args => { path => $p },
        describe  => sub { "read_file($p)" },
        explain_result => '',
        run       => sub {
            return { exit => 0, output => '' } unless -e $p;
            open my $fh, '<', $p or _refuse("read_file: cannot open '$p': $!");
            local $/;
            my $text = <$fh>;
            close $fh;
            return { exit => 0, output => $text // '' };
        },
    );
    return ref($result) eq 'HASH' ? $result->{output} : $result;
}

sub corrupt_file {
    my (%a) = @_;
    my $p = path($a{path});
    my ($offset, $length) = ($a{offset}, $a{length});
    _refuse("corrupt_file: length $length exceeds 1MB cap") if $length > 1_048_576;
    return _run_primitive(
        primitive => 'corrupt_file', class => 'destructive', args => { path => $p, offset => $offset, length => $length },
        describe  => sub { "corrupt_file($p, offset=$offset, length=$length)" },
        run       => sub {
            _refuse("corrupt_file: refusing a symlink: '$p'") if -l $p;
            _refuse("corrupt_file: '$p' does not exist") unless -f $p;
            my $size = -s $p;
            _refuse("corrupt_file: write would extend the file (size=$size, want $offset+$length)")
                if $offset + $length > $size;
            sysopen(my $fh, $p, O_WRONLY) or _refuse("corrupt_file: open '$p': $!");
            sysseek($fh, $offset, SEEK_SET) or _refuse("corrupt_file: seek: $!");
            my $garbage = random_bytes(bytes => $length);
            my $n = syswrite($fh, $garbage);
            close $fh;
            _refuse("corrupt_file: short write ($n of $length bytes)") unless defined($n) && $n == $length;
            return { exit => 0, output => "corrupted $length bytes at offset $offset" };
        },
    );
}

# ---------------------------------------------------------------------------
# Randomness - read directly from /dev/urandom. No exec needed at all for this, which is
# simpler and safer than shelling out to 'openssl rand' or 'head -c N /dev/urandom'.
# ---------------------------------------------------------------------------

sub random_bytes {
    my (%a) = @_;
    my $n = $a{bytes};
    open my $fh, '<:raw', '/dev/urandom' or croak "Harness::Safe: cannot open /dev/urandom: $!";
    my $buf = '';
    my $got = read($fh, $buf, $n);
    close $fh;
    croak "Harness::Safe: short read from /dev/urandom ($got of $n)" unless defined($got) && $got == $n;
    return $buf;
}

sub random_hex {
    my (%a) = @_;
    return unpack('H*', random_bytes(bytes => $a{bytes}));
}

# A real embedded-date timestamp suffix, matching TESTING.md's own literal
# `$(date +%Y-%m-%d_%H.%M.%S)` snapshot-naming convention - required by
# ZFS_Utils::parseSnapshotDateTime/findCommonBaseSnapshot, which only ever consider a
# snapshot as a base-selection candidate if its name embeds a YYYY-MM-DD date (see Parts
# 9/11/12). Second resolution only, matching parseSnapshotDateTime's own precision - callers
# that need two distinct suffixes must `sleep` at least 1 second between calls.
sub now_suffix {
    return strftime('%Y-%m-%d_%H.%M.%S', localtime());
}

# ---------------------------------------------------------------------------
# openssl / xz - single-process exec, no pipeline needed except where noted.
# ---------------------------------------------------------------------------

sub openssl_encrypt_to_file {
    my (%a) = @_; # key, iv, plaintext, out_path
    my $out = path($a{out_path});
    local @REDACT = (@REDACT, $a{key});
    my @argv = ($BIN{openssl}, 'enc', '-aes-256-cbc', '-K', $a{key}, '-iv', $a{iv});
    return _run_primitive(
        primitive => 'openssl_encrypt_to_file', class => 'mutating',
        args      => { out_path => $out, iv => $a{iv} },
        describe  => sub { join(' ', @argv) . " > $out" },
        run       => sub {
            pipe(my $rd, my $wr) or _refuse("pipe: $!");
            sysopen(my $outFh, $out, O_WRONLY | O_CREAT | O_EXCL, 0644)
                or _refuse("openssl_encrypt_to_file: cannot create '$out': $!");
            my $pid = fork();
            _refuse("fork: $!") unless defined $pid;
            if ($pid == 0) {
                close $wr;
                open(STDIN, '<&', $rd) or POSIX::_exit(126);
                open(STDOUT, '>&', $outFh) or POSIX::_exit(126);
                close $rd;
                { exec { $argv[0] } @argv; }
                POSIX::_exit(127);
            }
            close $rd; close $outFh;
            print {$wr} $a{plaintext};
            close $wr;
            waitpid($pid, 0);
            my $status = $?;
            _refuse("openssl_encrypt_to_file failed, exit $status") if $status != 0;
            return { exit => 0, output => "wrote $out" };
        },
    );
}

# The one genuine multi-process pipeline in the whole catalogue (Step 8.5): pipe a real ZFS
# send stream directly into openssl, with no intermediate buffering in this process.
sub pipeline_zfs_send_to_encrypted_file {
    my (%a) = @_; # snapshot (dataset@snap), key, iv, out_path
    my $snap = ds($a{snapshot});
    my $out  = path($a{out_path});
    local @REDACT = (@REDACT, $a{key});
    my @argv1 = ($BIN{zfs}, 'send', $snap);
    my @argv2 = ($BIN{openssl}, 'enc', '-aes-256-cbc', '-K', $a{key}, '-iv', $a{iv});
    return _run_primitive(
        primitive => 'pipeline_zfs_send_to_encrypted_file', class => 'mutating',
        args      => { snapshot => $snap, out_path => $out, iv => $a{iv} },
        describe  => sub { join(' ', @argv1) . ' | ' . join(' ', @argv2) . " > $out" },
        run       => sub {
            my ($s1, $s2) = _exec_pipeline(\@argv1, \@argv2, $out);
            _refuse("pipeline: zfs send exited $s1") if $s1 != 0;
            _refuse("pipeline: openssl exited $s2") if $s2 != 0;
            return { exit => 0, output => "wrote $out" };
        },
    );
}

# ---------------------------------------------------------------------------
# sneakernet / svn / perl -c / utilities - the higher-level things the catalogue drives.
# ---------------------------------------------------------------------------

# Re-validates the config immediately before every invocation - see TESTING_automation.md 6.5.
# $containmentCheck is a coderef supplied by the runner (Harness::Runner registers the actual
# check; kept out of this module so Harness::Safe has no YAMLPatch dependency of its own).
our $CONFIG_CONTAINMENT_CHECK;

sub run_sneakernet {
    my (%a) = @_; # role ('source'|'target'), servername, verbosity, sn_path, log_path, expect_exit
    _refuse('run_sneakernet: no containment check registered')
        unless ref($CONFIG_CONTAINMENT_CHECK) eq 'CODE';
    my $snPath = path($a{sn_path});
    my $logPath = defined $a{log_path} ? path($a{log_path}) : undef;
    my @argv = ($BIN{perl}, $snPath, '-s', $a{servername}, '-v', ($a{verbosity} // 1));
    return _run_primitive(
        primitive => 'run_sneakernet', class => 'mutating',
        args      => { servername => $a{servername}, verbosity => $a{verbosity} },
        describe  => sub { join(' ', @argv) },
        run       => sub {
            $CONFIG_CONTAINMENT_CHECK->();  # dies on failure - re-checked every invocation
            my ($out, $status) = _exec_argv(\@argv);
            my $exit = $status >> 8;
            if (defined $logPath) {
                sysopen(my $fh, $logPath, O_WRONLY | O_CREAT, 0644) or _refuse("cannot open log '$logPath': $!");
                seek($fh, 0, 2);
                print {$fh} $out;
                close $fh;
            }
            if (defined $a{expect_exit} && $a{expect_exit} ne 'any' && $exit != $a{expect_exit}) {
                _refuse("run_sneakernet($a{servername}): expected exit $a{expect_exit}, got $exit. Output:\n$out");
            }
            return { exit => $exit, output => $out };
        },
    );
}

# For Part 1's --version/--help only - these never touch config or ZFS, so the containment
# gate doesn't apply, but the flag itself is allow-listed rather than accepting arbitrary
# argv (this is not a general "run sneakernet with any flags" bypass of run_sneakernet above).
our @SNEAKERNET_INFO_FLAGS = ('--version', '--help');

sub run_sneakernet_info {
    my (%a) = @_; # sn_path, flag
    _refuse("run_sneakernet_info: '$a{flag}' is not an allow-listed info flag")
        unless grep { $_ eq $a{flag} } @SNEAKERNET_INFO_FLAGS;
    my $snPath = path($a{sn_path});
    my @argv = ($BIN{perl}, $snPath, $a{flag});
    return _run_primitive(
        primitive => 'run_sneakernet_info', class => 'readonly', args => { flag => $a{flag} },
        describe  => sub { join(' ', @argv) },
        explain_result => { exit => 0, output => '' },
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            return { exit => $status >> 8, output => $out };
        },
    );
}

# Restricted to the two existing repo utilities the catalogue actually needs to invoke
# (Part 16's checker, Part 17's builder). This is an allow-list, not a general "run any
# script under {CODE}/utilities" primitive - the generated OUTPUT of buildUpgradeOneShot.pl
# (a script designed to overwrite a real sneakernet install) is never on this list and has no
# path to execution through this module at all; see TESTING_automation.md constraint 4 rule 7.
our @UTILITY_ALLOW = qw(checkSneakernetFile buildUpgradeOneShot.pl);

sub run_utility {
    my (%a) = @_;
    my $script = $a{script};
    _refuse("run_utility: '$script' is not on the utility allow-list")
        unless grep { $_ eq $script } @UTILITY_ALLOW;
    my $utilPath = path($a{code_dir} . "/utilities/$script");
    my @argv = ($BIN{perl}, $utilPath, @{ $a{args} // [] });
    return _run_primitive(
        primitive => 'run_utility', class => 'mutating', args => { script => $script, args => join(' ', @{ $a{args} // [] }) },
        describe  => sub { join(' ', @argv) },
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            return { exit => $status >> 8, output => $out };
        },
    );
}

sub run_perl_c {
    my (%a) = @_;
    my $target = path($a{path});
    my @argv = ($BIN{perl}, '-c', $target);
    return _run_primitive(
        primitive => 'run_perl_c', class => 'readonly', args => { path => $target },
        describe  => sub { join(' ', @argv) },
        explain_result => { exit => 0, output => '' },
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            return { exit => $status >> 8, output => $out };
        },
    );
}

sub svn_status {
    my (%a) = @_;
    my $dir = path($a{dir});
    my @argv = ($BIN{svn}, 'status', $dir);
    my $result = _run_primitive(
        primitive => 'svn_status', class => 'readonly', args => { dir => $dir },
        describe  => sub { join(' ', @argv) },
        explain_result => [],
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            my @lines = grep { length } split /\n/, ($out // '');
            return { exit => 0, output => \@lines };
        },
    );
    my $listRef = ref($result) eq 'HASH' ? $result->{output} : $result;
    return @{ $listRef // [] };
}

sub svn_update {
    my (%a) = @_;
    my $dir = path($a{dir});
    my @argv = ($BIN{svn}, 'update', $dir);
    return _run_primitive(
        primitive => 'svn_update', class => 'mutating', args => { dir => $dir },
        describe  => sub { join(' ', @argv) },
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            _refuse("svn_update('$dir') failed: $out") if $status != 0;
            return { exit => 0, output => $out };
        },
    );
}

sub svnversion {
    my (%a) = @_;
    my $dir = path($a{dir});
    my @argv = ($BIN{svnversion}, $dir);
    my $result = _run_primitive(
        primitive => 'svnversion', class => 'readonly', args => { dir => $dir },
        describe  => sub { join(' ', @argv) },
        explain_result => '',
        run       => sub {
            my ($out, $status) = _exec_argv(\@argv);
            chomp(my $rev = $out // '');
            return { exit => 0, output => $rev };
        },
    );
    return ref($result) eq 'HASH' ? $result->{output} : $result;
}

# ---------------------------------------------------------------------------
# Mount-topology preflight - TESTING_automation.md 6.4. The guard's genuine blind spot: a
# lexically valid path() could still resolve into production if something is mounted where it
# shouldn't be. Run at startup and before every Part.
# ---------------------------------------------------------------------------

sub assert_mount_topology {
    for my $line (zfs_list_raw(target => $SANDBOX_DS, opts => ['-o', 'name,mountpoint,canmount', '-r'])) {
        my ($name, $mp, $canmount) = split /\t/, $line;
        next unless defined $name;
        _refuse("sandbox dataset '$name' has unexpected mountpoint '$mp'")
            unless $mp eq "/$name" || $mp eq 'none' || $mp eq '-';
    }
    for my $line (zfs_list_raw(target => 'storage', opts => ['-o', 'name,mountpoint', '-r'])) {
        my ($name, $mp) = split /\t/, $line;
        next unless defined $name;
        next if $name eq $SANDBOX_DS || index($name, "$SANDBOX_DS/") == 0;
        next unless defined $mp && length $mp;
        _refuse("foreign dataset '$name' is mounted at '$mp', inside the sandbox")
            if $mp eq $SANDBOX_PATH || index($mp, "$SANDBOX_PATH/") == 0;
    }
    return 1;
}

# ---------------------------------------------------------------------------
# Production tripwire - TESTING_automation.md 6.6.
# ---------------------------------------------------------------------------

sub production_fingerprint {
    my @names = zfs_list_raw(target => 'storage/backup', opts => ['-o', 'name', '-t', 'all', '-r']);
    return {
        sha   => sha256_hex(join("\n", @names)),
        count => scalar(@names),
    };
}

sub assert_production_unchanged {
    my ($baseline) = @_;
    my $now = production_fingerprint();
    if ($now->{sha} ne $baseline->{sha}) {
        _audit(class => 'n/a', primitive => 'production_tripwire', verdict => 'TRIPWIRE_FIRED',
               reason => "storage/backup fingerprint changed: $baseline->{count} -> $now->{count} entries");
        die "Harness::Safe: PRODUCTION TRIPWIRE FIRED - storage/backup changed during this run "
          . "($baseline->{count} -> $now->{count} snapshot/dataset entries). Stopping immediately. "
          . "Do not continue without a human looking at storage/backup first.\n";
    }
    return 1;
}

# ---------------------------------------------------------------------------
# Cron guard - constraint 5. FreeBSD localtime; the window is specified in wall-clock local
# time to match how /usr/local/etc/cron.d/replicate is scheduled.
# ---------------------------------------------------------------------------

sub in_cron_guard_window {
    my ($epoch) = @_;
    $epoch //= time();
    my @t = localtime($epoch);
    my $minutesSinceMidnight = $t[2] * 60 + $t[1];
    return $minutesSinceMidnight >= (3 * 60 + 45) && $minutesSinceMidnight <= (5 * 60);
}

sub assert_outside_cron_guard_window {
    _refuse('current time is inside the 03:45-05:00 production replicate cron guard window - refusing to start')
        if in_cron_guard_window();
    return 1;
}

1;

__END__
See sneakernet/Documentation/TESTING_automation.md sections 6 and 9 for the full design this
module implements. This module has no public documentation file of its own - it is internal
to the harness, not a reusable library like YAMLPatch.pm (see TESTING_automation.md 5, table
row for Harness::Safe: "Test-suite specific - deliberately so").