secure-testing/bin/gen-DSA.py

#!/usr/bin/python
# gen-DSA -- create a DSA template
# Copyright (C) 2011 Florian Weimer <fw@deneb.enyo.de>
#
# User interface based on a shell version written by
# Raphael Geissert <geissert@debian.org>.
# 
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

# This script is intended to be run on security-master to get an
# unprocessed dump of the contents of the embargoed and unembargoed
# queues.
#
# The script reads .deb and .changes files.  A caching database is
# written to ~/.cache.


import sys
import os.path
def setup_path():
    dirname = os.path.dirname
    base = dirname(dirname(os.path.realpath(sys.argv[0])))
    sys.path.insert(0, os.path.join(base, "lib", "python"))
setup_path()

from pwd import getpwuid
import re
import time

import bugs
import debian_support
import secmaster

def parsecommand():
    args = sys.argv[1:]
    if not args:
        usage()

    global opt_save
    if args[0] == "--save":
        opt_save = True
        del args[0]
    else:
        opt_save = False
    if len(args) < 3:
        usage()

    global opt_dsaid
    opt_dsaid = args[0]
    if opt_dsaid.upper().startswith("DSA-"):
        opt_dsaid = opt_dsaid[4:]
    if "-" not in opt_dsaid:
        opt_dsaid += "-1"

    global opt_package
    opt_package = args[1]
    if not opt_package:
        usage("package argument is empty")

    global opt_vulnerability
    opt_vulnerability = args[2]
    if not opt_vulnerability:
        usage("vulnerability argument is empty")

    global opt_cve
    if len(args) >= 4:
        re_cve = re.compile("(?i)CVE-\d{4}-\d{4,}")
        opt_cve = set()
        for cve in args[3].split():
            if not cve:
                continue
            cve = cve.upper()
            if not re_cve.match(cve):
                usage("malformed CVE name: " + repr(cve))
            if cve in opt_cve:
                usage("duplicate CVE: " + repr(cve))
            opt_cve.add(cve)
        opt_cve = tuple(sorted(opt_cve))
    else:
        opt_cve = ()

    global opt_bugs
    if len(args) >= 5:
        opt_bugs = set()
        for bug in args[3].split():
            if not bug:
                continue
            try:
                bug = int(bug)
                if bug <= 0:
                    raise ValueError
            except:
                usage("malformed bug number: " + repr(bug))
            if bug in opt_bugs:
                usage("duplicate bug number: " + repr(bug))
            opt_bugs.add(cve)
        opt_bugs = tuple(sorted(opt_bugs))
    else:
        opt_bugs = ()
    if len(args) >= 5:
        usage()

def usage(msg=None):
    if msg is not None:
        print >>sys.stderr, "error:", msg
    print >>sys.stderr, "usage:", sys.argv[0], \
        "[--save] DSA package vulnerability [CVE [bug number]]"
    print >>sys.stderr
    print >>sys.stderr, \
        "Multiple CVE and bug numbers can be separated by spaces"
    sys.exit(1)

def gecos():
    gecos = os.getenv("DEBFULLNAME")
    if gecos is not None:
        return gecos
    gecos = getpwuid(os.getuid()).pw_gecos
    return gecos.split(",")[0]

def debemail():
    for env in ("DEBEMAIL", "USER"):
        email = os.getenv(env)
        if email is not None:
            return email
    return "unknown"

def filledtemplate(values, re_var=re.compile(r"\$\$?([A-Z_]+)")):
    template = file(debian_support.findresource("doc", "DSA.template")).read()
    def repl(match):
        return values[match.group(1)]
    return re_var.sub(repl, template)

tm = time.gmtime(time.time())
def getdate(months=" January February March April May June July August September October November December".split()):
    return "{0} {1:0}, {2}".format(months[tm.tm_mon], tm.tm_mday, tm.tm_year)

dsa_list_path = debian_support.findresource(*"data DSA list".split())
def checklist():
    name = "DSA-" + opt_dsaid
    for bug in bugs.DSAFile(dsa_list_path):
        if bug.name == name or (bug.name + "-1") == name:
            print >>sys.stderr, "DSA already exists:", repr(opt_dsaid)
            sys.exit(1)

