Merge pull request #41 from bluelabsio/fix-lpass-to-read-1pass-notes

Update secret retrieval for onepassword, add support for notes field

db_facts/lpass.py

@@ -5,9 +5,11 @@
 # Even though this file says "lpass", all underlying calls to the
 # lpass CLI have been replaced with calls to the 1password CLI.
 
-from subprocess import check_output
+import json
+from subprocess import check_output, CalledProcessError
 from .db_facts_types import LastPassUsernamePassword, LastPassAWSIAM
 from .db_type import canonicalize_db_type, db_protocol
+import logging
 
 
 def pull_lastpass_username_password(lastpass_entry_name: str) -> LastPassUsernamePassword:
@@ -32,17 +34,27 @@ def lpass_field(name: str, field: str) -> str:
     # from lastpass to 1password. This command retrieves the fields in the
     # same format from 1password instead.
 
-    # Note this won't work for the way 1password stores notes and URLs, which
-    # is different from lpass. But as of now db-facts doesn't ever rely on
-    # these fields.
-    if field == 'notes' or field == 'url':
+    # Note this won't work for URLs, which 1password stores
+    # different from lpass. But as of now db-facts doesn't ever rely on
+    # this field.
+    if field == 'url':
         raise NotImplementedError(
             'Cannot retrieve notes or URL fields from 1password')
 
-    raw_output = check_output(
-        ['op', 'item', 'get', name, '--field', f'label={field}'])
+    # The field lastpass stored notes in was called "notes", but the field
+    # 1password stores notes in is called "notesPlain".
+    if field == 'notes':
+        field = 'notesPlain'
 
-    return raw_output.decode('utf-8').rstrip('\n')
+    try:
+        raw_output = check_output(
+            ['op', 'item', 'get', name, '--field', f'label={field}', '--format=json'])
+        parsed_output = json.loads(raw_output)
+        return parsed_output['value']
+    except (json.JSONDecodeError, TypeError, CalledProcessError) as e:
+        log = logging.getLogger(__name__)
+        log.error(f"Error retrieving entry from 1password cli: {e}")
+        raise
 
 
 def db_info_from_lpass(lpass_entry_name: str):

db_facts/version.py

@@ -1 +1 @@
-__version__ = '5.0.0'
+__version__ = '5.1.0'

requirements.txt

@@ -1,7 +1,7 @@
 # Run ./deps.sh after changing this file.
 
 # The following are dev-time dependencies
-nose>=0.0
+nose
 coverage>=0.0
 flake8
 mypy
@@ -15,3 +15,8 @@ sphinx>=3 # used to generate and upload docs - sphinx-autodoc-typehints requires
 sphinx-autodoc-typehints # used to handle type hints
 sphinx-rtd-theme # used to style docs for readthedocs.io
 recommonmark # used to be able to use sphinx with markdown
+
+boto3
+Jinja2
+setuptools
+pyyaml

tests/test_lpass.py

@@ -5,7 +5,9 @@
 # Even though this file says "lpass", all underlying calls to the
 # lpass CLI have been replaced with calls to the 1password CLI.
 
+import json
 import unittest
+from subprocess import CalledProcessError
 from unittest.mock import patch
 from db_facts import lpass
 
@@ -15,32 +17,65 @@ def test_lpass_field_url_raises(self):
         with self.assertRaises(NotImplementedError):
             lpass.lpass_field('my_name', 'url')
 
