API Reference

Generated from the source. Every docstring here is the one in the code.

Client

smbfence.client

Reaching the share, guarded.

Every path passes through paths.guard before it reaches the SMB layer, and every call into that layer is wrapped so its three failure modes arrive as one.

The backend is a Protocol rather than a direct smbclient import, so the whole client is testable without a server. That is not only convenient: code that can only be exercised against a real share tends not to be exercised at all, and this is code whose failure modes matter.

SmbBackend

Bases: Protocol

Everything this library needs from an SMB layer.

Declared in full, including session handling, so a backend missing a method is a type error rather than a connection silently left open.

Source code in src/smbfence/client.py

@runtime_checkable
class SmbBackend(Protocol):
    """Everything this library needs from an SMB layer.

    Declared in full, including session handling, so a backend missing a method
    is a type error rather than a connection silently left open.
    """

    def listdir(self, path: str) -> list[str]: ...
    def open_file(self, path: str, mode: Literal["rb", "wb"]) -> IO[bytes]: ...
    def rename(self, src: str, dst: str) -> None: ...
    def exists(self, path: str) -> bool: ...
    def register_session(self, host: str, username: str, password: str, port: int) -> None: ...
    def reset_connection_cache(self) -> None: ...

ShareClient

Reads and writes within a fixed set of allowed roots.

Source code in src/smbfence/client.py

class ShareClient:
    """Reads and writes within a fixed set of allowed roots."""

    def __init__(self, base: str, allowed: Sequence[str], backend: SmbBackend) -> None:
        self._base = base
        self._allowed = tuple(allowed)
        self._backend = backend

    def _resolve(self, path: str) -> str:
        """The full SMB path, or `PathNotAllowedError`."""
        safe = paths.to_windows(paths.guard(path, self._allowed))
        return f"{self._base}{paths.WINDOWS_SEPARATOR}{safe}"

    def list_files(
        self,
        directory: str,
        *,
        suffix: str | None = None,
        prefix: str | None = None,
        excluded_prefixes: Sequence[str] = (),
    ) -> list[str]:
        """The matching entries in an allowed directory, sorted."""
        full = self._resolve(directory)
        with translated(f"listing {directory}"):
            entries = self._backend.listdir(full)
        return names.select(
            entries, suffix=suffix, prefix=prefix, excluded_prefixes=excluded_prefixes
        )

    def read_file(self, path: str) -> bytes:
        """The bytes of one file in an allowed directory."""
        full = self._resolve(path)
        with translated(f"reading {path}"), self._backend.open_file(full, mode="rb") as handle:
            return handle.read()

    def write_file(self, path: str, content: bytes) -> None:
        """Publish `content` at `path`, atomically, never overwriting.

        Written beside itself and renamed into place, so a transfer that stops
        half way leaves nothing under the final name. Without that, a link
        dropping mid-write leaves a truncated file where a whole one is
        expected, and every retry afterwards reads it back, finds bytes that
        differ, and refuses for good.

        A file already in place with the same bytes is success: a retry of a
        completed write is not an error. One holding *different* bytes belongs
        to somebody else, and is left exactly where it is.
        """
        full = self._resolve(path)
        temp = full + TEMP_SUFFIX

        with translated(f"writing {path}"):
            if self._backend.exists(full):
                with self._backend.open_file(full, mode="rb") as handle:
                    existing = handle.read()
                if existing == content:
                    logger.debug("%s already in place with identical bytes", path)
                    return
                raise ConflictingContentError(
                    f"{path!r} is already in place with different bytes and was not overwritten"
                )

            with self._backend.open_file(temp, mode="wb") as handle:
                handle.write(content)
            self._backend.rename(temp, full)

list_files

list_files(
    directory: str,
    *,
    suffix: str | None = None,
    prefix: str | None = None,
    excluded_prefixes: Sequence[str] = (),
) -> list[str]

The matching entries in an allowed directory, sorted.

Source code in src/smbfence/client.py

def list_files(
    self,
    directory: str,
    *,
    suffix: str | None = None,
    prefix: str | None = None,
    excluded_prefixes: Sequence[str] = (),
) -> list[str]:
    """The matching entries in an allowed directory, sorted."""
    full = self._resolve(directory)
    with translated(f"listing {directory}"):
        entries = self._backend.listdir(full)
    return names.select(
        entries, suffix=suffix, prefix=prefix, excluded_prefixes=excluded_prefixes
    )

read_file

read_file(path: str) -> bytes

The bytes of one file in an allowed directory.

Source code in src/smbfence/client.py

def read_file(self, path: str) -> bytes:
    """The bytes of one file in an allowed directory."""
    full = self._resolve(path)
    with translated(f"reading {path}"), self._backend.open_file(full, mode="rb") as handle:
        return handle.read()

write_file

