Onepassword lookup add service accounts

plugins/lookup/onepassword.py

@@ -42,6 +42,8 @@
         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.
       vault:
         description: Vault containing the item to retrieve (case-insensitive). If absent will search all vaults.
     notes:
@@ -113,28 +115,30 @@
 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
 
     def _check_required_params(self, required_params):
-        non_empty_attrs = dict((param, getattr(self, param, None)) for param in required_params if getattr(self, param, None))
-        missing = set(required_params).difference(non_empty_attrs)
-        if missing:
-            prefix = "Unable to sign in to 1Password. Missing required parameter"
-            plural = ""
-            suffix = ": {params}.".format(params=", ".join(missing))
-            if len(missing) > 1:
-                plural = "s"
-
-            msg = "{prefix}{plural}{suffix}".format(prefix=prefix, plural=plural, suffix=suffix)
-            raise AnsibleLookupError(msg)
+        if not self.service_account_token:
+            non_empty_attrs = dict((param, getattr(self, param, None)) for param in required_params if getattr(self, param, None))
+            missing = set(required_params).difference(non_empty_attrs)
+            if missing:
+                prefix = "Unable to sign in to 1Password. Missing required parameter"
+                plural = ""
+                suffix = ": {params}. Or use a service_account_token.".format(params=", ".join(missing))
+                if len(missing) > 1:
+                    plural = "s"
+
+                msg = "{prefix}{plural}{suffix}".format(prefix=prefix, plural=plural, suffix=suffix)
+                raise AnsibleLookupError(msg)
 
     @abc.abstractmethod
     def _parse_field(self, data_json, field_name, section_title):
@@ -472,55 +476,77 @@ def _parse_field(self, data_json, field_name, section_title=None):
         return ""
 
     def assert_logged_in(self):
-        args = ["account", "list"]
-        if self.subdomain:
-            account = "{subdomain}.{domain}".format(subdomain=self.subdomain, domain=self.domain)
-            args.extend(["--account", account])
+        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)
 
-        rc, out, err = self._run(args)
+            return not bool(rc)
 
-        if out:
-            # Running 'op account get' if there are no accounts configured on the system drops into
-            # an interactive prompt. Only run 'op account get' after first listing accounts to see
-            # if there are any previously configured accounts.
-            args = ["account", "get"]
+        else:
+
+            args = ["account", "list"]
             if self.subdomain:
                 account = "{subdomain}.{domain}".format(subdomain=self.subdomain, domain=self.domain)
                 args.extend(["--account", account])
 
-            rc, out, err = self._run(args, ignore_errors=True)
+            rc, out, err = self._run(args)
 
-            return not bool(rc)
+            if out:
+                # Running 'op account get' if there are no accounts configured on the system drops into
+                # an interactive prompt. Only run 'op account get' after first listing accounts to see
+                # if there are any previously configured accounts.
+                args = ["account", "get"]
+                if self.subdomain:
+                    account = "{subdomain}.{domain}".format(subdomain=self.subdomain, domain=self.domain)
+                    args.extend(["--account", account])
+
+                rc, out, err = self._run(args, ignore_errors=True)
 
-        return False
+                return not bool(rc)
+
+            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:
+            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 token is not None:
+        if not self.service_account_token and token is not None:
             args += [to_bytes("--session=") + token]
-
-        return self._run(args)
+        if self.service_account_token:
+            environment_update = {"OP_SERVICE_ACCOUNT_TOKEN": self.service_account_token}
+            return self._run(args, environment_update=environment_update)
+        else:
+            return self._run(args)
 
     def signin(self):
         self._check_required_params(['master_password'])
@@ -533,12 +559,13 @@ 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 +578,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,12 +641,12 @@ 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 = []
         for term in terms:
             values.append(op.get_field(term, field, section, vault))
-
         return values

plugins/lookup/onepassword_raw.py

@@ -39,6 +39,8 @@
         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.
       vault:
         description: Vault containing the item to retrieve (case-insensitive). If absent will search all vaults.
     notes:
@@ -89,8 +91,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/modules/onepassword_info.py

@@ -78,6 +78,11 @@
                 description:
                     - 1Password username.
                     - Only required for initial sign in.
+            service_account_token:
+                type: str
+                description:
+                    - 1Password service account token.
+                    - Only required for initial sign in.
             master_password:
                 type: str
                 description:
@@ -375,6 +380,7 @@ def main():
                 username=dict(type='str'),
                 master_password=dict(required=True, type='str', no_log=True),
                 secret_key=dict(type='str', no_log=True),
+                service_account_token=dict(type='str', no_log=True),
             ), default=None),
             search_terms=dict(required=True, type='list', elements='dict'),
         ),