diff --git a/docs/user/reference/api.md b/docs/user/reference/api.md
index 824c73e..c04f65a 100644
--- a/docs/user/reference/api.md
+++ b/docs/user/reference/api.md
@@ -19,6 +19,14 @@ Anything else, including every name starting with `_`, is internal.
 
 ::: intercede.registry
 
+## Credentials
+
+::: intercede.auth.models
+
+::: intercede.auth.protocols
+
+::: intercede.auth.providers
+
 ## Exceptions
 
 ::: intercede.exceptions
diff --git a/src/intercede/_backends/_htcondor/__init__.py b/src/intercede/_backends/_htcondor/__init__.py
new file mode 100644
index 0000000..743a1fd
--- /dev/null
+++ b/src/intercede/_backends/_htcondor/__init__.py
@@ -0,0 +1,11 @@
+"""What interCEde knows about HTCondor, shared by every backend that drives it.
+
+Plain functions and data only, with no dependency on any backend class: the
+HTCondor-CE backend uses them directly, and the HTCondor scheduler of the
+composed backend (issue #21) will use the same ones over its transport. They
+share these functions, never a subtype.
+
+- ``submit``: a submit description built from a ``SubmissionSpec``
+- ``commands``: the argv of each HTCondor tool, and parsers for their output
+- ``status``: the state mapping from job ads to ``JobStatus``
+"""
diff --git a/src/intercede/_backends/_htcondor/commands.py b/src/intercede/_backends/_htcondor/commands.py
new file mode 100644
index 0000000..8c76276
--- /dev/null
+++ b/src/intercede/_backends/_htcondor/commands.py
@@ -0,0 +1,117 @@
+"""The argv of each HTCondor tool, and parsers for what they print.
+
+``remote`` is how a tool reaches its schedd: ``schedd_options(pool, name)``
+for a remote schedd, as an HTCondor-CE is reached, or ``()`` for the schedd
+of the host the tool runs on, as the composed backend reaches it.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Sequence
+
+from intercede.models import JobID
+
+# Ids per command line, so a status poll of thousands of jobs stays bounded.
+CHUNK = 100
+
+
+def schedd_options(pool: str, name: str) -> list[str]:
+    """Reach the schedd ``name`` through the collector ``pool`` (host:port)."""
+    return ["-pool", pool, "-name", name]
+
+
+def submit(
+    description: str, pool: str | None = None, name: str | None = None
+) -> list[str]:
+    """Submit a description file and spool its inputs to the schedd.
+
+    Spooling copies the inputs at submit time, so nothing on the submitting
+    host is needed afterwards. With ``pool`` and ``name`` the schedd is a
+    remote one (``-remote`` implies spooling); without, it is the local one.
+    """
+    if (pool is None) != (name is None):
+        raise ValueError("a remote schedd needs both pool and name")
+
+    command = ["condor_submit", "-terse"]
+
+    if pool is None or name is None:
+        command.append("-spool")
+    else:
+        command.extend(["-pool", pool, "-remote", name])
+
+    command.append(description)
+    return command
+
+
+def query(
+    ids: Sequence[JobID], attributes: Sequence[str], remote: Sequence[str]
+) -> list[str]:
+    """Read the ads of jobs still in the queue."""
+    return ["condor_q", *remote, *ids, "-attributes", ",".join(attributes), "-json"]
+
+
+def history(
+    ids: Sequence[JobID], attributes: Sequence[str], remote: Sequence[str]
+) -> list[str]:
+    """Read the ads of jobs that have left the queue."""
+    return [
+        "condor_history",
+        *remote,
+        *ids,
+        "-attributes",
+        ",".join(attributes),
+        "-json",
+    ]
+
+
+def remove(ids: Sequence[JobID], remote: Sequence[str]) -> list[str]:
+    """Remove jobs: cancels active ones, releases finished ones."""
+    return ["condor_rm", *remote, *ids]
+
+
+def transfer_data(ids: Sequence[JobID], remote: Sequence[str]) -> list[str]:
+    """Fetch the spooled output of jobs into the directory they were submitted from."""
+    return ["condor_transfer_data", *remote, *ids]
+
+
+def chunks(ids: Sequence[JobID]) -> list[Sequence[JobID]]:
+    """Split ids into groups of at most ``CHUNK``."""
+    return [ids[start : start + CHUNK] for start in range(0, len(ids), CHUNK)]
+
+
+_TERSE = re.compile(
+    r"\s*(?P<cluster>\d+)\.(?P<first>\d+)"
+    r"\s*-\s*"
+    r"(?P<last_cluster>\d+)\.(?P<last>\d+)\s*"
+)
+
+
+def parse_submit(raw: str) -> list[JobID]:
+    """Parse job IDs from condor_submit -terse output, e.g. '12.0 - 12.3'."""
+    lines = [line for line in raw.splitlines() if line.strip()]
+
+    if len(lines) != 1:
+        raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}")
+
+    match = _TERSE.fullmatch(lines[0])
+    if match is None:
+        raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}")
+
+    cluster = int(match["cluster"])
+    first = int(match["first"])
+    last_cluster = int(match["last_cluster"])
+    last = int(match["last"])
+
+    if cluster != last_cluster or last < first:
+        raise ValueError(f"unexpected condor_submit output: {raw.strip()!r}")
+
+    return [JobID(f"{cluster}.{proc}") for proc in range(first, last + 1)]
+
+
+_REMOVED = re.compile(r"^Job (\d+\.\d+) marked for removal", re.MULTILINE)
+
+
+def parse_remove(raw: str) -> set[JobID]:
+    """Return the ids ``condor_rm`` reports as removed; any other id was not."""
+    return {JobID(job) for job in _REMOVED.findall(raw)}
diff --git a/src/intercede/_backends/_htcondor/status.py b/src/intercede/_backends/_htcondor/status.py
new file mode 100644
index 0000000..40d8aca
--- /dev/null
+++ b/src/intercede/_backends/_htcondor/status.py
@@ -0,0 +1,124 @@
+"""From HTCondor job ads to ``JobStatus``.
+
+Reading only: the state is taken from what HTCondor records about a job, never
+from a policy interCEde sets on it. A finished job reads as FAILED when its
+payload exited non-zero or was killed by a signal, from ``ExitCode`` and
+``ExitBySignal``.
+
+Ported from DIRAC's ``BatchSystems/Condor.py`` (``STATES_MAP`` and
+``getCondorStatus``), with differences: DIRAC puts a payload that fails on
+hold to recognise it, where the exit code is read here instead; the states
+DIRAC leaves unmapped (6, 7) are mapped; and a removed job reports what it did
+before it was removed, so a purged job keeps its result.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from intercede.models import JobID, JobStatus
+
+# The ad attributes every status query asks for.
+STATE_ATTRIBUTES: Sequence[str] = (
+    "ClusterId",
+    "ProcId",
+    "JobStatus",
+    "LastJobStatus",
+    "ExitCode",
+    "ExitBySignal",
+    "HoldReasonCode",
+    "HoldReason",
+)
+
+# HTCondor's JobStatus codes.
+_IDLE, _RUNNING, _REMOVED, _COMPLETED, _HELD, _TRANSFERRING, _SUSPENDED = range(1, 8)
+
+_STATES: Mapping[int, JobStatus] = {
+    _IDLE: JobStatus.WAITING,
+    _RUNNING: JobStatus.RUNNING,
+    _TRANSFERRING: JobStatus.RUNNING,  # transferring output after running
+    _SUSPENDED: JobStatus.WAITING,
+}
+
+_HOLD_SPOOLING = 16  # HoldReasonCode: input files are still being spooled
+
+
+def job_id(ad: Mapping[str, Any]) -> JobID:
+    """Return the ``cluster.proc`` id of a job ad."""
+    return JobID(f"{ad['ClusterId']}.{ad['ProcId']}")
+
+
+def job_status(ad: Mapping[str, Any]) -> JobStatus:
+    """Map one job ad to the common status."""
+    code = ad.get("JobStatus")
+    if code == _COMPLETED:
+        return _result(ad)
+    if code == _REMOVED:
+        # purge() removes finished jobs to release them, after which the
+        # history only says "removed": the state it left tells the result.
+        return (
+            _result(ad) if ad.get("LastJobStatus") == _COMPLETED else JobStatus.ABORTED
+        )
+    if code == _HELD:
+        # Held by the site or by HTCondor, not by interCEde: the job is stuck,
+        # except while its inputs are still being spooled.
+        return (
+            JobStatus.WAITING
+            if ad.get("HoldReasonCode") == _HOLD_SPOOLING
+            else JobStatus.ABORTED
+        )
+    return (
+        _STATES.get(code, JobStatus.UNKNOWN)
+        if isinstance(code, int)
+        else JobStatus.UNKNOWN
+    )
+
+
+def is_removed(ad: Mapping[str, Any]) -> bool:
+    """Tell whether a job was removed, even while it is still in the queue.
+
+    ``condor_rm`` only marks a job for removal: the schedd drops it from the
+    queue a moment later. In between it is already released.
+
+    Args:
+        ad: A job ad with ``JobStatus``.
+
+    Returns:
+        True when the job has been removed.
+
+    """
+    return ad.get("JobStatus") == _REMOVED
+
+
+def _result(ad: Mapping[str, Any]) -> JobStatus:
+    """Tell DONE from FAILED for a job whose payload has finished."""
+    if ad.get("ExitBySignal") is True:
+        return JobStatus.FAILED
+    exit_code = ad.get("ExitCode")
+    return (
+        JobStatus.FAILED
+        if isinstance(exit_code, int) and exit_code != 0
+        else JobStatus.DONE
+    )
+
+
+def hold_reason(ad: Mapping[str, Any]) -> str | None:
+    """Return why a held job is held, if it is."""
+    reason = ad.get("HoldReason")
+    return reason if ad.get("JobStatus") == _HELD and isinstance(reason, str) else None
+
+
+def parse_ads(raw: str) -> list[dict[str, Any]]:
+    """Parse ``condor_q -json`` or ``condor_history -json`` output.
+
+    ``condor_q`` prints nothing at all when no job matches, and
+    ``condor_history`` prints an empty list.
+    """
+    if not raw.strip():
+        return []
+    ads = json.loads(raw)
+    if not isinstance(ads, list) or not all(isinstance(ad, dict) for ad in ads):
+        raise ValueError("expected a JSON list of job ads")
+    return ads
diff --git a/src/intercede/_backends/_htcondor/submit.py b/src/intercede/_backends/_htcondor/submit.py
new file mode 100644
index 0000000..0fa5a0f
--- /dev/null
+++ b/src/intercede/_backends/_htcondor/submit.py
@@ -0,0 +1,219 @@
+"""Submit descriptions: a ``SubmissionSpec`` written in HTCondor's submit language.
+
+One description submits every copy of a spec as one cluster, one process per
+copy. What differs between copies (environment, secrets, working directory) is
+expanded per process through macros indexed by ``$(ProcId)``, so the copies
+stay one bulk submission.
+
+A description only says what to run and where its files go. It sets no
+policy on the job's lifetime: no holds, no automatic removal, no retries.
+Those are the consumer's decisions. The one exception is that a finished job
+stays in the queue until it is purged, so its output can be collected again.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
+
+
+def quote_arguments(arguments: Sequence[str]) -> str:
+    """Quote command arguments using HTCondor's new syntax."""
+    quoted_arguments = [_quote_item(argument) for argument in arguments]
+
+    joined_arguments = " ".join(quoted_arguments)
+    return f'"{joined_arguments}"'
+
+
+def quote_environment(environment: Mapping[str, str]) -> str:
+    """Quote environment variables using HTCondor's new syntax."""
+    for name in environment:
+        _validate_environment_name(name)
+
+    quoted_variables = [
+        f"{name}={_quote_item(value)}" for name, value in sorted(environment.items())
+    ]
+
+    joined_variables = " ".join(quoted_variables)
+    return f'"{joined_variables}"'
+
+
+def _validate_environment_name(name: str) -> None:
+    """Raise ValueError if the environment variable name is invalid."""
+    is_valid = bool(name) and "=" not in name and name.replace("_", "").isalnum()
+
+    if not is_valid:
+        raise ValueError(f"{name!r} is not a valid environment variable name")
+
+
+def _quote_item(value: str) -> str:
+    """Escape and quote a single argument or environment variable value."""
+    _refuse_newlines(value)
+
+    escaped_value = value.replace('"', '""')
+    escaped_value = escaped_value.replace("'", "''")
+    escaped_value = _escape_macros(escaped_value)
+
+    return f"'{escaped_value}'"
+
+
+def _escape_macros(value: str) -> str:
+    # "$(...)" is macro syntax in a submit description; $(DOLLAR) is a literal "$".
+    return value.replace("$", "$(DOLLAR)")
+
+
+def _refuse_newlines(value: str) -> None:
+    if "\n" in value or "\r" in value:
+        raise ValueError("a submit description value cannot contain a line break")
+
+
+@dataclass(frozen=True)
+class CopyPlan:
+    """What one process of the cluster gets on its own."""
+
+    initialdir: str  # absolute; outputs come back here (SUBMIT_Iwd)
+    environment: Mapping[str, str] = field(default_factory=dict)
+    # Inputs of this copy only, relative to ``initialdir`` (e.g. "secrets/").
+    inputs: Sequence[str] = ()
+
+
+@dataclass(frozen=True)
+class SubmitPlan:
+    """Everything a submit description says, resolved to HTCondor's terms."""
+
+    executable: str
+    transfer_executable: bool
+    arguments: Sequence[str]
+    copies: Sequence[CopyPlan]
+    inputs: Sequence[str] = ()  # shared by every copy: absolute paths or URLs
+    stdout: str = "$(ClusterId).$(ProcId).out"
+    stderr: str = "$(ClusterId).$(ProcId).err"
+    log: str = "$(ClusterId).$(ProcId).log"
+    # None: every file the payload creates comes back; empty: none at all.
+    outputs: Sequence[str] | None = ()
+    output_remaps: Mapping[str, str] = field(default_factory=dict)  # name -> URL
+    cpus: int = 1
+    memory_mb: int | None = None
+    wall_time_s: int | None = None
+    queue: str | None = None
+
+
+def render(plan: SubmitPlan) -> str:
+    """Write ``plan`` as a submit description that queues one process per copy."""
+    _validate_plan(plan)
+
+    lines = _render_basics(plan)
+    lines.extend(_render_copies(plan))
+    lines.extend(_render_file_transfer(plan))
+    lines.extend(_render_resources(plan))
+    lines.extend(_render_queue(plan))
+
+    return "\n".join(lines) + "\n"
+
+
+def _join(items: Sequence[str]) -> str:
+    for item in items:
+        _refuse_newlines(item)
+        if "," in item:
+            raise ValueError(f"{item!r}: HTCondor file lists cannot hold a comma")
+    return ", ".join(_escape_macros(item) for item in items)
+
+
+def _bool(value: bool) -> str:
+    return "true" if value else "false"
+
+
+def _classad_string(value: str) -> str:
+    _refuse_newlines(value)
+    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
+    return '"' + _escape_macros(escaped) + '"'
+
+
+def _validate_plan(plan: SubmitPlan) -> None:
+    if not plan.copies:
+        raise ValueError("A submit description needs at least one copy")
+
+
+def _render_basics(plan: SubmitPlan) -> list[str]:
+    lines = [
+        "universe = vanilla",
+        f"executable = {_escape_macros(plan.executable)}",
+        f"transfer_executable = {_bool(plan.transfer_executable)}",
+    ]
+
+    if plan.arguments:
+        lines.append(f"arguments = {quote_arguments(plan.arguments)}")
+
+    return lines
+
+
+def _render_copies(plan: SubmitPlan) -> list[str]:
+    lines = []
+
+    for index, copy in enumerate(plan.copies):
+        _refuse_newlines(copy.initialdir)
+        lines.extend(
+            [
+                f"ic_dir_{index} = {_escape_macros(copy.initialdir)}",
+                f"ic_env_{index} = {quote_environment(copy.environment)}",
+                f"ic_in_{index} = {_join(copy.inputs)}",
+            ]
+        )
+
+    lines.extend(
+        ["initialdir = $(ic_dir_$(ProcId))", "environment = $(ic_env_$(ProcId))"]
+    )
+
+    return lines
+
+
+def _render_file_transfer(plan: SubmitPlan) -> list[str]:
+    shared_inputs = _join(plan.inputs)
+    copy_inputs = "$(ic_in_$(ProcId))"
+    inputs = f"{shared_inputs}, {copy_inputs}" if shared_inputs else copy_inputs
+
+    lines = [
+        "should_transfer_files = YES",
+        "when_to_transfer_output = ON_EXIT_OR_EVICT",
+        f"transfer_input_files = {inputs}",
+        f"output = {plan.stdout}",
+        f"error = {plan.stderr}",
+        f"log = {plan.log}",
+    ]
+
+    if plan.outputs is not None:
+        output_files = _join(plan.outputs) or '""'
+        lines.append(f"transfer_output_files = {output_files}")
+
+    if plan.output_remaps:
+        remaps = "; ".join(
+            f"{name} = {url}" for name, url in sorted(plan.output_remaps.items())
+        )
+        lines.append(f'transfer_output_remaps = "{_escape_macros(remaps)}"')
+
+    return lines
+
+
+def _render_resources(plan: SubmitPlan) -> list[str]:
+    lines = [f"request_cpus = {plan.cpus}", f"+xcount = {plan.cpus}"]
+
+    if plan.memory_mb is not None:
+        lines.extend(
+            [f"request_memory = {plan.memory_mb}", f"+maxMemory = {plan.memory_mb}"]
+        )
+
+    if plan.wall_time_s is not None:
+        lines.append(f"+BatchRuntime = {plan.wall_time_s}")
+
+    if plan.queue is not None:
+        lines.append(f"+queue = {_classad_string(plan.queue)}")
+
+    return lines
+
+
+def _render_queue(plan: SubmitPlan) -> list[str]:
+    return [
+        # Keep completed jobs until purge() releases them
+        "leave_in_queue = (JobStatus == 4)",
+        f"queue {len(plan.copies)}",
+    ]
diff --git a/src/intercede/_backends/credentials.py b/src/intercede/_backends/credentials.py
new file mode 100644
index 0000000..05a68e3
--- /dev/null
+++ b/src/intercede/_backends/credentials.py
@@ -0,0 +1,192 @@
+"""What a backend asks its credential provider for, when, and what it accepts back.
+
+``credential_requirements`` turns a configuration into what to ask for. This is
+also the one place freshness is decided. A backend holds a ``CredentialCache``,
+and before each operation asks it for that operation's credential: the provider
+is asked only when there is none yet or it is about to expire. Where credentials
+come from, and how they are renewed, stays the consumer's business, inside its
+provider.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from intercede.auth.models import (
+    BearerToken,
+    Credential,
+    CredentialRequirements,
+    SSHPassword,
+    SSHPrivateKey,
+)
+from intercede.auth.protocols import CredentialProvider
+from intercede.config import (
+    ARCConfig,
+    BackendConfig,
+    BatchSSHConfig,
+    HTCondorCEConfig,
+)
+from intercede.exceptions import AuthenticationError, InterCEdeError
+
+# What each configured SSH auth mode asks the credential provider for.
+# "agent" is absent on purpose: an SSH agent holds the key, so nothing is asked.
+AUTH_KINDS: dict[str, type[Credential]] = {
+    "key": SSHPrivateKey,
+    "password": SSHPassword,
+}
+
+# Ask again this long before a credential expires, so it cannot run out in the
+# middle of an operation. A library default; whether a provider should be able
+# to change it is still open.
+DEFAULT_MARGIN = timedelta(minutes=5)
+
+
+def credential_requirements(
+    config: BackendConfig, scopes: frozenset[str] = frozenset()
+) -> CredentialRequirements | None:
+    """Say what a backend must ask its provider for.
+
+    Args:
+        config: The backend's validated configuration.
+        scopes: The scopes of the operation about to run, for token-based
+            backends; ignored by the others.
+
+    Returns:
+        What to ask for, or None when the backend needs no credential: it runs
+        on this host, or an SSH agent holds the key.
+    """
+    if isinstance(config, ARCConfig | HTCondorCEConfig):
+        # The endpoint is the resource, so a token is minted for this CE's audience.
+        return CredentialRequirements(
+            resource=config.endpoint,
+            kind=BearerToken,
+            scopes=scopes,
+        )
+    if isinstance(config, BatchSSHConfig) and config.auth != "agent":
+        # The resource is the login, ``user@host:port``: a provider finds the key
+        # or password by it, and the same host can take different users.
+        return CredentialRequirements(
+            resource=f"{config.user}@{config.host}:{config.port}",
+            kind=AUTH_KINDS[config.auth],
+        )
+    return None
+
+
+def is_stale(
+    credential: Credential | None,
+    requirements: CredentialRequirements,
+    *,
+    margin: timedelta = DEFAULT_MARGIN,
+    now: datetime | None = None,
+) -> bool:
+    """Tell whether ``credential`` cannot serve ``requirements`` for ``margin`` more.
+
+    A credential that never expires is never stale. Only the kind and the expiry
+    are checked, not the scopes, since a token is opaque to interCEde: ``credential``
+    must have been given for these same requirements, as ``CredentialCache`` ensures.
+
+    Args:
+        credential: The one the backend holds, or None before its first request.
+        requirements: What the backend needs for the operation about to run.
+        margin: How long the credential must stay valid from now.
+        now: The current time; tests pass a fixed one.
+
+    Returns:
+        True when there is no credential yet, when it is of another kind than
+        required, or when it expires within ``margin``.
+    """
+    if not isinstance(credential, requirements.kind):
+        return True
+    if credential.expires_at is None:
+        return False
+    return credential.expires_at - margin <= (now or datetime.now(UTC))
+
+
+async def refresh_if_stale(
+    current: Credential | None,
+    requirements: CredentialRequirements,
+    provider: CredentialProvider,
+    *,
+    margin: timedelta = DEFAULT_MARGIN,
+    now: datetime | None = None,
+) -> Credential:
+    """Return ``current`` while it is fresh, otherwise a new one from ``provider``.
+
+    Args:
+        current: The credential the backend holds, or None before its first request.
+        requirements: What the backend needs for the operation about to run.
+        provider: The consumer's provider, asked only when ``current`` is stale.
+        margin: How long the credential must stay valid from now.
+        now: The current time; tests pass a fixed one.
+
+    Returns:
+        The credential to authenticate with: ``current``, or the provider's new one.
+
+    Raises:
+        AuthenticationError: The provider failed, returned something other than
+            the required kind, or returned a credential that has already expired.
+            Always a failure of the whole operation, never a per-job outcome.
+    """
+    if not is_stale(current, requirements, margin=margin, now=now):
+        assert current is not None  # is_stale() is True for None
+        return current
+
+    try:
+        fresh = await provider.get(requirements)
+    except InterCEdeError:
+        raise  # already typed, e.g. an AuthenticationError from the provider
+    except Exception as exc:
+        # Only the type: the provider's message may hold a token or a secret URL.
+        # The exception itself stays available as __cause__.
+        raise AuthenticationError(
+            f"the credential provider failed for {requirements.resource} "
+            f"with {type(exc).__name__}"
+        ) from exc
+
+    if not isinstance(fresh, requirements.kind):
+        raise AuthenticationError(
+            f"the credential provider returned a {type(fresh).__name__} "
+            f"for {requirements.resource}, not a {requirements.kind.__name__}"
+        )
+    # Expiring within the margin is accepted (it is asked again next time);
+    # already expired is not, since it cannot authenticate anything.
+    if is_stale(fresh, requirements, margin=timedelta(0), now=now):
+        raise AuthenticationError(
+            f"the credential provider returned an expired credential "
+            f"for {requirements.resource}"
+        )
+    return fresh
+
+
+class CredentialCache:
+    """Keeps one credential per requirements for one backend, so each scope gets its own token.
+
+    The provider is asked again only when that credential is missing or about to expire.
+    """
+
+    def __init__(self, provider: CredentialProvider) -> None:
+        self._provider = provider
+        self._held: dict[CredentialRequirements, Credential] = {}
+
+    async def fresh(self, requirements: CredentialRequirements) -> Credential:
+        """Return the credential for these requirements, asking the provider if needed.
+
+        Args:
+            requirements: What the backend needs for the operation about to run.
+
+        Returns:
+            The credential to authenticate with.
+
+        Raises:
+            AuthenticationError: The provider failed, returned something other than
+                the required kind, or returned a credential that has already expired.
+        """
+        credential = await refresh_if_stale(
+            self._held.get(requirements), requirements, self._provider
+        )
+        self._held[requirements] = credential
+        return credential
+
+    def clear(self) -> None:
+        """Forget every credential, as a backend does when it closes."""
+        self._held.clear()
diff --git a/src/intercede/_backends/fake.py b/src/intercede/_backends/fake.py
index 53c4d59..609fb37 100644
--- a/src/intercede/_backends/fake.py
+++ b/src/intercede/_backends/fake.py
@@ -13,6 +13,7 @@
 from typing import Self
 
 from intercede._backends.config import FakeConfig
+from intercede.auth.protocols import CredentialProvider
 from intercede.exceptions import (
     InterCEdeError,
     ResourceUnavailableError,
@@ -94,11 +95,15 @@ def __init__(
         self._closed = False
 
     @classmethod
-    def from_settings(cls, settings: FakeConfig) -> Self:
+    def from_settings(
+        cls, settings: FakeConfig, credentials: CredentialProvider
+    ) -> Self:
         """Build the fake from its settings; how the registry constructs it.
 
         Args:
             settings: The fake's settings: only the resource it reports.