write_file(path: str, content: bytes) -> None

Publish content at path, atomically, never overwriting.

Written beside itself and renamed into place, so a transfer that stops half way leaves nothing under the final name. Without that, a link dropping mid-write leaves a truncated file where a whole one is expected, and every retry afterwards reads it back, finds bytes that differ, and refuses for good.

A file already in place with the same bytes is success: a retry of a completed write is not an error. One holding different bytes belongs to somebody else, and is left exactly where it is.

Source code in src/smbfence/client.py

def write_file(self, path: str, content: bytes) -> None:
    """Publish `content` at `path`, atomically, never overwriting.

    Written beside itself and renamed into place, so a transfer that stops
    half way leaves nothing under the final name. Without that, a link
    dropping mid-write leaves a truncated file where a whole one is
    expected, and every retry afterwards reads it back, finds bytes that
    differ, and refuses for good.

    A file already in place with the same bytes is success: a retry of a
    completed write is not an error. One holding *different* bytes belongs
    to somebody else, and is left exactly where it is.
    """
    full = self._resolve(path)
    temp = full + TEMP_SUFFIX

    with translated(f"writing {path}"):
        if self._backend.exists(full):
            with self._backend.open_file(full, mode="rb") as handle:
                existing = handle.read()
            if existing == content:
                logger.debug("%s already in place with identical bytes", path)
                return
            raise ConflictingContentError(
                f"{path!r} is already in place with different bytes and was not overwritten"
            )

        with self._backend.open_file(temp, mode="wb") as handle:
            handle.write(content)
        self._backend.rename(temp, full)

share_connection

share_connection(
    config: ShareConfig,
    backend: SmbBackend,
    *,
    allowed: Sequence[str] | None = None,
) -> Iterator[ShareClient]

A client for the configured share, with the session dropped afterwards.

The connection is dropped after every use. A hand-run script exits and takes its sockets with it; a long-lived process does not, and smbclient caches connections per host. A session that died quietly between two runs would be handed back to the next one, turning a recoverable blip into a fault that persists until somebody restarts the process.

Source code in src/smbfence/client.py

@contextmanager
def share_connection(
    config: ShareConfig, backend: SmbBackend, *, allowed: Sequence[str] | None = None
) -> Iterator[ShareClient]:
    """A client for the configured share, with the session dropped afterwards.

    The connection is dropped after every use. A hand-run script exits and takes
    its sockets with it; a long-lived process does not, and `smbclient` caches
    connections per host. A session that died quietly between two runs would be
    handed back to the next one, turning a recoverable blip into a fault that
    persists until somebody restarts the process.
    """
    config.require()
    roots = tuple(allowed if allowed is not None else config.allowed_paths)

    with translated(f"connecting to {config.host}"):
        backend.register_session(
            config.host,
            username=config.user,
            password=config.password.get_secret_value(),
            port=config.port,
        )
    try:
        yield ShareClient(base=paths.unc(config.host, config.share), allowed=roots, backend=backend)
    finally:
        backend.reset_connection_cache()

Configuration

smbfence.config

Share credentials, read from the environment.

Held as a Pydantic model so the password is a SecretStr and cannot reach a log line by accident, and so a partially configured share is a distinguishable state rather than a connection that fails for no stated reason.

ShareConfig

Bases: BaseSettings

What is needed to reach one share.

Source code in src/smbfence/config.py

class ShareConfig(BaseSettings):
    """What is needed to reach one share."""

    model_config = SettingsConfigDict(env_prefix="SMB_", env_file="secrets.env", extra="ignore")

    host: str = ""
    share: str = ""
    user: str = ""
    password: SecretStr = SecretStr("")
    port: int = 445
    allowed_paths: Sequence[str] = Field(default=())

    @property
    def is_configured(self) -> bool:
        """Whether there are credentials for the share at all.

        Distinct from reachability: this answers "should anything be attempted",
        which a caller often needs before it has a connection to ask.
        """
        return bool(self.host and self.share and self.user and self.password.get_secret_value())

    def require(self) -> None:
        """Raise `ShareConfigError` unless every credential is present."""
        if not self.is_configured:
            raise ShareConfigError(MISSING_CREDENTIALS)

is_configured property

is_configured: bool

Whether there are credentials for the share at all.

Distinct from reachability: this answers "should anything be attempted", which a caller often needs before it has a connection to ask.

require

require() -> None

Raise ShareConfigError unless every credential is present.

Source code in src/smbfence/config.py

def require(self) -> None:
    """Raise `ShareConfigError` unless every credential is present."""
    if not self.is_configured:
        raise ShareConfigError(MISSING_CREDENTIALS)

Paths

smbfence.paths

Share paths, as pure functions.

