keepass_to_pass: add integration

README.md

@@ -41,7 +41,7 @@ Github issues and feature requests welcomed.
 | API             | flask                                                       |
 | Chat            | slack                                                       |
 | Data Scraping   | beautifulsoup<br/>facebook groups<br/>instagram<br/>scrapy  |
-| Databases       | elasticsearch<br/>neo4j<br/>splunk                          |
+| Databases       | elasticsearch<br/>neo4j<br/>splunk<br/>pass                 |
 | Data Store      | minio<br/>swift                                             |
 | Devices         | snmp                                                        |
 | Google Cloud    | google auth api<br/>google people api<br/>google sheets api |

automon/integrations/keepass_to_pass/__init__.py

@@ -0,0 +1,4 @@
+from .client import (
+    KeepassClient,
+    PassClient
+)

automon/integrations/keepass_to_pass/client.py

@@ -0,0 +1,80 @@
+import os
+import csv
+
+from automon import log
+from automon.helpers import Run
+
+logger = log.logging.getLogger(__name__)
+logger.setLevel(level=log.DEBUG)
+
+
+class KeepassClient(object):
+
+    def __init__(self):
+        self.database_csv_path = None
+
+    def read_database_csv(self, database_csv_path: str, delimiter: str = ',', quotechar: str = '"') -> csv.reader:
+        if self.set_database_csv_path(database_csv_path=database_csv_path):
+            logger.debug(f'KeepassClient :: read_database_csv :: read')
+            with open(self.database_csv_path, 'r') as csvfile:
+                return [KeepassEntry(pw) for pw in csv.reader(csvfile, delimiter=delimiter, quotechar=quotechar)][1:]
+
+        raise OSError(f'KeepassClient :: read_database_csv :: ERROR :: {database_csv_path}')
+
+    def set_database_csv_path(self, database_csv_path: str) -> True:
+        if os.path.exists(database_csv_path):
+            self.database_csv_path = database_csv_path
+            logger.debug(
+                f'KeepassClient :: set_database_csv_path :: {database_csv_path} :: {os.stat(database_csv_path)}')
+            return True
+
+        raise FileNotFoundError(f'KeepassClient :: set_database_csv_path :: ERROR :: {database_csv_path}')
+
+
+class KeepassEntry(object):
+    ENTRY_FORMAT: dict = {
+        'Group': 0,
+        'Title': 1,
+        'Username': 2,
+        'Password': 3,
+        'URL': 4,
+        'Notes': 5,
+        'TOPT': 6,
+        'Icon': 7,
+        'Last Modified': 8,
+        'Created': 9,
+    }
+
+    def __init__(self, cvs_row: list):
+        self.Group: str = cvs_row[self.ENTRY_FORMAT['Group']]
+        self.Title: str = cvs_row[self.ENTRY_FORMAT['Title']]
+        self.Username: str = cvs_row[self.ENTRY_FORMAT['Username']]
+        self.Password: str = cvs_row[self.ENTRY_FORMAT['Password']]
+        self.URL: str = cvs_row[self.ENTRY_FORMAT['URL']]
+        self.Notes: str = cvs_row[self.ENTRY_FORMAT['Notes']]
+        self.TOPT: str = cvs_row[self.ENTRY_FORMAT['TOPT']]
+        self.Icon: str = cvs_row[self.ENTRY_FORMAT['Icon']]
+        self.LastModified: str = cvs_row[self.ENTRY_FORMAT['Last Modified']]
+        self.Created: str = cvs_row[self.ENTRY_FORMAT['Created']]
+
+    def __str__(self):
+        return f'{self.Group} :: {self.Title} :: {self.Username} :: {self.URL}'
+
+    def to_pass(self):
+        return f"""{self.Password}
+Title: {self.Title}
+Username: {self.Username}
+URL: {self.URL}
+TOPT: {self.TOPT}
+Icon: {self.Icon}
+Last Modified: {self.LastModified}
+Created: {self.Created}
+Notes: {self.Notes}
+"""
+
+
+class PassClient(object):
+    TEMPLATE: str = f''
+
+    def __init__(self):
+        self._runner = Run()

automon/integrations/keepass_to_pass/tests/test_keepass.py

@@ -0,0 +1,23 @@
+import unittest
+
+from automon.integrations.keepass_to_pass import KeepassClient
+
+test = KeepassClient()
+
+
+class TestKeepassClient(unittest.TestCase):
+    def test_set_database_csv_path(self):
+        with self.assertRaises(FileNotFoundError):
+            test.set_database_csv_path('AAAA.XXXCSV')
+
+        open('BBBB.XXXCSV', 'w').close()
+
+        self.assertTrue(test.set_database_csv_path('BBBB.XXXCSV'))
+
+    def test_read_database_csv(self):
+        with self.assertRaises(OSError):
+            test.read_database_csv('AAAA.XXXCSV')
+
+
+if __name__ == '__main__':
+    unittest.main()