+            credentials: Never asked: nothing leaves the process, so the fake
+                needs no credentials.
 
         Returns:
             A new fake, not yet opened.
@@ -379,4 +384,6 @@ def _write_job_files(
 _kill: type[Killable] = FakeBackend
 _purge: type[Purgeable] = FakeBackend
 _diag: type[Diagnosable] = FakeBackend
-_built_by_registry: Callable[[FakeConfig], JobBackend] = FakeBackend.from_settings
+_built_by_registry: Callable[[FakeConfig, CredentialProvider], JobBackend] = (
+    FakeBackend.from_settings
+)
diff --git a/src/intercede/auth/__init__.py b/src/intercede/auth/__init__.py
new file mode 100644
index 0000000..0b886c3
--- /dev/null
+++ b/src/intercede/auth/__init__.py
@@ -0,0 +1,28 @@
+"""Backend credentials: typed, declared, supplied by a provider.
+
+A consumer implements a ``CredentialProvider``, which returns the one credential
+a backend asks for. Configuration never holds a secret: it only says how a backend
+logs in, and the provider supplies the credential at run time.
+"""
+
+from __future__ import annotations
+
+from intercede.auth.models import (
+    BearerToken,
+    Credential,
+    CredentialRequirements,
+    SSHPassword,
+    SSHPrivateKey,
+)
+from intercede.auth.protocols import CredentialProvider
+from intercede.auth.providers import StaticCredentials
+
+__all__ = [
+    "BearerToken",
+    "Credential",
+    "CredentialProvider",
+    "CredentialRequirements",
+    "SSHPassword",
+    "SSHPrivateKey",
+    "StaticCredentials",
+]
diff --git a/src/intercede/auth/models.py b/src/intercede/auth/models.py
new file mode 100644
index 0000000..868c5b9
--- /dev/null
+++ b/src/intercede/auth/models.py
@@ -0,0 +1,106 @@
+"""The public credential models a consumer of interCEde can use.
+
+Credentials are immutable values. Secret material is held in pydantic's secret
+types, so a credential cannot be printed or logged by accident: ``repr``,
+``str`` and JSON all show ``**********``, and the content is only readable on
+purpose, with ``get_secret_value()``.
+"""
+
+from __future__ import annotations
+
+from typing import Self
+
+from pydantic import (
+    AwareDatetime,
+    ConfigDict,
+    SecretBytes,
+    SecretStr,
+    model_validator,
+)
+
+from intercede.models import _Model
+
+__all__ = [
+    "BearerToken",
+    "Credential",
+    "CredentialRequirements",
+    "SSHPassword",
+    "SSHPrivateKey",
+]
+
+
+# --------------------------------------------------------------------------
+# Credential values
+# --------------------------------------------------------------------------
+
+
+class Credential(_Model):
+    """Base of every credential: the one field the freshness check needs.
+
+    Not used on its own: a provider returns one of the concrete kinds below.
+    """
+
+    # Added to _Model's settings: a value given by mistake may be the secret
+    # itself, so no error message repeats one.
+    model_config = ConfigDict(hide_input_in_errors=True, use_attribute_docstrings=True)
+
+    expires_at: AwareDatetime | None = None
+    """When the credential stops working, with a timezone, so it can be compared
+    with the current time. None means it never expires, as for SSH secrets."""
+
+
+class BearerToken(Credential):
+    """A token for the channel to a CE. Opaque to interCEde."""
+
+    value: SecretStr
+    """The token itself."""
+
+
+class SSHPrivateKey(Credential):
+    """An SSH private key: the key itself, never a path on some host."""
+
+    pem: SecretBytes
+    """The private key, PEM-encoded."""
+
+    passphrase: SecretStr | None = None
+    """The passphrase the key is encrypted with, or None if it is not."""
+
+
+class SSHPassword(Credential):
+    """A password for an SSH login."""
+
+    value: SecretStr
+    """The password itself."""
+
+
+# --------------------------------------------------------------------------
+# What a backend asks for
+# --------------------------------------------------------------------------
+
+
+class CredentialRequirements(_Model):
+    """What a backend asks its provider for.
+
+    The backend computes it from its own configuration and keeps it to itself:
+    a consumer only meets it as the argument to its provider's ``get``.
+    """
+
+    model_config = ConfigDict(use_attribute_docstrings=True)
+
+    resource: str
+    """Which backend instance is asking: the endpoint it talks to, the same
+    string as ``JobHandle.resource``. For a token this is also the audience,
+    in the form that CE expects."""
+
+    kind: type[Credential]
+    """The kind of credential needed: a backend authenticates with exactly one."""
+
+    scopes: frozenset[str] = frozenset()
+    """For a token, the scopes of the operation about to run; empty for the
+    other kinds."""
+
+    @model_validator(mode="after")
+    def _concrete_kind(self) -> Self:
+        if self.kind is Credential:
+            raise ValueError("kind must be a concrete kind, not the Credential base")
+        return self
diff --git a/src/intercede/auth/protocols.py b/src/intercede/auth/protocols.py
new file mode 100644
index 0000000..077b15c
--- /dev/null
+++ b/src/intercede/auth/protocols.py
@@ -0,0 +1,49 @@
+"""The public protocol for a ``CredentialProvider``."""
+
+from __future__ import annotations
+
+from types import TracebackType
+from typing import Protocol, Self, runtime_checkable
+
+from intercede.auth.models import Credential, CredentialRequirements
+
+__all__ = ["CredentialProvider"]
+
+
+@runtime_checkable
+class CredentialProvider(Protocol):
+    """The protocol any consumer's own credential provider must satisfy.
+
+    The consumer creates the provider, shares it between backends and closes
+    it: interCEde never closes a provider, because it did not create it. A
+    backend calls ``get`` only when its credential for those requirements is
+    missing or close to expiry, so the consumer decides where credentials come
+    from and interCEde decides when to ask. Requirements with other scopes get a
+    credential of their own.
+    """
+
+    async def get(self, requirements: CredentialRequirements) -> Credential:
+        """Return the credential a backend needs.
+
+        Args:
+            requirements: The resource, the kind of credential it takes, and
+                the scopes of the operation about to run.
+
+        Returns:
+            One credential of ``requirements.kind``.
+
+        Raises:
+            Exception: Anything raised here fails the whole operation: the
+                backend reports it as an ``AuthenticationError``, never as a
+                per-job outcome.
+        """
+        ...
+
+    async def __aenter__(self) -> Self: ...
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None: ...
diff --git a/src/intercede/auth/providers.py b/src/intercede/auth/providers.py
new file mode 100644
index 0000000..f6c2fe1
--- /dev/null
+++ b/src/intercede/auth/providers.py
@@ -0,0 +1,77 @@
+"""Ready-made credential providers."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from types import TracebackType
+from typing import Self
+
+from intercede.auth.models import Credential, CredentialRequirements
+from intercede.auth.protocols import CredentialProvider
+from intercede.exceptions import AuthenticationError
+
+__all__ = ["StaticCredentials"]
+
+
+class StaticCredentials:
+    """A provider holding fixed credentials: files read once, or fixed values.
+
+    For standalone use, tests and the integration stacks, where nothing issues
+    or renews credentials. It holds at most one credential of each kind, since
+    one provider can serve several backends, and each backend receives only the
+    one it asks for. It holds nothing to release, so closing does nothing, but
+    it is still a context manager, like every provider::
+
+        async with StaticCredentials([SSHPrivateKey(pem=key)]) as credentials:
+            backend = registry.get_backend(config, credentials)
+    """
+
+    def __init__(self, credentials: Iterable[Credential] = ()) -> None:
+        # Validated now, so a duplicate kind or plain data fails at construction.
+        self._credentials: dict[type[Credential], Credential] = {}
+        for credential in credentials:
+            # A plain dict is ambiguous: {"value": ...} fits both a BearerToken
+            # and an SSHPassword. Guessing the kind of a secret is never right.
+            # Never repeat the item in the message: it may be the secret itself.
+            if not isinstance(credential, Credential) or type(credential) is Credential:
+                raise ValueError(
+                    "StaticCredentials holds concrete credential objects, "
+                    "e.g. BearerToken(value=...), not plain data"
+                )
+            kind = type(credential)
+            if kind in self._credentials:
+                raise ValueError(f"more than one credential of kind {kind.__name__}")
+            self._credentials[kind] = credential
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        return None
+
+    async def get(self, requirements: CredentialRequirements) -> Credential:
+        """Return the held credential of the requested kind.
+
+        Args:
+            requirements: What the backend asks for.
+
+        Returns:
+            The held credential of ``requirements.kind``.
+
+        Raises:
+            AuthenticationError: It holds no credential of that kind.
+        """
+        credential = self._credentials.get(requirements.kind)
+        if credential is None:
+            raise AuthenticationError(
+                f"no {requirements.kind.__name__} for {requirements.resource}"
+            )
+        return credential
+
+
+_provider: type[CredentialProvider] = StaticCredentials
diff --git a/src/intercede/config.py b/src/intercede/config.py
index e9be28a..8724d57 100644
--- a/src/intercede/config.py
+++ b/src/intercede/config.py
@@ -16,7 +16,7 @@
     known_hosts = "batch.example.org ssh-ed25519 AAAA..."
 
 Configuration describes the resource and is never secret: keys, passwords,
-tokens and proxies come from the credential provider at run time. Error
+and tokens come from the credential provider at run time. Error
 messages therefore never repeat the values they were given. The kinds are a
 closed set: they list what interCEde ships.
 """
@@ -101,16 +101,6 @@ class ARCConfig(_QueueConfig):
     endpoint: str
     """The CE's REST endpoint, e.g. ``"https://arc.example.org:443/arex"``."""
 
-    auth: frozenset[Literal["token", "proxy"]] = frozenset({"token"})
-    """Which credentials the provider must supply, all of them together: a token
-    for the channel, a proxy for delegation, or both."""
-
-    @model_validator(mode="after")
-    def _auth_not_empty(self) -> Self:
-        if not self.auth:
-            raise ValueError("auth must name at least one of 'token', 'proxy'")
-        return self
-
 
 class HTCondorCEConfig(_QueueConfig):
     """An HTCondor-CE."""
@@ -121,18 +111,6 @@ class HTCondorCEConfig(_QueueConfig):
     endpoint: str
     """The CE's collector, as ``"host"`` or ``"host:port"``, e.g. ``"ce.example.org:9619"``."""
 
-    auth: frozenset[Literal["token", "proxy"]] = frozenset({"token"})
-    """Which credentials the provider must supply. The token authenticates the
-    channel; ``"proxy"`` adds an X.509 proxy for sites that account on VOMS attributes."""
-
-    @model_validator(mode="after")
-    def _token_required(self) -> Self:
-        if "token" not in self.auth:
-            raise ValueError(
-                "an HTCondor-CE authenticates with a token; auth must include 'token'"
-            )
-        return self
-
 
 # -- a cloud provider that boots one VM per job --------------------------------
 
diff --git a/src/intercede/registry.py b/src/intercede/registry.py
index 063afe1..6acf729 100644
--- a/src/intercede/registry.py
+++ b/src/intercede/registry.py
@@ -7,6 +7,8 @@
 
 Every registered class provides a ``from_settings`` classmethod, which is how
 the registry builds it from a validated configuration model.
+The parts that authenticate (single-class backends and transports) also
+receive the consumer's credential provider; a scheduler only builds commands.
 What it builds is checked against the relevant protocol before it is returned,
 so a broken entry fails here, with a clear error, and not at the first call.
 """
@@ -20,6 +22,7 @@
 
 from intercede._backends.batch.protocols import Scheduler, Transport
 from intercede._backends.config import FakeConfig
+from intercede.auth.protocols import CredentialProvider
 from intercede.config import (
     _MODELS as _CONFIG_MODELS,
 )
@@ -29,7 +32,7 @@
     BatchSSHConfig,
     validate_config,
 )
-from intercede.exceptions import RegistryError
+from intercede.exceptions import ConfigurationError, RegistryError
 from intercede.protocols import JobBackend
 
 __all__ = ["available", "get_backend"]
@@ -53,25 +56,39 @@
 }
 
 
-def get_backend(config: BackendConfig | Mapping[str, Any]) -> JobBackend:
+def get_backend(
+    config: BackendConfig | Mapping[str, Any], credentials: CredentialProvider
+) -> JobBackend:
     """Build the backend a configuration describes.
 
-    The backend is returned unopened: use it as ``async with registry.get_backend(config) as ce``.
-    Construction does no I/O, so building a backend for an unreachable resource succeeds.
+    The backend is returned unopened: use it as
+    ``async with registry.get_backend(config, credentials) as ce``. Construction
+    does no I/O and asks the provider for nothing, so building a backend for an
+    unreachable resource succeeds. The backend asks for credentials when it
+    first needs them, and again when they are about to expire.
 
     Args:
         config: A ``BackendConfig``, or the same data as a mapping, for example
             a parsed TOML table.
+        credentials: The consumer's provider. The consumer opens and closes it;
+            interCEde never does. One provider is normally shared by every
+            backend a task drives.
 
     Returns:
         The backend, not yet opened.
 
     Raises:
-        ConfigurationError: The configuration is invalid.
+        ConfigurationError: The configuration is invalid, or ``credentials``
+            is not a ``CredentialProvider``.
         RegistryError: The configuration is valid, but this version of
             interCEde does not provide what it asks for.
     """
-    return _build(validate_config(config))
+    validated = validate_config(config)
+    if not isinstance(credentials, CredentialProvider):
+        raise ConfigurationError(
+            f"credentials must be a CredentialProvider, not a {type(credentials).__name__}"
+        )
+    return _build(validated, credentials)
 
 
 def available() -> dict[str, tuple[str, ...]]:
@@ -123,11 +140,14 @@ def _choices(annotation: Any) -> set[str]:
     return values
 
 
-def _build(config: BackendConfig | FakeConfig) -> JobBackend:
+def _build(
+    config: BackendConfig | FakeConfig, credentials: CredentialProvider
+) -> JobBackend:
     """Build a backend from a validated configuration. Internal: the test suites call it directly.
 
     Args:
         config: The validated configuration of one backend, or the fake's settings.
+        credentials: The consumer's provider, for the parts that authenticate.
 
     Returns:
         The backend, not yet opened. A ``batch`` backend is composed of the
@@ -139,14 +159,14 @@ def _build(config: BackendConfig | FakeConfig) -> JobBackend:
             what was built does not satisfy its protocol.
     """
     if not isinstance(config, BatchLocalConfig | BatchSSHConfig):
-        built = _load("backend", config.kind).from_settings(config)
+        built = _load("backend", config.kind).from_settings(config, credentials)
         result: JobBackend = _checked("backend", config.kind, built, JobBackend)
         return result
 
     transport: Transport = _checked(
         "transport",
         config.transport,
-        _load("transport", config.transport).from_settings(config),
+        _load("transport", config.transport).from_settings(config, credentials),
         Transport,
     )
     scheduler: Scheduler = _checked(
diff --git a/tests/integration/stacks/_template/config/basic/intercede-client.toml b/tests/integration/stacks/_template/config/basic/intercede-client.toml
index 785b352..54f2be0 100644
--- a/tests/integration/stacks/_template/config/basic/intercede-client.toml
+++ b/tests/integration/stacks/_template/config/basic/intercede-client.toml
@@ -9,5 +9,5 @@ endpoint = "CHANGEME" # e.g. "https://localhost:443/arex" or "localhost:2222"
 queue = "CHANGEME"
 
 [credentials]
-type = "CHANGEME"              # token | proxy | ssh-key
+type = "CHANGEME"              # token | ssh-key
 path = "credentials/CHANGEME/" # under the ./credentials bind mount
diff --git a/tests/unit/_registry_dummies.py b/tests/unit/_registry_dummies.py
index 5dc02fe..345b4eb 100644
--- a/tests/unit/_registry_dummies.py
+++ b/tests/unit/_registry_dummies.py
@@ -19,10 +19,13 @@ async def __aexit__(self, *exc_info: object) -> None: ...
 class DummyTransport(_Lifecycle):
     def __init__(self, settings: Any) -> None:
         self.settings = settings
+        self.credentials: Any = None
 
     @classmethod
-    def from_settings(cls, settings: Any) -> DummyTransport:
-        return cls(settings)
+    def from_settings(cls, settings: Any, credentials: Any) -> DummyTransport:
+        built = cls(settings)
+        built.credentials = credentials
+        return built
 
     async def run(self, argv: Any) -> Any: ...
 
@@ -74,10 +77,13 @@ class DummyBackend(_Lifecycle):
 
     def __init__(self, settings: Any) -> None:
         self.settings = settings
+        self.credentials: Any = None
 
     @classmethod
-    def from_settings(cls, settings: Any) -> DummyBackend:
-        return cls(settings)
+    def from_settings(cls, settings: Any, credentials: Any) -> DummyBackend:
+        built = cls(settings)
+        built.credentials = credentials
+        return built
 
     async def submit(self, spec: Any, copies: Any = 1) -> Any: ...
 
diff --git a/tests/unit/backends/__init__.py b/tests/unit/backends/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/unit/backends/_htcondor/__init__.py b/tests/unit/backends/_htcondor/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/unit/backends/_htcondor/test_core.py b/tests/unit/backends/_htcondor/test_core.py
new file mode 100644
index 0000000..185d2e6
--- /dev/null
+++ b/tests/unit/backends/_htcondor/test_core.py
@@ -0,0 +1,360 @@
+"""The shared HTCondor knowledge: submit descriptions, commands and the state mapping.
+
+Expected values were checked against HTCondor 25 (condor_submit -dry-run and a
+live schedd): what is asserted here is what HTCondor actually does with them.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from intercede import JobStatus
+from intercede._backends._htcondor import commands, status, submit
+from intercede._backends._htcondor.submit import CopyPlan, SubmitPlan, render
+
+# --------------------------------------------------------------------------
+# Quoting
+# --------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("arguments", "quoted"),
+    [
+        (["plain"], "\"'plain'\""),
+        (["two words"], "\"'two words'\""),
+        (["it's"], "\"'it''s'\""),
+        (['say "hi"'], '"\'say ""hi""\'"'),
+        (["cost $(HOME) $5"], "\"'cost $(DOLLAR)(HOME) $(DOLLAR)5'\""),
+        ([""], "\"''\""),
+        (["a", "b"], "\"'a' 'b'\""),
+    ],
+    ids=[
+        "plain",
+        "space",
+        "single-quote",
+        "double-quote",
+        "dollar",
+        "empty",
+        "several",
+    ],
+)
+def test_quote_arguments(arguments: list[str], quoted: str):
+    assert submit.quote_arguments(arguments) == quoted
+
+
+def test_quote_environment_sorts_and_quotes():
+    quoted = submit.quote_environment({"STAMP": "a b", "X": "$(Y)", "A": "it's"})
+    assert quoted == "\"A='it''s' STAMP='a b' X='$(DOLLAR)(Y)'\""
+
+
+@pytest.mark.parametrize("name", ["", "A=B", "WITH SPACE", "DASH-ED", "$X"])
+def test_environment_names_must_be_identifiers(name: str):
+    with pytest.raises(ValueError, match="not a valid environment variable name"):
+        submit.quote_environment({name: "x"})
+
+
+@pytest.mark.parametrize("value", ["line\nbreak", "carriage\rreturn"])
+def test_line_breaks_are_refused(value: str):
+    with pytest.raises(ValueError, match="line break"):
+        submit.quote_arguments([value])
+    with pytest.raises(ValueError, match="line break"):
+        submit.quote_environment({"X": value})
+
+
+# --------------------------------------------------------------------------
+# Submit descriptions
+# --------------------------------------------------------------------------
+
+
+def plan(**changes: Any) -> SubmitPlan:
+    values: dict[str, Any] = {
+        "executable": "/work/payload.sh",
+        "transfer_executable": True,
+        "arguments": ["a"],
+        "copies": [
+            CopyPlan("/work/0", {"STAMP": "s0"}, ["secrets/"]),
+            CopyPlan("/work/1", {"STAMP": "s1"}),
+        ],
+    }
+    return SubmitPlan(**{**values, **changes})
+
+
+def lines(text: str) -> list[str]:
+    return text.splitlines()
+
+
+class TestRender:
+    def test_one_process_per_copy(self):
+        assert lines(render(plan()))[-1] == "queue 2"
+
+    def test_per_copy_values_are_indexed_by_proc_id(self):
+        text = render(plan())
+        assert "ic_dir_0 = /work/0" in lines(text)
+        assert "ic_dir_1 = /work/1" in lines(text)
+        assert "ic_env_0 = \"STAMP='s0'\"" in lines(text)
+        assert "ic_in_0 = secrets/" in lines(text)
+        assert "ic_in_1 = " in lines(text)
+        assert "initialdir = $(ic_dir_$(ProcId))" in lines(text)
+        assert "environment = $(ic_env_$(ProcId))" in lines(text)
+
+    def test_inputs_shared_and_per_copy(self):
+        text = render(plan(inputs=["/work/inputs/", "https://example.org/in.dat"]))
+        assert (
+            "transfer_input_files = /work/inputs/, https://example.org/in.dat, $(ic_in_$(ProcId))"
+            in lines(text)
+        )
+
+    @pytest.mark.parametrize(
+        ("outputs", "expected"),
+        [
+            (
+                [],
+                'transfer_output_files = ""',
+            ),  # nothing but stdout, stderr and the log
+            (["result.txt", "dir/x"], "transfer_output_files = result.txt, dir/x"),
+            (None, None),  # every file the payload creates
+        ],
+        ids=["none", "listed", "all"],
+    )
+    def test_outputs(self, outputs: list[str] | None, expected: str | None):
+        text = render(plan(outputs=outputs))
+        found = [
+            line for line in lines(text) if line.startswith("transfer_output_files")
+        ]
+        assert found == ([expected] if expected else [])
+
+    def test_output_remaps_upload_to_urls(self):
+        text = render(
+            plan(
+                outputs=["a", "b"],
+                output_remaps={"b": "https://s/b", "a": "https://s/a"},
+            )
+        )
+        assert 'transfer_output_remaps = "a = https://s/a; b = https://s/b"' in lines(
+            text
+        )
+
+    def test_resources_for_a_schedd_and_for_a_ce_router(self):
+        text = render(plan(cpus=8, memory_mb=4000, wall_time_s=7200, queue="long"))
+        for line in [
+            "request_cpus = 8",
+            "+xcount = 8",
+            "request_memory = 4000",
+            "+maxMemory = 4000",
+            "+BatchRuntime = 7200",
+            '+queue = "long"',
+        ]:
+            assert line in lines(text)
+
+    def test_finished_jobs_stay_until_purged(self):
+        # The one lifetime rule interCEde sets: output can be collected
+        # again until the consumer releases the job.
+        assert "leave_in_queue = (JobStatus == 4)" in lines(render(plan()))
+
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "on_exit_hold",
+            "on_exit_remove",
+            "periodic_hold",
+            "periodic_release",
+            "periodic_remove",
+            "max_retries",
+            "requirements",
+        ],
+    )
+    def test_sets_no_policy_on_the_job(self, command: str):
+        # Holding, releasing, removing or retrying jobs is the consumer's
+        # policy, never interCEde's.
+        assert not any(line.startswith(command) for line in lines(render(plan())))
+
+    def test_macros_in_paths_are_escaped(self):
+        assert "ic_dir_0 = /work/$(DOLLAR)0" in lines(
+            render(plan(copies=[CopyPlan("/work/$0")]))
+        )
+
+    @pytest.mark.parametrize("field", ["inputs", "outputs"])
+    def test_file_lists_cannot_hold_a_comma(self, field: str):
+        with pytest.raises(ValueError, match="comma"):
+            render(plan(**{field: ["a,b"]}))
+
+    def test_paths_cannot_hold_a_line_break(self):
+        with pytest.raises(ValueError, match="line break"):
+            render(plan(copies=[CopyPlan("/work/a\nqueue 100")]))
+
+
+# --------------------------------------------------------------------------
+# Commands
+# --------------------------------------------------------------------------
+
+
+class TestCommands:
+    def test_submit_to_the_local_schedd_spools(self):
+        assert commands.submit("/w/job.sub") == [
+            "condor_submit",
+            "-terse",
+            "-spool",
+            "/w/job.sub",
+        ]
+
+    @pytest.mark.parametrize(
+        ("count", "sizes"),
+        [(100, [100]), (101, [100, 1]), (250, [100, 100, 50])],
+    )
+    def test_chunks(self, count: int, sizes: list[int]):
+        ids = [status.JobID(f"1.{i}") for i in range(count)]
+        chunked = commands.chunks(ids)
+        assert [len(chunk) for chunk in chunked] == sizes
+        assert [i for chunk in chunked for i in chunk] == ids
+
+
+class TestParseSubmit:
+    @pytest.mark.parametrize(
+        ("raw", "ids"),
+        [
+            ("1.0 - 1.1\n", ["1.0", "1.1"]),
+            ("  42.0 - 42.0  ", ["42.0"]),
+            ("\n7.3 - 7.5\n\n", ["7.3", "7.4", "7.5"]),
+        ],
+    )
+    def test_terse_output(self, raw: str, ids: list[str]):
+        assert commands.parse_submit(raw) == ids
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            "",
+            "Submitting job(s).\n1 job(s) submitted to cluster 1.",
+            "1.0 - 2.0",
+            "1.3 - 1.1",
+            "1.0 - 1.1\n2.0 - 2.1",
+            "x",
+        ],
+        ids=["empty", "verbose", "two-clusters", "backwards", "two-lines", "garbage"],
+    )
+    def test_anything_else_is_refused(self, raw: str):
+        with pytest.raises(ValueError, match="unexpected condor_submit output"):
+            commands.parse_submit(raw)
+
+
+def test_parse_remove():
+    raw = "Job 1.0 marked for removal\nJob 1.2 marked for removal\nCouldn't find/remove all jobs matching 1.1\n"
+    assert commands.parse_remove(raw) == {"1.0", "1.2"}
+    assert commands.parse_remove("") == set()
+
+
+# --------------------------------------------------------------------------
+# State mapping
+# --------------------------------------------------------------------------
+
+
+def ad(**fields: Any) -> dict[str, Any]:
+    return {"ClusterId": 3, "ProcId": 1, **fields}
+
+
+class TestJobStatus:
+    @pytest.mark.parametrize(
+        ("code", "expected"),
+        [
+            (1, JobStatus.WAITING),
+            (2, JobStatus.RUNNING),
+            (4, JobStatus.DONE),
+            (6, JobStatus.RUNNING),
+            (7, JobStatus.WAITING),
+        ],
+        ids=["idle", "running", "completed", "transferring-output", "suspended"],
+    )
+    def test_plain_states(self, code: int, expected: JobStatus):
+        assert status.job_status(ad(JobStatus=code)) == expected
+
+    @pytest.mark.parametrize(
+        "code",
+        [8, None],
+        ids=["unexpected", "missing"],
+    )
+    def test_unknown_codes(self, code: Any):
+        assert status.job_status(ad(JobStatus=code)) == JobStatus.UNKNOWN
+
+    @pytest.mark.parametrize(
+        ("fields", "expected"),
+        [
+            ({"ExitCode": 0, "ExitBySignal": False}, JobStatus.DONE),
+            ({"ExitCode": 3, "ExitBySignal": False}, JobStatus.FAILED),
+            ({"ExitBySignal": True, "ExitSignal": 9}, JobStatus.FAILED),
+            ({}, JobStatus.DONE),
+        ],
+        ids=["exit-0", "exit-3", "signal", "no-exit-info"],
+    )
+    def test_finished_payload_result_comes_from_its_exit(
+        self, fields: dict[str, Any], expected: JobStatus
+    ):
+        # Read from what HTCondor records; no hold is put on the job to tell.
+        assert status.job_status(ad(JobStatus=4, **fields)) == expected
+
+    def test_spooling_is_waiting(self):
+        assert (
+            status.job_status(ad(JobStatus=5, HoldReasonCode=16)) == JobStatus.WAITING
+        )
+
+    @pytest.mark.parametrize(
+        "fields",
+        [{"HoldReasonCode": 1}, {"HoldReasonCode": 3, "HoldReasonSubCode": 55}, {}],
+        ids=["by-user", "by-site-policy", "no-reason"],
+    )
+    def test_any_other_hold_is_aborted(self, fields: dict[str, Any]):
+        assert status.job_status(ad(JobStatus=5, **fields)) == JobStatus.ABORTED
+
+    @pytest.mark.parametrize(
+        ("fields", "expected"),
+        [
+            ({"LastJobStatus": 4, "ExitCode": 0}, JobStatus.DONE),
+            ({"LastJobStatus": 4}, JobStatus.DONE),
+            ({"LastJobStatus": 4, "ExitCode": 2}, JobStatus.FAILED),
+            ({"LastJobStatus": 4, "ExitBySignal": True}, JobStatus.FAILED),
+            ({"LastJobStatus": 5, "ExitCode": 3}, JobStatus.ABORTED),
+            ({"LastJobStatus": 2}, JobStatus.ABORTED),
+            ({"LastJobStatus": 1}, JobStatus.ABORTED),
+            ({}, JobStatus.ABORTED),
+        ],
+        ids=[
+            "purged-done",
+            "purged-done-no-code",
+            "purged-failed",
+            "purged-signal",
+            "removed-while-held",
+            "killed-running",
+            "killed-idle",
+            "bare",
+        ],
+    )
+    def test_removed_jobs_keep_their_result(
+        self, fields: dict[str, Any], expected: JobStatus
+    ):
+        assert status.job_status(ad(JobStatus=3, **fields)) == expected
+
+
+class TestParseAds:
+    def test_condor_q_prints_nothing_for_no_match(self):
+        assert status.parse_ads("") == []
+        assert status.parse_ads("\n  \n") == []
+
+    def test_condor_history_prints_an_empty_list(self):
+        assert status.parse_ads("[\n]\n") == []
+
+    @pytest.mark.parametrize("raw", ["{}", "[1, 2]", '"text"', "not json"])
+    def test_anything_else_is_refused(self, raw: str):
+        with pytest.raises(ValueError):
+            status.parse_ads(raw)
+
+    def test_every_status_query_asks_for_what_the_mapping_reads(self):
+        assert {
+            "ClusterId",
+            "ProcId",
+            "JobStatus",
+            "LastJobStatus",
+            "ExitCode",
+            "ExitBySignal",
+            "HoldReasonCode",
+        } <= set(status.STATE_ATTRIBUTES)
diff --git a/tests/unit/backends/fake/__init__.py b/tests/unit/backends/fake/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/unit/test_fake_backend.py b/tests/unit/backends/fake/test_fake_backend.py
similarity index 100%
rename from tests/unit/test_fake_backend.py
rename to tests/unit/backends/fake/test_fake_backend.py
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
index b243067..e3b6575 100644
--- a/tests/unit/conftest.py
+++ b/tests/unit/conftest.py
@@ -13,11 +13,12 @@
 
 from intercede import JobBackend
 from intercede._backends.config import FakeConfig
