Skip to content

Index

axm_vault

axm-vault.

Catalog-resolver secrets manager (keyring + SecretStr) for AXM

SERVICE = 'axm-vault' module-attribute

The fixed keyring service name under which every AXM secret is stored.

Layer = Literal['env', 'file', 'keyring', 'default', 'prompt']

Resolution layer a credential may be sourced from.

Provenance = dict[str, dict[str, str | bool]]

Per-credential report: {"group.name": {"layer": str, "present": bool}}.

For a SECRET spec whose keyring backend is unavailable (headless host), the entry additionally carries "keyring": "unavailable" so the doctor surfaces the outage rather than silently reporting the credential as merely missing.

Catalog

Bases: BaseModel

An in-memory index of credential groups, keyed by group id.

Source code in packages/axm-vault/src/axm_vault/catalog.py
Python
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class Catalog(BaseModel):  # type: ignore[explicit-any]
    """An in-memory index of credential groups, keyed by group id."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    groups_: tuple[CredentialGroup, ...] = ()

    def __init__(
        self, groups: tuple[CredentialGroup, ...] = (), **data: object
    ) -> None:
        super().__init__(groups_=tuple(groups), **data)

    @model_validator(mode="after")
    def _validate_names(self) -> Catalog:
        """Reject group ids / spec names that no axm-config segment can hold.

        Every ``group.id`` namespaces the group's axm-config writes
        (``set_(group.id, ...)``) and every SECRET/CONFIG spec name is used as
        an axm-config key; both must round-trip through ``axm_config.set_``.
        The charsets are axm-config's (namespace ``^[a-z0-9]+(\\.[a-z0-9]+)*$``,
        key ``^[a-z0-9]+(_[a-z0-9]+)*$``) so validation delegates to the
        canonical :func:`axm_config.validate_segment` rather than mirroring the
        patterns (the mirror had already diverged). An id/name that could never
        round-trip is a structural error caught here (the same path
        :func:`load_catalog` takes) instead of surfacing as a ``ConfigError``
        mid-``run_setup``. NONSENSITIVE spec names are env-only and exempt; the
        group id is checked regardless. axm-config's ``ConfigError`` is
        normalised to ``ValueError`` so the failure arrives as pydantic's
        ``ValidationError`` like any other model-validation error.
        """
        try:
            for group in self.groups_:
                axm_config.validate_segment(group.id, kind="namespace")
                for spec in group.specs:
                    if spec.sensitivity in _STORABLE:
                        axm_config.validate_segment(spec.name, kind="key")
        except axm_config.ConfigError as exc:
            raise ValueError(str(exc)) from exc
        return self

    def group(self, gid: str) -> CredentialGroup:
        """Return the group identified by ``gid``.

        Raises:
            KeyError: if no group with that id is registered.
        """
        for candidate in self.groups_:
            if candidate.id == gid:
                return candidate
        raise KeyError(f"no credential group with id {gid!r}")

    def groups(self) -> list[CredentialGroup]:
        """Return every registered group."""
        return list(self.groups_)

    def for_package(self, package: str) -> list[CredentialGroup]:
        """Return the groups contributed by ``package``."""
        return [g for g in self.groups_ if g.package == package]

    def all_specs(self) -> list[tuple[str, CredentialSpec]]:
        """Return every ``(group_id, spec)`` pair across all groups."""
        return [(g.id, spec) for g in self.groups_ for spec in g.specs]
all_specs()

Return every (group_id, spec) pair across all groups.

Source code in packages/axm-vault/src/axm_vault/catalog.py
Python
 98
 99
100
def all_specs(self) -> list[tuple[str, CredentialSpec]]:
    """Return every ``(group_id, spec)`` pair across all groups."""
    return [(g.id, spec) for g in self.groups_ for spec in g.specs]
for_package(package)

Return the groups contributed by package.

Source code in packages/axm-vault/src/axm_vault/catalog.py
Python
94
95
96
def for_package(self, package: str) -> list[CredentialGroup]:
    """Return the groups contributed by ``package``."""
    return [g for g in self.groups_ if g.package == package]
group(gid)

Return the group identified by gid.

Raises:

Type Description
KeyError

if no group with that id is registered.
Source code in packages/axm-vault/src/axm_vault/catalog.py
Python
79
80
81
82
83
84
85
86
87
88
def group(self, gid: str) -> CredentialGroup:
    """Return the group identified by ``gid``.

    Raises:
        KeyError: if no group with that id is registered.
    """
    for candidate in self.groups_:
        if candidate.id == gid:
            return candidate
    raise KeyError(f"no credential group with id {gid!r}")
groups()

Return every registered group.

Source code in packages/axm-vault/src/axm_vault/catalog.py
Python
90
91
92
def groups(self) -> list[CredentialGroup]:
    """Return every registered group."""
    return list(self.groups_)

CredentialGroup

Bases: BaseModel

A bundle of credential specs declared by a package.

Source code in packages/axm-vault/src/axm_vault/models.py
Python
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class CredentialGroup(BaseModel):  # type: ignore[explicit-any]
    """A bundle of credential specs declared by a package."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    id: str
    package: str
    title: str
    specs: tuple[CredentialSpec, ...]
    multi: bool = False

    def spec(self, name: str) -> CredentialSpec:
        """Return the spec named ``name``.

        Raises:
            KeyError: if no spec with that name exists in the group.
        """
        for candidate in self.specs:
            if candidate.name == name:
                return candidate
        raise KeyError(name)
