Add support for keys as notes (#53)

Adding support to notes; Improving log messages; Improving error handling for multiple folders

README.md

@@ -3,7 +3,6 @@
 ## Requirements
 * You need to have the [Bitwarden CLI tool](https://bitwarden.com/help/cli/) installed and available in the `$PATH` as `bw`. See below for detailed instructions.
 * `ssh-agent` must be running in the current session.
-* the bitwarden free version it's not supported.
 
 ## Installation
 Just save the file `bw_add_sshkeys.py` in a folder where it can by found when calling it from the command line. On linux you can see these folders by running `echo $PATH` from the command line. To install for a single user, you can - for example - save the script under `~/.local/bin/` and make it executable by running `chmod +x ~/.local/bin/bw_add_sshkeys.py`.
@@ -19,13 +18,19 @@ Fetches SSH keys stored in Bitwarden vault and adds them to `ssh-agent`.
 2. Enter your Bitwarden credentials, if a Bitwarden vault session is not already set.
 3. (optional) Enter your SSH keys' passphrases if they're not stored in your Bitwarden.
 
-## Storing the keys in BitWarden
+## Storing the keys in Bitwarden
 1. Create a folder called `ssh-agent` (can be overridden on the command line).
 2. Add an new secure note to that folder.
-3. Upload the private key as an attachment.
-4. Add the custom field `private` (can be overridden on the command line), containing the file name of the private key attachment.
-5. (optional) If your key is encrypted with passphrase and you want it to decrypt automatically, save passphrase into custom field `passphrase` (field name can be overriden on the command line). You can create this field as `hidden` if you don't want the passphrase be displayed by default.
-6. Repeat steps 2-5 for each subsequent key
+3. Add the private key to the secure note:
+    - Add the private key directly on the `notes` field
+
+        OR
+
+    - Upload the private key as an attachment (requires Bitwarden Premium)
+        - You'll need to add a custom field `private` containing the file name of the private key attachment.
+        - The field name can be overridden on the command line
+4. (optional) If your key is encrypted with passphrase and you want it to decrypt automatically, save passphrase into custom field `passphrase` (field name can be overriden on the command line). You can create this field as `hidden` if you don't want the passphrase be displayed by default.
+5. Repeat steps 2-4 for each subsequent key
 
 ## Command line overrides
 * `--debug`/`-d` - Show debug output

bw_add_sshkeys.py

@@ -62,16 +62,12 @@ def get_folders(session: str, foldername: str) -> str:
 
     folders = json.loads(proc_folders.stdout)
 
-    if not folders:
-        logging.error('"%s" folder not found', foldername)
-        return ""
+    try:
+        return str([k["id"] for k in folders if k["name"] == foldername][0])
+    except IndexError:
+        pass
 
-    # Do we have any folders
-    if len(folders) != 1:
-        logging.error('%d folders with the name "%s" found', len(folders), foldername)
-        return ""
-
-    return str(folders[0]["id"])
+    raise RuntimeError('"%s" folder not found' % foldername)
 
 
 def folder_items(session: str, folder_id: str) -> list[dict[str, Any]]:
@@ -103,79 +99,115 @@ def add_ssh_keys(
     Function to attempt to get keys from a vault item
     """
     for item in items:
+        logging.info("----------------------------------")
+        logging.info('Processing item "%s"', item["name"])
         try:
-            private_key_file = [
-                k["value"] for k in item["fields"] if k["name"] == keyname
-            ][0]
-        except IndexError:
-            logging.warning('No "%s" field found for item %s', keyname, item["name"])
-            continue
-        except KeyError as error:
-            logging.debug(
-                'No key "%s" found in item %s - skipping', error.args[0], item["name"]
-            )
+            ssh_key = fetch_key(session, item, keyname)
+        except RuntimeError as error:
+            logging.error(str(error))
             continue
-        logging.debug("Private key file declared")
 
         private_key_pw = ""
-        try:
-            private_key_pw = [
-                k["value"] for k in item["fields"] if k["name"] == pwkeyname
-            ][0]
-            logging.debug("Passphrase declared")
-        except IndexError:
-            logging.warning('No "%s" field found for item %s', pwkeyname, item["name"])
-        except KeyError as error:
-            logging.debug(
-                'No key "%s" found in item %s - skipping', error.args[0], item["name"]
-            )
 
-        try:
-            private_key_id = [
-                k["id"]
-                for k in item["attachments"]
-                if k["fileName"] == private_key_file
-            ][0]
-        except IndexError:
-            logging.warning(
-                'No attachment called "%s" found for item %s',
-                private_key_file,
-                item["name"],
-            )
-            continue
-        logging.debug("Private key ID found")
+        if "fields" in item:
+            try:
+                private_key_pw = [
+                    k["value"] for k in item["fields"] if k["name"] == pwkeyname
+                ][0]
+                logging.debug("Passphrase declared")
+            except IndexError:
+                logging.warning(
+                    'No "%s" field found for item %s', pwkeyname, item["name"]
+                )
 
         try:
-            ssh_add(session, item["id"], private_key_id, private_key_pw)
+            ssh_add(ssh_key, private_key_pw)
         except subprocess.SubprocessError:
-            logging.warning("Could not add key to the SSH agent")
+            logging.warning('Could not add key "%s" to the SSH agent', item["name"])
+
 
+def fetch_key(session: str, item: dict[str, Any], keyname: str) -> str:
+    if "fields" in item and "attachments" in item:
+        logging.debug(
+            "Item %s has custom fields and attachments - searching for %s",
+            item["name"],
+            keyname,
+        )
+        try:
+            return fetch_from_attachment(session, item, keyname)
+        except RuntimeWarning as warning:
+            logging.warning(str(warning))
+        except RuntimeError as error:
+            logging.error(str(error))
 
-def ssh_add(session: str, item_id: str, key_id: str, key_pw: str = "") -> None:
+    logging.debug("Couldn't find an ssh key in attachments - falling back to notes")
+
+    # no way to validate the key without extra dependencies
+    # maybe check if the key starts with '----'?
+    # or just pass it to ssh-agent, and let it fail?
+    if isinstance(item["notes"], str):
+        return item["notes"]
+
+    raise RuntimeError("Could not find an SSH key on item %s" % item["name"])
+
+
+def fetch_from_attachment(session: str, item: dict[str, Any], keyname: str) -> str:
     """
     Function to get the key contents from the Bitwarden vault
     """
-    logging.debug("Item ID: %s", item_id)
-    logging.debug("Key ID: %s", key_id)
-
-    proc_attachment = subprocess.run(
-        [
-            "bw",
-            "get",
-            "attachment",
-            key_id,
-            "--itemid",
-            item_id,
-            "--raw",
-            "--session",
-            session,
-        ],
-        stdout=subprocess.PIPE,
-        universal_newlines=True,
-        check=True,
-    )
-    ssh_key = proc_attachment.stdout
+    private_key_file = ""
+    try:
+        private_key_file = [k["value"] for k in item["fields"] if k["name"] == keyname][
+            0
+        ]
+    except IndexError:
+        raise RuntimeWarning(
+            'No "%s" field found for item %s' % (keyname, item["name"])
+        )
+
+    logging.debug("Private key file declared")
+
+    try:
+        private_key_id = [
+            k["id"] for k in item["attachments"] if k["fileName"] == private_key_file
+        ][0]
+    except IndexError:
+        raise RuntimeWarning(
+            'No attachment called "%s" found for item %s'
+            % (private_key_file, item["name"])
+        )
+
+    logging.debug("Private key ID found")
+    logging.debug("Item ID: %s", item["id"])
+    logging.debug("Key ID: %s", private_key_id)
+
+    try:
+        proc_attachment = subprocess.run(
+            [
+                "bw",
+                "get",
+                "attachment",
+                private_key_id,
+                "--itemid",
+                item["id"],
+                "--raw",
+                "--session",
+                session,
+            ],
+            stdout=subprocess.PIPE,
+            universal_newlines=True,
+            check=True,
+        )
+    except subprocess.CalledProcessError:
+        raise RuntimeError("Could not get attachment from Bitwarden")
+
+    return proc_attachment.stdout
 
+
+def ssh_add(ssh_key: str, key_pw: str = "") -> None:
+    """
+    Adds the key to the agent
+    """
     if key_pw:
         envdict = dict(
             os.environ,
@@ -186,6 +218,12 @@ def ssh_add(session: str, item_id: str, key_id: str, key_pw: str = "") -> None:
         envdict = dict(os.environ, SSH_ASKPASS_REQUIRE="never")
 
     logging.debug("Running ssh-add")
+
+    # if the key doesn't end with a line break, let's add it
+    if not ssh_key.endswith("\n"):
+        logging.debug("Adding a line break at the end of the key")
+        ssh_key += "\n"
+
     # CAVEAT: `ssh-add` provides no useful output, even with maximum verbosity
     subprocess.run(
         ["ssh-add", "-"],
@@ -249,7 +287,7 @@ def main() -> None:
         else:
             loglevel = logging.INFO
 
-        logging.basicConfig(level=loglevel)
+        logging.basicConfig(format="%(levelname)-8s %(message)s", level=loglevel)
 
         try:
             logging.info("Getting Bitwarden session")
@@ -264,9 +302,11 @@ def main() -> None:
 
             logging.info("Attempting to add keys to ssh-agent")
             add_ssh_keys(session, items, args.customfield, args.passphrasefield)
+        except RuntimeError as error:
+            logging.critical(str(error))
         except subprocess.CalledProcessError as error:
             if error.stderr:
-                logging.error('"%s" error: %s', error.cmd[0], error.stderr)
+                logging.critical('"%s" error: %s', error.cmd[0], error.stderr)
             logging.debug("Error running %s", error.cmd)
 
     if os.environ.get("SSH_ASKPASS") and os.environ.get(