common.GnuPG

33 - def __init__(self, use_agent = False):

34 GnuPGInterface.GnuPG.__init__(self) 35 self.use_agent = use_agent 36 self._setup_my_options() 37 self.always_trust = False

38

39 - def _setup_my_options(self):

40 self.options.armor = 1 41 self.options.meta_interactive = 0 42 self.options.extra_args.append('--no-secmem-warning') 43 # disable photo viewer when verifying keys 44 self.options.extra_args.append('--verify-options') 45 self.options.extra_args.append('no-show-photo') 46 if self.use_agent: 47 self.options.extra_args.append('--use-agent')

48

49 - def _read_response(self, child_stdout):

50 # Internal method: reads all the output from GPG, taking notice 51 # only of lines that begin with the magic [GNUPG:] prefix. 52 # (See doc/DETAILS in the GPG distribution for info on GPG's 53 # output when --status-fd is specified.) 54 # 55 # Returns a dictionary, mapping GPG's keywords to the arguments 56 # for that keyword. 57 58 resp = {} 59 while True: 60 line = helpers.temp_failure_retry(child_stdout.readline) 61 if line == "": break 62 line = line.rstrip() 63 if line[0:9] == '[GNUPG:] ': 64 # Chop off the prefix 65 line = line[9:] 66 L = line.split(None, 1) 67 keyword = L[0] 68 if len(L) > 1: 69 resp[ keyword ] = L[1] 70 else: 71 resp[ keyword ] = "" 72 return resp

73

74 - def encrypt(self, str_, recipients, always_trust=False):

75 self.options.recipients = recipients # a list! 76 77 opt = ['--encrypt'] 78 if always_trust or self.always_trust: 79 opt.append('--always-trust') 80 proc = self.run(opt, create_fhs=['stdin', 'stdout', 'status', 81 'stderr']) 82 proc.handles['stdin'].write(str_) 83 try: 84 proc.handles['stdin'].close() 85 except IOError: 86 pass 87 88 output = proc.handles['stdout'].read() 89 try: 90 proc.handles['stdout'].close() 91 except IOError: 92 pass 93 94 stat = proc.handles['status'] 95 resp = self._read_response(stat) 96 try: 97 proc.handles['status'].close() 98 except IOError: 99 pass 100 101 error = proc.handles['stderr'].read() 102 proc.handles['stderr'].close() 103 104 try: proc.wait() 105 except IOError: pass 106 if 'INV_RECP' in resp and resp['INV_RECP'].split()[0] == '10': 107 # unusable recipient "Key not trusted" 108 return '', 'NOT_TRUSTED' 109 if 'BEGIN_ENCRYPTION' in resp and 'END_ENCRYPTION' in resp: 110 # Encryption succeeded, even if there is output on stderr. Maybe 111 # verbose is on 112 error = '' 113 return self._stripHeaderFooter(output), helpers.decode_string(error)

114

115 - def decrypt(self, str_, keyID):

116 proc = self.run(['--decrypt', '-q', '-u %s'%keyID], create_fhs=['stdin', 'stdout']) 117 enc = self._addHeaderFooter(str_, 'MESSAGE') 118 proc.handles['stdin'].write(enc) 119 proc.handles['stdin'].close() 120 121 output = proc.handles['stdout'].read() 122 proc.handles['stdout'].close() 123 124 try: proc.wait() 125 except IOError: pass 126 return output

127

128 - def sign(self, str_, keyID):

129 proc = self.run(['-b', '-u %s'%keyID], create_fhs=['stdin', 'stdout', 'status', 'stderr']) 130 proc.handles['stdin'].write(str_) 131 try: 132 proc.handles['stdin'].close() 133 except IOError: 134 pass 135 136 output = proc.handles['stdout'].read() 137 try: 138 proc.handles['stdout'].close() 139 proc.handles['stderr'].close() 140 except IOError: 141 pass 142 143 stat = proc.handles['status'] 144 resp = self._read_response(stat) 145 try: 146 proc.handles['status'].close() 147 except IOError: 148 pass 149 150 try: proc.wait() 151 except IOError: pass 152 if 'GOOD_PASSPHRASE' in resp or 'SIG_CREATED' in resp: 153 return self._stripHeaderFooter(output) 154 if 'KEYEXPIRED' in resp: 155 return 'KEYEXPIRED' 156 return 'BAD_PASSPHRASE'

157

158 - def verify(self, str_, sign):

159 if str_ is None: 160 return '' 161 f = tmpfile() 162 fd = f.fileno() 163 f.write(str_) 164 f.seek(0) 165 166 proc = self.run(['--verify', '--enable-special-filenames', '-', '-&%s'%fd], create_fhs=['stdin', 'status', 'stderr']) 167 168 f.close() 169 sign = self._addHeaderFooter(sign, 'SIGNATURE') 170 proc.handles['stdin'].write(sign) 171 proc.handles['stdin'].close() 172 proc.handles['stderr'].close() 173 174 stat = proc.handles['status'] 175 resp = self._read_response(stat) 176 proc.handles['status'].close() 177 178 try: proc.wait() 179 except IOError: pass 180 181 keyid = '' 182 if 'GOODSIG' in resp: 183 keyid = resp['GOODSIG'].split()[0] 184 return keyid

185

186 - def get_keys(self, secret = False):

187 if secret: 188 opt = '--list-secret-keys' 189 else: 190 opt = '--list-keys' 191 proc = self.run(['--with-colons', opt], 192 create_fhs=['stdout']) 193 output = proc.handles['stdout'].read() 194 proc.handles['stdout'].close() 195 196 try: proc.wait() 197 except IOError: pass 198 199 keys = {} 200 lines = output.split('\n') 201 for line in lines: 202 sline = line.split(':') 203 if (sline[0] == 'sec' and secret) or \ 204 (sline[0] == 'pub' and not secret): 205 # decode escaped chars 206 name = eval('"' + sline[9].replace('"', '\\"') + '"') 207 # make it unicode instance 208 keys[sline[4][8:]] = helpers.decode_string(name) 209 return keys

210

211 - def get_secret_keys(self):

212 return self.get_keys(True)

213

214 - def _stripHeaderFooter(self, data):

215 """ 216 Remove header and footer from data 217 """ 218 if not data: return '' 219 lines = data.split('\n') 220 while lines[0] != '': 221 lines.remove(lines[0]) 222 while lines[0] == '': 223 lines.remove(lines[0]) 224 i = 0 225 for line in lines: 226 if line: 227 if line[0] == '-': break 228 i = i+1 229 line = '\n'.join(lines[0:i]) 230 return line

231

232 - def _addHeaderFooter(self, data, type_):

233 """ 234 Add header and footer from data 235 """ 236 out = "-----BEGIN PGP %s-----\n" % type_ 237 out = out + "Version: PGP\n" 238 out = out + "\n" 239 out = out + data + "\n" 240 out = out + "-----END PGP %s-----\n" % type_ 241 return out

Source Code for Module common.GnuPG