Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,13 @@ repos:
- id: 'mixed-line-ending'
- id: 'trailing-whitespace'

- repo: 'https://github.com/astral-sh/ruff-pre-commit'
rev: 'v0.16.1'
hooks:
- id: 'ruff-check'
args: ['--fix']
- id: 'ruff-format'

- repo: 'https://github.com/PyCQA/bandit'
rev: '1.9.4'
hooks:
Expand Down
19 changes: 10 additions & 9 deletions plugins/filter/combine_lod.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

from ansible.errors import AnsibleFilterError

DOCUMENTATION = r'''
DOCUMENTATION = r"""
name: combine_lod
version_added: "3.0.0"
short_description: Merge lists of dictionaries by a unique key
Expand Down Expand Up @@ -45,9 +45,9 @@
- Pass a list when no single key is unique on its own (e.g. C(["server_name", "server_port"]) for vHosts where the same hostname can appear on multiple ports).
type: raw
default: name
'''
"""

EXAMPLES = r'''
EXAMPLES = r"""
# create two lists of dictionaries
- set_fact:
# this list could be in the role defaults
Expand Down Expand Up @@ -152,14 +152,14 @@
# value: 1
# - name: net.core.somaxconn
# value: 2048
'''
"""

RETURN = r'''
RETURN = r"""
_value:
description: Resulting merged list of dictionaries.
type: list
elements: dictionary
'''
"""


def combine_lod(*args, **kwargs):
Expand All @@ -184,7 +184,9 @@ def combine_lod(*args, **kwargs):
for lod in list(args):
for item in lod:
if not isinstance(item, collections.abc.MutableMapping):
raise AnsibleFilterError('found a non-dictionary item in the list, this is not supported')
raise AnsibleFilterError(
'found a non-dictionary item in the list, this is not supported'
)

# A unique_key is the item's identity, so every key must be set
# explicitly and may not be left to a default applied elsewhere:
Expand Down Expand Up @@ -223,8 +225,7 @@ def combine_lod(*args, **kwargs):
return list(result.values())



class FilterModule(object):
class FilterModule:
"""Register custom filter plugins in Ansible"""

def filters(self):
Expand Down
36 changes: 18 additions & 18 deletions plugins/filter/platform_select.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@

from ansible.errors import AnsibleFilterError

DOCUMENTATION = r'''
DOCUMENTATION = r"""
name: platform_select
version_added: "6.0.2"
short_description: Pick the value matching the target host from a platform-keyed dictionary
Expand All @@ -34,9 +34,9 @@
description: Value to return when no key in I(_input) matches the target host. If omitted, an unmatched call raises C(AnsibleFilterError).
type: raw
required: false
'''
"""

EXAMPLES = r'''
EXAMPLES = r"""
# in a role's vars/main.yml (auto-loaded at play parse, so visible to roles
# that run earlier in the same play via the `__dependent_var` pattern):
mariadb_server__python__modules__dependent_var:
Expand Down Expand Up @@ -67,13 +67,13 @@
+ (apache_httpd__python__modules__dependent_var
| linuxfabrik.lfops.platform_select(ansible_facts, default=[]))
}}'
'''
"""

RETURN = r'''
RETURN = r"""
_value:
description: The value associated with the most specific matching key in I(_input), or the supplied I(default) if no key matches.
type: raw
'''
"""


_SENTINEL = object()
Expand All @@ -87,13 +87,13 @@ def platform_select(values, ansible_facts, default=_SENTINEL):
"""
if not isinstance(values, dict):
raise AnsibleFilterError(
"platform_select: input must be a dict keyed by platform identifier, "
f"got {type(values).__name__}"
'platform_select: input must be a dict keyed by platform identifier, '
f'got {type(values).__name__}'
)
if not isinstance(ansible_facts, dict):
raise AnsibleFilterError(
"platform_select: ansible_facts must be a dict, "
f"got {type(ansible_facts).__name__}"
'platform_select: ansible_facts must be a dict, '
f'got {type(ansible_facts).__name__}'
)

os_family = ansible_facts.get('os_family')
Expand All @@ -106,10 +106,10 @@ def platform_select(values, ansible_facts, default=_SENTINEL):
# least to most specific and the later (more specific) call wins.
candidates = [
f'{distribution}{version}' if distribution and version else None,
f'{distribution}{major}' if distribution and major else None,
f'{distribution}{major}' if distribution and major else None,
distribution,
f'{os_family}{version}' if os_family and version else None,
f'{os_family}{major}' if os_family and major else None,
f'{os_family}{version}' if os_family and version else None,
f'{os_family}{major}' if os_family and major else None,
os_family,
]
for key in candidates:
Expand All @@ -119,14 +119,14 @@ def platform_select(values, ansible_facts, default=_SENTINEL):
if default is not _SENTINEL:
return default
raise AnsibleFilterError(
f"platform_select: no key in the input dict matched the target host "
f"(os_family={os_family!r}, distribution={distribution!r}, "
f"distribution_major_version={major!r}, distribution_version={version!r}); "
f"input keys: {sorted(values)}"
f'platform_select: no key in the input dict matched the target host '
f'(os_family={os_family!r}, distribution={distribution!r}, '
f'distribution_major_version={major!r}, distribution_version={version!r}); '
f'input keys: {sorted(values)}'
)


class FilterModule(object):
class FilterModule:
"""Register custom filter plugins in Ansible"""

def filters(self):
Expand Down
42 changes: 27 additions & 15 deletions plugins/lookup/bitwarden_item.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

__metaclass__ = type

DOCUMENTATION = r'''
DOCUMENTATION = r"""
lookup: bitwarden_item

