mirror of
https://github.com/ansible-collections/community.general.git
synced 2026-06-04 23:37:12 +00:00
Lookup plugins: prevent using _terms for positional arguments, and reject positional arguments completely for lookups that don't use them (#12060)
Prevent using _terms for positional arguments. Also rejecting positional arguments completely for lookups that don't use them.
This commit is contained in:
parent
f97ff78c20
commit
2d8e6cb851
37 changed files with 151 additions and 1 deletions
2
.github/BOTMETA.yml
vendored
2
.github/BOTMETA.yml
vendored
|
|
@ -1503,6 +1503,8 @@ files:
|
|||
maintainers: vbotka
|
||||
$plugin_utils/_keys_filter.py:
|
||||
maintainers: vbotka
|
||||
$plugin_utils/_lookup.py:
|
||||
maintainers: felixfontein
|
||||
$plugin_utils/_unsafe.py:
|
||||
maintainers: felixfontein
|
||||
$plugin_utils/_tags.py:
|
||||
|
|
|
|||
6
changelogs/fragments/lookups.yml
Normal file
6
changelogs/fragments/lookups.yml
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
breaking_changes:
|
||||
- "all lookup plugins - if a keyword argument ``_terms`` is passed, the plugin rejects the call and tells the user to use positional arguments instead. ``_terms`` was never used, and is used in documentation as a placeholder for positional arguments (https://github.com/ansible-collections/community.general/pull/12060)."
|
||||
- "github_app_access_token lookup plugin - the plugin no longer accepts positional arguments. They were never used anyway (https://github.com/ansible-collections/community.general/pull/12060)."
|
||||
- "random_pet lookup plugin - the plugin no longer accepts positional arguments. They were never used anyway (https://github.com/ansible-collections/community.general/pull/12060)."
|
||||
- "random_string lookup plugin - the plugin no longer accepts positional arguments. They were never used anyway (https://github.com/ansible-collections/community.general/pull/12060)."
|
||||
- "random_words lookup plugin - the plugin no longer accepts positional arguments. They were never used anyway (https://github.com/ansible-collections/community.general/pull/12060)."
|
||||
|
|
@ -81,12 +81,15 @@ from ansible.errors import AnsibleLookupError
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
display = Display()
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
not_exist = self.get_option("not_exist")
|
||||
|
||||
result = []
|
||||
|
|
|
|||
|
|
@ -125,6 +125,8 @@ from ansible.module_utils.common.text.converters import to_bytes, to_text
|
|||
from ansible.parsing.ajson import AnsibleJSONDecoder
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class BitwardenException(AnsibleError):
|
||||
pass
|
||||
|
|
@ -254,6 +256,7 @@ class Bitwarden:
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms=None, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
field = self.get_option("field")
|
||||
search_field = self.get_option("search")
|
||||
collection_id = self.get_option("collection_id")
|
||||
|
|
|
|||
|
|
@ -75,6 +75,7 @@ from ansible.parsing.ajson import AnsibleJSONDecoder
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.module_utils._version import LooseVersion
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class BitwardenSecretsManagerException(AnsibleLookupError):
|
||||
|
|
@ -147,6 +148,7 @@ class BitwardenSecretsManager:
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
bws_access_token = self.get_option("bws_access_token")
|
||||
|
||||
return [_bitwarden_secrets_manager.get_secret(term, bws_access_token) for term in terms]
|
||||
|
|
|
|||
|
|
@ -49,6 +49,8 @@ from ansible.errors import AnsibleError
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.listify import listify_lookup_plugin_terms
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
"""
|
||||
|
|
@ -69,6 +71,7 @@ class LookupModule(LookupBase):
|
|||
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
terms = self._lookup_variables(terms)
|
||||
|
||||
|
|
|
|||
|
|
@ -87,6 +87,8 @@ class LookupModule(LookupBase):
|
|||
if not HAS_CHEF:
|
||||
raise AnsibleError("PyChef needed for lookup plugin, try `pip install pychef`")
|
||||
|
||||
# TODO: use new-style option parsing
|
||||
|
||||
for term in terms:
|
||||
self.parse_kv_args(parse_kv(term))
|
||||
|
||||
|
|
|
|||
|
|
@ -64,6 +64,8 @@ import yaml
|
|||
from ansible.errors import AnsibleLookupError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
FQCN_RE = re.compile(r"^[A-Za-z0-9_]+\.[A-Za-z0-9_]+$")
|
||||
|
||||
|
||||
|
|
@ -103,6 +105,7 @@ class LookupModule(LookupBase):
|
|||
def run(self, terms, variables=None, **kwargs):
|
||||
result = []
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
not_found = self.get_option("result_not_found")
|
||||
no_version = self.get_option("result_no_version")
|
||||
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ description:
|
|||
requirements:
|
||||
- 'py-consul python library U(https://github.com/criteo/py-consul?tab=readme-ov-file#installation)'
|
||||
options:
|
||||
_raw:
|
||||
_terms:
|
||||
description: List of key(s) to retrieve.
|
||||
type: list
|
||||
elements: string
|
||||
|
|
@ -126,6 +126,8 @@ from ansible.errors import AnsibleAssertionError, AnsibleError
|
|||
from ansible.module_utils.common.text.converters import to_text
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
try:
|
||||
import consul
|
||||
|
||||
|
|
@ -143,6 +145,7 @@ class LookupModule(LookupBase):
|
|||
|
||||
# get options
|
||||
self.set_options(direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
scheme = self.get_option("scheme")
|
||||
host = self.get_option("host")
|
||||
|
|
|
|||
|
|
@ -94,6 +94,8 @@ _raw:
|
|||
from ansible.errors import AnsibleError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
CREDSTASH_INSTALLED = False
|
||||
|
||||
try:
|
||||
|
|
@ -110,6 +112,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError("The credstash lookup plugin requires credstash to be installed.")
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
version = self.get_option("version")
|
||||
region = self.get_option("region")
|
||||
|
|
|
|||
|
|
@ -179,6 +179,7 @@ class LookupModule(LookupBase):
|
|||
"""
|
||||
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
# TODO: use new-style option parsing
|
||||
display.vvvv(f"{terms}")
|
||||
if isinstance(terms, list):
|
||||
return_values = []
|
||||
|
|
|
|||
|
|
@ -124,6 +124,8 @@ from ansible.errors import AnsibleLookupError
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.template import Templar
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
try:
|
||||
from ansible.template import trust_as_template as _trust_as_template
|
||||
|
||||
|
|
@ -195,6 +197,7 @@ class LookupModule(LookupBase):
|
|||
def run(self, terms, variables=None, **kwargs):
|
||||
"""Generate list."""
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
result = []
|
||||
if len(terms) > 0:
|
||||
|
|
|
|||
|
|
@ -247,6 +247,8 @@ from ansible.module_utils.parsing.convert_bool import boolean
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
try:
|
||||
import dns.exception
|
||||
import dns.name
|
||||
|
|
@ -371,6 +373,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError("The dig lookup requires the python 'dnspython' library and it is not installed")
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
# Create Resolver object so that we can set NS if necessary
|
||||
myres = dns.resolver.Resolver(configure=True)
|
||||
|
|
|
|||
|
|
@ -65,6 +65,8 @@ except ImportError:
|
|||
from ansible.errors import AnsibleError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
# ==============================================================
|
||||
# DNSTXT: DNS TXT records
|
||||
#
|
||||
|
|
@ -76,6 +78,7 @@ from ansible.plugins.lookup import LookupBase
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
if HAVE_DNS is False:
|
||||
raise AnsibleError("Can't LOOKUP(dnstxt): module dns.resolver is not installed")
|
||||
|
|
|
|||
|
|
@ -85,6 +85,8 @@ EXAMPLES = r"""
|
|||
|
||||
from ansible.errors import AnsibleError, AnsibleOptionsError
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
sdk_is_missing = False
|
||||
|
||||
try:
|
||||
|
|
@ -115,6 +117,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError("python-dsv-sdk==0.0.1 must be installed to use this plugin")
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
vault = LookupModule.Client(
|
||||
{
|
||||
|
|
|
|||
|
|
@ -72,6 +72,8 @@ import json
|
|||
from ansible.module_utils.urls import open_url
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
# this can be made configurable, not should not use ansible.cfg
|
||||
#
|
||||
# Made module configurable from playbooks:
|
||||
|
|
@ -156,6 +158,7 @@ class Etcd:
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
validate_certs = self.get_option("validate_certs")
|
||||
url = self.get_option("url")
|
||||
|
|
|
|||
|
|
@ -151,6 +151,8 @@ from ansible.module_utils.common.text.converters import to_native
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
try:
|
||||
import etcd3
|
||||
|
||||
|
|
@ -185,6 +187,7 @@ def etcd3_client(client_params):
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
if not HAS_ETCD:
|
||||
display.error(missing_required_lib("etcd3"))
|
||||
|
|
|
|||
|
|
@ -128,6 +128,7 @@ _raw:
|
|||
description: Time of last metadata update or creation (depends on OS).
|
||||
type: float
|
||||
"""
|
||||
|
||||
import grp
|
||||
import os
|
||||
import pwd
|
||||
|
|
@ -146,6 +147,8 @@ from ansible.module_utils.common.text.converters import to_native, to_text
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
display = Display()
|
||||
|
||||
|
||||
|
|
@ -218,6 +221,7 @@ def file_props(root, path):
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
basedir = self.get_basedir(variables)
|
||||
|
||||
|
|
|
|||
|
|
@ -34,10 +34,13 @@ _raw:
|
|||
- Flattened list.
|
||||
type: list
|
||||
"""
|
||||
|
||||
from ansible.errors import AnsibleError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.listify import listify_lookup_plugin_terms
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
def _check_list_of_one_list(self, term):
|
||||
|
|
@ -81,5 +84,6 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError("with_flattened expects a list")
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
return self._do_flatten(terms, variables)
|
||||
|
|
|
|||
|
|
@ -105,6 +105,8 @@ from ansible.module_utils.urls import open_url
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_no_terms
|
||||
|
||||
display = Display()
|
||||
|
||||
|
||||
|
|
@ -208,6 +210,7 @@ class LookupModule(LookupBase):
|
|||
)
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_no_terms(self, terms=terms, direct=kwargs)
|
||||
|
||||
if not (self.get_option("key_path") or self.get_option("private_key")):
|
||||
raise AnsibleOptionsError("One of key_path or private_key is required")
|
||||
|
|
|
|||
|
|
@ -14,6 +14,13 @@ requirements:
|
|||
short_description: Grab secrets from the OS keyring
|
||||
description:
|
||||
- Allows you to access data stored in the OS provided keyring/keychain.
|
||||
options:
|
||||
_terms:
|
||||
description:
|
||||
- List of pairs of service and user name.
|
||||
- Every entry must be of the form V(servicename username).
|
||||
type: list
|
||||
elements: str
|
||||
"""
|
||||
|
||||
EXAMPLES = r"""
|
||||
|
|
@ -41,6 +48,8 @@ HAS_KEYRING = True
|
|||
from ansible.errors import AnsibleError
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
try:
|
||||
import keyring
|
||||
except ImportError:
|
||||
|
|
@ -57,6 +66,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError("Can't LOOKUP(keyring): missing required python library 'keyring'")
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
display.vvvv(f"keyring: {keyring.get_keyring()}")
|
||||
ret = []
|
||||
|
|
|
|||
|
|
@ -45,6 +45,8 @@ from ansible.errors import AnsibleError
|
|||
from ansible.module_utils.common.text.converters import to_bytes, to_text
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class LPassException(AnsibleError):
|
||||
pass
|
||||
|
|
@ -89,6 +91,7 @@ class LPass:
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
field = self.get_option("field")
|
||||
|
||||
lp = LPass()
|
||||
|
|
|
|||
|
|
@ -62,6 +62,8 @@ from ansible.errors import AnsibleError
|
|||
from ansible.module_utils.common.text.converters import to_native, to_text
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
HAVE_LMDB = True
|
||||
try:
|
||||
import lmdb
|
||||
|
|
@ -87,6 +89,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError("Can't LOOKUP(lmdb_kv): this module requires lmdb to be installed")
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
db = self.get_option("db")
|
||||
|
||||
|
|
|
|||
|
|
@ -275,6 +275,8 @@ from ansible.errors import AnsibleError
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
|
|
@ -294,6 +296,7 @@ def _verify_and_get_type(variable: t.Any) -> str:
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms: list[str], variables: dict[str, t.Any], **kwargs) -> list[t.Any]:
|
||||
self.set_options(direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
initial_value = self.get_option("initial_value", None)
|
||||
self._override_behavior = self.get_option("override", "error")
|
||||
self._pattern_type = self.get_option("pattern_type", "regex")
|
||||
|
|
|
|||
|
|
@ -89,6 +89,7 @@ from ansible.module_utils.common.text.converters import to_bytes, to_text
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.module_utils._onepassword import OnePasswordConfig
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
def _lower_if_possible(value):
|
||||
|
|
@ -731,6 +732,7 @@ class OnePass:
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
field = self.get_option("field")
|
||||
section = self.get_option("section")
|
||||
|
|
|
|||
|
|
@ -46,6 +46,7 @@ _raw:
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.lookup.onepassword import OnePass, OnePassCLIv2
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class OnePassCLIv2Doc(OnePassCLIv2):
|
||||
|
|
@ -57,6 +58,7 @@ class OnePassCLIv2Doc(OnePassCLIv2):
|
|||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
vault = self.get_option("vault")
|
||||
subdomain = self.get_option("subdomain")
|
||||
|
|
|
|||
|
|
@ -57,11 +57,13 @@ import json
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.lookup.onepassword import OnePass
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
vault = self.get_option("vault")
|
||||
subdomain = self.get_option("subdomain")
|
||||
|
|
|
|||
|
|
@ -56,6 +56,7 @@ from ansible_collections.community.general.plugins.lookup.onepassword import (
|
|||
OnePass,
|
||||
OnePassCLIv2,
|
||||
)
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
|
|
@ -82,6 +83,7 @@ class LookupModule(LookupBase):
|
|||
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
ssh_format = self.get_option("ssh_format")
|
||||
vault = self.get_option("vault")
|
||||
|
|
|
|||
|
|
@ -255,6 +255,7 @@ from ansible.utils.display import Display
|
|||
from ansible.utils.encrypt import random_password
|
||||
|
||||
from ansible_collections.community.general.plugins.module_utils._filelock import FileLock
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
display = Display()
|
||||
|
||||
|
|
@ -540,6 +541,7 @@ class LookupModule(LookupBase):
|
|||
|
||||
def run(self, terms, variables, **kwargs):
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
self.setup(variables)
|
||||
result = []
|
||||
|
||||
|
|
|
|||
|
|
@ -75,6 +75,8 @@ except ImportError:
|
|||
from ansible.errors import AnsibleError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_no_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
|
|
@ -82,6 +84,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError('Python petname library is required. Please install using "pip install petname"')
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_no_terms(self, terms=terms, direct=kwargs)
|
||||
words = self.get_option("words")
|
||||
length = self.get_option("length")
|
||||
prefix = self.get_option("prefix")
|
||||
|
|
|
|||
|
|
@ -173,6 +173,8 @@ from ansible.errors import AnsibleLookupError
|
|||
from ansible.module_utils.common.text.converters import to_bytes, to_text
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_no_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
@staticmethod
|
||||
|
|
@ -192,6 +194,7 @@ class LookupModule(LookupBase):
|
|||
special_chars = string.punctuation
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_no_terms(self, terms=terms, direct=kwargs)
|
||||
|
||||
length = self.get_option("length")
|
||||
base64_flag = self.get_option("base64")
|
||||
|
|
|
|||
|
|
@ -79,6 +79,8 @@ _raw:
|
|||
from ansible.errors import AnsibleLookupError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_no_terms
|
||||
|
||||
try:
|
||||
from xkcdpass import xkcd_password as xp
|
||||
|
||||
|
|
@ -95,6 +97,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleLookupError('Python xkcdpass library is required. Please install using "pip install xkcdpass"')
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_no_terms(self, terms=terms, direct=kwargs)
|
||||
method = self.get_option("case")
|
||||
delimiter = self.get_option("delimiter")
|
||||
max_length = self.get_option("max_length")
|
||||
|
|
|
|||
|
|
@ -85,6 +85,8 @@ from ansible.errors import AnsibleError
|
|||
from ansible.module_utils.common.text.converters import to_text
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
def run(self, terms, variables, **kwargs):
|
||||
|
|
@ -93,6 +95,7 @@ class LookupModule(LookupBase):
|
|||
|
||||
# get options
|
||||
self.set_options(direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
# setup connection
|
||||
host = self.get_option("host")
|
||||
|
|
|
|||
|
|
@ -64,6 +64,8 @@ from ansible.errors import AnsibleError
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
ANOTHER_LIBRARY_IMPORT_ERROR: ImportError | None
|
||||
try:
|
||||
from pam.revbits_ansible.server import SecretServer
|
||||
|
|
@ -85,6 +87,7 @@ class LookupModule(LookupBase):
|
|||
if ANOTHER_LIBRARY_IMPORT_ERROR:
|
||||
raise AnsibleError("revbits_ansible must be installed to use this plugin") from ANOTHER_LIBRARY_IMPORT_ERROR
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
secret_server = LookupModule.Client(
|
||||
{
|
||||
"base_url": self.get_option("base_url"),
|
||||
|
|
|
|||
|
|
@ -44,6 +44,8 @@ from ansible.errors import AnsibleAssertionError, AnsibleError
|
|||
from ansible.module_utils.common.text.converters import to_bytes, to_text
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
def read_shelve(self, shelve_filename, key):
|
||||
|
|
@ -56,9 +58,12 @@ class LookupModule(LookupBase):
|
|||
return res
|
||||
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
# TODO: use new-style option parsing
|
||||
if not isinstance(terms, list):
|
||||
terms = [terms]
|
||||
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
ret = []
|
||||
|
||||
for term in terms:
|
||||
|
|
|
|||
|
|
@ -305,6 +305,8 @@ from ansible.errors import AnsibleError, AnsibleOptionsError
|
|||
from ansible.plugins.lookup import LookupBase
|
||||
from ansible.utils.display import Display
|
||||
|
||||
from ansible_collections.community.general.plugins.plugin_utils._lookup import check_for_wrong_terms
|
||||
|
||||
try:
|
||||
from delinea.secrets.server import (
|
||||
AccessTokenAuthorizer,
|
||||
|
|
@ -465,6 +467,7 @@ class LookupModule(LookupBase):
|
|||
raise AnsibleError("python-tss-sdk must be installed to use this plugin")
|
||||
|
||||
self.set_options(var_options=variables, direct=kwargs)
|
||||
check_for_wrong_terms(self, direct=kwargs)
|
||||
|
||||
tss = TSSClient.from_params(
|
||||
base_url=self.get_option("base_url"),
|
||||
|
|
|
|||
38
plugins/plugin_utils/_lookup.py
Normal file
38
plugins/plugin_utils/_lookup.py
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
# Copyright (c) 2026 Felix Fontein <felix@fontein.de>
|
||||
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
# Note that this plugin util is **PRIVATE** to the collection. It can have breaking changes at any time.
|
||||
# Do not use this from other collections or standalone plugins/modules!
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import typing as t
|
||||
|
||||
from ansible.errors import AnsibleLookupError
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
from collections.abc import Mapping, Sequence
|
||||
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
|
||||
|
||||
def check_for_wrong_terms(plugin: LookupBase, *, direct: Mapping[str, t.Any]) -> None:
|
||||
# Note that we don't check "terms" here since the keyword argument "terms"
|
||||
# is mapped by Python to the run() positional argument "terms".
|
||||
for opt in ("_terms",):
|
||||
if opt in direct:
|
||||
raise AnsibleLookupError(
|
||||
f"The {opt!r} keyword argument is not supported, you must provide terms as positional arguments: use"
|
||||
f" lookup({plugin.ansible_name!r}, arg1, arg2) instead of lookup({plugin.ansible_name!r}, {opt}=[arg1, arg2])"
|
||||
)
|
||||
|
||||
|
||||
def check_for_no_terms(plugin: LookupBase, *, terms: Sequence[t.Any], direct: Mapping[str, t.Any]) -> None:
|
||||
if terms:
|
||||
raise AnsibleLookupError("The lookup plugin does not accept positional arguments")
|
||||
# Note that we don't check "terms" here since the keyword argument "terms"
|
||||
# is mapped by Python to the run() positional argument "terms".
|
||||
for opt in ("_terms",):
|
||||
if opt in direct:
|
||||
raise AnsibleLookupError(f"The {opt!r} keyword argument is not supported")
|
||||
Loading…
Add table
Add a link
Reference in a new issue