filesystem

A trusted clauz3 layer for reading, searching, listing, writing and managing files, with path policies an agent can prove before any file is touched.

Unlike the layers under examples/, these effects are real, not mocks. The prover never runs their bodies: it records each call as a fact and proves the contract. The bodies run only under clauz3 run, after a program is proved and approved.

Install

clauz3 install stdlib:filesystem

This copies the library into .clauz3/tools/filesystem/. Then prove a program against it:

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

Migrating from coreutils, editor, find, grep and text

These libraries were merged into filesystem, and an old copy defines effects with the same names (remove, append_file, find, ...). Proving it beside the new filesystem is refused as a duplicate effect, and installing the new filesystem next to one is refused with the same advice. Remove the old libraries (from the Libraries view, or delete them under .clauz3/tools/ and, for user-wide installs, ~/.local/share/clauz3/tools/), then run clauz3 install --force stdlib:filesystem.

Programs that imported them need the new names. edit_file is gone; use write_file. Contracts not listed keep their names under fs:

Old	New
grep.searches_at_most	fs.reads_at_most
find.searches_at_most	fs.lists_at_most
find.never_read_under	fs.never_list_under (fs.never_read_under covers contents only, not listings)
editor.no_edits	fs.read_only
editor.only_edit_under, coreutils.only_create_under, only_copy_under, only_move_under, only_chmod_under	fs.only_write_under
editor.never_edit_under, coreutils.never_create_under	fs.never_write_under
editor.edits_at_most	fs.writes_at_most
editor.must_not_replace, must_not_append	fs.must_not_write
editor.replace_length_at_most, append_length_at_most	fs.write_length_at_most
coreutils.never_copy_from_under, never_move_from_under	fs.never_read_under

Contracts cover events, not functions

Each effect declares, with @performs, the events it performs on the filesystem (events.py):

Effect	Events
read_file(path), read_lines(path), read_line(path, line_number)	Read(path)
grep(pattern, path)	Read(path)
find(path, args)	List(path)
write_file(path, content), append_file(path, text)	WriteText(path, text)
make_dir(path)	DirectoryCreate(path) (a Create)
touch(path)	Create(path)
remove(path)	Delete(path)
copy(src, dest)	RecursiveRead(src), RecursiveWrite(dest)
move(src, dest)	RecursiveRead(src), Delete(src), RecursiveWrite(dest)
chmod(path, mode)	Chmod(path, mode)

The contracts are stated over events, so one guarantee covers every effect that performs the event. never_read_under("/repo/PRECIOUS.md") refuses read_file, grep and read_line on that file, and a copy or move from it. Moving the file and reading it under a new name is refused at the move.

Events form a hierarchy, and a contract over an event covers its subclasses. RecursiveRead is a kind of Read. Read and List are kinds of Observe. WriteText, Create, RecursiveWrite, and Chmod are kinds of Write; Delete is a RecursiveWrite, and DirectoryCreate is a Create. So read_only() forbids deletes and chmods as well as writes. Each event still counts once in its ancestor relation: a move counts as two modifications.

Exact files or a directory

Use the same exact-file contract for one file or several:

fs.only_reads_files([clauz3.FilePath("README.md")])
fs.only_reads_files([clauz3.FilePath("README.md"), clauz3.FilePath("LICENSE")])
fs.only_reads_files_under(clauz3.DirectoryPath("data"))
fs.only_writes_files([clauz3.FilePath("test.md")])
fs.only_writes_files_under(clauz3.DirectoryPath("data"))

Import FilePath and DirectoryPath from clauz3, or use the qualified names above. Plain strings and the wrong path type are refused by these contracts. README.md means that exact name, never README.md.bak. An empty file list permits no corresponding reads or modifications. Directory data permits data/a.txt and subdirectories, never database/a.txt; a trailing slash is optional. Both contracts accept bare and leading ./ spellings of relative paths. DirectoryPath(".") permits only relative paths; DirectoryPath("/") permits only absolute paths.

Exact-file read access forbids directory listings and all copy/move sources: those tools can recursively read a directory, and a name alone cannot prove it is a file. The directory contract allows these operations inside its boundary. Read contracts do not restrict writes; combine with read_only() or an appropriate write policy. Path types express lexical intent, without checking existence or resolving symlinks or hard links. These contracts cover filesystem-library events, not file access by unrelated libraries.

Exact-file write scope allows write_file, append_file, touch, and chmod at the listed names. It refuses copy, move, and remove because their arguments may be directories, and refuses explicit make_dir. Implicit parent directories created by writing, appending, or touching a file are allowed; these are not separate modification events. only_writes_files_under covers all modification paths, including deletes and both ends of moves, within a directory boundary. Its implicit parent-directory creation can extend above that boundary. Write scopes do not restrict reads; add no_reads() when needed.

For a request such as “write hello to test.md”, use only_writes_files([FilePath("test.md")]). The legacy only_write_under("test.md") is a literal prefix and also allows test.md.bak; it remains for compatibility. write_file can overwrite an existing file. Neither typed scope asserts that a file is new or checks its physical type, symlinks, or hard links.

Effects