short_description: Fetch (or create) a Bitwarden login item
Expand Down Expand Up @@ -93,9 +93,9 @@
description: Username for the login item. Used both as a search filter and, if the item has to be created, as the C(login.username) value.
required: False
type: str
'''
"""

EXAMPLES = r'''
EXAMPLES = r"""
- name: 'The normal way using this lookup plugin. Search for the Bitwarden item using hostname, purpose and username. If not found, creates a new item called `appsrv01 - MariaDB`. Returns the password item, including a `username` and a `password` subkey.'
ansible.builtin.debug:
msg: "{{ lookup('linuxfabrik.lfops.bitwarden_item',
Expand Down Expand Up @@ -170,9 +170,9 @@
'collection_id': '16ea112a-dd5f-4f68-9dfb-95a9f302a8a5',
},
) }}"
'''
"""

RETURN = r'''
RETURN = r"""
collectionIds:
description: List of collection IDs in which the item is.
type: list
Expand Down Expand Up @@ -278,26 +278,29 @@
type: str
returned: always
sample: 'root'
'''
"""

from ansible.errors import AnsibleError
from ansible.plugins.lookup import LookupBase
from ansible.utils.display import Display
from ansible_collections.linuxfabrik.lfops.plugins.module_utils.bitwarden import \
Bitwarden
from ansible_collections.linuxfabrik.lfops.plugins.module_utils.bitwarden import (
Bitwarden,
)

display = Display() # log prefix "lfbwlp" = Linuxfabrik Bitwarden Lookup Plugin

# https://docs.ansible.com/ansible/latest/dev_guide/developing_plugins.html#developing-lookup-plugins
# inspired by the lookup plugins lastpass (same topic) and redis (more modern)

class LookupModule(LookupBase):

class LookupModule(LookupBase):
def run(self, terms, variables=None, **kwargs):
bw = Bitwarden()

if not bw.is_unlocked:
raise AnsibleError('Not logged into Bitwarden, or Bitwarden Vault is locked. Please run `bw login` and `bw unlock` first.')
raise AnsibleError(
'Not logged into Bitwarden, or Bitwarden Vault is locked. Please run `bw login` and `bw unlock` first.'
)
display.vvv('lfbwlp - run - bitwarden vault is unlocked')

bw.sync()
Expand All @@ -314,12 +317,17 @@ def run(self, terms, variables=None, **kwargs):
notes = term.get('notes', 'Generated by Ansible.')
organization_id = term.get('organization_id', None)
password_length = term.get('password_length', 60)
password_choice = term.get('password_choice', '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ')
password_choice = term.get(
'password_choice',
'0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ',
)
purpose = term.get('purpose', None)
uris = term.get('uris', [])
username = term.get('username', None)
except Exception as e:
raise AnsibleError(f'Encountered exception while fetching {term}: {e}')
raise AnsibleError(
f'Encountered exception while fetching {term}: {e}'
) from e

if id_:
result = bw.get_item_by_id(id_)
Expand All @@ -329,17 +337,21 @@ def run(self, terms, variables=None, **kwargs):
result['username'] = result['login']['username']
result['password'] = result['login']['password']
ret.append(result)
continue # done here, go to next term
continue # done here, go to next term
else:
# item not found by ID. if there is an ID given we expect it to exist
raise AnsibleError(f'Item with id {id_} not found.')

name = Bitwarden.get_pretty_name(name, hostname, purpose)
display.vvv(f'lfbwlp - run - get item: {name}')
result = bw.get_items(name, username, folder_id, collection_id, organization_id)
result = bw.get_items(
name, username, folder_id, collection_id, organization_id
)

if len(result) > 1:
raise AnsibleError('Found multiple Bitwarden items with the same name/title and username, cannot decide which one to use. Aborting.')
raise AnsibleError(
'Found multiple Bitwarden items with the same name/title and username, cannot decide which one to use. Aborting.'
)

if len(result) == 1:
display.vvv('lfbwlp - run - found existing item')
Expand Down
Loading