Coverage for pass_audit/passwordstore.py: 96%

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

2# pass audit - Password Store Extension (https://www.passwordstore.org/) 

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

4# 

5 

6import os 

7import shutil 

8from subprocess import Popen, PIPE # nosec 

9from pathlib import Path 

10 

11 

12class PasswordStoreError(Exception): 

13 """Error in the execution of password store.""" 

14 

15 

16class PasswordStore(): 

17 """Simple Password Store wrapper for python. 

18 

19 Based on the PasswordStore class from pass-import. 

20 See https://github.com/roddhjav/pass-import for more information. 

21 

22 """ 

23 

24 def __init__(self, prefix=None): 

25 self._binary = shutil.which('pass') 

26 self._gpgbinary = shutil.which('gpg2') or shutil.which('gpg') 

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

28 self._setenv('PASSWORD_STORE_DIR') 

29 self._setenv('PASSWORD_STORE_KEY') 

30 self._setenv('PASSWORD_STORE_GIT', 'GIT_DIR') 

31 self._setenv('PASSWORD_STORE_GPG_OPTS') 

32 self._setenv('PASSWORD_STORE_X_SELECTION', 'X_SELECTION') 

33 self._setenv('PASSWORD_STORE_CLIP_TIME', 'CLIP_TIME') 

34 self._setenv('PASSWORD_STORE_UMASK') 

35 self._setenv('PASSWORD_STORE_GENERATED_LENGTH', 'GENERATED_LENGTH') 

36 self._setenv('PASSWORD_STORE_CHARACTER_SET', 'CHARACTER_SET') 

37 self._setenv('PASSWORD_STORE_CHARACTER_SET_NO_SYMBOLS', 

38 'CHARACTER_SET_NO_SYMBOLS') 

39 self._setenv('PASSWORD_STORE_ENABLE_EXTENSIONS') 

40 self._setenv('PASSWORD_STORE_EXTENSIONS_DIR', 'EXTENSIONS') 

41 self._setenv('PASSWORD_STORE_SIGNING_KEY') 

42 self._setenv('GNUPGHOME') 

43 

44 if prefix: 

45 self.prefix = prefix 

46 if 'PASSWORD_STORE_DIR' not in self.env or self.prefix is None: 

47 raise PasswordStoreError("pass prefix unknown") 

48 

49 def _setenv(self, var, env=None): 

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

51 if env is None: 

52 env = var 

53 if env in os.environ: 

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

55 

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

57 """Call to a command.""" 

58 if isinstance(data, bytes): 

59 nline = False 

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

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

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

63 res = process.wait() 

64 return res, stdout, stderr 

65 

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

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

68 command = [self._binary] 

69 command.extend(arg) 

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

71 if res: 

72 raise PasswordStoreError(f"{stderr} {stdout}") 

73 return stdout 

74 

75 @property 

76 def prefix(self): 

77 """Get password store prefix from PASSWORD_STORE_DIR.""" 

78 return self.env['PASSWORD_STORE_DIR'] 

79 

80 @prefix.setter 

81 def prefix(self, value): 

82 self.env['PASSWORD_STORE_DIR'] = value 

83 

84 def list(self, path='', filename='*'): 

85 """List the paths in the password store repository.""" 

86 prefix = os.path.join(self.prefix, path) 

87 if os.path.isfile(prefix + '.gpg'): 

88 paths = [path] 

89 else: 

90 paths = [] 

91 for ppath in Path(prefix).rglob(f'{filename}.gpg'): 

92 file = os.sep + str(ppath)[len(self.prefix) + 1:] 

93 if f"{os.sep}." not in file: 

94 file = os.path.splitext(file)[0][1:] 

95 paths.append(file) 

96 paths.sort() 

97 return paths 

98 

99 def show(self, path): 

100 """Decrypt path and read the credentials in the password file.""" 

101 entry = {} 

102 entry['group'] = os.path.dirname(path) 

103 entry['title'] = os.path.basename(path) 

104 try: 

105 data = self._command(['show', path]).split('\n') 

106 except UnicodeDecodeError: 

107 entry['data'] = self._command(['show', path], nline=False) 

108 return entry 

109 

110 data.pop() 

111 if data: 

112 line = data.pop(0) 

113 if ': ' in line: 

114 (key, value) = line.split(': ', 1) 

115 entry[key] = value 

116 else: 

117 entry['password'] = line 

118 for line in data: 

119 if ': ' in line: 

120 (key, value) = line.split(': ', 1) 

121 entry[key] = value 

122 elif line.startswith('otpauth://'): 

123 entry['otpauth'] = line 

124 elif 'comments' in entry: 

125 entry['comments'] += '\n' + line 

126 return entry 

127 

128 def exist(self): 

129 """Check if the password store is initialized.""" 

130 return os.path.isfile(os.path.join(self.prefix, '.gpg-id')) 

131 

132 def isvalid(self): 

133 """Ensure the GPG keyring is usable.""" 

134 trusted = ['m', 'f', 'u', 'w', 's'] 

135 with open(os.path.join(self.prefix, '.gpg-id'), 'r') as file: 

136 gpgids = file.read().split('\n') 

137 gpgids.pop() 

138 

139 cmd = [ 

140 self._gpgbinary, 

141 '--with-colons', 

142 '--batch', 

143 '--list-keys', 

144 '--', 

145 ] 

146 for gpgid in gpgids: 

147 res, out, _ = self._call(cmd + [gpgid]) 

148 if res: 

149 return False 

150 for line in out.split('\n'): 

151 record = line.split(':') 

152 if record[0] == 'pub': 

153 trust = record[1] 

154 if trust not in trusted: 

155 return False 

156 

157 cmd = [ 

158 self._gpgbinary, 

159 '--with-colons', 

160 '--batch', 

161 '--list-secret-keys', 

162 '--', 

163 ] 

164 for gpgid in gpgids: 

165 res, _, _ = self._call(cmd + [gpgid]) 

166 if res == 0: 

167 return True 

168 return False