Import from tools.filesystem.trusted.effects. Every path argument must be non-empty and have no .. component, no . component after the first (/./, a trailing /.), no empty component (//) and no backslash. A leading ./ is allowed.

• read_file(path) -> str: read a UTF-8 text file.
• read_lines(path) -> list[str]: read UTF-8 logical lines without terminators, using Python str.splitlines(). Blank lines and a final unterminated line count; an empty file returns [].
• read_line(path, line_number) -> str: read one one-based line without its terminator; empty at end of file. line_number must be positive.
• grep(pattern, path) -> list[str]: the lines of path containing the substring pattern.
• find(path, args) -> list[str]: list paths under path using -name, -path, -type f/d/l, -mindepth and -maxdepth, optionally negated with -not. Other arguments raise ValueError before find runs.
• write_file(path, content): write UTF-8 text, creating parent directories.
• append_file(path, text): append UTF-8 text, creating the file as needed.
• make_dir(path), touch(path): create a directory (with parents) or an empty file.
• remove(path): delete a file, or a directory and everything under it.
• copy(src, dest), move(src, dest): copy or move a file or directory tree.
• chmod(path, mode): set permission bits.

Contracts

Import with from tools.filesystem.trusted import contracts as fs:

Contract	Over
fs.no_guarantees()	nothing
fs.only_reads_files(files: list[FilePath])	exact file reads; no listings or copy/move
fs.only_reads_files_under(directory: DirectoryPath)	Observe within a directory boundary
fs.only_writes_files(files: list[FilePath])	exact modification paths; no recursive operations or explicit directory creation
fs.only_writes_files_under(directory: DirectoryPath)	Write within a directory boundary
fs.no_reads(), legacy fs.only_read_under(root)	Observe (reads and listings)
fs.never_read_under(prefix), fs.reads_at_most(count)	Read
fs.never_list_under(prefix), fs.lists_at_most(count)	List
fs.read_only(), legacy fs.only_write_under(root), fs.never_write_under(prefix), fs.writes_at_most(count)	Write (every modification)
fs.must_not_write(substring), fs.write_length_at_most(max_chars)	WriteText
fs.no_deletes(), fs.only_delete_under(root), fs.never_delete_under(prefix), fs.deletes_at_most(count)	Delete
fs.no_creates(), fs.creates_at_most(count)	Create
fs.no_chmods(), fs.only_modes(allowlist)	Chmod
fs.only_pattern(pattern)	grep calls

The older only_read_under, only_write_under, only_delete_under, and never_* helpers are literal path-prefix policies over the symbolic call argument: only_read_under("test.md") also admits test.md.bak. The effects refuse .., /./ and // first, so a path cannot satisfy a prefix and then traverse outside it at runtime, or be respelled around a never_* prefix. A prefix that no effect path can be written as, such as data//secret, or an empty prefix, is refused instead of holding vacuously. The contracts also check each event's path the way the effects do, so a tool from another layer that performs these events cannot step around a prefix with //, /./ or ...

The check is still lexical. A relative prefix is compared in both its bare and ./ spellings, and . counts as an ancestor of every relative path, so never_delete_under("data") refuses remove("."). A never_* prefix refuses any path written the other way, relative under an absolute prefix or absolute under a relative one, because whether it reaches the prefix depends on the working directory: never_delete_under("/home/user/.ssh") refuses remove("."). A path contains a prefix only at a / boundary, so never_write_under("/repo/notes.md.orig") allows /repo/notes.md. Backslashes are refused, since the library is POSIX. Symbolic links, hard links and case-insensitive filesystems are not modelled.

Delete means remove or a move's source. Overwriting a file with write_file, copy or move is a Write, not a Delete, so protect contents with never_write_under. Parent directories created on the way by write_file, touch or copy are not Create events.

Every contract covers this library's tools only. git, rsync and db also touch files, and their effects are not filesystem events, so read_only() says nothing about a git commit. The approval text names "filesystem tools" for that reason.

A path may name a directory, and copy, move and remove act on everything under one. The never_* contracts therefore also refuse an event whose path is a prefix of the forbidden one: never_delete_under("/home/user/.ssh") refuses remove("/home/user"). Listing is not reading, so never_read_under allows find over a directory containing the prefix. It reveals names, not contents; use never_list_under to keep find away from a prefix, or only_read_under to confine listings too.

Example

import clauz3
from tools.filesystem.trusted import contracts as fs
from tools.filesystem.trusted.effects import copy, grep, read_file, write_file


@clauz3.guarantee(fs.only_reads_files_under(clauz3.DirectoryPath("/repo")))
@clauz3.guarantee(fs.only_write_under("/repo/build"))
@clauz3.guarantee(fs.never_read_under("/repo/.env"))
def main() -> None:
    source = read_file("/repo/src/app.py")
    print(grep("TODO", "/repo/src/app.py"))
    write_file("/repo/build/app.txt", source)
    copy("/repo/src/app.py", "/repo/build/app.py")

These guarantees confine the program to reading under /repo and modifying under /repo/build, and keep it away from /repo/.env whichever tool it uses.

Counting lines

import clauz3
from tools.filesystem.trusted import contracts as fs
from tools.filesystem.trusted.effects import read_lines

@clauz3.guarantee(fs.only_reads_files([clauz3.FilePath("./data/input.txt")]))
@clauz3.guarantee(fs.read_only())
def main() -> None:
    print(len(read_lines("./data/input.txt")))

This counts logical lines, not LF bytes as wc -l does. For hello without a terminator, the logical count is 1 and wc -l is 0. A trailing separator does not add a phantom line; hello\n\n has two logical lines.

Sessions expose this as the count-lines recipe through list_skills and read_skill when the installed filesystem library matches the bundled version. After upgrading clauz3, an older copied filesystem library hides all recipes that require it, including read-file, write-and-read, and directory-search recipes, until that library is updated. The installed effects remain usable through tools() and run; only the tested recipe previews are withheld. Existing copied libraries need an operator-authorized update to gain new effects; updating clauz3 alone does not rewrite project libraries. For an unmodified bundled installation, the operator can run clauz3 install --force stdlib:filesystem in the project and restart the session. --force overwrites the installed filesystem library; preserve any local customizations before choosing that update.

Tests

just -f tests/Justfile test

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

Source

src/clauz3/stdlib/filesystem/tools/filesystem/trusted/events.py

"""What filesystem tools do to the filesystem.

Every effect in ``effects.py`` declares the events it performs, and the
contracts in ``contracts.py`` are stated over these events, so a policy covers
every tool that performs the event. ``copy`` reads its source and writes its
destination, so ``never_read_under`` sees the copy and ``only_write_under`` sees
the destination.

The classes form a hierarchy, and a relation over a class covers its
subclasses::

    FsEvent(path)
    ├── Observe          anything that reveals filesystem state
    │   ├── Read         file contents
    │   │   └── RecursiveRead   copy/move source, possibly a whole directory
    │   └── List         names under a directory
    └── Write            any modification
        ├── WriteText(text)   text written to a file
        ├── Create            a new directory or empty file
        │   └── DirectoryCreate  explicit directory creation
        ├── RecursiveWrite    modification that may cover descendants
        │   └── Delete        removal (move removes its source)
        └── Chmod(mode)       permission bits

A path may name a directory, and copy, move and remove act on everything under
one. Contracts that forbid a prefix therefore refuse an event whose path is a
prefix of it too: ``remove("/repo")`` deletes ``/repo/PRECIOUS``.
"""

from clauz3.events import Event


class FsEvent(Event):
    """An event on the filesystem at ``path``."""

    path: str


class Observe(FsEvent):
    """Filesystem state is revealed to the program."""


class Read(Observe):
    """The contents at ``path`` are read."""


class RecursiveRead(Read):
    """Read a path that may be a directory and all its descendants."""


class List(Observe):
    """The names under the directory ``path`` are listed."""


class Write(FsEvent):
    """The filesystem at ``path`` is modified."""


class WriteText(Write):
    """``text`` is written to the file at ``path``, replacing or appending."""

    text: str


class Create(Write):
    """A directory or empty file is created at ``path``."""


class DirectoryCreate(Create):
    """Explicitly create a directory, with parents."""


class RecursiveWrite(Write):
    """Modify a path that may cover a directory and its descendants."""


class Delete(RecursiveWrite):
    """``path``, and everything under it, is removed."""


class Chmod(Write):
    """The permission bits of ``path`` are set to ``mode``."""

    mode: int

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

"""Trusted filesystem effects: read, search, list, write, edit and manage files.

These are real (non-mock) side-effecting functions. The prover never executes
their bodies: it records each call as an effect fact, proves the declared
preconditions, and reads the call as the events in ``@performs`` (see
``events.py``). The bodies run only under ``clauz3 run``, after a program has
been proved and approved.

The bodies use ``pathlib``/``shutil``/``os`` directly, not a shell, so no
argument is ever interpreted by a shell. ``find`` is the exception: it runs the
``find`` binary with an allowlisted argument grammar.

The ``@deal.has(...)`` markers are deal's own: they buy its runtime guard and
the ``clauz3 lint`` check on these bodies, and play no part in what a contract
can say.
"""

import os
import shutil
import subprocess
from pathlib import Path

import deal
from tools.filesystem.trusted.events import (
    Chmod,
    Create,
    Delete,
    DirectoryCreate,
    List,
    Read,
    RecursiveRead,
    RecursiveWrite,
    WriteText,
)

import clauz3
from clauz3._path_safety import is_plain_path
from clauz3.events import performs

DEFAULT_TIMEOUT_SECONDS = 120


# -- reading ----------------------------------------------------------------


@clauz3.requires(
    lambda path: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path: len(path) > 0, message="path must be non-empty")
@deal.has("read")
@clauz3.trusted
@performs(lambda path: Read(path))
def read_file(path: str) -> str:
    """Read and return the UTF-8 text contents of ``path``."""
    return Path(path).read_text(encoding="utf-8")


@clauz3.requires(
    lambda path: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path: len(path) > 0, message="path must be non-empty")
@deal.has("read")
@clauz3.trusted
@performs(lambda path: Read(path))
def read_lines(path: str) -> list[str]:
    """Read UTF-8 logical lines without terminators, using str.splitlines.

    Empty files return []; blank lines are retained; a final unterminated
    line counts. This is not wc -l, which counts LF bytes. Decode and I/O
    errors propagate, as with read_file.
    """
    return Path(path).read_text(encoding="utf-8").splitlines()


@clauz3.requires(
    lambda path, line_number: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(
    lambda path, line_number: len(path) >= 1 and line_number >= 1,
    message="path must be non-empty and line_number must be positive",
)
@deal.has("read")
@clauz3.trusted
@performs(lambda path: Read(path))
def read_line(path: str, line_number: int) -> str:
    """Read a one-based line, without its terminator; return empty at EOF.

    Decode UTF-8 with universal newline handling. Preserve all other whitespace.
    """
    with Path(path).open(encoding="utf-8") as stream:
        for number, line in enumerate(stream, start=1):
            if number == line_number:
                return line.removesuffix("\n")
    return ""


@clauz3.requires(
    lambda pattern, path: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(
    lambda pattern, path: len(pattern) >= 1 and len(path) >= 1,
    message="pattern and path must be non-empty",
)
@deal.has("read")
@clauz3.trusted
@performs(lambda path: Read(path))
def grep(pattern: str, path: str) -> list[str]:
    """Return the lines in ``path`` that contain the substring ``pattern``."""
    text = Path(path).read_text(encoding="utf-8")
    return [line for line in text.splitlines() if pattern in line]


def _filters(args: list[str]) -> list[str]:
    """Parse conjunctions of filters, optionally negating one ordinary filter."""
    result: list[str] = []
    depths: list[str] = []
    index = 0
    while index < len(args):
        negate = args[index] == "-not"
        if negate:
            index += 1
        if index + 1 >= len(args):
            raise ValueError(
                "find expects filter/value pairs, optionally prefixed by -not"
            )
        option, value = args[index : index + 2]
        if "\0" in value:
            raise ValueError("find filter values must not contain NUL")
        if option in {"-name", "-path"}:
            pass
        elif option == "-type":
            if value not in {"f", "d", "l"}:
                raise ValueError("find -type supports only f, d, or l")
        elif option in {"-mindepth", "-maxdepth"}:
            if negate:
                raise ValueError("find depth limits cannot be negated")
            if not value or not value.isascii() or not value.isdecimal():
                raise ValueError("find depth must be a non-negative decimal integer")
        else:
            raise ValueError(f"unsupported find filter: {option}")
        if negate:
            result.append("-not")
        if option in {"-mindepth", "-maxdepth"}:
            depths.extend((option, value))
        else:
            result.extend((option, value))
        index += 2
    # GNU find expects global depth options before ordinary expression tests.
    return depths + result


@clauz3.requires(
    lambda path, args: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path, args: len(path) > 0, message="path must be non-empty")
@deal.has("read", "io", "syscall")
@clauz3.trusted
@performs(lambda path: List(path))
def find(path: str, args: list[str]) -> list[str]:
    """List paths using -name, -path, -type f/d/l, -mindepth, and -maxdepth.

    Filters are combined with AND. -not may negate a name/path/type filter.
    Invalid arguments raise ValueError before starting a subprocess. Traversal
    uses -P for encountered symlinks; root-path resolution can still follow
    symlinks, including roots ending in "/" or "/.". Output is NUL-delimited.
    Contracts constrain the lexical starting path, not resolved confinement.
    """
    filters = _filters(args)
    # Do not rely on Deal's optional runtime checks at this external boundary.
    if not path or "\0" in path or not is_plain_path(path):
        raise ValueError(
            "find requires a non-empty path without .., . or empty segments, "
            "backslashes or NUL"
        )
    # Prefix relative roots so find cannot parse an option or expression
    # such as '-L', '!', or '(' in place of the one selected starting path.
    root = (
        path
        if path == "." or os.path.isabs(path) or path.startswith("./")
        else "./" + path
    )
    completed = subprocess.run(  # noqa: S603
        ["find", "-P", root, *filters, "-print0"],
        capture_output=True,
        timeout=DEFAULT_TIMEOUT_SECONDS,
        check=True,
    )
    return [os.fsdecode(item) for item in completed.stdout.split(b"\0") if item]


# -- writing ----------------------------------------------------------------


@clauz3.requires(
    lambda path, content: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path, content: len(path) > 0, message="path must be non-empty")
@deal.has("write")
@clauz3.trusted
@performs(lambda path, content: WriteText(path, content))
def write_file(path: str, content: str) -> None:
    """Write ``content`` to ``path`` as UTF-8 text, creating parent dirs."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(content, encoding="utf-8")


@clauz3.requires(
    lambda path, text: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path, text: len(path) > 0, message="path must be non-empty")
@deal.has("write")
@clauz3.trusted
@performs(lambda path, text: WriteText(path, text))
def append_file(path: str, text: str) -> None:
    """Append ``text`` to the UTF-8 contents of ``path``.

    Creates parent directories and the file as needed.
    """
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    with target.open("a", encoding="utf-8") as fh:
        fh.write(text)


# -- managing files and directories ------------------------------------------


@clauz3.requires(
    lambda path: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path: len(path) > 0, message="path must be non-empty")
@deal.has("write", "io")
@clauz3.trusted
@performs(lambda path: DirectoryCreate(path))
def make_dir(path: str) -> None:
    """Create directory ``path`` (with parents)."""
    Path(path).mkdir(parents=True, exist_ok=True)


@clauz3.requires(
    lambda path: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path: len(path) > 0, message="path must be non-empty")
@deal.has("write", "io")
@clauz3.trusted
@performs(lambda path: Create(path))
def touch(path: str) -> None:
    """Create empty file ``path`` (or bump its mtime)."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.touch()


@clauz3.requires(
    lambda path: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path: len(path) > 0, message="path must be non-empty")
@deal.has("write", "io")
@clauz3.trusted
@performs(lambda path: Delete(path))
def remove(path: str) -> None:
    """Delete ``path`` — a file, or a directory and everything under it."""
    target = Path(path)
    if target.is_dir() and not target.is_symlink():
        shutil.rmtree(target)
    else:
        target.unlink(missing_ok=True)


@clauz3.requires(
    lambda src, dest: is_plain_path(src) and is_plain_path(dest),
    message="src and dest must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(
    lambda src, dest: len(src) > 0 and len(dest) > 0,
    message="src and dest must be non-empty",
)
@deal.has("read", "write", "io")
@clauz3.trusted
@performs(lambda src, dest: [RecursiveRead(src), Delete(src), RecursiveWrite(dest)])
def move(src: str, dest: str) -> None:
    """Move ``src`` to ``dest``.

    A move is read as reading and deleting its source and writing its
    destination, so a file cannot be moved out from under a read policy and
    read under its new name.
    """
    shutil.move(src, dest)


@clauz3.requires(
    lambda src, dest: is_plain_path(src) and is_plain_path(dest),
    message="src and dest must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(
    lambda src, dest: len(src) > 0 and len(dest) > 0,
    message="src and dest must be non-empty",
)
@deal.has("read", "write", "io")
@clauz3.trusted
@performs(lambda src, dest: [RecursiveRead(src), RecursiveWrite(dest)])
def copy(src: str, dest: str) -> None:
    """Copy ``src`` to ``dest`` (file or directory tree)."""
    if Path(src).is_dir():
        shutil.copytree(src, dest, dirs_exist_ok=True)
    else:
        Path(dest).parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src, dest)


@clauz3.requires(
    lambda path, mode: is_plain_path(path),
    message="path must have no .., . or empty segments, or backslashes",
)
@clauz3.requires(lambda path, mode: len(path) > 0, message="path must be non-empty")
@deal.has("io")
@clauz3.trusted
@performs(lambda path, mode: Chmod(path, mode))
def chmod(path: str, mode: int) -> None:
    """Set the permission bits of ``path`` to ``mode``."""
    os.chmod(path, mode)

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

"""Filesystem policies, stated over the events the tools perform.

Each contract below is a query over an event class from ``events.py``, so it
governs every tool that performs that event: ``never_read_under`` covers
``read_file``, ``read_lines``, ``read_line``, ``grep`` and the source of
``copy`` and ``move``; ``only_write_under`` covers every modification,
including deletes, creates and chmods.

Every contract covers this library's tools only. Effects of other libraries
that touch files (git, rsync, db) are not events here, and the approval text
says so by naming "filesystem tools".

Use ``only_reads_files`` / ``only_writes_files`` for exact files and their
``*_files_under`` counterparts for a directory boundary.
The legacy prefix helpers use literal string prefixes:
``only_read_under("test.md")`` also admits ``test.md.bak``. The ``never_*``
contracts also refuse an event whose path is a prefix of the forbidden one,
since that path may be a directory containing it. The effects refuse ``..``,
``/./`` and ``//`` so a path cannot be respelled around a prefix, but the check
stays lexical. A relative prefix is compared in both its bare and ``./``
spellings, and ``.`` counts as an ancestor of every relative path, but a
relative prefix says nothing about an absolute spelling of the same file.
Symbolic links, hard links and case-insensitive filesystems are not modelled.

``Delete`` means ``remove`` or a ``move``'s source. Overwriting a file with
``write_file``, ``copy`` or ``move`` is a ``Write``, not a ``Delete``: protect
contents with ``never_write_under``. Parent directories that ``write_file``,
``touch`` and ``copy`` create on the way are not ``Create`` events; they lie on
the path of the ``Write`` that created them.

A few contracts are about one tool's own arguments (``only_pattern``,
``only_modes``) and are stated over that tool or event alone.
"""

from tools.filesystem.trusted.effects import grep
from tools.filesystem.trusted.events import (
    Chmod,
    Create,
    Delete,
    DirectoryCreate,
    FsEvent,
    List,
    Observe,
    Read,
    RecursiveRead,
    RecursiveWrite,
    Write,
    WriteText,
)

from clauz3._path_safety import is_plain_path
from clauz3.paths import DirectoryPath, FilePath
from clauz3.spec import ConjunctionSpec, ContractSpec, contract, effect
from clauz3.spec import no_guarantees as core_no_guarantees

Grep = effect(grep)


def _checked(prefix: str) -> str:
    """Refuse a prefix no effect path can be written as, or the empty prefix.

    Every effect refuses ``..``, ``/./``, ``//`` and backslashes, so a policy
    over ``data//secret`` would match nothing while ``data/secret`` stayed
    unconstrained, and ``only_read_under("")`` would admit every path. Raising
    here makes the proof a refusal, not a vacuous pass.

    >>> _checked("data/./secret")
    Traceback (most recent call last):
    ...
    ValueError: path prefix 'data/./secret' is not written the way effect paths are
    """
    if not (prefix and is_plain_path(prefix)):
        raise ValueError(
            f"path prefix {prefix!r} is not written the way effect paths are"
        )
    return prefix


# Each path contract also checks the event path is plain, the condition
# ``is_plain_path`` states for the filesystem effects' preconditions. A tool
# from another layer may perform the same events without that precondition,
# and a respelled path (``/home/user//.ssh``) would otherwise step around a
# prefix. Contract lambdas cannot call helpers, so the clauses are inline.


def _under(event: type[FsEvent], root: str) -> ContractSpec:
    _checked(root)
    return event.all(
        lambda e: (
            "//" not in e.path
            and "\\" not in e.path
            # Shortcut first disjuncts, as in is_plain_path, keep the solver
            # from re-deriving the detail from an agent's simpler guard.
            and (
                "/." not in e.path
                or ("/./" not in e.path and not e.path.endswith("/."))
            )
            and (
                ".." not in e.path
                or (
                    "/../" not in e.path
                    and not e.path.startswith("../")
                    and not e.path.endswith("/..")
                    and e.path != ".."
                    and ":.." not in e.path
                )
            )
            and e.path.startswith(root)
        )
    )


def _spellings(prefix: str) -> tuple[str, str]:
    """A prefix written bare and with ``./``; the effects admit both spellings.

    ``.`` and ``./`` are the working directory, the bare prefix ``""``.

    >>> _spellings("./data/.env"), _spellings("data"), _spellings("/etc")
    (('data/.env', './data/.env'), ('data', './data'), ('/etc', '/etc'))
    >>> _spellings("."), _spellings("./")
    (('', './'), ('', './'))
    """
    if prefix.startswith("/"):
        return prefix, prefix
    bare = "" if prefix == "." else prefix.removeprefix("./")
    return bare, "./" + bare


def _never_under(event: type[FsEvent], prefix: str) -> ContractSpec:
    """Refuse an event at, under, or containing either spelling of ``prefix``.

    An event path written the other way (relative under an absolute prefix,
    or absolute under a relative one) is refused outright: whether it reaches
    the prefix depends on the working directory, which a proof cannot see.
    An event path contains the prefix only at a ``/`` boundary, so
    ``never_write_under("/repo/notes.md.orig")`` allows ``/repo/notes.md``.
    """
    bare, dotted = _spellings(_checked(prefix))
    absolute = prefix.startswith("/")
    return event.all(
        lambda e: (
            "//" not in e.path
            and "\\" not in e.path
            # Shortcut first disjuncts, as in is_plain_path, keep the solver
            # from re-deriving the detail from an agent's simpler guard.
            and (
                "/." not in e.path
                or ("/./" not in e.path and not e.path.endswith("/."))
            )
            and (
                ".." not in e.path
                or (
                    "/../" not in e.path
                    and not e.path.startswith("../")
                    and not e.path.endswith("/..")
                    and e.path != ".."
                    and ":.." not in e.path
                )
            )
            and e.path.startswith("/") == absolute
            # at or under the prefix
            and not e.path.startswith(bare)
            and not e.path.startswith(dotted)
            # a directory containing it
            and not bare.startswith(e.path + "/")
            and not dotted.startswith(e.path + "/")
            and not (e.path.endswith("/") and bare.startswith(e.path))
            and not (e.path.endswith("/") and dotted.startswith(e.path))
        )
    )


@contract(description="No restrictions are imposed on filesystem effects.")
def no_guarantees() -> ContractSpec:
    """Make no guarantees about filesystem effects."""
    return core_no_guarantees()


# -- reading -----------------------------------------------------------------


@contract(description="No filesystem tool reads a file or lists a directory.")
def no_reads() -> ContractSpec:
    """Guarantee the program neither reads files nor lists directories."""
    return Observe.empty()


@contract(
    description=(
        "Filesystem tools read only the exact files {files}; "
        "directory listings and copy or move sources are forbidden."
    )
)
def only_reads_files(files: list[FilePath]) -> ContractSpec:
    """Allow reads of these exact files (one or many), never a name prefix.

    Bare and leading ./ spellings are equivalent. An empty list permits no
    reads or listings. Copy and move may recursively read directories, so this
    contract refuses them even when their source looks like a file name.
    Paths are lexical: no existence, symlink or hard-link checks are implied.
    """
    if type(files) is not list or any(type(path) is not FilePath for path in files):
        raise TypeError("files must be a list of FilePath literals")
    names = [name for path in files for name in _spellings(path.value)]
    return ConjunctionSpec(
        (
            Observe.all(lambda e: e.path in names),
            List.empty(),
            RecursiveRead.empty(),
        )
    )


@contract(
    description=(
        "Filesystem tools read or list only paths in the directory {directory} "
        "or its subdirectories (paths compared lexically)."
    )
)
def only_reads_files_under(directory: DirectoryPath) -> ContractSpec:
    """Confine reads, listings, and copy/move sources to a directory boundary.

    DirectoryPath("data") permits data and data/a.txt, but not database/a.txt.
    A trailing slash is optional; ./data and data are equivalent. DirectoryPath
    (".") permits relative paths only; DirectoryPath("/") permits absolute
    paths only. Paths are lexical, without existence or symlink checks.
    """
    return _in_directory(Observe, directory)


def _in_directory(event: type[FsEvent], directory: DirectoryPath) -> ContractSpec:
    """Compare an event path with a validated lexical directory boundary."""
    if type(directory) is not DirectoryPath:
        raise TypeError("directory must be a DirectoryPath literal")
    root = directory.value.rstrip("/") or "/"
    bare, dotted = _spellings(root)
    roots = [bare, dotted]
    bare_prefix, dotted_prefix = bare + "/", dotted + "/"
    cwd = root in (".", "./")
    absolute_root = root == "/"
    return event.all(
        lambda e: (
            len(e.path) > 0
            and "//" not in e.path
            and "\\" not in e.path
            and (
                "/." not in e.path
                or ("/./" not in e.path and not e.path.endswith("/."))
            )
            and (
                ".." not in e.path
                or (
                    "/../" not in e.path
                    and not e.path.startswith("../")
                    and not e.path.endswith("/..")
                    and e.path != ".."
                    and ":.." not in e.path
                )
            )
            and (
                (cwd and not e.path.startswith("/"))
                or (absolute_root and e.path.startswith("/"))
                or (
                    not cwd
                    and not absolute_root
                    and (
                        e.path in roots
                        or e.path.startswith(bare_prefix)
                        or e.path.startswith(dotted_prefix)
                    )
                )
            )
        )
    )


@contract(description="Every file a filesystem tool reads or lists starts with {root}.")
def only_read_under(root: str) -> ContractSpec:
    """Legacy literal-prefix policy, including suffixes such as README.md.bak.

    Prefer only_reads_files for exact names or only_reads_files_under for a
    directory boundary. This helper retains its original prefix semantics.
    """
    return _under(Observe, root)


@contract(
    description=(
        "No filesystem tool reads, copies or moves a path starting with "
        "{prefix}, or a directory containing it (paths compared as written)."
    )
)
def never_read_under(prefix: str) -> ContractSpec:
    """Guarantee no file contents under ``prefix`` reach the program or a copy.

    Listing a directory that contains ``prefix`` is allowed: it reveals the
    name, not the contents. Use ``only_read_under`` to confine listings too.
    """
    return _never_under(Read, prefix)


@contract(description="Filesystem tools read files at most {count} times.")
def reads_at_most(count: int) -> ContractSpec:
    """Guarantee at most ``count`` reads (a copy or move reads its source)."""
    return Read.count() <= count


@contract(
    description=(
        "No filesystem tool lists a directory starting with {prefix}, or one "
        "containing it (paths compared as written)."
    )
)
def never_list_under(prefix: str) -> ContractSpec:
    """Guarantee no ``find`` walk starts at, under, or above ``prefix``.

    The listing counterpart of ``never_read_under``, which covers contents
    only. Walking a directory that contains ``prefix`` reveals its names, so
    that is refused too. This is what ``find.never_read_under`` used to say.
    """
    return _never_under(List, prefix)


@contract(description="Filesystem tools list directories at most {count} times.")
def lists_at_most(count: int) -> ContractSpec:
    """Guarantee at most ``count`` ``find`` walks occur."""
    return List.count() <= count


@contract(description="Every grep search uses exactly the pattern {pattern}.")
def only_pattern(pattern: str) -> ContractSpec:
    """Guarantee every ``grep`` search uses exactly this pattern."""
    return Grep.all(lambda e: e.pattern == pattern)


# -- writing -----------------------------------------------------------------


@contract(
    description=(
        "No filesystem tool modifies anything. Effects of other libraries, such"
        " as git or rsync, are not covered."
    )
)
def read_only() -> ContractSpec:
    """Guarantee no write, append, create, delete, move, copy or chmod occurs."""
    return Write.empty()


@contract(
    description=(
        "Filesystem tools modify only the exact paths {files}; "
        "copy, move, remove and explicit directory creation are forbidden. "
        "Parent directories may be created when writing or touching a file."
    )
)
def only_writes_files(files: list[FilePath]) -> ContractSpec:
    """Confine modifications to exact file names (one or many), never prefixes.

    Allows write_file, append_file, touch and chmod at the listed paths.
    Copy, move and remove may modify whole directories, so they are refused;
    make_dir is refused too. An empty list permits no modifications. Parent
    directories implicitly created by write_file, append_file or touch are
    allowed and are not separate write events. Bare and leading ./ spellings
    are equivalent. Paths are lexical: existence, file type and symlink or
    hard-link targets are not checked. Reads are not restricted by this contract.
    """
    if type(files) is not list or any(type(path) is not FilePath for path in files):
        raise TypeError("files must be a list of FilePath literals")
    names = [name for path in files for name in _spellings(path.value)]
    return ConjunctionSpec(
        (
            Write.all(lambda e: e.path in names),
            RecursiveWrite.empty(),
            DirectoryCreate.empty(),
        )
    )


@contract(
    description=(
        "Filesystem tools modify only paths in the directory {directory} "
        "or its subdirectories (paths compared lexically). "
        "Implicit creation of parent directories is allowed."
    )
)
def only_writes_files_under(directory: DirectoryPath) -> ContractSpec:
    """Confine all modification paths to a directory boundary, including deletes.

    Covers write, append, touch, make_dir, chmod, copy destinations, and both
    the source and destination of moves. Reads are unrestricted. DirectoryPath
    ("data") permits data/a but not database/a. Bare and ./ paths are equivalent;
    a trailing slash is optional. Dot permits relative paths, slash absolute
    paths. Implicit parent-directory creation may reach above the boundary.
    Paths are lexical, without existence, symlink or hard-link checks.
    """
    return _in_directory(Write, directory)


@contract(
    description=(
        "Every modification by a filesystem tool is to a path starting with {root}."
    )
)
def only_write_under(root: str) -> ContractSpec:
    """Legacy literal-prefix policy, including suffixes such as test.md.bak.

    Prefer only_writes_files for exact files or only_writes_files_under for
    a directory boundary. This helper retains its original prefix semantics.
    A move deletes its source, so moving a file out of ``root`` is refused too.
    """
    return _under(Write, root)


@contract(
    description=(
        "No filesystem tool modifies a path starting with {prefix}, or a "
        "directory containing it (paths compared as written)."
    )
)
def never_write_under(prefix: str) -> ContractSpec:
    """Guarantee no modification, of any kind, touches ``prefix``."""
    return _never_under(Write, prefix)


@contract(description="Filesystem tools make at most {count} modifications.")
def writes_at_most(count: int) -> ContractSpec:
    """Guarantee at most ``count`` modifications (a move counts twice)."""
    return Write.count() <= count


@contract(
    description="No text a filesystem tool writes to a file contains {substring}."
)
def must_not_write(substring: str) -> ContractSpec:
    """Guarantee no written or appended text contains ``substring``.

    A secrets / banned-content guard before content reaches disk. Copies and
    moves write existing contents, not text, so they are not covered.
    """
    return WriteText.all(lambda e: substring not in e.text)


@contract(
    description=(
        "Text a filesystem tool writes to a file is at most {max_chars} characters."
    )
)
def write_length_at_most(max_chars: int) -> ContractSpec:
    """Guarantee every written or appended text is at most ``max_chars`` long."""
    return WriteText.all(lambda e: len(e.text) <= max_chars)


# -- deleting ----------------------------------------------------------------


@contract(description="No filesystem tool deletes or moves anything away.")
def no_deletes() -> ContractSpec:
    """Guarantee nothing is removed, including by moving it.

    Overwriting a file is a write, not a delete; see ``never_write_under``.
    """
    return Delete.empty()


@contract(
    description="Everything a filesystem tool deletes or moves away starts with {root}."
)
def only_delete_under(root: str) -> ContractSpec:
    """Guarantee every removed (or moved) path is under ``root``."""
    return _under(Delete, root)


@contract(
    description=(
        "No filesystem tool deletes or moves a path starting with {prefix}, or "
        "a directory containing it (paths compared as written)."
    )
)
def never_delete_under(prefix: str) -> ContractSpec:
    """Guarantee nothing under ``prefix`` is removed or moved.

    The blast-radius guard for handing an agent ``rm``: e.g.
    ``never_delete_under("/home/user/.ssh")``.
    """
    return _never_under(Delete, prefix)


@contract(description="Filesystem tools delete or move away at most {count} paths.")
def deletes_at_most(count: int) -> ContractSpec:
    """Guarantee at most ``count`` removals (a move counts as one)."""
    return Delete.count() <= count


# -- creating ----------------------------------------------------------------


@contract(description="No filesystem tool creates a directory or empty file.")
def no_creates() -> ContractSpec:
    """Guarantee no ``make_dir`` or ``touch`` occurs."""
    return Create.empty()


@contract(
    description="Filesystem tools create at most {count} directories or empty files."
)
def creates_at_most(count: int) -> ContractSpec:
    """Guarantee at most ``count`` directories or empty files are created."""
    return Create.count() <= count


# -- permissions -------------------------------------------------------------


@contract(description="No filesystem tool changes permission bits.")
def no_chmods() -> ContractSpec:
    """Guarantee the program changes no permission bits."""
    return Chmod.empty()


@contract(
    description="Every filesystem chmod sets a mode in {allowlist}, written in decimal."
)
def only_modes(allowlist: list[int]) -> ContractSpec:
    """Guarantee every chmod sets a mode in ``allowlist``.

    Bitwise masking is not expressible in a contract lambda, so permission
    safety is stated as an explicit allowlist of whole modes, e.g.
    ``only_modes([0o644, 0o755, 0o600, 0o700])`` — which by construction admits
    no world-writable, setuid, or setgid bits.
    """
    return Chmod.all(lambda e: e.mode in allowlist)

All cases

Browse the full set on the all-cases page.