Onepassword lookup add service accounts

changelogs/fragments/6660-onepassword-lookup-service-account.yaml

@@ -0,0 +1,3 @@
+minor_changes:
+  - onepassword lookup plugin - add service account support (https://github.com/ansible-collections/community.general/issues/6635, https://github.com/ansible-collections/community.general/pull/6660).
+  - onepassword_raw lookup plugin - add service account support (https://github.com/ansible-collections/community.general/issues/6635, https://github.com/ansible-collections/community.general/pull/6660).

plugins/lookup/onepassword.py

@@ -42,6 +42,12 @@
         description: The username used to sign in.
       secret_key:
         description: The secret key used when performing an initial sign in.
+      service_account_token:
+        description:
+          - The access key for a service account.
+          - Only works with 1Password CLI version 2 or later.
+        type: str
+        version_added: 7.1.0
       vault:
         description: Vault containing the item to retrieve (case-insensitive). If absent will search all vaults.
     notes:
@@ -113,12 +119,13 @@
 class OnePassCLIBase(with_metaclass(abc.ABCMeta, object)):
     bin = "op"
 
-    def __init__(self, subdomain=None, domain="1password.com", username=None, secret_key=None, master_password=None):
+    def __init__(self, subdomain=None, domain="1password.com", username=None, secret_key=None, master_password=None, service_account_token=None):
         self.subdomain = subdomain
         self.domain = domain
         self.username = username
         self.master_password = master_password
         self.secret_key = secret_key
+        self.service_account_token = service_account_token
 
         self._path = None
         self._version = None
@@ -295,6 +302,10 @@ def assert_logged_in(self):
         return not bool(rc)
 
     def full_signin(self):
+        if self.service_account_token:
+            raise AnsibleLookupError(
+                "1Password CLI version 1 does not support Service Accounts. Please use version 2 or later.")
+
         required_params = [
             "subdomain",
             "username",
@@ -472,6 +483,14 @@ def _parse_field(self, data_json, field_name, section_title=None):
         return ""
 
     def assert_logged_in(self):
+        if self.service_account_token:
+            # maybe not necessary
+            args = ["account", "get"]
+            environment_update = {"OP_SERVICE_ACCOUNT_TOKEN": self.service_account_token}
+            rc, out, err = self._run(args, ignore_errors=False, environment_update=environment_update)
+
+            return not bool(rc)
+
         args = ["account", "list"]
         if self.subdomain:
             account = "{subdomain}.{domain}".format(subdomain=self.subdomain, domain=self.domain)
@@ -495,28 +514,44 @@ def assert_logged_in(self):
         return False
 
     def full_signin(self):
-        required_params = [
-            "subdomain",
-            "username",
-            "secret_key",
-            "master_password",
-        ]
-        self._check_required_params(required_params)
+        if self.service_account_token:
+            self._check_required_params(['service_account_token'])
+            environment_update = {"OP_SERVICE_ACCOUNT_TOKEN": self.service_account_token}
+            args = [
+                "whoami",
+            ]
 
-        args = [
-            "account", "add", "--raw",
-            "--address", "{0}.{1}".format(self.subdomain, self.domain),
-            "--email", to_bytes(self.username),
-            "--signin",
-        ]
+            return self._run(args, environment_update=environment_update)
+        else:
+            required_params = [
+                "subdomain",
+                "username",
+                "secret_key",
+                "master_password",
+            ]
+            self._check_required_params(required_params)
 
-        environment_update = {"OP_SECRET_KEY": self.secret_key}
-        return self._run(args, command_input=to_bytes(self.master_password), environment_update=environment_update)
+            args = [
+                "account", "add", "--raw",
+                "--address", "{0}.{1}".format(self.subdomain, self.domain),
+                "--email", to_bytes(self.username),
+                "--signin",
+            ]
+
+            environment_update = {"OP_SECRET_KEY": self.secret_key}
+
+            return self._run(args, command_input=to_bytes(self.master_password), environment_update=environment_update)
 
     def get_raw(self, item_id, vault=None, token=None):
         args = ["item", "get", item_id, "--format", "json"]
         if vault is not None:
             args += ["--vault={0}".format(vault)]
+        if not self.service_account_token and token is not None:
+            args += [to_bytes("--session=") + token]
+        if self.service_account_token:
+            environment_update = {"OP_SERVICE_ACCOUNT_TOKEN": self.service_account_token}
+            return self._run(args, environment_update=environment_update)
+
         if token is not None:
             args += [to_bytes("--session=") + token]
 
@@ -533,12 +568,14 @@ def signin(self):
 
 
 class OnePass(object):
-    def __init__(self, subdomain=None, domain="1password.com", username=None, secret_key=None, master_password=None):
+    def __init__(self, subdomain=None, domain="1password.com", username=None, secret_key=None, master_password=None,
+                 service_account_token=None):
         self.subdomain = subdomain
         self.domain = domain
         self.username = username
         self.secret_key = secret_key
         self.master_password = master_password
+        self.service_account_token = service_account_token
 
         self.logged_in = False
         self.token = None
@@ -551,7 +588,7 @@ def _get_cli_class(self):
         for cls in OnePassCLIBase.__subclasses__():
             if cls.supports_version == version.split(".")[0]:
                 try:
-                    return cls(self.subdomain, self.domain, self.username, self.secret_key, self.master_password)
+                    return cls(self.subdomain, self.domain, self.username, self.secret_key, self.master_password, self.service_account_token)
                 except TypeError as e:
                     raise AnsibleLookupError(e)
 
@@ -614,8 +651,9 @@ def run(self, terms, variables=None, **kwargs):
         username = self.get_option("username")
         secret_key = self.get_option("secret_key")
         master_password = self.get_option("master_password")
+        service_account_token = self.get_option("service_account_token")
 
-        op = OnePass(subdomain, domain, username, secret_key, master_password)
+        op = OnePass(subdomain, domain, username, secret_key, master_password, service_account_token)
         op.assert_logged_in()
 
         values = []

plugins/lookup/onepassword_raw.py

@@ -39,6 +39,12 @@
         description: The username used to sign in.
       secret_key:
         description: The secret key used when performing an initial sign in.
+      service_account_token:
+        description:
+          - The access key for a service account.
+          - Only works with 1Password CLI version 2 or later.
+        type: string
+        version_added: 7.1.0
       vault:
         description: Vault containing the item to retrieve (case-insensitive). If absent will search all vaults.
     notes:
@@ -89,8 +95,9 @@ def run(self, terms, variables=None, **kwargs):
         username = self.get_option("username")
         secret_key = self.get_option("secret_key")
         master_password = self.get_option("master_password")
+        service_account_token = self.get_option("service_account_token")
 
-        op = OnePass(subdomain, domain, username, secret_key, master_password)
+        op = OnePass(subdomain, domain, username, secret_key, master_password, service_account_token)
         op.assert_logged_in()
 
         values = []