+from intercede.auth import StaticCredentials
 from intercede.registry import _build
 
 # Built through the registry, so the suite also covers the registry path.
 BACKENDS: dict[str, Callable[[], JobBackend]] = {
-    "fake": lambda: _build(FakeConfig()),
+    "fake": lambda: _build(FakeConfig(), StaticCredentials()),
 }
 
 
diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py
new file mode 100644
index 0000000..3977caa
--- /dev/null
+++ b/tests/unit/test_auth.py
@@ -0,0 +1,112 @@
+"""Credential values, requirements and the static provider."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import datetime
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
+from intercede.auth import (
+    BearerToken,
+    Credential,
+    CredentialProvider,
+    CredentialRequirements,
+    SSHPassword,
+    SSHPrivateKey,
+    StaticCredentials,
+)
+from intercede.exceptions import AuthenticationError
+
+SECRET = "TOPSECRET"  # noqa: S105  (a fake secret the tests look for)
+
+# One instance of every kind, each holding SECRET in all of its secret fields.
+EVERY_KIND: dict[type[Credential], Credential] = {
+    BearerToken: BearerToken(value=SECRET),
+    SSHPrivateKey: SSHPrivateKey(pem=SECRET.encode(), passphrase=SECRET),
+    SSHPassword: SSHPassword(value=SECRET),
+}
+
+
+@pytest.mark.parametrize("kind", list(EVERY_KIND), ids=lambda kind: kind.__name__)
+def test_a_secret_never_shows(kind: type[Credential]):
+    credential = EVERY_KIND[kind]
+    for shown in (
+        repr(credential),
+        str(credential),
+        credential.model_dump_json(),
+        repr(StaticCredentials([credential])),
+    ):
+        assert SECRET not in shown
+
+
+@pytest.mark.parametrize(
+    "build",
+    [
+        lambda: BearerToken(value="t", token=SECRET),
+        lambda: SSHPrivateKey(pem=b"k", passphrase=[SECRET]),
+        lambda: StaticCredentials([{"value": SECRET}]),
+    ],
+    ids=["unknown-field", "wrong-type", "plain-data-in-provider"],
+)
+def test_a_mistake_never_repeats_the_secret(build: Callable[[], Any]):
+    with pytest.raises(ValueError) as exc_info:  # also pydantic's ValidationError
+        build()
+    assert SECRET not in str(exc_info.value)
+
+
+def test_expiry_must_have_a_timezone():
+    # A naive time could not be compared with the current time in UTC.
+    with pytest.raises(ValidationError, match="timezone"):
+        BearerToken(value="t", expires_at=datetime(2030, 1, 1))  # noqa: DTZ001
+
+
+class TestCredentialRequirements:
+    def test_names_one_kind(self):
+        requirements = CredentialRequirements(resource="r", kind=BearerToken)
+        assert requirements.kind is BearerToken
+
+    def test_needs_a_kind(self):
+        with pytest.raises(ValidationError, match="kind"):
+            CredentialRequirements(resource="r")  # type: ignore[call-arg]
+
+    @pytest.mark.parametrize("kind", [Credential, str, "BearerToken"])
+    def test_the_kind_is_a_concrete_credential_class(self, kind: Any):
+        with pytest.raises(ValidationError):
+            CredentialRequirements(resource="r", kind=kind)
+
+
+class TestStaticCredentials:
+    def test_is_a_provider(self):
+        assert isinstance(StaticCredentials(), CredentialProvider)
+
+    async def test_hands_out_only_the_requested_kind(self):
+        token, password = BearerToken(value="t"), SSHPassword(value="pw")
+        provider = StaticCredentials([token, password])
+        requirements = CredentialRequirements(resource="ce:9619", kind=BearerToken)
+        async with provider:
+            assert await provider.get(requirements) is token
+
+    async def test_a_kind_it_does_not_hold_is_named(self):
+        # A token and a password both have a single "value" field: it goes by
+        # kind, not by shape.
+        provider = StaticCredentials([SSHPassword(value="pw")])
+        requirements = CredentialRequirements(resource="ce:9619", kind=BearerToken)
+        with pytest.raises(AuthenticationError, match="no BearerToken for ce:9619"):
+            await provider.get(requirements)
+
+    @pytest.mark.parametrize(
+        "credentials",
+        [
+            [BearerToken(value="a"), BearerToken(value="b")],
+            [{"value": "x"}],
+            ["token"],
+            [Credential()],
+        ],
+        ids=["two-of-a-kind", "plain-data", "a-string", "the-base-class"],
+    )
+    def test_refuses_bad_input_at_construction(self, credentials: Any):
+        with pytest.raises(ValueError):
+            StaticCredentials(credentials)
diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py
index 2856b53..36f8da6 100644
--- a/tests/unit/test_config.py
+++ b/tests/unit/test_config.py
@@ -64,7 +64,7 @@ def test_defaults_are_filled_in():
     assert (ssh.port, ssh.auth, ssh.connect_timeout_s) == (22, "key", 30)
     assert ssh.transfer_timeout_s == 3600
     ce = validate_config(CE)
-    assert isinstance(ce, HTCondorCEConfig) and ce.auth == {"token"}
+    assert isinstance(ce, HTCondorCEConfig) and ce.max_output_files == 10_000
 
 
 def test_a_model_passes_through_unchanged():
@@ -95,6 +95,8 @@ def test_an_unknown_choice(self, data: dict[str, Any], message: str):
         ("data", "field", "where"),
         [
             ({**ARC, "host": "h"}, "host", "kind='arc'"),
+            ({**ARC, "auth": "token"}, "auth", "kind='arc'"),
+            ({**CE, "auth": "proxy"}, "auth", "kind='htcondor-ce'"),
             ({**CLOUD, "queue": "short"}, "queue", "kind='cloud'"),
             (
                 {**LOCAL, "known_hosts": "h k"},
@@ -127,9 +129,6 @@ def test_every_missing_field_is_named_at_once(self):
     @pytest.mark.parametrize(
         ("data", "message"),
         [
-            ({**ARC, "auth": []}, "auth must name at least one"),
-            ({**CE, "auth": ["proxy"]}, "must include 'token'"),
-            ({**ARC, "auth": ["key"]}, "auth"),
             ({**SSH, "auth": "token"}, "auth"),
             (
                 {**SSH, "known_hosts": "\n\t "},
@@ -143,9 +142,6 @@ def test_every_missing_field_is_named_at_once(self):
             ({**LOCAL, "max_output_files": 0}, "max_output_files"),
         ],
         ids=[
-            "arc-no-auth",
-            "htcondor-ce-no-token",
-            "ce-ssh-auth",
             "ssh-ce-auth",
             "no-host-keys",
             "launcher-no-image",
diff --git a/tests/unit/test_credentials.py b/tests/unit/test_credentials.py
new file mode 100644
index 0000000..26aef4d
--- /dev/null
+++ b/tests/unit/test_credentials.py
@@ -0,0 +1,260 @@
+"""When a backend asks its provider again, and what it accepts back."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from types import TracebackType
+from typing import Any, Self
+
+import pytest
+
+from intercede import AuthenticationError, InterCEdeError, validate_config
+from intercede._backends.credentials import (
+    DEFAULT_MARGIN,
+    CredentialCache,
+    credential_requirements,
+    is_stale,
+    refresh_if_stale,
+)
+from intercede.auth import (
+    BearerToken,
+    Credential,
+    CredentialRequirements,
+    SSHPassword,
+    SSHPrivateKey,
+)
+
+NOW = datetime(2030, 1, 1, 12, 0, tzinfo=UTC)
+TOKEN_ONLY = CredentialRequirements(resource="ce:9619", kind=BearerToken)
+
+
+def token(expires_in: timedelta | None = None) -> BearerToken:
+    return BearerToken(
+        value="t", expires_at=None if expires_in is None else NOW + expires_in
+    )
+
+
+class Recording:
+    """A provider that returns whatever the test gives it, and counts calls."""
+
+    def __init__(self, result: Any = None, error: BaseException | None = None) -> None:
+        self.result = result
+        self.error = error
+        self.asked: list[CredentialRequirements] = []
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None: ...
+
+    async def get(self, requirements: CredentialRequirements) -> Any:
+        self.asked.append(requirements)
+        if self.error is not None:
+            raise self.error
+        return self.result
+
+
+class TestIsStale:
+    @pytest.mark.parametrize(
+        ("credential", "stale"),
+        [
+            (None, True),  # nothing asked for yet
+            (token(), False),  # never expires
+            (token(DEFAULT_MARGIN + timedelta(seconds=1)), False),
+            (token(DEFAULT_MARGIN), True),  # inside the margin
+            (token(-timedelta(hours=1)), True),  # expired
+            (SSHPassword(value="p"), True),  # not what is required
+        ],
+        ids=[
+            "none",
+            "never-expires",
+            "outside-margin",
+            "at-margin",
+            "expired",
+            "another-kind",
+        ],
+    )
+    def test_one_credential(self, credential: Credential | None, stale: bool):
+        assert is_stale(credential, TOKEN_ONLY, now=NOW) is stale
+
+
+class TestRefreshIfStale:
+    async def test_fresh_credential_is_kept_without_asking(self):
+        current = token(timedelta(hours=1))
+        provider = Recording()
+        assert await refresh_if_stale(current, TOKEN_ONLY, provider, now=NOW) is current
+        assert provider.asked == []
+
+    async def test_stale_credential_is_replaced(self):
+        fresh = token(timedelta(hours=1))
+        provider = Recording(fresh)
+        current = token(timedelta(minutes=1))
+        assert await refresh_if_stale(current, TOKEN_ONLY, provider, now=NOW) is fresh
+        assert provider.asked == [TOKEN_ONLY]
+
+    async def test_short_lived_credentials_are_accepted(self):
+        # Inside the margin but not expired: usable now, asked again next time.
+        fresh = token(timedelta(minutes=1))
+        assert (
+            await refresh_if_stale(None, TOKEN_ONLY, Recording(fresh), now=NOW) is fresh
+        )
+
+
+class TestProviderFailures:
+    """Every failure is a whole-operation AuthenticationError."""
+
+    async def test_provider_raising_is_an_authentication_error(self):
+        provider = Recording(error=RuntimeError("no token for s3cret-url"))
+        with pytest.raises(AuthenticationError) as exc_info:
+            await refresh_if_stale(None, TOKEN_ONLY, provider, now=NOW)
+        assert isinstance(exc_info.value.__cause__, RuntimeError)
+        message = str(exc_info.value)
+        assert "ce:9619" in message and "RuntimeError" in message
+        assert "s3cret" not in message  # the provider's text may hold a secret
+
+    async def test_our_own_errors_pass_through_unchanged(self):
+        error = InterCEdeError("x")
+        with pytest.raises(InterCEdeError) as exc_info:
+            await refresh_if_stale(None, TOKEN_ONLY, Recording(error=error), now=NOW)
+        assert exc_info.value is error
+
+    @pytest.mark.parametrize(
+        ("result", "returned"),
+        [
+            (None, "NoneType"),
+            ([token()], "list"),
+            (SSHPassword(value="p"), "SSHPassword"),
+        ],
+        ids=["nothing", "several", "another-kind"],
+    )
+    async def test_anything_but_the_required_kind(self, result: Any, returned: str):
+        with pytest.raises(
+            AuthenticationError, match=f"returned a {returned} .*not a BearerToken"
+        ):
+            await refresh_if_stale(None, TOKEN_ONLY, Recording(result), now=NOW)
+
+    async def test_an_expired_credential(self):
+        provider = Recording(token(-timedelta(minutes=1)))
+        with pytest.raises(AuthenticationError, match="expired"):
+            await refresh_if_stale(None, TOKEN_ONLY, provider, now=NOW)
+
+
+READ = CredentialRequirements(
+    resource="ce:9619", kind=BearerToken, scopes=frozenset({"compute.read"})
+)
+CREATE = CredentialRequirements(
+    resource="ce:9619", kind=BearerToken, scopes=frozenset({"compute.create"})
+)
+
+
+class PerScope(Recording):
+    """A provider that mints a token named after the scopes it is asked for."""
+
+    def __init__(self, short_lived: CredentialRequirements | None = None) -> None:
+        super().__init__()
+        self.short_lived = short_lived
+
+    async def get(self, requirements: CredentialRequirements) -> BearerToken:
+        self.asked.append(requirements)
+        soon = datetime.now(UTC) + timedelta(minutes=1)
+        return BearerToken(
+            value=",".join(requirements.scopes),
+            expires_at=soon if requirements == self.short_lived else None,
+        )
+
+
+def value(credential: Credential) -> str:
+    assert isinstance(credential, BearerToken)
+    return credential.value.get_secret_value()
+
+
+class TestCredentialCache:
+    async def test_a_token_for_one_scope_is_not_used_for_another(self):
+        cache = CredentialCache(PerScope())
+        assert value(await cache.fresh(READ)) == "compute.read"
+        assert value(await cache.fresh(CREATE)) == "compute.create"
+
+    async def test_the_provider_is_asked_once_per_requirement(self):
+        provider = PerScope()
+        cache = CredentialCache(provider)
+        for requirements in (READ, CREATE, READ, CREATE):
+            await cache.fresh(requirements)
+        assert provider.asked == [READ, CREATE]
+
+    async def test_only_the_credential_about_to_expire_is_asked_for_again(self):
+        provider = PerScope(short_lived=READ)
+        cache = CredentialCache(provider)
+        for requirements in (READ, CREATE, READ, CREATE):
+            await cache.fresh(requirements)
+        assert provider.asked == [READ, CREATE, READ]
+
+    async def test_clearing_forgets_every_credential(self):
+        provider = PerScope()
+        cache = CredentialCache(provider)
+        await cache.fresh(READ)
+        cache.clear()
+        await cache.fresh(READ)
+        assert provider.asked == [READ, READ]
+
+    async def test_a_provider_failure_is_an_authentication_error(self):
+        cache = CredentialCache(Recording(error=RuntimeError("s3cret-url")))
+        with pytest.raises(AuthenticationError):
+            await cache.fresh(READ)
+
+    async def test_a_secret_never_shows(self):
+        cache = CredentialCache(PerScope())
+        await cache.fresh(READ)
+        assert "compute.read" not in repr(cache)
+
+
+# --------------------------------------------------------------------------
+# From a configuration to what to ask the provider for
+# --------------------------------------------------------------------------
+
+ARC: dict[str, Any] = {"kind": "arc", "endpoint": "https://arc.example.org:443"}
+CE: dict[str, Any] = {"kind": "htcondor-ce", "endpoint": "ce.example.org:9619"}
+SSH: dict[str, Any] = {
+    "kind": "batch",
+    "transport": "ssh",
+    "scheduler": "slurm",
+    "host": "batch.example.org",
+    "user": "dirac",
+    "known_hosts": "batch.example.org ssh-ed25519 AAAAC3Nz...",
+}
+
+
+class TestCredentialRequirements:
+    @pytest.mark.parametrize("base", [ARC, CE], ids=["arc", "htcondor-ce"])
+    def test_a_ce_asks_for_a_token_for_its_endpoint(self, base: dict[str, Any]):
+        scopes = frozenset({"compute.create"})
+        requirements = credential_requirements(validate_config(base), scopes)
+        assert requirements is not None
+        assert requirements.kind is BearerToken
+        assert requirements.resource == base["endpoint"]
+        assert requirements.scopes == scopes
+
+    @pytest.mark.parametrize(
+        ("auth", "kind"), [("key", SSHPrivateKey), ("password", SSHPassword)]
+    )
+    def test_ssh_asks_for_the_login(self, auth: str, kind: type):
+        config = validate_config({**SSH, "auth": auth, "port": 2222})
+        requirements = credential_requirements(config)
+        assert requirements is not None
+        assert requirements.kind is kind
+        assert requirements.resource == "dirac@batch.example.org:2222"
+
+    @pytest.mark.parametrize(
+        "data",
+        [
+            {**SSH, "auth": "agent"},
+            {"kind": "batch", "transport": "local", "scheduler": "slurm"},
+        ],
+        ids=["ssh-agent", "local"],
+    )
+    def test_nothing_to_ask_for(self, data: dict[str, Any]):
+        assert credential_requirements(validate_config(data)) is None
diff --git a/tests/unit/test_registry.py b/tests/unit/test_registry.py
index eebab72..1fa1c59 100644
--- a/tests/unit/test_registry.py
+++ b/tests/unit/test_registry.py
@@ -18,9 +18,11 @@
     registry,
 )
 from intercede._backends.config import FakeConfig