spec(name)

Return the spec named name.

Raises:

Type Description
KeyError

if no spec with that name exists in the group.
Source code in packages/axm-vault/src/axm_vault/models.py
Python
62
63
64
65
66
67
68
69
70
71
def spec(self, name: str) -> CredentialSpec:
    """Return the spec named ``name``.

    Raises:
        KeyError: if no spec with that name exists in the group.
    """
    for candidate in self.specs:
        if candidate.name == name:
            return candidate
    raise KeyError(name)

CredentialSpec

Bases: BaseModel

Schema for a single credential — value-less by construction.

Source code in packages/axm-vault/src/axm_vault/models.py
Python
36
37
38
39
40
41
42
43
44
45
46
47
48
class CredentialSpec(BaseModel):  # type: ignore[explicit-any]
    """Schema for a single credential — value-less by construction."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    name: str
    env: str
    kind: str
    sensitivity: Sensitivity = Sensitivity.SECRET
    required: bool = True
    default: str | None = None
    prompt: str | None = None
    aliases: tuple[str, ...] = ()

KeyringStore

Store and retrieve secrets in the OS keyring under :data:SERVICE.

The store is stateless: every call delegates to the process-wide keyring backend, so tests can swap in an in-memory backend via keyring.set_keyring(...) without touching the real Keychain.

Source code in packages/axm-vault/src/axm_vault/store.py
Python
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class KeyringStore:
    """Store and retrieve secrets in the OS keyring under :data:`SERVICE`.

    The store is stateless: every call delegates to the process-wide
    ``keyring`` backend, so tests can swap in an in-memory backend via
    ``keyring.set_keyring(...)`` without touching the real Keychain.
    """

    @staticmethod
    def username(group: str, name: str, instance: str | None = None) -> str:
        """Compose the keyring ``username`` for a credential.

        The instance segment is omitted cleanly when ``instance`` is ``None``
        (``{group}.{name}``) and inserted between group and name otherwise
        (``{group}.{instance}.{name}``).

        ``.`` is the structural separator, so each segment is percent-escaped
        before joining: a literal ``.`` (or ``%``) inside any segment is
        encoded so that distinct ``(group, name, instance)`` tuples can never
        collapse onto the same username (e.g. ``("a.b", "c")`` vs
        ``("a", "b.c")``). The encoding is reversible and leaves dot-free
        segments untouched, so existing usernames are unchanged and the
        reserved ``.prev`` rotation slot stays distinct from its base name.
        """
        parts = [group, instance, name] if instance is not None else [group, name]
        return ".".join(_escape_segment(part) for part in parts)

    def set(
        self, group: str, name: str, value: str, instance: str | None = None
    ) -> None:
        """Store ``value`` under ``(group, name[, instance])`` in the keyring.

        Raises :class:`KeyringUnavailableError` when no usable backend exists
        (headless host) rather than surfacing a raw backend traceback.
        """
        with _keyring_guard():
            keyring.set_password(SERVICE, self.username(group, name, instance), value)

    def get(self, group: str, name: str, instance: str | None = None) -> str | None:
        """Return the stored secret, or ``None`` if no such credential exists.

        Raises :class:`KeyringUnavailableError` when no usable backend exists
        (headless host) rather than surfacing a raw backend traceback.
        """
        with _keyring_guard():
            return keyring.get_password(SERVICE, self.username(group, name, instance))

    def delete(self, group: str, name: str, instance: str | None = None) -> None:
        """Remove the credential from the keyring (no-op if already absent).

        The real macOS Keychain backend raises
        :class:`keyring.errors.PasswordDeleteError` (``Item not found``) when
        the credential is absent; it is suppressed here so the call is a true
        no-op, as the contract promises and as rotation/cleanup callers rely on.
        """
        with _keyring_guard(), contextlib.suppress(PasswordDeleteError):
            keyring.delete_password(SERVICE, self.username(group, name, instance))
delete(group, name, instance=None)

Remove the credential from the keyring (no-op if already absent).

The real macOS Keychain backend raises :class:keyring.errors.PasswordDeleteError (Item not found) when the credential is absent; it is suppressed here so the call is a true no-op, as the contract promises and as rotation/cleanup callers rely on.

Source code in packages/axm-vault/src/axm_vault/store.py
Python
128
129
130
131
132
133
134
135
136
137
def delete(self, group: str, name: str, instance: str | None = None) -> None:
    """Remove the credential from the keyring (no-op if already absent).

    The real macOS Keychain backend raises
    :class:`keyring.errors.PasswordDeleteError` (``Item not found``) when
    the credential is absent; it is suppressed here so the call is a true
    no-op, as the contract promises and as rotation/cleanup callers rely on.
    """
    with _keyring_guard(), contextlib.suppress(PasswordDeleteError):
        keyring.delete_password(SERVICE, self.username(group, name, instance))
get(group, name, instance=None)

Return the stored secret, or None if no such credential exists.

Raises :class:KeyringUnavailableError when no usable backend exists (headless host) rather than surfacing a raw backend traceback.

Source code in packages/axm-vault/src/axm_vault/store.py
Python
119
120
121
122
123
124
125
126
def get(self, group: str, name: str, instance: str | None = None) -> str | None:
    """Return the stored secret, or ``None`` if no such credential exists.

    Raises :class:`KeyringUnavailableError` when no usable backend exists
    (headless host) rather than surfacing a raw backend traceback.
    """
    with _keyring_guard():
        return keyring.get_password(SERVICE, self.username(group, name, instance))
set(group, name, value, instance=None)

Store value under (group, name[, instance]) in the keyring.

Raises :class:KeyringUnavailableError when no usable backend exists (headless host) rather than surfacing a raw backend traceback.

Source code in packages/axm-vault/src/axm_vault/store.py
Python
108
109
110
111
112
113
114
115
116
117
def set(
    self, group: str, name: str, value: str, instance: str | None = None
) -> None:
    """Store ``value`` under ``(group, name[, instance])`` in the keyring.

    Raises :class:`KeyringUnavailableError` when no usable backend exists
    (headless host) rather than surfacing a raw backend traceback.
    """
    with _keyring_guard():
        keyring.set_password(SERVICE, self.username(group, name, instance), value)
username(group, name, instance=None) staticmethod

Compose the keyring username for a credential.

The instance segment is omitted cleanly when instance is None ({group}.{name}) and inserted between group and name otherwise ({group}.{instance}.{name}).

. is the structural separator, so each segment is percent-escaped before joining: a literal . (or %) inside any segment is encoded so that distinct (group, name, instance) tuples can never collapse onto the same username (e.g. ("a.b", "c") vs ("a", "b.c")). The encoding is reversible and leaves dot-free segments untouched, so existing usernames are unchanged and the reserved .prev rotation slot stays distinct from its base name.

Source code in packages/axm-vault/src/axm_vault/store.py
Python
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@staticmethod
def username(group: str, name: str, instance: str | None = None) -> str:
    """Compose the keyring ``username`` for a credential.

    The instance segment is omitted cleanly when ``instance`` is ``None``
    (``{group}.{name}``) and inserted between group and name otherwise
    (``{group}.{instance}.{name}``).

    ``.`` is the structural separator, so each segment is percent-escaped
    before joining: a literal ``.`` (or ``%``) inside any segment is
    encoded so that distinct ``(group, name, instance)`` tuples can never
    collapse onto the same username (e.g. ``("a.b", "c")`` vs
    ``("a", "b.c")``). The encoding is reversible and leaves dot-free
    segments untouched, so existing usernames are unchanged and the
    reserved ``.prev`` rotation slot stays distinct from its base name.
    """
    parts = [group, instance, name] if instance is not None else [group, name]
    return ".".join(_escape_segment(part) for part in parts)

MissingCredentialError

Bases: Exception

A required credential could not be resolved from any layer.

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
49
50
class MissingCredentialError(Exception):
    """A required credential could not be resolved from any layer."""

Resolved

Bases: BaseModel

The outcome of resolving a single credential.

Carries the resolved value, the layer it was sourced from and the originating spec — but never masks: callers wrap secrets themselves (e.g. via :func:~axm_vault.secrets.as_secret).

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
53
54
55
56
57
58
59
60
61
62
63
64
65
class Resolved(BaseModel):  # type: ignore[explicit-any]
    """The outcome of resolving a single credential.

    Carries the resolved ``value``, the ``layer`` it was sourced from and the
    originating ``spec`` — but never masks: callers wrap secrets themselves
    (e.g. via :func:`~axm_vault.secrets.as_secret`).
    """

    model_config = ConfigDict(frozen=True)

    value: str
    layer: Layer
    spec: CredentialSpec

Resolver

Walk the layer precedence to resolve a credential value.

The resolver is stateless and cheap to instantiate; a process-wide :data:resolver singleton is provided for the common case. Set interactive=True to enable the prompt layer (off by default so the resolver is safe in non-interactive contexts).

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class Resolver:
    """Walk the layer precedence to resolve a credential value.

    The resolver is stateless and cheap to instantiate; a process-wide
    :data:`resolver` singleton is provided for the common case. Set
    ``interactive=True`` to enable the ``prompt`` layer (off by default so the
    resolver is safe in non-interactive contexts).
    """

    PRECEDENCE: tuple[Layer, ...] = ("env", "file", "keyring", "default", "prompt")

    def __init__(self, *, interactive: bool = False) -> None:
        self._interactive = interactive
        self._keyring = KeyringStore()

    def resolve(
        self, group: CredentialGroup, name: str, instance: str | None = None
    ) -> Resolved:
        """Resolve ``group.name`` by walking :data:`PRECEDENCE`.

        Returns the first layer that yields a value. Raises
        :class:`MissingCredentialError` when a *required* spec resolves to
        nothing across every layer.
        """
        spec = group.spec(name)
        for layer in self.PRECEDENCE:
            value = self._try(layer, spec, group, instance)
            if value is not None:
                return Resolved(value=value, layer=layer, spec=spec)
        if spec.required:
            raise MissingCredentialError(f"{group.id}.{name}")
        return Resolved(value=spec.default or "", layer="default", spec=spec)

    def probe(
        self,
        layer: Layer,
        spec: CredentialSpec,
        group: CredentialGroup,
        instance: str | None = None,
    ) -> bool:
        """Report whether ``layer`` supplies ``spec``, value-free.

        Reduces the layer's value to a boolean the instant it is read, so the
        value never escapes — the seam the *doctor* uses to build provenance
        without violating the never-leak invariant.
        """
        return self._try(layer, spec, group, instance) is not None

    def _try(
        self,
        layer: Layer,
        spec: CredentialSpec,
        group: CredentialGroup,
        instance: str | None,
    ) -> str | None:
        """Return the value ``layer`` provides for ``spec``, or ``None``."""
        match layer:
            case "env":
                return self._from_env(spec)
            case "file":
                return self._from_file(group, spec)
            case "keyring":
                return self._from_keyring(spec, group, instance)
            case "default":
                return spec.default
            case "prompt":
                return self._from_prompt(spec)
            case _:  # pragma: no cover - exhaustive over Layer
                return None

    @staticmethod
    def _from_env(spec: CredentialSpec) -> str | None:
        """Read ``spec.env`` then each alias (canonical wins over alias).

        An empty-string env var (``VAR=``) is treated as *absent*, not as an
        empty value: it must not win the env layer nor mask a value declared in
        a lower layer (file/keyring/default). A legitimately empty credential
        is vanishingly rare, so treating ``""`` as unset is the safer default.
        """
        for key in (spec.env, *spec.aliases):
            value = os.environ.get(key)
            if value:
                return value
        return None

    @staticmethod
    def _from_file(group: CredentialGroup, spec: CredentialSpec) -> str | None:
        """Read ``spec.name`` from the ``~/.axm/<group>.toml`` file *only*.

        Uses :class:`~axm_config.store.NamespaceStore` (file-only) rather than
        ``axm_config.get`` (which mixes an env tier under its own naming),
        so a value present only in the environment never surfaces here — the
        env tier belongs exclusively to the ``env`` layer.
        """
        value = NamespaceStore().read(group.id).get(spec.name)
        return value if isinstance(value, str) else None

    def _from_keyring(
        self, spec: CredentialSpec, group: CredentialGroup, instance: str | None
    ) -> str | None:
        """Consult the keyring only for SECRET specs.

        Degrades gracefully when the OS keyring is unavailable (headless host):
        the layer is skipped (returns ``None``) so resolution falls through to
        the lower layers (default) instead of crashing.
        """
        if spec.sensitivity is not Sensitivity.SECRET:
            return None
        try:
            return self._keyring.get(group.id, spec.name, instance)
        except KeyringUnavailableError:
            return None

    def keyring_available(self) -> bool:
        """Report whether the OS keyring backend is usable, value-free.

        Probes the backend with a read that resolves no real credential; a
        :class:`KeyringUnavailableError` means no usable backend (headless
        host). The *doctor* uses this to flag ``keyring unavailable`` without
        ever reading a secret value.
        """
        try:
            self._keyring.get("__axm_vault__", "__probe__")
        except KeyringUnavailableError:
            return False
        return True

    def _from_prompt(self, spec: CredentialSpec) -> str | None:
        """Prompt interactively when enabled and the spec declares one.

        A SECRET spec is read through :func:`getpass.getpass` so the typed
        value is never echoed to the terminal (never-leak invariant); a
        non-secret spec uses a visible :func:`input` prompt.
        """
        if not (self._interactive and spec.prompt):
            return None
        reader = getpass if spec.sensitivity is Sensitivity.SECRET else input
        return reader(spec.prompt) or None
keyring_available()

Report whether the OS keyring backend is usable, value-free.

Probes the backend with a read that resolves no real credential; a :class:KeyringUnavailableError means no usable backend (headless host). The doctor uses this to flag keyring unavailable without ever reading a secret value.

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
181
182
183
184
185
186
187
188
189
190
191
192
193
def keyring_available(self) -> bool:
    """Report whether the OS keyring backend is usable, value-free.

    Probes the backend with a read that resolves no real credential; a
    :class:`KeyringUnavailableError` means no usable backend (headless
    host). The *doctor* uses this to flag ``keyring unavailable`` without
    ever reading a secret value.
    """
    try:
        self._keyring.get("__axm_vault__", "__probe__")
    except KeyringUnavailableError:
        return False
    return True
probe(layer, spec, group, instance=None)

Report whether layer supplies spec, value-free.

Reduces the layer's value to a boolean the instant it is read, so the value never escapes — the seam the doctor uses to build provenance without violating the never-leak invariant.

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def probe(
    self,
    layer: Layer,
    spec: CredentialSpec,
    group: CredentialGroup,
    instance: str | None = None,
) -> bool:
    """Report whether ``layer`` supplies ``spec``, value-free.

    Reduces the layer's value to a boolean the instant it is read, so the
    value never escapes — the seam the *doctor* uses to build provenance
    without violating the never-leak invariant.
    """
    return self._try(layer, spec, group, instance) is not None
resolve(group, name, instance=None)

Resolve group.name by walking :data:PRECEDENCE.

Returns the first layer that yields a value. Raises :class:MissingCredentialError when a required spec resolves to nothing across every layer.

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def resolve(
    self, group: CredentialGroup, name: str, instance: str | None = None
) -> Resolved:
    """Resolve ``group.name`` by walking :data:`PRECEDENCE`.

    Returns the first layer that yields a value. Raises
    :class:`MissingCredentialError` when a *required* spec resolves to
    nothing across every layer.
    """
    spec = group.spec(name)
    for layer in self.PRECEDENCE:
        value = self._try(layer, spec, group, instance)
        if value is not None:
            return Resolved(value=value, layer=layer, spec=spec)
    if spec.required:
        raise MissingCredentialError(f"{group.id}.{name}")
    return Resolved(value=spec.default or "", layer="default", spec=spec)

Sensitivity

Bases: StrEnum

Classification of how sensitive a credential value is.

Source code in packages/axm-vault/src/axm_vault/models.py
Python
24
25
26
27
28
29
class Sensitivity(StrEnum):
    """Classification of how sensitive a credential value is."""

    SECRET = "secret"  # noqa: S105 # enum label, not a password value
    CONFIG = "config"
    NONSENSITIVE = "nonsensitive"

VaultDoctorTool

Report credential provenance (layer + presence), never a value.

Source code in packages/axm-vault/src/axm_vault/tools.py
Python
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class VaultDoctorTool:
    """Report credential provenance (layer + presence), never a value."""

    agent_hint = (
        "Report which layer (env/file/keyring/default/missing) supplies each "
        "credential and whether it is present — values are NEVER returned."
    )
    domain = "vault"
    tags = frozenset({"vault", "credentials", "doctor", "provenance"})

    @property
    def name(self) -> str:
        """Unique tool identifier."""
        return "vault_doctor"

    def execute(
        self, *, package: str | None = None, instance: str | None = None
    ) -> ToolResult:
        """Return value-free provenance for the catalog (or one package)."""
        try:
            data = doctor_data(package, instance=instance)
        except Exception as exc:  # noqa: BLE001 # MCP boundary: any error -> failure
            return ToolResult(success=False, error=str(exc))
        return ToolResult(success=True, data=dict(data))
name property

Unique tool identifier.

execute(*, package=None, instance=None)

Return value-free provenance for the catalog (or one package).

Source code in packages/axm-vault/src/axm_vault/tools.py
Python
39
40
41
42
43
44
45
46
47
def execute(
    self, *, package: str | None = None, instance: str | None = None
) -> ToolResult:
    """Return value-free provenance for the catalog (or one package)."""
    try:
        data = doctor_data(package, instance=instance)
    except Exception as exc:  # noqa: BLE001 # MCP boundary: any error -> failure
        return ToolResult(success=False, error=str(exc))
    return ToolResult(success=True, data=dict(data))

VaultSetTool

Store a credential by group.name — keyring (SECRET) or config.

NONSENSITIVE credentials are environment-only and are rejected outright: storing them would create a second, stale source of truth. The stored value is never echoed back — only the storage target is reported.

Source code in packages/axm-vault/src/axm_vault/tools.py
Python
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class VaultSetTool:
    """Store a credential by ``group.name`` — keyring (SECRET) or config.

    NONSENSITIVE credentials are environment-only and are rejected outright:
    storing them would create a second, stale source of truth. The stored
    value is never echoed back — only the storage target is reported.
    """

    agent_hint = (
        "Store a credential: SECRET -> OS keyring, CONFIG -> axm-config; "
        "NONSENSITIVE is env-only and rejected. The value is never echoed."
    )
    domain = "vault"
    tags = frozenset({"vault", "credentials", "set"})

    @property
    def name(self) -> str:
        """Unique tool identifier."""
        return "vault_set"

    def execute(
        self,
        *,
        group: str = "",
        name: str = "",
        value: str = "",
        instance: str | None = None,
    ) -> ToolResult:
        """Route ``group.name`` to its store by sensitivity; never echo value."""
        try:
            spec = load_catalog().group(group).spec(name)
            target = self._store(spec.sensitivity, group, name, value, instance)
        except Exception as exc:  # noqa: BLE001 # MCP boundary: any error -> failure
            return ToolResult(success=False, error=str(exc))
        return ToolResult(success=True, data={"stored": target})

    @staticmethod
    def _store(
        sensitivity: Sensitivity,
        group: str,
        name: str,
        value: str,
        instance: str | None,
    ) -> str:
        """Persist ``value`` to the backend for ``sensitivity``; return target.

        Raises:
            ValueError: when ``sensitivity`` is NONSENSITIVE (env-only).
        """
        target = f"{group}.{name}"
        match sensitivity:
            case Sensitivity.SECRET:
                KeyringStore().set(group, name, value, instance)
                return f"keyring:{target}"
            case Sensitivity.CONFIG:
                axm_config.set_(group, name, value)
                return f"config:{target}"
            case Sensitivity.NONSENSITIVE:
                msg = (
                    f"{target} is NONSENSITIVE (environment-only); "
                    "set it via its env var, it is never stored"
                )
                raise ValueError(msg)
name property

Unique tool identifier.

execute(*, group='', name='', value='', instance=None)

Route group.name to its store by sensitivity; never echo value.

Source code in packages/axm-vault/src/axm_vault/tools.py
Python
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def execute(
    self,
    *,
    group: str = "",
    name: str = "",
    value: str = "",
    instance: str | None = None,
) -> ToolResult:
    """Route ``group.name`` to its store by sensitivity; never echo value."""
    try:
        spec = load_catalog().group(group).spec(name)
        target = self._store(spec.sensitivity, group, name, value, instance)
    except Exception as exc:  # noqa: BLE001 # MCP boundary: any error -> failure
        return ToolResult(success=False, error=str(exc))
    return ToolResult(success=True, data={"stored": target})

as_secret(value)

Coerce a raw secret string into a :class:~pydantic.SecretStr.

None passes through unchanged (for optional secrets), and an existing :class:~pydantic.SecretStr is returned as-is (idempotent).

Source code in packages/axm-vault/src/axm_vault/secrets.py
Python
22
23
24
25
26
27
28
29
30
def as_secret(value: str | SecretStr | None) -> SecretStr | None:
    """Coerce a raw secret string into a :class:`~pydantic.SecretStr`.

    ``None`` passes through unchanged (for optional secrets), and an
    existing :class:`~pydantic.SecretStr` is returned as-is (idempotent).
    """
    if value is None or isinstance(value, SecretStr):
        return value
    return SecretStr(value)

atomic_write(path, data, *, encoding='utf-8')

Write data to path atomically (temp file + os.replace).

The write goes to a temporary file in the destination directory, is flushed and fsync-ed, then atomically renamed over path so a concurrent reader never observes a partially written file. After the rename the parent directory is itself fsync-ed so the new directory entry is durable — without that, a crash right after os.replace could lose the rename even though the file content reached disk. The destination directory must already exist — vault does not create it (that is axm-config's responsibility). Intended for the OAuth refresh-token rotation case.

Source code in packages/axm-vault/src/axm_vault/store.py
Python
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def atomic_write(path: Path | str, data: str, *, encoding: str = "utf-8") -> None:
    """Write ``data`` to ``path`` atomically (temp file + ``os.replace``).

    The write goes to a temporary file in the destination directory, is flushed
    and ``fsync``-ed, then atomically renamed over ``path`` so a concurrent
    reader never observes a partially written file. After the rename the parent
    directory is itself ``fsync``-ed so the new directory entry is durable —
    without that, a crash right after ``os.replace`` could lose the rename even
    though the file content reached disk. The destination directory must
    already exist — vault does not create it (that is ``axm-config``'s
    responsibility). Intended for the OAuth refresh-token rotation case.
    """
    target = Path(path)
    directory = target.parent
    fd, tmp_name = tempfile.mkstemp(
        dir=directory, prefix=f".{target.name}.", suffix=".tmp"
    )
    try:
        with os.fdopen(fd, "w", encoding=encoding) as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(tmp_name, target)
        # Refresh-token leak guard: enforce owner-only perms regardless of
        # umask. mkstemp already creates the temp file 0600; re-chmod after
        # replace mirrors axm-config's store.py so the final inode is 0600.
        os.chmod(target, 0o600)
        # Crash-safety: fsync the parent directory so the new directory entry
        # (the rename) is durable, not just the file content fsync-ed above.
        dir_fd = os.open(directory, os.O_RDONLY)
        try:
            os.fsync(dir_fd)
        finally:
            os.close(dir_fd)
    except BaseException:
        with contextlib.suppress(FileNotFoundError):
            os.unlink(tmp_name)
        raise

bind(model, group, instance=None)

Build model from the resolved values of every spec in group.

Each field is keyed by spec.name; SECRET specs are wrapped with :func:~axm_vault.secrets.as_secret so the consumer model holds SecretStr. A missing required spec propagates :class:MissingCredentialError. The return type is the concrete model type (generic over type[_ModelT]), so a caller need not cast the result back to its model.

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def bind[ModelT: BaseModel](
    model: type[ModelT], group: str, instance: str | None = None
) -> ModelT:
    """Build ``model`` from the resolved values of every spec in ``group``.

    Each field is keyed by ``spec.name``; SECRET specs are wrapped with
    :func:`~axm_vault.secrets.as_secret` so the consumer model holds
    ``SecretStr``. A missing *required* spec propagates
    :class:`MissingCredentialError`. The return type is the concrete ``model``
    type (generic over ``type[_ModelT]``), so a caller need not cast the
    result back to its model.
    """
    grp = load_catalog().group(group)
    data: dict[str, object] = {}
    for spec in grp.specs:
        resolved = resolver.resolve(grp, spec.name, instance)
        value = _resolved_value(resolved)
        if spec.sensitivity is Sensitivity.SECRET:
            data[spec.name] = as_secret(value)
        else:
            data[spec.name] = value
    return model.model_validate(data)

doctor_data(package=None, *, catalog=None, instance=None)

Report the winning layer and presence of every credential, value-free.

Parameters:

Name Type Description Default
package str | None

When given, restrict the report to credential groups contributed by that package; otherwise cover the whole catalog.

 None
catalog Catalog | None

Catalog to inspect; defaults to the discovered :func:~axm_vault.catalog.load_catalog result.

 None
instance str | None

Optional multi-instance segment forwarded to the probe.

 None

Returns:

Name Type Description
A Provenance

data:Provenance mapping "group.name" to {layer, present}.
Provenance

layer is the first probed layer to supply the credential, or
Provenance

"missing" when none does; present mirrors that. The value
Provenance

itself is NEVER included (security invariant).
Source code in packages/axm-vault/src/axm_vault/doctor.py
Python
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def doctor_data(
    package: str | None = None,
    *,
    catalog: Catalog | None = None,
    instance: str | None = None,
) -> Provenance:
    """Report the winning layer and presence of every credential, value-free.

    Args:
        package: When given, restrict the report to credential groups
            contributed by that package; otherwise cover the whole catalog.
        catalog: Catalog to inspect; defaults to the discovered
            :func:`~axm_vault.catalog.load_catalog` result.
        instance: Optional multi-instance segment forwarded to the probe.

    Returns:
        A :data:`Provenance` mapping ``"group.name"`` to ``{layer, present}``.
        ``layer`` is the first probed layer to supply the credential, or
        ``"missing"`` when none does; ``present`` mirrors that. The value
        itself is NEVER included (security invariant).
    """
    cat = catalog if catalog is not None else load_catalog()
    groups = cat.for_package(package) if package is not None else cat.groups()
    resolver = Resolver()
    keyring_ok = resolver.keyring_available()
    report: Provenance = {}
    for group in groups:
        for spec in group.specs:
            report[f"{group.id}.{spec.name}"] = _probe(
                resolver, group, spec, instance, keyring_ok=keyring_ok
            )
    return report

get(group, name, instance=None)

Resolve group.name via the singleton :data:resolver.

Convenience over resolver.resolve(load_catalog().group(group), name) returning just the resolved value.

Source code in packages/axm-vault/src/axm_vault/resolver.py
Python
227
228
229
230
231
232
233
234
def get(group: str, name: str, instance: str | None = None) -> str:
    """Resolve ``group.name`` via the singleton :data:`resolver`.

    Convenience over ``resolver.resolve(load_catalog().group(group), name)``
    returning just the resolved value.
    """
    grp = load_catalog().group(group)
    return resolver.resolve(grp, name, instance).value

load_catalog() cached

Discover and index all axm.credentials groups.

Reads the axm.credentials entry-points, calls each (a callable returning list[CredentialGroup]) and indexes the groups by id. Returns an empty catalog when no entry-point is registered — the nominal state for vault itself. Cached so discovery runs once.

Source code in packages/axm-vault/src/axm_vault/catalog.py
Python
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@cache
def load_catalog() -> Catalog:
    """Discover and index all ``axm.credentials`` groups.

    Reads the ``axm.credentials`` entry-points, calls each (a callable
    returning ``list[CredentialGroup]``) and indexes the groups by id.
    Returns an empty catalog when no entry-point is registered — the
    nominal state for vault itself. Cached so discovery runs once.
    """
    index: dict[str, CredentialGroup] = {}
    for endpoint in entry_points(group=CREDENTIALS_GROUP):
        provider = endpoint.load()
        for group in provider():
            index[group.id] = group
    return Catalog(groups=tuple(index.values()))

redact(text, *secrets)

Mask occurrences of known secret substrings for log scrubbing.

Each qualifying secret found in text is replaced by :data:MASK. A :class:~pydantic.SecretStr argument is accepted and unwrapped via :meth:~pydantic.SecretStr.get_secret_value. Secrets are masked longest-first so a short secret that is a prefix of a longer one cannot leave the longer one's tail unmasked. Secrets shorter than :data:MIN_REDACT_LEN (and empty ones) are ignored, since masking a tiny substring over-redacts surrounding text without protecting anything.

This is best-effort log scrubbing, not a security boundary: it only masks the exact substrings it is given, in the casing they are given, and cannot catch transformed or partial echoes. The authoritative never-leak surface is :class:~pydantic.SecretStr.

Source code in packages/axm-vault/src/axm_vault/secrets.py
Python
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def redact(text: str, *secrets: str | SecretStr) -> str:
    """Mask occurrences of known secret substrings for log scrubbing.

    Each qualifying secret found in ``text`` is replaced by :data:`MASK`. A
    :class:`~pydantic.SecretStr` argument is accepted and unwrapped via
    :meth:`~pydantic.SecretStr.get_secret_value`. Secrets are masked
    longest-first so a short secret that is a prefix of a longer one cannot
    leave the longer one's tail unmasked. Secrets shorter than
    :data:`MIN_REDACT_LEN` (and empty ones) are ignored, since masking a tiny
    substring over-redacts surrounding text without protecting anything.

    This is **best-effort** log scrubbing, not a security boundary: it only
    masks the exact substrings it is given, in the casing they are given, and
    cannot catch transformed or partial echoes. The authoritative never-leak
    surface is :class:`~pydantic.SecretStr`.
    """
    plain = [s.get_secret_value() if isinstance(s, SecretStr) else s for s in secrets]
    for secret in sorted(plain, key=len, reverse=True):
        if len(secret) >= MIN_REDACT_LEN:
            text = text.replace(secret, MASK)
    return text

rotate_secret(group, name, value, instance=None)

Rotate a keyring secret, retaining the previous value for one cycle.

The prior cycle's backup is purged first, the current value (if any) is copied to the reserved {name}.prev slot, then value is written over {name}. Retention is strictly one cycle: a stale .prev never lingers across rotations. Keeping the previous secret one rotation lets a caller fall back during an in-flight credential roll. No value is ever returned or logged.

Raises:

Type Description
ValueError

if name already ends with the reserved .prev suffix — that namespace is owned by the rotation backup slot and must not collide with a real spec name or instance.
KeyringUnavailableError

when the OS keyring backend is unavailable.
Source code in packages/axm-vault/src/axm_vault/store.py
Python
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def rotate_secret(
    group: str, name: str, value: str, instance: str | None = None
) -> None:
    """Rotate a keyring secret, retaining the previous value for one cycle.

    The prior cycle's backup is purged first, the current value (if any) is
    copied to the reserved ``{name}.prev`` slot, then ``value`` is written over
    ``{name}``. Retention is strictly one cycle: a stale ``.prev`` never
    lingers across rotations. Keeping the previous secret one rotation lets a
    caller fall back during an in-flight credential roll. No value is ever
    returned or logged.

    Raises:
        ValueError: if ``name`` already ends with the reserved ``.prev``
            suffix — that namespace is owned by the rotation backup slot and
            must not collide with a real spec name or instance.
        KeyringUnavailableError: when the OS keyring backend is unavailable.
    """
    if name.endswith(_PREV_SUFFIX):
        raise ValueError(
            f"cannot rotate {name!r}: the {_PREV_SUFFIX!r} suffix is reserved "
            "for the rotation backup slot"
        )
    store = KeyringStore()
    prev_name = f"{name}{_PREV_SUFFIX}"
    current = store.get(group, name, instance)
    # Purge the prior cycle's backup before writing the new one so retention
    # stays strictly one cycle (a no-op when absent).
    store.delete(group, prev_name, instance)
    if current is not None:
        store.set(group, prev_name, current, instance)
    store.set(group, name, value, instance)

run_setup(only=None)

Prompt for and store every storable credential in the catalog.

Parameters:

Name Type Description Default
only str | None

When given, restrict setup to the single group.name (or a bare name) matching this string; otherwise cover every spec.

 None

The function refuses to run without an interactive TTY (writes :class:SystemExit 1 to stderr). NONSENSITIVE specs are skipped (environment-only); SECRET specs are read with :func:getpass.getpass and CONFIG specs with :func:input. A blank answer keeps the existing value, making the driver idempotent across re-runs.

Source code in packages/axm-vault/src/axm_vault/setup.py
Python
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def run_setup(only: str | None = None) -> None:
    """Prompt for and store every storable credential in the catalog.

    Args:
        only: When given, restrict setup to the single ``group.name`` (or a
            bare ``name``) matching this string; otherwise cover every spec.

    The function refuses to run without an interactive TTY (writes
    :class:`SystemExit` ``1`` to stderr). NONSENSITIVE specs are skipped
    (environment-only); SECRET specs are read with :func:`getpass.getpass`
    and CONFIG specs with :func:`input`. A blank answer keeps the existing
    value, making the driver idempotent across re-runs.
    """
    if not sys.stdin.isatty():
        print(
            "axm-vault setup requires an interactive terminal (TTY).",
            file=sys.stderr,
        )
        raise SystemExit(1)
    keyring = KeyringStore()
    for group in load_catalog().groups():
        for spec in group.specs:
            if only is not None and only not in (f"{group.id}.{spec.name}", spec.name):
                continue
            _setup_spec(keyring, group, spec)