def newdsalist(months=" Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split()):
    lines = ["[{0:0} {1} {2}] DSA-{3} {4} - {5}\n".format(
            tm.tm_mday, months[tm.tm_mon], tm.tm_year,
            opt_dsaid, opt_package, opt_vulnerability)]
    if opt_cve:
        lines.append("\t{" + " ".join(opt_cve) + "}\n")
    lines = lines + list(file(dsa_list_path))
    return lines
    
def checkfile():
    global dsa_file
    dsa_file = "DSA-" + opt_dsaid
    if os.path.exists(dsa_file):
        print >>sys.stderr, "DSA file already exists:", repr(dsa_file)
        sys.exit(1)

def search_queue(queue, package, codename, missing=None):
    """Search QUEUE for PACKAGE in distribution CODENAME.
    Returns MISSING if not found (or raises exception of None).
    If found, returns the maximum version number encountered.
    """
    cname = debian_support.releasecodename(codename)
    versions = [debian_support.Version(pkg.source_version)
                for pkg, dists in queue
                if pkg.source == package and cname in dists]
    if versions:
        versions.sort()
        return str(versions[-1])
    if missing is None:
        raise IOError("could not find version for package {0!r} in {1!r}".
                      format(package, codename))
    return missing

def searchlist_bugs(cve):
    """Returns a list of bug objects pertinent to CVE."""
    result = []
    for bug in bugs.CVEFile(debian_support.findresource(
            *"data CVE list".split())):
        if bug.name in cve:
            result.append(bug)
    for bug in bugs.DTSAFile(debian_support.findresource(
            *"data DTSA list".split())):
        found = False
        for xref in bug.xref:
            if xref in cve:
                found = True
        if found:
            result.append(bug)
    return result

def searchlist(buglist, package, codename):
    "Returns the highest matching version number from BUGLIST."
    relcode = debian_support.releasecodename
    codename = relcode(codename)
    versions = []
    vzero = debian_support.Version("0") # filter out <not-affected> etc.
    for bug in buglist:
        for ann in bug.notes:
            rel = relcode(ann.release or "sid")
            if isinstance(ann, bugs.PackageNoteParsed) \
                    and ann.fixed_version is not None \
                    and ann.fixed_version != vzero \
                    and ann.package == package and rel == codename:
                versions.append(ann.fixed_version)
    if versions:
        versions.sort()
        return str(versions[-1])
    return "<missing>"

parsecommand()
checklist()
checkfile()

queue = secmaster.listqueue()
stable_version = search_queue(queue, opt_package, "stable")
oldstable_version = search_queue(queue, opt_package, "oldstable", "<missing>")
testing_version = search_queue(queue, opt_package, "testing", False)

buglist = searchlist_bugs(opt_cve)
if testing_version is False:
    testing_version = searchlist(buglist, opt_package, "testing")
unstable_version = searchlist(buglist, opt_package, "unstable")

template = filledtemplate({
        "DEBFULLNAME" : gecos(),
        "DEBEMAIL" : debemail(),
        "DSAID" : opt_dsaid,
        "PACKAGE" : opt_package,
        "SPACEDDATE" : getdate().ljust(22),
        "VULNERABILITY" : opt_vulnerability,
        "REMLOCAL" : os.getenv("REMLOCAL", "remote"),
        "DEBIANSPECIFIC" : os.getenv("DEBIANSPECIFIC", "no"),
        "CVE" : " ".join(opt_cve),
        "BUGNUM" : " ".join(opt_bugs),
        "OLDSTABLE" : debian_support.releasecodename("oldstable"),
        "STABLE" : debian_support.releasecodename("stable"),
        "TESTING" : debian_support.releasecodename("testing"),
        "OLDSTABLE_VERSION" : oldstable_version,
        "STABLE_VERSION" : stable_version,
        "TESTING_VERSION" : testing_version,
        "UNSTABLE_VERSION" : unstable_version,
        "SPACEDDEBFULLNAME" : gecos().rjust(24),
        })
if opt_save:
    updatedlist = newdsalist()
    out = file(dsa_file, "w+")
    out.write(template)
    out.close()
    debian_support.replaceFile(updatedlist, dsa_list_path)
else:
    print template,
1	#!/usr/bin/python
2	# gen-DSA -- create a DSA template
3	# Copyright (C) 2011 Florian Weimer <fw@deneb.enyo.de>
4	#
5	# User interface based on a shell version written by
6	# Raphael Geissert <geissert@debian.org>.
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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
21
22	# This script is intended to be run on security-master to get an
23	# unprocessed dump of the contents of the embargoed and unembargoed
24	# queues.
25	#
26	# The script reads .deb and .changes files. A caching database is
27	# written to ~/.cache.
28
29
30	import sys
31	import os.path
32	def setup_path():
33	dirname = os.path.dirname
34	base = dirname(dirname(os.path.realpath(sys.argv[0])))
35	sys.path.insert(0, os.path.join(base, "lib", "python"))
36	setup_path()
37
38	from pwd import getpwuid
39	import re
40	import time
41
42	import bugs
43	import debian_support
44	import secmaster
45
46	def parsecommand():
47	args = sys.argv[1:]
48	if not args:
49	usage()
50
51	global opt_save
52	if args[0] == "--save":
53	opt_save = True
54	del args[0]
55	else:
56	opt_save = False
57	if len(args) < 3:
58	usage()
59
60	global opt_dsaid
61	opt_dsaid = args[0]
62	if opt_dsaid.upper().startswith("DSA-"):
63	opt_dsaid = opt_dsaid[4:]
64	if "-" not in opt_dsaid:
65	opt_dsaid += "-1"
66
67	global opt_package
68	opt_package = args[1]
69	if not opt_package:
70	usage("package argument is empty")
71
72	global opt_vulnerability
73	opt_vulnerability = args[2]
74	if not opt_vulnerability:
75	usage("vulnerability argument is empty")
76
77	global opt_cve
78	if len(args) >= 4:
79	re_cve = re.compile("(?i)CVE-\d{4}-\d{4,}")
80	opt_cve = set()
81	for cve in args[3].split():
82	if not cve:
83	continue
84	cve = cve.upper()
85	if not re_cve.match(cve):
86	usage("malformed CVE name: " + repr(cve))
87	if cve in opt_cve:
88	usage("duplicate CVE: " + repr(cve))
89	opt_cve.add(cve)
90	opt_cve = tuple(sorted(opt_cve))
91	else:
92	opt_cve = ()
93
94	global opt_bugs
95	if len(args) >= 5:
96	opt_bugs = set()
97	for bug in args[3].split():
98	if not bug:
99	continue
100	try:
101	bug = int(bug)
102	if bug <= 0:
103	raise ValueError
104	except:
105	usage("malformed bug number: " + repr(bug))
106	if bug in opt_bugs:
107	usage("duplicate bug number: " + repr(bug))
108	opt_bugs.add(cve)
109	opt_bugs = tuple(sorted(opt_bugs))
110	else:
111	opt_bugs = ()
112	if len(args) >= 5:
113	usage()
114
115	def usage(msg=None):
116	if msg is not None:
117	print >>sys.stderr, "error:", msg
118	print >>sys.stderr, "usage:", sys.argv[0], \
119	"[--save] DSA package vulnerability [CVE [bug number]]"
120	print >>sys.stderr
121	print >>sys.stderr, \
122	"Multiple CVE and bug numbers can be separated by spaces"
123	sys.exit(1)
124
125	def gecos():
126	gecos = os.getenv("DEBFULLNAME")
127	if gecos is not None:
128	return gecos
129	gecos = getpwuid(os.getuid()).pw_gecos
130	return gecos.split(",")[0]
131
132	def debemail():
133	for env in ("DEBEMAIL", "USER"):
134	email = os.getenv(env)
135	if email is not None:
136	return email
137	return "unknown"
138
139	def filledtemplate(values, re_var=re.compile(r"\$\$?([A-Z_]+)")):
140	template = file(debian_support.findresource("doc", "DSA.template")).read()
141	def repl(match):
142	return values[match.group(1)]
143	return re_var.sub(repl, template)
144
145	tm = time.gmtime(time.time())
146	def getdate(months=" January February March April May June July August September October November December".split()):
147	return "{0} {1:0}, {2}".format(months[tm.tm_mon], tm.tm_mday, tm.tm_year)
148
149	dsa_list_path = debian_support.findresource(*"data DSA list".split())
150	def checklist():
151	name = "DSA-" + opt_dsaid
152	for bug in bugs.DSAFile(dsa_list_path):
153	if bug.name == name or (bug.name + "-1") == name:
154	print >>sys.stderr, "DSA already exists:", repr(opt_dsaid)
155	sys.exit(1)
156
157	def newdsalist(months=" Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split()):
158	lines = ["[{0:0} {1} {2}] DSA-{3} {4} - {5}\n".format(
159	tm.tm_mday, months[tm.tm_mon], tm.tm_year,
160	opt_dsaid, opt_package, opt_vulnerability)]
161	if opt_cve:
162	lines.append("\t{" + " ".join(opt_cve) + "}\n")
163	lines = lines + list(file(dsa_list_path))
164	return lines
165
166	def checkfile():
167	global dsa_file
168	dsa_file = "DSA-" + opt_dsaid
169	if os.path.exists(dsa_file):
170	print >>sys.stderr, "DSA file already exists:", repr(dsa_file)
171	sys.exit(1)
172
173	def search_queue(queue, package, codename, missing=None):
174	"""Search QUEUE for PACKAGE in distribution CODENAME.
175	Returns MISSING if not found (or raises exception of None).
176	If found, returns the maximum version number encountered.
177	"""
178	cname = debian_support.releasecodename(codename)
179	versions = [debian_support.Version(pkg.source_version)
180	for pkg, dists in queue
181	if pkg.source == package and cname in dists]
182	if versions:
183	versions.sort()
184	return str(versions[-1])
185	if missing is None:
186	raise IOError("could not find version for package {0!r} in {1!r}".
187	format(package, codename))
188	return missing
189
190	def searchlist_bugs(cve):
191	"""Returns a list of bug objects pertinent to CVE."""
192	result = []
193	for bug in bugs.CVEFile(debian_support.findresource(
194	*"data CVE list".split())):
195	if bug.name in cve:
196	result.append(bug)
197	for bug in bugs.DTSAFile(debian_support.findresource(
198	*"data DTSA list".split())):
199	found = False
200	for xref in bug.xref:
201	if xref in cve:
202	found = True
203	if found:
204	result.append(bug)
205	return result
206
207	def searchlist(buglist, package, codename):
208	"Returns the highest matching version number from BUGLIST."
209	relcode = debian_support.releasecodename
210	codename = relcode(codename)
211	versions = []
212	vzero = debian_support.Version("0") # filter out <not-affected> etc.
213	for bug in buglist:
214	for ann in bug.notes:
215	rel = relcode(ann.release or "sid")
216	if isinstance(ann, bugs.PackageNoteParsed) \
217	and ann.fixed_version is not None \
218	and ann.fixed_version != vzero \
219	and ann.package == package and rel == codename:
220	versions.append(ann.fixed_version)
221	if versions:
222	versions.sort()
223	return str(versions[-1])
224	return "<missing>"
225
226	parsecommand()
227	checklist()
228	checkfile()
229
230	queue = secmaster.listqueue()
231	stable_version = search_queue(queue, opt_package, "stable")
232	oldstable_version = search_queue(queue, opt_package, "oldstable", "<missing>")
233	testing_version = search_queue(queue, opt_package, "testing", False)
234
235	buglist = searchlist_bugs(opt_cve)
236	if testing_version is False:
237	testing_version = searchlist(buglist, opt_package, "testing")
238	unstable_version = searchlist(buglist, opt_package, "unstable")
239
240	template = filledtemplate({
241	"DEBFULLNAME" : gecos(),
242	"DEBEMAIL" : debemail(),
243	"DSAID" : opt_dsaid,
244	"PACKAGE" : opt_package,
245	"SPACEDDATE" : getdate().ljust(22),
246	"VULNERABILITY" : opt_vulnerability,
247	"REMLOCAL" : os.getenv("REMLOCAL", "remote"),
248	"DEBIANSPECIFIC" : os.getenv("DEBIANSPECIFIC", "no"),
249	"CVE" : " ".join(opt_cve),
250	"BUGNUM" : " ".join(opt_bugs),
251	"OLDSTABLE" : debian_support.releasecodename("oldstable"),
252	"STABLE" : debian_support.releasecodename("stable"),
253	"TESTING" : debian_support.releasecodename("testing"),
254	"OLDSTABLE_VERSION" : oldstable_version,
255	"STABLE_VERSION" : stable_version,
256	"TESTING_VERSION" : testing_version,
257	"UNSTABLE_VERSION" : unstable_version,
258	"SPACEDDEBFULLNAME" : gecos().rjust(24),
259	})
260	if opt_save:
261	updatedlist = newdsalist()
262	out = file(dsa_file, "w+")
263	out.write(template)
264	out.close()
265	debian_support.replaceFile(updatedlist, dsa_list_path)
266	else:
267	print template,