mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 10:59:59 +00:00
fix(api): resolve credentials against the declaration an inventory override hides
API is a plain mapping, so an inventory that configures three providers replaces the whole declaration from group_vars/all/18_api.yml instead of merging into it. inventories/development/default.yml declares openai, anthropic and openrouter, so github, google and jira did not exist in CI at all and lookup('api', 'github.client_id') raised.
A raising lookup inside a value the applications renderer is templating does not fail the render: _templar_render_best_effort returns the string untouched and, for a value carrying a non-env lookup(, refuses the fallback as well. The raw template text reached plugin_value.enabled | bool, which coerced it to False and emitted the deprecation warning that fails the deploy gate. Run 34893363518, job 104153194109 lost web-app-nextcloud to exactly that, for three integrations at once. The configvalue entries would have stored template text as the client id.
utils/api.py merges the scope's override over the declaration, so an override narrows values and never the provider set, and holds the enable rule: every credential a provider declares must carry a value. The api lookup reads through it; the new api_enabled lookup gates on it and returns a real boolean instead of a string a filter has to coerce.
The same class sat in group_vars/all/16_ai.yml and roles/svc-ai-litellm/meta/secrets.yml as bare API.<provider>.<key> dereferences; both now go through the lookups.
Two lints keep it shut. test_addon_enabled_renders_to_bool renders the whole applications map against a scope with API blanked, the most hostile override an inventory may legally write, and requires every addon gate to come out a boolean. test_api_accessed_via_lookup forbids a bare API dereference in Jinja under roles/, group_vars/ and inventories/.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
e90cac4ee2
commit
15a6251d72
12 files changed
+462
-20
No files matched your search
@@ -33,9 +33,9 @@ AI_LOCAL_MODELS:
|
||||
source: https://huggingface.co/nomic-ai/nomic-embed-text-v1.5-GGUF
|
||||
|
||||
AI_REMOTE_ALIASES: >-
|
||||
{{ ['openai/gpt-4o-mini'] * ((API.openai.api_key | string | length > 0) | int)
|
||||
+ ['anthropic/claude-sonnet-4-5'] * ((API.anthropic.api_key | string | length > 0) | int)
|
||||
+ ['openrouter/auto'] * ((API.openrouter.api_key | string | length > 0) | int) }}
|
||||
{{ ['openai/gpt-4o-mini'] * (lookup('api_enabled', 'openai') | int)
|
||||
+ ['anthropic/claude-sonnet-4-5'] * (lookup('api_enabled', 'anthropic') | int)
|
||||
+ ['openrouter/auto'] * (lookup('api_enabled', 'openrouter') | int) }}
|
||||
|
||||
LITELLM_CHAT_MODEL: "{{ (lookup('config', 'svc-ai-ollama', 'services.ollama.preload_models') | map(attribute='alias') | reject('search', 'embed') | list | first |
|
||||
default('')) if 'svc-ai-ollama' in LITELLM_BACKEND_ROLES else ((lookup('config', 'svc-ai-lmstudio', 'services.lmstudio.preload_models') | map(attribute='alias')
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
---
|
||||
# Central SPOT for proprietary / external (non-self-hosted) API credentials.
|
||||
# Each entry is empty by default; an integration that depends on a provider is
|
||||
# only activated when its credentials are present (see the addons that gate
|
||||
# their `enabled` on `API.<provider>`). Override per provider in the (vault-
|
||||
# encrypted) inventory; access ONLY via lookup('api', '<provider>.<key>').
|
||||
# nocheck: unused-var: consumed by the api lookup plugin (plugins/lookup/api.py) via lookup('api', ...), not referenced directly in YAML/Jinja.
|
||||
# Each entry is empty by default. Override per provider in the (vault-
|
||||
# encrypted) inventory; read ONLY via lookup('api', '<provider>.<key>') and
|
||||
# gate ONLY on lookup('api_enabled', '<provider>') (see utils/api.py).
|
||||
# nocheck: unused-var: consumed by the api lookup plugins (plugins/lookup/api.py, plugins/lookup/api_enabled.py) via lookup('api', ...), not referenced directly in YAML/Jinja.
|
||||
API:
|
||||
anthropic:
|
||||
api_key: ""
|
||||
|
||||
@@ -5,13 +5,16 @@ from typing import Any
|
||||
from ansible.errors import AnsibleError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from utils.api import resolve_api
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
"""
|
||||
lookup('api', '<provider>.<key>') e.g. lookup('api', 'github.client_id')
|
||||
|
||||
Single point of truth for proprietary / external API credentials. Resolves a
|
||||
dotted path against the global ``API`` mapping (group_vars/all/18_api.yml).
|
||||
dotted path against the effective ``API`` mapping: the declaration in
|
||||
group_vars/all/18_api.yml with the calling scope's override merged over it.
|
||||
Routing every access through this plugin keeps one place to change how
|
||||
proprietary API credentials are sourced (vault, env, secret manager, a
|
||||
different provider abstraction) without touching the call sites.
|
||||
@@ -31,11 +34,10 @@ class LookupModule(LookupBase):
|
||||
if not path:
|
||||
raise AnsibleError("lookup('api'): empty key path.")
|
||||
|
||||
api = variables.get("API", {})
|
||||
if templar is not None:
|
||||
api = templar.template(api)
|
||||
if not isinstance(api, dict):
|
||||
raise AnsibleError("lookup('api'): global 'API' is not a mapping.")
|
||||
try:
|
||||
api = resolve_api(variables, templar=templar)
|
||||
except TypeError as error:
|
||||
raise AnsibleError(f"lookup('api'): {error}") from error
|
||||
|
||||
node: Any = api
|
||||
for part in path.split("."):
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ansible.errors import AnsibleError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from utils.api import provider_enabled, resolve_api
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
"""
|
||||
lookup('api_enabled', '<provider>') e.g. lookup('api_enabled', 'github')
|
||||
|
||||
True when every credential the provider declares in group_vars/all/18_api.yml
|
||||
carries a value in the effective ``API`` mapping. Integrations that depend on
|
||||
a proprietary provider gate on this instead of spelling the emptiness test
|
||||
out per call site, so an added credential key extends the gate everywhere and
|
||||
the result is a real boolean rather than a string a filter has to coerce.
|
||||
"""
|
||||
|
||||
def run(self, terms, variables: dict[str, Any] | None = None, **kwargs):
|
||||
if not terms or len(terms) != 1:
|
||||
raise AnsibleError(
|
||||
"lookup('api_enabled') takes exactly one term: '<provider>' "
|
||||
"(e.g. 'github')."
|
||||
)
|
||||
|
||||
templar = getattr(self, "_templar", None)
|
||||
variables = variables or getattr(templar, "available_variables", {}) or {}
|
||||
|
||||
provider = str(terms[0]).strip()
|
||||
if not provider:
|
||||
raise AnsibleError("lookup('api_enabled'): empty provider name.")
|
||||
|
||||
try:
|
||||
api = resolve_api(variables, templar=templar)
|
||||
return [provider_enabled(api, provider)]
|
||||
except (KeyError, TypeError) as error:
|
||||
raise AnsibleError(f"lookup('api_enabled'): {error}") from error
|
||||
@@ -15,14 +15,14 @@ credentials:
|
||||
OpenAI moved from sk- to sk-proj- without notice, and a schema that pinned the prefix would reject a valid key"
|
||||
type: "string"
|
||||
regex: "\\S{20,}"
|
||||
default: "{{ API.openai.api_key }}"
|
||||
default: "{{ lookup('api', 'openai.api_key') }}"
|
||||
anthropic_api_key:
|
||||
description: "Anthropic API key; empty leaves the anthropic/* models unpublished"
|
||||
type: "string"
|
||||
regex: "\\S{20,}"
|
||||
default: "{{ API.anthropic.api_key }}"
|
||||
default: "{{ lookup('api', 'anthropic.api_key') }}"
|
||||
openrouter_api_key:
|
||||
description: "OpenRouter API key; empty leaves openrouter/auto unpublished"
|
||||
type: "string"
|
||||
regex: "\\S{20,}"
|
||||
default: "{{ API.openrouter.api_key }}"
|
||||
default: "{{ lookup('api', 'openrouter.api_key') }}"
|
||||
@@ -1,5 +1,5 @@
|
||||
---
|
||||
enabled: "{{ (lookup('api', 'github.client_id') | length > 0) and (lookup('api', 'github.client_secret') | length > 0) }}"
|
||||
enabled: "{{ lookup('api_enabled', 'github') }}"
|
||||
required: true
|
||||
mechanism: plugin
|
||||
source: upstream
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
---
|
||||
enabled: "{{ (lookup('api', 'google.client_id') | length > 0) and (lookup('api', 'google.client_secret') | length > 0) }}"
|
||||
enabled: "{{ lookup('api_enabled', 'google') }}"
|
||||
required: true
|
||||
mechanism: plugin
|
||||
source: upstream
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
---
|
||||
enabled: "{{ (lookup('api', 'jira.client_id') | length > 0) and (lookup('api', 'jira.client_secret') | length > 0) }}"
|
||||
enabled: "{{ lookup('api_enabled', 'jira') }}"
|
||||
required: true
|
||||
mechanism: plugin
|
||||
source: upstream
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Every addon ``enabled`` must render to a boolean, even on a bare scope.
|
||||
|
||||
Rationale
|
||||
=========
|
||||
``meta/addons/*.yml`` is rendered inside the applications SPOT and consumed as
|
||||
``plugin_value.enabled | bool``. An expression that cannot resolve does not
|
||||
fail: ``utils.templating.ansible._templar_render_best_effort`` returns the
|
||||
string untouched, and for a value carrying a non-env ``lookup(`` it refuses the
|
||||
fallback as well. The template text reaches the filter, which coerces it to
|
||||
False and emits::
|
||||
|
||||
[DEPRECATION WARNING]: The `bool` filter coerced invalid value
|
||||
"{{ ... }}" (str) to False.
|
||||
|
||||
The deploy gate fails on that warning, and the addon is off in every run that
|
||||
did not warn loudly enough to be noticed. Run 34893363518, job 104153194109
|
||||
lost web-app-nextcloud this way.
|
||||
|
||||
The check renders the whole applications map against a scope carrying only what
|
||||
every run has -- group_vars plus the deployed role names -- and blanks ``API``
|
||||
to the most hostile override an inventory may legally write. Anything that
|
||||
survives as template text resolves only against something the render scope does
|
||||
not carry.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from typing import Any
|
||||
|
||||
from ansible.parsing.dataloader import DataLoader
|
||||
from ansible.template import Templar
|
||||
|
||||
from utils.cache.applications import (
|
||||
_MERGED_APPLICATIONS_CACHE,
|
||||
get_merged_applications,
|
||||
)
|
||||
from utils.cache.yaml import load_yaml_any
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
|
||||
_ADDONS_GLOB = "*/meta/addons/*.yml"
|
||||
_BOOL_LITERALS = frozenset(
|
||||
{"true", "false", "yes", "no", "on", "off", "1", "0", "y", "n"}
|
||||
)
|
||||
|
||||
|
||||
def _role_names() -> list[str]:
|
||||
return sorted(
|
||||
path.name for path in (PROJECT_ROOT / "roles").iterdir() if path.is_dir()
|
||||
)
|
||||
|
||||
|
||||
def _hostile_scope() -> dict[str, Any]:
|
||||
variables: dict[str, Any] = {}
|
||||
for path in sorted((PROJECT_ROOT / "group_vars" / "all").glob("*.yml")):
|
||||
data = load_yaml_any(path) or {}
|
||||
if isinstance(data, dict):
|
||||
variables.update(data)
|
||||
variables["group_names"] = _role_names()
|
||||
variables["API"] = {}
|
||||
return variables
|
||||
|
||||
|
||||
def _rendered_addons() -> dict[str, dict[str, Any]]:
|
||||
variables = _hostile_scope()
|
||||
templar = Templar(loader=DataLoader(), variables=dict(variables))
|
||||
templar.available_variables = dict(variables)
|
||||
|
||||
_MERGED_APPLICATIONS_CACHE.clear()
|
||||
try:
|
||||
applications = get_merged_applications(
|
||||
variables=dict(variables), templar=templar
|
||||
)
|
||||
finally:
|
||||
_MERGED_APPLICATIONS_CACHE.clear()
|
||||
|
||||
return {
|
||||
app_id: config["addons"]
|
||||
for app_id, config in applications.items()
|
||||
if isinstance(config, dict) and isinstance(config.get("addons"), dict)
|
||||
}
|
||||
|
||||
|
||||
def _verdict(value: Any) -> str:
|
||||
if isinstance(value, bool):
|
||||
return ""
|
||||
if not isinstance(value, str):
|
||||
return f"is {type(value).__name__}, not a boolean"
|
||||
if "{{" in value or "{%" in value:
|
||||
return f"stayed template text: {value}"
|
||||
if value.strip().lower() not in _BOOL_LITERALS:
|
||||
return f"is not a boolean literal: {value!r}"
|
||||
return ""
|
||||
|
||||
|
||||
class TestAddonEnabledRendersToBool(unittest.TestCase):
|
||||
def test_every_addon_enabled_resolves(self) -> None:
|
||||
findings = [
|
||||
f" - roles/{app_id}/meta/addons/{addon_id}.yml: 'enabled' {verdict}"
|
||||
for app_id, addons in _rendered_addons().items()
|
||||
for addon_id, spec in sorted(addons.items())
|
||||
if isinstance(spec, dict)
|
||||
for verdict in [_verdict(spec.get("enabled"))]
|
||||
if verdict
|
||||
]
|
||||
|
||||
if findings:
|
||||
self.fail(
|
||||
f"{len(findings)} addon gate(s) do not render to a boolean on a "
|
||||
"scope without inventory extras, so '| bool' coerces the leftover "
|
||||
"text to False and the deploy fails on the deprecation warning:\n"
|
||||
+ "\n".join(findings)
|
||||
+ "\n\nResolve the value through a lookup that carries its own "
|
||||
"source instead of a variable the render scope may not hold: "
|
||||
"lookup('api_enabled', '<provider>') gates on credentials, "
|
||||
"lookup('config', ...) on another application's settings."
|
||||
)
|
||||
|
||||
def test_the_scan_reaches_templated_gates(self) -> None:
|
||||
"""A scope that resolved nothing, or addons without gates, passes for free."""
|
||||
templated = [
|
||||
path
|
||||
for path in sorted((PROJECT_ROOT / "roles").glob(_ADDONS_GLOB))
|
||||
if "{{" in str((load_yaml_any(path) or {}).get("enabled", ""))
|
||||
]
|
||||
self.assertTrue(
|
||||
templated,
|
||||
"no addon declares a templated 'enabled', so the rule above is vacuous",
|
||||
)
|
||||
|
||||
rendered = _rendered_addons()
|
||||
self.assertTrue(rendered, "no application carries addons; nothing was checked")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,105 @@
|
||||
"""``API`` is reachable through its lookups only, never as a bare variable.
|
||||
|
||||
Rationale
|
||||
=========
|
||||
A bare ``API.github.client_id`` resolves against whatever the scope happens to
|
||||
hold, and an inventory override replaces the declaration rather than merging
|
||||
into it -- ``utils/api.py`` carries the mechanism and the consequences.
|
||||
``lookup('api', '<provider>.<key>')`` and ``lookup('api_enabled', '<provider>')``
|
||||
merge the override over ``group_vars/all/18_api.yml``, so an override narrows
|
||||
values and never the provider set. Routing every access through them also keeps
|
||||
one place to change how credentials are sourced.
|
||||
|
||||
Scope
|
||||
=====
|
||||
Jinja in ``roles/``, ``group_vars/`` and ``inventories/``. An ``API:`` key
|
||||
being *declared* is not an access; only a dereference inside ``{{ }}`` or
|
||||
``{% %}`` is. Test fixtures legitimately build the bare form to exercise the
|
||||
renderer, so ``tests/`` is out of scope.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unittest
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from utils.cache.files import iter_project_files_with_content
|
||||
|
||||
from . import PROJECT_ROOT
|
||||
|
||||
_SCOPES = ("roles", "group_vars", "inventories")
|
||||
_SUFFIXES = (".yml", ".yaml", ".j2")
|
||||
_JINJA_BLOCK = re.compile(r"\{\{(.*?)\}\}|\{%(.*?)%\}", re.DOTALL)
|
||||
_BARE_API = re.compile(r"(?<![\w.'\"])API\s*[.\[]")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Finding:
|
||||
path: str
|
||||
line: int
|
||||
expression: str
|
||||
|
||||
|
||||
def _scoped_files() -> list[tuple[str, str]]:
|
||||
prefixes = tuple(f"{scope}/" for scope in _SCOPES)
|
||||
scoped: list[tuple[str, str]] = []
|
||||
for path, text in iter_project_files_with_content(
|
||||
extensions=_SUFFIXES, exclude_tests=True
|
||||
):
|
||||
rel = Path(path).relative_to(PROJECT_ROOT).as_posix()
|
||||
if rel.startswith(prefixes):
|
||||
scoped.append((rel, text))
|
||||
return sorted(scoped)
|
||||
|
||||
|
||||
def _findings() -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
for rel, text in _scoped_files():
|
||||
for match in _JINJA_BLOCK.finditer(text):
|
||||
expression = match.group(1) or match.group(2) or ""
|
||||
if not _BARE_API.search(expression):
|
||||
continue
|
||||
line = text.count("\n", 0, match.start()) + 1
|
||||
findings.append(Finding(rel, line, " ".join(expression.split())))
|
||||
return findings
|
||||
|
||||
|
||||
class TestApiAccessedViaLookup(unittest.TestCase):
|
||||
def test_no_jinja_dereferences_api_directly(self) -> None:
|
||||
findings = _findings()
|
||||
|
||||
if findings:
|
||||
self.fail(
|
||||
f"{len(findings)} expression(s) read 'API' as a bare variable, which "
|
||||
"breaks against an inventory that overrides only some providers:\n"
|
||||
+ "\n".join(
|
||||
f" - {f.path}:{f.line}: {{{{ {f.expression} }}}}" for f in findings
|
||||
)
|
||||
+ "\n\nRead the value with lookup('api', '<provider>.<key>') and gate "
|
||||
"on lookup('api_enabled', '<provider>')."
|
||||
)
|
||||
|
||||
def test_the_scan_reaches_jinja_in_every_scope(self) -> None:
|
||||
"""A suffix or scope typo would make the rule above pass forever."""
|
||||
scanned = _scoped_files()
|
||||
for scope in _SCOPES:
|
||||
with self.subTest(scope=scope):
|
||||
blocks = sum(
|
||||
len(_JINJA_BLOCK.findall(text))
|
||||
for rel, text in scanned
|
||||
if rel.startswith(f"{scope}/")
|
||||
)
|
||||
self.assertTrue(blocks, f"no Jinja found under {scope}/")
|
||||
|
||||
def test_the_pattern_separates_a_dereference_from_a_declaration(self) -> None:
|
||||
self.assertTrue(_BARE_API.search("API.github.client_id | length"))
|
||||
self.assertTrue(_BARE_API.search("API['github']"))
|
||||
self.assertFalse(_BARE_API.search("lookup('api', 'github.client_id')"))
|
||||
self.assertFalse(_BARE_API.search("LITELLM_API.url"))
|
||||
self.assertFalse(_BARE_API.search("item.API.key"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,51 @@
|
||||
import unittest
|
||||
|
||||
from utils.api import declared_api, provider_enabled, resolve_api
|
||||
|
||||
|
||||
class TestResolveApi(unittest.TestCase):
|
||||
def test_a_partial_override_keeps_the_declared_providers(self) -> None:
|
||||
declared = declared_api()
|
||||
self.assertIn("github", declared)
|
||||
|
||||
resolved = resolve_api({"API": {"openai": {"api_key": "sk-test"}}})
|
||||
|
||||
self.assertEqual(set(resolved), set(declared))
|
||||
self.assertEqual(resolved["openai"]["api_key"], "sk-test")
|
||||
self.assertEqual(resolved["github"]["client_id"], "")
|
||||
|
||||
def test_an_override_wins_per_key(self) -> None:
|
||||
resolved = resolve_api({"API": {"github": {"client_id": "ID42"}}})
|
||||
|
||||
self.assertEqual(resolved["github"]["client_id"], "ID42")
|
||||
self.assertEqual(resolved["github"]["client_secret"], "")
|
||||
|
||||
def test_a_missing_api_falls_back_to_the_declaration(self) -> None:
|
||||
self.assertEqual(resolve_api({}), declared_api())
|
||||
|
||||
def test_a_non_mapping_api_is_an_error(self) -> None:
|
||||
with self.assertRaises(TypeError):
|
||||
resolve_api({"API": "nope"})
|
||||
|
||||
|
||||
class TestProviderEnabled(unittest.TestCase):
|
||||
def test_every_credential_must_carry_a_value(self) -> None:
|
||||
api = {"github": {"client_id": "ID42", "client_secret": "SEC42"}}
|
||||
|
||||
self.assertTrue(provider_enabled(api, "github"))
|
||||
|
||||
def test_one_empty_credential_disables_the_provider(self) -> None:
|
||||
api = {"github": {"client_id": "ID42", "client_secret": " "}}
|
||||
|
||||
self.assertFalse(provider_enabled(api, "github"))
|
||||
|
||||
def test_a_provider_without_credentials_is_disabled(self) -> None:
|
||||
self.assertFalse(provider_enabled({"github": {}}, "github"))
|
||||
|
||||
def test_an_undeclared_provider_is_an_error(self) -> None:
|
||||
with self.assertRaises(KeyError):
|
||||
provider_enabled({}, "github")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
"""Proprietary API credential SPOT, read from group_vars/all/18_api.yml.
|
||||
|
||||
``API`` is a plain mapping, so an inventory that configures three providers
|
||||
replaces the whole declaration instead of merging into it -- see
|
||||
``inventories/development/default.yml``, which declares openai, anthropic and
|
||||
openrouter and therefore hides github, google and jira. Resolving a hidden
|
||||
provider then raises, and a raising lookup inside a value the applications
|
||||
renderer is templating leaves the raw ``{{ ... }}`` text in place: the consumer
|
||||
stores template text, and ``| bool`` coerces it to False with a deprecation
|
||||
warning that fails the deploy on the warning gate.
|
||||
|
||||
Merging the scope's override over the declared defaults keeps every declared
|
||||
provider resolvable, so an override narrows values, never the provider set.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
from utils import PROJECT_ROOT
|
||||
from utils.cache.base import _deep_merge
|
||||
from utils.cache.files import read_text
|
||||
|
||||
_API_SPOT_FILE = str(PROJECT_ROOT / "group_vars" / "all" / "18_api.yml")
|
||||
|
||||
|
||||
def declared_api() -> dict[str, Any]:
|
||||
"""The ``API`` mapping as declared in the group_vars SPOT.
|
||||
|
||||
Returns:
|
||||
Provider name to credential mapping, every value a plain string.
|
||||
|
||||
Raises:
|
||||
TypeError: the SPOT does not declare ``API`` as a mapping.
|
||||
"""
|
||||
document = yaml.safe_load(read_text(_API_SPOT_FILE)) or {}
|
||||
declared = document.get("API")
|
||||
if not isinstance(declared, Mapping):
|
||||
raise TypeError(f"{_API_SPOT_FILE} does not declare 'API' as a mapping.")
|
||||
return dict(declared)
|
||||
|
||||
|
||||
def resolve_api(
|
||||
variables: Mapping[str, Any] | None, templar: Any = None
|
||||
) -> dict[str, Any]:
|
||||
"""The effective ``API`` mapping for a render scope.
|
||||
|
||||
Args:
|
||||
variables: the scope's variables; its ``API`` entry overrides the SPOT.
|
||||
templar: optional templar; renders the override and the merged values,
|
||||
so an inventory may source a credential from an env lookup.
|
||||
|
||||
Returns:
|
||||
The SPOT declaration with the scope's override merged over it.
|
||||
|
||||
Raises:
|
||||
TypeError: the scope's ``API`` is present but not a mapping.
|
||||
"""
|
||||
declared = declared_api()
|
||||
override: Any = (variables or {}).get("API")
|
||||
if override is not None and not isinstance(override, Mapping) and templar:
|
||||
override = _templated(templar, override)
|
||||
if override is None:
|
||||
return declared
|
||||
if not isinstance(override, Mapping):
|
||||
raise TypeError("'API' is defined but is not a mapping.")
|
||||
merged = _deep_merge(declared, dict(override))
|
||||
if templar is None:
|
||||
return merged
|
||||
rendered = _templated(templar, merged)
|
||||
return dict(rendered) if isinstance(rendered, Mapping) else merged
|
||||
|
||||
|
||||
def _templated(templar: Any, value: Any) -> Any:
|
||||
from ansible.errors import AnsibleError
|
||||
|
||||
try:
|
||||
return templar.template(value, fail_on_undefined=False)
|
||||
except (AnsibleError, TypeError, ValueError):
|
||||
return value
|
||||
|
||||
|
||||
def provider_enabled(api: Mapping[str, Any], provider: str) -> bool:
|
||||
"""Whether every credential of a provider carries a value.
|
||||
|
||||
Args:
|
||||
api: an effective ``API`` mapping, as returned by ``resolve_api``.
|
||||
provider: the provider name, e.g. ``github``.
|
||||
|
||||
Returns:
|
||||
True when the provider declares at least one credential and none of
|
||||
them is empty.
|
||||
|
||||
Raises:
|
||||
KeyError: the provider is not declared in the SPOT.
|
||||
"""
|
||||
credentials = api.get(provider)
|
||||
if not isinstance(credentials, Mapping):
|
||||
raise KeyError(
|
||||
f"unknown API provider {provider!r}; declare it under 'API' in "
|
||||
f"{_API_SPOT_FILE}."
|
||||
)
|
||||
if not credentials:
|
||||
return False
|
||||
return all(str(value).strip() for value in credentials.values())
|
||||
Reference in new issue
Block a user