A share path arrives spelled whichever way its caller happens to hold it: with backslashes from a Windows tool, with forward slashes from a config file, with or without a leading separator. Comparing two of those spellings directly is the bug this module exists to prevent, because the comparison that matters is whether a path is inside an allowed root, and "clients/acme" must match "\clients\acme\" for that check to mean anything.

Nothing here touches a share, so all of it is tested without one.

normalise

normalise(path: str) -> str

One spelling for a share path, whichever separator the caller used.

Source code in src/smbfence/paths.py

def normalise(path: str) -> str:
    """One spelling for a share path, whichever separator the caller used."""
    return path.replace(WINDOWS_SEPARATOR, POSIX_SEPARATOR).strip(POSIX_SEPARATOR)

to_windows

to_windows(path: str) -> str

The normalised path spelled the way SMB expects it.

Source code in src/smbfence/paths.py

def to_windows(path: str) -> str:
    """The normalised path spelled the way SMB expects it."""
    return normalise(path).replace(POSIX_SEPARATOR, WINDOWS_SEPARATOR)

unc

unc(host: str, share: str) -> str

The UNC prefix every resolved path is built on.

Source code in src/smbfence/paths.py

def unc(host: str, share: str) -> str:
    """The UNC prefix every resolved path is built on."""
    return f"{WINDOWS_SEPARATOR * 2}{host}{WINDOWS_SEPARATOR}{share}"

is_within

is_within(path: str, allowed: Sequence[str]) -> bool

Whether a path sits inside one of the allowed roots.

A root matches itself and anything beneath it. The separator in the prefix test is what stops clients/acme-holdings from matching the root clients/acme — without it, any root would also admit every sibling whose name it happens to prefix.

Source code in src/smbfence/paths.py

def is_within(path: str, allowed: Sequence[str]) -> bool:
    """Whether a path sits inside one of the allowed roots.

    A root matches itself and anything beneath it. The separator in the prefix
    test is what stops `clients/acme-holdings` from matching the root
    `clients/acme` — without it, any root would also admit every sibling whose
    name it happens to prefix.
    """
    candidate = normalise(path)
    return any(
        candidate == normalise(root) or candidate.startswith(normalise(root) + POSIX_SEPARATOR)
        for root in allowed
    )

guard

guard(path: str, allowed: Sequence[str]) -> str

The normalised path, or PathNotAllowedError if it is out of bounds.

Every path reaching the share passes through here. Returning the normalised form rather than None is what makes that enforceable: a caller cannot use a path it did not get back from this function.

Source code in src/smbfence/paths.py

def guard(path: str, allowed: Sequence[str]) -> str:
    """The normalised path, or `PathNotAllowedError` if it is out of bounds.

    Every path reaching the share passes through here. Returning the normalised
    form rather than `None` is what makes that enforceable: a caller cannot use
    a path it did not get back from this function.
    """
    if not is_within(path, allowed):
        raise PathNotAllowedError(PATH_NOT_ALLOWED.format(path=path))
    return normalise(path)

Names

smbfence.names

Choosing entries out of a directory listing, as pure functions.

A share filed by hand from both Windows and macOS accumulates metadata files next to the real ones: .DS_Store and ._-prefixed resource forks from macOS, Thumbs.db and desktop.ini from Windows. They appear in every folder and they are never what a caller asked for.

The second rule here is less obvious. Selecting by suffix and excluding by prefix are different questions, and folding them into one test lets an excluded file through the moment a caller asks by suffix alone. Where a share holds two kinds of file under one extension — a booked export and a provisional one, say — that mistake imports data that was never meant to be read as final.

Nothing here touches a share, so all of it is tested without one.

is_noise

is_noise(name: str) -> bool

Whether an entry is filesystem metadata rather than a file anyone filed.

Source code in src/smbfence/names.py

def is_noise(name: str) -> bool:
    """Whether an entry is filesystem metadata rather than a file anyone filed."""
    lowered = name.lower()
    return lowered in NOISE_NAMES or lowered.startswith(APPLE_DOUBLE_PREFIX)

matches

matches(
    name: str,
    *,
    suffix: str | None = None,
    prefix: str | None = None,
    excluded_prefixes: Sequence[str] = (),
) -> bool

Whether an entry is one a caller asked for.

Exclusions are applied before the suffix test, never merged with it, so a caller asking only by suffix still does not receive an excluded file.

Source code in src/smbfence/names.py

def matches(
    name: str,
    *,
    suffix: str | None = None,
    prefix: str | None = None,
    excluded_prefixes: Sequence[str] = (),
) -> bool:
    """Whether an entry is one a caller asked for.

    Exclusions are applied before the suffix test, never merged with it, so a
    caller asking only by suffix still does not receive an excluded file.
    """
    if is_noise(name):
        return False

    lowered = name.lower()
    if any(lowered.startswith(excluded.lower()) for excluded in excluded_prefixes):
        return False
    if prefix is not None and not lowered.startswith(prefix.lower()):
        return False
    return not (suffix is not None and not lowered.endswith(suffix.lower()))