+from intercede.auth import BearerToken, Credential, StaticCredentials
 from intercede.registry import _build, _load
 
 DUMMIES = "tests.unit._registry_dummies"
+CREDENTIALS = StaticCredentials()
 
 CLOUD: dict[str, Any] = {
     "kind": "cloud",
@@ -63,7 +65,7 @@ def composed(monkeypatch: pytest.MonkeyPatch) -> None:
 
 class TestBuilds:
     async def test_the_fake_for_the_test_suites(self):
-        async with _build(FakeConfig(resource="fake://test")) as backend:
+        async with _build(FakeConfig(resource="fake://test"), CREDENTIALS) as backend:
             assert isinstance(backend, JobBackend)
             assert backend.resource == "fake://test"
 
@@ -72,29 +74,59 @@ async def test_the_fake_for_the_test_suites(self):
         [CLOUD, CloudConfig(**CLOUD)],
         ids=["mapping", "object"],
     )
-    def test_a_single_class_backend_gets_its_settings(
+    def test_a_single_class_backend_gets_its_settings_and_provider(
         self, monkeypatch: pytest.MonkeyPatch, config: Any
     ):
         monkeypatch.setitem(registry._BACKENDS, "cloud", f"{DUMMIES}:DummyBackend")
-        built: Any = registry.get_backend(config)
+        provider = StaticCredentials()
+        built: Any = registry.get_backend(config, provider)
         assert type(built).__name__ == "DummyBackend"
         assert isinstance(built.settings, CloudConfig)
+        assert built.credentials is provider
 
     def test_a_batch_backend_is_composed_of_its_parts(self, composed: None):
-        built: Any = registry.get_backend({**SSH_SLURM, "queue": "short"})
+        provider = StaticCredentials()
+        built: Any = registry.get_backend({**SSH_SLURM, "queue": "short"}, provider)
         assert type(built).__name__ == "DummyBatchBackend"
-        # Transport and scheduler each get the whole batch configuration.
+        # Transport and scheduler each get the whole batch configuration; the
+        # transport, which authenticates, also gets the provider.
         assert isinstance(built.transport.settings, BatchSSHConfig)
         assert built.transport.settings.host == "batch.example.org"
+        assert built.transport.credentials is provider
         assert built.scheduler.settings.queue == "short"
 
+    async def test_building_asks_the_provider_nothing(
+        self, composed: None, monkeypatch: pytest.MonkeyPatch
+    ):
+        monkeypatch.setitem(registry._BACKENDS, "cloud", f"{DUMMIES}:DummyBackend")
+
+        class Watching(StaticCredentials):
+            asked = 0
+
+            async def get(self, requirements: Any) -> Credential:
+                Watching.asked += 1
+                return await super().get(requirements)
+
+        registry.get_backend(SSH_SLURM, Watching())
+        registry.get_backend(CLOUD, Watching())
+        assert Watching.asked == 0  # construction does no I/O
+
 
 class TestRefused:
     def test_invalid_configuration_keeps_the_details(self):
         with pytest.raises(ConfigurationError) as exc_info:
-            registry.get_backend({"kind": "arc"})
+            registry.get_backend({"kind": "arc"}, CREDENTIALS)
         assert "endpoint: required for kind='arc'" in str(exc_info.value)
 
+    @pytest.mark.parametrize(
+        "credentials",
+        [None, BearerToken(value="t")],
+        ids=["none", "credential-not-provider"],
+    )
+    def test_credentials_must_be_a_provider(self, composed: None, credentials: Any):
+        with pytest.raises(ConfigurationError, match="must be a CredentialProvider"):
+            registry.get_backend(LOCAL_SLURM, credentials)
+
     @pytest.mark.parametrize(
         "config",
         [{"kind": "arc", "endpoint": "https://arc.example.org:443"}, CLOUD],
@@ -104,18 +136,18 @@ def test_valid_but_not_shipped_yet(self, config: dict[str, Any]):
         with pytest.raises(
             RegistryError, match=f"backend '{config['kind']}' is not available"
         ):
-            registry.get_backend(config)
+            registry.get_backend(config, CREDENTIALS)
 
     def test_composed_part_not_shipped_yet(self):
         with pytest.raises(RegistryError, match="transport 'ssh' is not available"):
-            registry.get_backend(SSH_SLURM)
+            registry.get_backend(SSH_SLURM, CREDENTIALS)
 
     def test_an_entry_that_cannot_be_imported(
         self, composed: None, monkeypatch: pytest.MonkeyPatch
     ):
         monkeypatch.setitem(registry._TRANSPORTS, "ssh", "tests.unit.does_not_exist:X")
         with pytest.raises(RegistryError, match="cannot be imported") as exc_info:
-            registry.get_backend(SSH_SLURM)
+            registry.get_backend(SSH_SLURM, CREDENTIALS)
         assert isinstance(exc_info.value.__cause__, ImportError)
 
     @pytest.mark.parametrize(
@@ -138,7 +170,7 @@ def test_what_breaks_its_protocol_is_refused(
             getattr(registry, table), name, f"{DUMMIES}:BuildsTheWrongThing"
         )
         with pytest.raises(RegistryError, match=f"which is not a {protocol}"):
-            registry.get_backend(LOCAL_SLURM)
+            registry.get_backend(LOCAL_SLURM, CREDENTIALS)
 
 
 class TestAvailable:
@@ -170,5 +202,5 @@ def test_only_the_requested_entry_is_imported(
         monkeypatch.delitem(sys.modules, DUMMIES, raising=False)
         registry.available()
         assert DUMMIES not in sys.modules  # listing imports nothing
-        registry.get_backend(LOCAL_SLURM)
+        registry.get_backend(LOCAL_SLURM, CREDENTIALS)
         assert DUMMIES in sys.modules