-    def test_lpass_field_notes_raises(self):
-        with self.assertRaises(NotImplementedError):
-            lpass.lpass_field('my_name', 'notes')
+    @patch('db_facts.lpass.check_output', side_effect=CalledProcessError(1, "mocked_op"))
+    def test_lpass_error_with_underlying_process(self, mock_check_output):
+        with self.assertRaises(CalledProcessError):
+            lpass.lpass_field('my_name', 'field1')
+
+    @patch('db_facts.lpass.check_output')
+    def test_lpass_bad_json_from_process(self, mock_check_output):
+        return_json = "{"
+        mock_check_output.return_value = return_json
+        with self.assertRaises(json.JSONDecodeError):
+            lpass.lpass_field('my_name', 'field1')
+
+    @patch('db_facts.lpass.check_output')
+    def test_lpass_field_notes_uses_notesPlain(self, mock_check_output):
+        notes_json = json.dumps({'field': 'are you sick of json.dumps yet'})
+        return_json = {
+            'id': 'notesPlain',
+            'value': notes_json,
+        }
+        mock_check_output.return_value = json.dumps(return_json).encode('utf-8')
+        out = lpass.lpass_field('my_name', 'notes')
+        mock_check_output.assert_called_with(
+            ['op', 'item', 'get', 'my_name', '--field', 'label=notesPlain', '--format=json'])
+        assert out == notes_json
 
     @patch('db_facts.lpass.check_output')
     def test_lpass_field_username(self, mock_check_output):
-        mock_check_output.return_value = "fakeuser\n".encode("utf-8")
+        return_json = {
+            'id': 'username',
+            'value': 'fakeuser'
+        }
+        mock_check_output.return_value = json.dumps(return_json).encode('utf-8')
         out = lpass.lpass_field('my_name', 'username')
         mock_check_output.assert_called_with(
-            ['op', 'item', 'get', 'my_name', '--field', 'label=username'])
+            ['op', 'item', 'get', 'my_name', '--field', 'label=username', '--format=json'])
         assert out == "fakeuser"
 
     @patch('db_facts.lpass.check_output')
     def test_lpass_field_password(self, mock_check_output):
-        mock_check_output.return_value = "fakepassword\n".encode("utf-8")
+        return_json = {
+            'id': 'password',
+            'value': 'fakepassword',
+        }
+        mock_check_output.return_value = json.dumps(return_json).encode('utf-8')
         out = lpass.lpass_field('my_name', 'password')
         mock_check_output.assert_called_with(
-            ['op', 'item', 'get', 'my_name', '--field', 'label=password'])
+            ['op', 'item', 'get', 'my_name', '--field', 'label=password', '--format=json'])
         assert out == "fakepassword"
 
     @patch('db_facts.lpass.check_output')
     def test_lpass_field_field1(self, mock_check_output):
-        mock_check_output.return_value = "fakefield1\n".encode("utf-8")
+        return_json = {
+            'id': 'field1',
+            'value': 'fakefield1',
+        }
+        mock_check_output.return_value = json.dumps(return_json).encode('utf-8')
         out = lpass.lpass_field('my_name', 'field1')
         mock_check_output.assert_called_with(
-            ['op', 'item', 'get', 'my_name', '--field', 'label=field1'])
+            ['op', 'item', 'get', 'my_name', '--field', 'label=field1', '--format=json'])
         assert out == "fakefield1"
 
     @patch('db_facts.lpass.check_output')
@@ -51,15 +86,17 @@ def fake_check_output(args):
             assert args[2] == 'get'
             assert args[3] == 'my_lpass_name'
             assert args[4] == '--field'
+            assert args[6] == '--format=json'
             ret = {
-                "label=username": 'fakeuser',
-                "label=password": 'fakepassword',
-                "label=Hostname": 'fakehost',
-                "label=Port": '123',
-                "label=Type": 'faketype',
-                "label=Database": 'fakedatabase',
+                "label=username": {'value': 'fakeuser'},
+                "label=password": {'value': 'fakepassword'},
+                "label=Hostname": {'value': 'fakehost'},
+                "label=Port": {'value': '123'},
+                "label=Type": {'value': 'faketype'},
+                "label=Database": {'value': 'fakedatabase'},
             }
-            return (ret[args[5]] + "\n").encode('utf-8')
+            return json.dumps(ret[args[5]]).encode('utf-8')
+
         mock_check_output.side_effect = fake_check_output
         db_info = lpass.db_info_from_lpass('my_lpass_name')
         expected_db_info = {