select

select(
    names: Iterable[str],
    *,
    suffix: str | None = None,
    prefix: str | None = None,
    excluded_prefixes: Sequence[str] = (),
) -> list[str]

The matching entries, sorted.

Sorted because a directory listing arrives in whatever order the server chose, and a caller that processes files in listing order would otherwise behave differently against two servers holding identical files.

Source code in src/smbfence/names.py

def select(
    names: Iterable[str],
    *,
    suffix: str | None = None,
    prefix: str | None = None,
    excluded_prefixes: Sequence[str] = (),
) -> list[str]:
    """The matching entries, sorted.

    Sorted because a directory listing arrives in whatever order the server
    chose, and a caller that processes files in listing order would otherwise
    behave differently against two servers holding identical files.
    """
    return sorted(
        name
        for name in names
        if matches(name, suffix=suffix, prefix=prefix, excluded_prefixes=excluded_prefixes)
    )

Errors

smbfence.errors

The errors a caller has to handle, and the translation into them.

There are four, and they are deliberately few. A caller reaching a file share wants to distinguish "I was not configured", "I asked for something I am not allowed to have", "the share did not answer" and "something is already there and it is not mine" — and nothing finer. Everything the underlying library raises is translated into one of these, so no caller has to import smbprotocol's exception tree to write a correct except clause.

ShareError

Bases: Exception

Base for everything this library raises.

Source code in src/smbfence/errors.py

class ShareError(Exception):
    """Base for everything this library raises."""

ShareConfigError

Bases: ShareError

Raised when the share is not configured.

Distinct from unreachable: nothing was attempted, so nothing can be retried until a human supplies credentials.

Source code in src/smbfence/errors.py

class ShareConfigError(ShareError):
    """Raised when the share is not configured.

    Distinct from unreachable: nothing was attempted, so nothing can be retried
    until a human supplies credentials.
    """

PathNotAllowedError

Bases: ShareError

Raised when a path falls outside every configured root.

Source code in src/smbfence/errors.py

class PathNotAllowedError(ShareError):
    """Raised when a path falls outside every configured root."""

ShareUnreachableError

Bases: ShareError

Raised when the credentials are there and the share still did not answer.

A share that will not answer is a fault, not an empty folder. Downstream the two are indistinguishable — both end with nothing new — so the failure is raised as itself rather than returned as an empty list.

Source code in src/smbfence/errors.py

class ShareUnreachableError(ShareError):
    """Raised when the credentials are there and the share still did not answer.

    A share that will not answer is a fault, not an empty folder. Downstream the
    two are indistinguishable — both end with nothing new — so the failure is
    raised as itself rather than returned as an empty list.
    """

ConflictingContentError

Bases: ShareError

Raised when a file is already in place with different bytes.

A retry of a completed write is success. A file under the same name holding different bytes is somebody else's, and overwriting it is the one unrecoverable thing this library could do.

Source code in src/smbfence/errors.py

class ConflictingContentError(ShareError):
    """Raised when a file is already in place with different bytes.

    A retry of a completed write is success. A file under the same name holding
    *different* bytes is somebody else's, and overwriting it is the one
    unrecoverable thing this library could do.
    """

translated

translated(what: str) -> Iterator[None]

Turn whatever the SMB layer raises into ShareUnreachableError.

Three exception types reach this point and the third is not obvious. smbprotocol catches the socket error during transport connection and re-raises it as a plain ValueError naming the server, so a host that does not resolve arrives as neither an OSError nor an SMBException. Without this, an unreachable host raises straight past a caller's except clause — and the caller records nothing, reports nothing, and leaves whatever it feeds looking as though the share had simply been empty.

ValueError is broad, so keep the guarded block narrow: build paths outside it and put only the calls into the SMB layer inside.

Source code in src/smbfence/errors.py

@contextmanager
def translated(what: str) -> Iterator[None]:
    """Turn whatever the SMB layer raises into `ShareUnreachableError`.

    Three exception types reach this point and the third is not obvious.
    `smbprotocol` catches the socket error during transport connection and
    re-raises it as a plain `ValueError` naming the server, so a host that does
    not resolve arrives as neither an `OSError` nor an `SMBException`. Without
    this, an unreachable host raises straight past a caller's `except` clause —
    and the caller records nothing, reports nothing, and leaves whatever it
    feeds looking as though the share had simply been empty.

    `ValueError` is broad, so keep the guarded block narrow: build paths
    outside it and put only the calls into the SMB layer inside.
    """
    try:
        yield
    except (OSError, SMBException, ValueError) as exc:
        raise ShareUnreachableError(f"{what}: {exc}") from exc