]> git.decadent.org.uk Git - dak.git/blob - daklib/gpg.py
Remove unused function.
[dak.git] / daklib / gpg.py
1 """Utilities for signed files
2
3 @contact: Debian FTP Master <ftpmaster@debian.org>
4 @copyright: 2011  Ansgar Burchardt <ansgar@debian.org>
5 @license: GNU General Public License version 2 or later
6 """
7
8 # This program is free software; you can redistribute it and/or modify
9 # it under the terms of the GNU General Public License as published by
10 # the Free Software Foundation; either version 2 of the License, or
11 # (at your option) any later version.
12
13 # This program is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16 # GNU General Public License for more details.
17
18 # You should have received a copy of the GNU General Public License
19 # along with this program; if not, write to the Free Software
20 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
21
22 import apt_pkg
23 import datetime
24 import errno
25 import fcntl
26 import os
27 import select
28
29 try:
30     _MAXFD = os.sysconf("SC_OPEN_MAX")
31 except:
32     _MAXFD = 256
33
34 class GpgException(Exception):
35     pass
36
37 class _Pipe(object):
38     """context manager for pipes
39
40     Note: When the pipe is closed by other means than the close_r and close_w
41     methods, you have to set self.r (self.w) to None.
42     """
43     def __enter__(self):
44         (self.r, self.w) = os.pipe()
45         return self
46     def __exit__(self, type, value, traceback):
47         self.close_w()
48         self.close_r()
49         return False
50     def close_r(self):
51         """close reading side of the pipe"""
52         if self.r:
53             os.close(self.r)
54             self.r = None
55     def close_w(self):
56         """close writing part of the pipe"""
57         if self.w:
58             os.close(self.w)
59             self.w = None
60
61 class SignedFile(object):
62     """handle files signed with PGP
63
64     The following attributes are available:
65       contents            - string with the content (after removing PGP armor)
66       valid               - Boolean indicating a valid signature was found
67       fingerprint         - fingerprint of the key used for signing
68       primary_fingerprint - fingerprint of the primary key associated to the key used for signing
69     """
70     def __init__(self, data, keyrings, require_signature=True, gpg="/usr/bin/gpg"):
71         """
72         @param data: string containing the message
73         @param keyrings: sequence of keyrings
74         @param require_signature: if True (the default), will raise an exception if no valid signature was found
75         @param gpg: location of the gpg binary
76         """
77         self.gpg = gpg
78         self.keyrings = keyrings
79
80         self.valid = False
81         self.fingerprint = None
82         self.primary_fingerprint = None
83
84         self._verify(data, require_signature)
85
86     def _verify(self, data, require_signature):
87         with _Pipe() as stdin:
88          with _Pipe() as contents:
89           with _Pipe() as status:
90            with _Pipe() as stderr:
91             pid = os.fork()
92             if pid == 0:
93                 self._exec_gpg(stdin.r, contents.w, stderr.w, status.w)
94             else:
95                 stdin.close_r()
96                 contents.close_w()
97                 stderr.close_w()
98                 status.close_w()
99
100                 read = self._do_io([contents.r, stderr.r, status.r], {stdin.w: data})
101                 stdin.w = None # was closed by _do_io
102
103                 (pid_, exit_code, usage_) = os.wait4(pid, 0)
104
105                 self.contents = read[contents.r]
106                 self.status   = read[status.r]
107                 self.stderr   = read[stderr.r]
108
109                 if self.status == "":
110                     raise GpgException("No status output from GPG. (GPG exited with status code %s)\n%s" % (exit_code, self.stderr))
111
112                 for line in self.status.splitlines():
113                     self._parse_status(line)
114
115                 if require_signature and not self.valid:
116                     raise GpgException("No valid signature found. (GPG exited with status code %s)\n%s" % (exit_code, self.stderr))
117
118     def _do_io(self, read, write):
119         for fd in write.keys():
120             old = fcntl.fcntl(fd, fcntl.F_GETFL)
121             fcntl.fcntl(fd, fcntl.F_SETFL, old | os.O_NONBLOCK)
122
123         read_lines = dict( (fd, []) for fd in read )
124         write_pos = dict( (fd, 0) for fd in write )
125
126         read_set = list(read)
127         write_set = write.keys()
128         while len(read_set) > 0 or len(write_set) > 0:
129             r, w, x_ = select.select(read_set, write_set, ())
130             for fd in r:
131                 data = os.read(fd, 4096)
132                 if data == "":
133                     read_set.remove(fd)
134                 read_lines[fd].append(data)
135             for fd in w:
136                 data = write[fd][write_pos[fd]:]
137                 if data == "":
138                     os.close(fd)
139                     write_set.remove(fd)
140                 else:
141                     bytes_written = os.write(fd, data)
142                     write_pos[fd] += bytes_written
143
144         return dict( (fd, "".join(read_lines[fd])) for fd in read_lines.keys() )
145
146     def _parse_date(self, value):
147         """parse date string in YYYY-MM-DD format
148
149         @rtype:   L{datetime.datetime}
150         @returns: datetime objects for 0:00 on the given day
151         """
152         year, month, day = value.split('-')
153         date = datetime.date(int(year), int(month), int(day))
154         time = datetime.time(0, 0)
155         return datetime.datetime.combine(date, time)
156
157     def _parse_status(self, line):
158         fields = line.split()
159         if fields[0] != "[GNUPG:]":
160             raise GpgException("Unexpected output on status-fd: %s" % line)
161
162         # VALIDSIG    <fingerprint in hex> <sig_creation_date> <sig-timestamp>
163         #             <expire-timestamp> <sig-version> <reserved> <pubkey-algo>
164         #             <hash-algo> <sig-class> <primary-key-fpr>
165         if fields[1] == "VALIDSIG":
166             self.valid = True
167             self.fingerprint = fields[2]
168             self.primary_fingerprint = fields[11]
169             self.signature_timestamp = self._parse_date(fields[3])
170
171         if fields[1] == "BADARMOR":
172             raise GpgException("Bad armor.")
173
174         if fields[1] == "NODATA":
175             raise GpgException("No data.")
176
177         if fields[1] == "DECRYPTION_FAILED":
178             raise GpgException("Decryption failed.")
179
180         if fields[1] == "ERROR":
181             raise GpgException("Other error: %s %s" % (fields[2], fields[3]))
182
183     def _exec_gpg(self, stdin, stdout, stderr, statusfd):
184         try:
185             if stdin != 0:
186                 os.dup2(stdin, 0)
187             if stdout != 1:
188                 os.dup2(stdout, 1)
189             if stderr != 2:
190                 os.dup2(stderr, 2)
191             if statusfd != 3:
192                 os.dup2(statusfd, 3)
193             for fd in range(4):
194                 old = fcntl.fcntl(fd, fcntl.F_GETFD)
195                 fcntl.fcntl(fd, fcntl.F_SETFD, old & ~fcntl.FD_CLOEXEC)
196             os.closerange(4, _MAXFD)
197
198             args = [self.gpg, "--status-fd=3", "--no-default-keyring"]
199             for k in self.keyrings:
200                 args.append("--keyring=%s" % k)
201             args.extend(["--decrypt", "-"])
202
203             os.execvp(self.gpg, args)
204         finally:
205             os._exit(1)
206
207     def contents_sha1(self):
208         return apt_pkg.sha1sum(self.contents)
209
210 # vim: set sw=4 et: