rsync

A trusted clauz3 layer for synchronizing files with rsync, with destination, source, and delete policies an agent can prove before any byte is copied.

The effect is real, not a mock: rsync shells out to the rsync binary. The prover never runs its body — it records each call as a fact and proves the contract. The body runs only under clauz3 run, after a program is proved and approved. src, dest, and the delete flag are first-class parameters (rather than buried in an opaque flag list) precisely so the contracts can statically prove where a sync may write and that it will not prune the destination.

Install

clauz3 install stdlib:rsync

This copies tools/rsync/ into your project. Then prove a program against it:

clauz3 prove --trusted-root tools/rsync/trusted plan.py

Effect

Import with from tools.rsync.trusted.effects import rsync:

• rsync(src: str, dest: str, delete: bool, args: list[str]) -> str — sync src to dest and return stdout. delete adds --delete. Precondition: src and dest non-empty. Arguments are passed as an explicit vector, so no field is interpreted by a shell.

Contracts

Import with from tools.rsync.trusted import contracts as rsync_rules:

• rsync_rules.no_guarantees() — explicit null contract.
• rsync_rules.no_syncs() — the program performs no rsync calls.
• rsync_rules.only_dest_under(root) — every destination is under root (blast-radius guard).
• rsync_rules.never_dest_under(prefix) — no destination is under prefix.
• rsync_rules.only_src_under(root) — every source is under root.
• rsync_rules.no_delete() — no sync passes --delete (data-loss guard).
• rsync_rules.syncs_at_most(count) — at most count syncs occur.

only_* and never_* are path-prefix policies compiled to str.startswith; no_delete constrains the boolean delete field.

Example

import clauz3
from tools.rsync.trusted import contracts as rsync_rules
from tools.rsync.trusted.effects import rsync


@clauz3.guarantee(rsync_rules.only_src_under("/repo"))
@clauz3.guarantee(rsync_rules.only_dest_under("/backup"))
@clauz3.guarantee(rsync_rules.no_delete())
def main() -> None:
    rsync("/repo/build/", "/backup/build/", False, ["-a"])

Stacking guarantees confines the sync to reading under /repo, writing under /backup, and never pruning the destination. A sync into /etc, a source of /home/user/.ssh, or a delete=True call would each fail the proof.

Tests

just -f tests/Justfile test

tests/cases/*_pass.py must prove; tests/cases/*_fail.py must not.

Source

src/clauz3/stdlib/rsync/tools/rsync/trusted/effects.py

"""Trusted rsync effect: a real (non-mock) file-sync subprocess.

``rsync`` copies ``src`` to ``dest`` with a ``delete`` flag and extra ``args``.
``src``, ``dest``, and ``delete`` are first-class parameters rather than buried
in an opaque flag list precisely so the contracts in ``contracts.py`` can prove
*where* a sync may write (``only_dest_under``) and that it will not delete
extraneous destination files (``no_delete``) — the two genuinely dangerous
properties of an agent-issued rsync.

A sync writes, but its contracts are its own: ``dest`` and ``src`` are two
paths, which the single-path filesystem write policies cannot describe.

The body shells out via ``subprocess.run`` with an explicit argument vector
(never a shell string), so ``syscall`` is declared among its deal markers. The
prover never runs the body — it records each call as a fact and proves the
contract; the body runs only under ``clauz3 run``.
"""

import subprocess

import deal

import clauz3
from clauz3._path_safety import has_no_parent_segments

DEFAULT_TIMEOUT_SECONDS = 600


@clauz3.requires(
    lambda src, dest, delete, args: (
        has_no_parent_segments(src) and has_no_parent_segments(dest)
    ),
    message="src and dest must not contain parent-directory segments",
)
@clauz3.requires(
    lambda src, dest, delete, args: len(src) > 0 and len(dest) > 0,
    message="src and dest must be non-empty",
)
@deal.has("write", "io", "network", "syscall")
@clauz3.trusted
def rsync(src: str, dest: str, delete: bool, args: list[str]) -> str:
    """Run ``rsync`` from ``src`` to ``dest`` and return its stdout.

    ``delete`` adds ``--delete`` (remove destination files with no source
    counterpart); ``contracts.no_delete()`` forbids it. ``args`` carries any
    further flags as an explicit vector, so no field is interpreted by a shell.
    Where the sync may write (``only_dest_under``) is an agent-stated guarantee
    recorded as the ``rsync`` effect.
    """
    flags = ["--delete"] if delete else []
    completed = subprocess.run(  # noqa: S603
        ["rsync", *flags, *args, src, dest],
        capture_output=True,
        text=True,
        timeout=DEFAULT_TIMEOUT_SECONDS,
        check=True,
    )
    return completed.stdout

src/clauz3/stdlib/rsync/tools/rsync/trusted/contracts.py

"""rsync destination and delete policies built from generic trusted effect facts.

Every ``rsync`` call is recorded as the ``rsync`` effect with its ``src``,
``dest``, and ``delete`` flag (see ``effects.py``). These contracts let an
agent state where a sync may write, whether it may delete extraneous
destination files, and how many syncs may occur.

``only_dest_under`` is the headline blast-radius guard: it pins every
destination under a fixed prefix, so a sync cannot scribble over an arbitrary
location. ``no_delete`` is the headline data-loss guard: it forbids the
``--delete`` flag, so a sync can only add and update, never prune.
"""

from tools.rsync.trusted.effects import rsync

from clauz3.spec import ContractSpec, contract, effect
from clauz3.spec import no_guarantees as core_no_guarantees

Rsync = effect(rsync)


@contract
def no_guarantees() -> ContractSpec:
    """Make no guarantees about rsync effects."""
    return core_no_guarantees()


@contract
def no_syncs() -> ContractSpec:
    """Guarantee the program performs no rsync calls."""
    return Rsync.empty()


@contract
def only_dest_under(root: str) -> ContractSpec:
    """Guarantee every sync writes to a destination under ``root``."""
    return Rsync.all(lambda e: e.dest.startswith(root))


@contract
def never_dest_under(prefix: str) -> ContractSpec:
    """Guarantee no sync writes to a destination under ``prefix``."""
    return Rsync.all(lambda e: not e.dest.startswith(prefix))


@contract
def only_src_under(root: str) -> ContractSpec:
    """Guarantee every sync reads from a source under ``root``."""
    return Rsync.all(lambda e: e.src.startswith(root))


@contract
def no_delete() -> ContractSpec:
    """Guarantee no sync passes ``--delete`` (so none can prune the destination)."""
    return Rsync.all(lambda e: not e.delete)


@contract
def syncs_at_most(count: int) -> ContractSpec:
    """Guarantee at most ``count`` rsync calls occur."""
    return Rsync.count() <= count

All cases

Browse the full set on the all-cases page.