Coverage for pass_import/formats/cli.py: 100%

1# -*- encoding: utf-8 -*- 

2# pass import - Passwords importer swiss army knife 

3# Copyright (C) 2017-2024 Alexandre PUJOL <alexandre@pujol.io>. 

4# 

5 

6import os 

7import shutil 

8from abc import abstractmethod 

9from subprocess import PIPE, Popen # nosec 

10 

11from pass_import.core import Cap 

12from pass_import.errors import PMError 

13from pass_import.manager import PasswordExporter, PasswordImporter 

14 

15 

16class CLI(PasswordImporter, PasswordExporter): 

17 """Base class for CLI based importer and exporter.""" 

18 cap = Cap.IMPORT | Cap.EXPORT 

19 format = 'cli' 

20 command = '' 

21 

22 def __init__(self, prefix=None, settings=None): 

23 self._binary = shutil.which(self.command) 

24 if self._binary is None: 

25 raise PMError(f"{self.command} is required.") # pragma: no cover 

26 

27 self.env = dict(**os.environ) 

28 super().__init__(prefix, settings) 

29 

30 def _setenv(self, var, env=None, value=None): 

31 """Add var in the environment variables dictionary.""" 

32 if env is None: 

33 env = var 

34 if env in os.environ: 

35 self.env[var] = os.environ[env] 

36 if value is not None: 

37 self.env[var] = value 

38 

39 def _call(self, command, data=None, nline=True): 

40 """Call to a command.""" 

41 if isinstance(data, bytes): 

42 nline = False 

43 with Popen(command, universal_newlines=nline, env=self.env, stdin=PIPE, 

44 stdout=PIPE, stderr=PIPE, shell=False) as process: 

45 (stdout, stderr) = process.communicate(data) 

46 res = process.wait() 

47 return res, stdout, stderr 

48 

49 def _command(self, arg, data=None, nline=True): 

50 """Call to the password manager cli command.""" 

51 command = [self._binary] 

52 command.extend(arg) 

53 res, stdout, stderr = self._call(command, data, nline) 

54 if res: 

55 raise PMError(f"{stderr} {stdout}") 

56 return stdout 

57 

58 def exist(self): 

59 """Nothing to do.""" 

60 return True 

61 

62 @abstractmethod 

63 def parse(self): 

64 """Parse the password manager repository and retrieve passwords.""" 

65 

66 @abstractmethod 

67 def insert(self, entry): 

68 """Insert a password entry into the password repository."""