passwordstore: Make compatible with shims (#4780) (#4846)

* passwordstore: Make compatible with shims, add backend config

This allows using the passwordstore plugin with scripts that wrap other
password managers. Also adds an explicit configuration (`backend` in
`ini` and `passwordstore_backend` in `vars`) to set the backend to `pass`
(the default) or `gopass`, which allows using gopass as the backend
without the need of a wrapper script. Please be aware that gopass
support is currently limited, but will work for basic operations.

Includes integrations tests.

Resolves #4766

* Apply suggestions from code review

(cherry picked from commit 006f3bfa89)

Co-authored-by: grembo <freebsd@grem.de>

changelogs/fragments/4780-passwordstore-wrapper-compat.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - passwordstore lookup plugin - allow using alternative password managers by detecting wrapper scripts, allow explicit configuration of pass and gopass backends (https://github.com/ansible-collections/community.general/issues/4766).

plugins/lookup/passwordstore.py

@@ -106,6 +106,22 @@ DOCUMENTATION = '''
         type: str
         default: 15m
         version_added: 4.5.0
+      backend:
+        description:
+          - Specify which backend to use.
+          - Defaults to C(pass), passwordstore.org's original pass utility.
+          - C(gopass) support is incomplete.
+        ini:
+          - section: passwordstore_lookup
+            key: backend
+        vars:
+          - name: passwordstore_backend
+        type: str
+        default: pass
+        choices:
+          - pass
+          - gopass
+        version_added: 5.2.0
 '''
 EXAMPLES = """
 ansible.cfg: |
@@ -231,6 +247,24 @@ def check_output2(*popenargs, **kwargs):
 
 
 class LookupModule(LookupBase):
+    def __init__(self, loader=None, templar=None, **kwargs):
+
+        super(LookupModule, self).__init__(loader, templar, **kwargs)
+        self.realpass = None
+
+    def is_real_pass(self):
+        if self.realpass is None:
+            try:
+                self.passoutput = to_text(
+                    check_output2([self.pass_cmd, "--version"], env=self.env),
+                    errors='surrogate_or_strict'
+                )
+                self.realpass = 'pass: the standard unix password manager' in self.passoutput
+            except (subprocess.CalledProcessError) as e:
+                raise AnsibleError(e)
+
+        return self.realpass
+
     def parse_params(self, term):
         # I went with the "traditional" param followed with space separated KV pairs.
         # Waiting for final implementation of lookup parameter parsing.
@@ -270,10 +304,12 @@ class LookupModule(LookupBase):
             self.env = os.environ.copy()
             self.env['LANGUAGE'] = 'C'  # make sure to get errors in English as required by check_output2
 
+            if self.backend == 'gopass':
+                self.env['GOPASS_NO_REMINDER'] = "YES"
+            elif os.path.isdir(self.paramvals['directory']):
                 # Set PASSWORD_STORE_DIR
-            if os.path.isdir(self.paramvals['directory']):
                 self.env['PASSWORD_STORE_DIR'] = self.paramvals['directory']
-            else:
+            elif self.is_real_pass():
                 raise AnsibleError('Passwordstore directory \'{0}\' does not exist'.format(self.paramvals['directory']))
 
             # Set PASSWORD_STORE_UMASK if umask is set
@@ -288,7 +324,9 @@ class LookupModule(LookupBase):
     def check_pass(self):
         try:
             self.passoutput = to_text(
-                check_output2(["pass", "show", self.passname], env=self.env),
+                check_output2([self.pass_cmd, 'show'] +
+                              (['--password'] if self.backend == 'gopass' else []) +
+                              [self.passname], env=self.env),
                 errors='surrogate_or_strict'
             ).splitlines()
             self.password = self.passoutput[0]
@@ -302,8 +340,10 @@ class LookupModule(LookupBase):
                     if ':' in line:
                         name, value = line.split(':', 1)
                         self.passdict[name.strip()] = value.strip()
-            if os.path.isfile(os.path.join(self.paramvals['directory'], self.passname + ".gpg")):
-                # Only accept password as found, if there a .gpg file for it (might be a tree node otherwise)
+            if (self.backend == 'gopass' or
+                    os.path.isfile(os.path.join(self.paramvals['directory'], self.passname + ".gpg"))
+                    or not self.is_real_pass()):
+                # When using real pass, only accept password as found if there is a .gpg file for it (might be a tree node otherwise)
                 return True
         except (subprocess.CalledProcessError) as e:
             # 'not in password store' is the expected error if a password wasn't found
@@ -339,7 +379,7 @@ class LookupModule(LookupBase):
         if self.paramvals['backup']:
             msg += "lookup_pass: old password was {0} (Updated on {1})\n".format(self.password, datetime)
         try:
-            check_output2(['pass', 'insert', '-f', '-m', self.passname], input=msg, env=self.env)
+            check_output2([self.pass_cmd, 'insert', '-f', '-m', self.passname], input=msg, env=self.env)
         except (subprocess.CalledProcessError) as e:
             raise AnsibleError(e)
         return newpass
@@ -351,7 +391,7 @@ class LookupModule(LookupBase):
         datetime = time.strftime("%d/%m/%Y %H:%M:%S")
         msg = newpass + '\n' + "lookup_pass: First generated by ansible on {0}\n".format(datetime)
         try:
-            check_output2(['pass', 'insert', '-f', '-m', self.passname], input=msg, env=self.env)
+            check_output2([self.pass_cmd, 'insert', '-f', '-m', self.passname], input=msg, env=self.env)
         except (subprocess.CalledProcessError) as e:
             raise AnsibleError(e)
         return newpass
@@ -380,6 +420,8 @@ class LookupModule(LookupBase):
             yield
 
     def setup(self, variables):
+        self.backend = self.get_option('backend')
+        self.pass_cmd = self.backend  # pass and gopass are commands as well
         self.locked = None
         timeout = self.get_option('locktimeout')
         if not re.match('^[0-9]+[smh]$', timeout):
@@ -402,6 +444,7 @@ class LookupModule(LookupBase):
         }
 
     def run(self, terms, variables, **kwargs):
+        self.set_options(var_options=variables, direct=kwargs)
         self.setup(variables)
         result = []
 

tests/integration/targets/lookup_passwordstore/tasks/tests.yml

@@ -19,6 +19,44 @@
     - "~/.gnupg"
     - "~/.password-store"
 
+- name: Get path of pass executable
+  command: which pass
+  register: result
+
+- name: Store path of pass executable
+  set_fact:
+     passpath: "{{ result.stdout }}"
+
+- name: Move original pass into place if there was a leftover
+  command:
+    argv:
+      - mv
+      - "{{ passpath }}.testorig"
+      - "{{ passpath }}"
+  args:
+    removes: "{{ passpath }}.testorig"
+
+# having gopass is not required for this test, but we store
+# its path in case it is installed, so we can restore it
+- name: Try to find gopass in path
+  command: which gopass
+  register: result
+  ignore_errors: yes
+
+- name: Store path of gopass executable
+  set_fact:
+     gopasspath: "{{ (result.rc == 0) |
+       ternary(result.stdout, (passpath | dirname, 'gopass') | path_join) }}"
+
+- name: Move original gopass into place if there was a leftover
+  command:
+    argv:
+      - mv
+      - "{{ gopasspath }}.testorig"
+      - "{{ gopasspath }}"
+  args:
+    removes: "{{ gopasspath }}.testorig"
+
 # How to generate a new GPG key:
 #   gpg2 --batch --gen-key input  # See templates/input
 #   gpg2 --list-secret-keys --keyid-format LONG
@@ -151,3 +189,163 @@
   assert:
     that:
       - readyamlpass == 'testpassword\nrandom additional line'
+
+- name: Create a password in a folder
+  set_fact:
+    newpass: "{{ lookup('community.general.passwordstore', 'folder/test-pass length=8 create=yes') }}"
+
+- name: Fetch password from folder
+  set_fact:
+    readpass: "{{ lookup('community.general.passwordstore', 'folder/test-pass') }}"
+
+- name: Verify password from folder
+  assert:
+    that:
+      - readpass == newpass
+
+- name: Try to read folder as passname
+  set_fact:
+    newpass: "{{ lookup('community.general.passwordstore', 'folder') }}"
+  ignore_errors: true
+  register: eval_error
+
+- name: Make sure reading folder as passname failed
+  assert:
+    that:
+      - eval_error is failed
+      - '"passname folder not found" in eval_error.msg'
+
+- name: Change passwordstore location explicitly
+  set_fact:
+    passwordstore: "{{ lookup('env','HOME') }}/.password-store"
+
+- name: Make sure password store still works with explicit location set
+  set_fact:
+    newpass: "{{ lookup('community.general.passwordstore', 'test-pass') }}"
+
+- name: Change passwordstore location to a non-existent place
+  set_fact:
+    passwordstore: "somenonexistentplace"
+
+- name: Try reading from non-existent passwordstore location
+  set_fact:
+    newpass: "{{ lookup('community.general.passwordstore', 'test-pass') }}"
+  ignore_errors: true
+  register: eval_error
+
+- name: Make sure reading from non-existent passwordstore location failed
+  assert:
+    that:
+      - eval_error is failed
+      - >-
+        "Passwordstore directory 'somenonexistentplace' does not exist" in eval_error.msg
+
+- name: Test pass compatibility shim detection
+  block:
+  - name: Move original pass out of the way
+    command:
+      argv:
+        - mv
+        - "{{ passpath }}"
+        - "{{ passpath }}.testorig"
+    args:
+      creates: "{{ passpath }}.testorig"
+
+  - name: Create dummy pass script
+    ansible.builtin.copy:
+      content: |
+        #!/bin/sh
+        echo "shim_ok"
+      dest: "{{ passpath }}"
+      mode: '0755'
+
+  - name: Try reading from non-existent passwordstore location with different pass utility
+    set_fact:
+      newpass: "{{ lookup('community.general.passwordstore', 'test-pass') }}"
+    environment:
+      PATH: "/tmp"
+
+  - name: Verify password received from shim
+    assert:
+      that:
+        - newpass == "shim_ok"
+
+  - name: Try to read folder as passname with a different pass utility
+    set_fact:
+      newpass: "{{ lookup('community.general.passwordstore', 'folder') }}"
+
+  - name: Verify password received from shim
+    assert:
+      that:
+        - newpass == "shim_ok"
+
+  always:
+  - name: Move original pass back into place
+    command:
+      argv:
+        - mv
+        - "{{ passpath }}.testorig"
+        - "{{ passpath }}"
+    args:
+      removes: "{{ passpath }}.testorig"
+
+- name: Very basic gopass compatibility test
+  vars:
+    passwordstore_backend: "gopass"
+  block:
+  - name: check if gopass executable exists
+    stat:
+      path: "{{ gopasspath }}"
+    register: gopass_check
+
+  - name: Move original gopass out of the way
+    command:
+      argv:
+        - mv
+        - "{{ gopasspath }}"
+        - "{{ gopasspath }}.testorig"
+    args:
+      creates: "{{ gopasspath }}.testorig"
+    when: gopass_check.stat.exists == true
+
+  - name: Create mocked gopass script
+    ansible.builtin.copy:
+      content: |
+        #!/bin/sh
+        if [ "$GOPASS_NO_REMINDER" != "YES" ]; then
+          exit 1
+        fi
+        if [ "$1" = "--version" ]; then
+          exit 2
+        fi
+        if [ "$1" = "show" ] && [ "$2" != "--password" ]; then
+          exit 3
+        fi
+        echo "gopass_ok"
+      dest: "{{ gopasspath }}"
+      mode: '0755'
+
+  - name: Try to read folder as passname using gopass
+    set_fact:
+      newpass: "{{ lookup('community.general.passwordstore', 'folder') }}"
+
+  - name: Verify password received from gopass
+    assert:
+      that:
+        - newpass == "gopass_ok"
+
+  always:
+  - name: Remove mocked gopass
+    ansible.builtin.file:
+      path: "{{ gopasspath }}"
+      state: absent
+
+  - name: Move original gopass back into place
+    command:
+      argv:
+        - mv
+        - "{{ gopasspath }}.testorig"
+        - "{{ gopasspath }}"
+    args:
+      removes: "{{ gopasspath }}.testorig"
+    when: gopass_check.stat.exists == true