secure-testing/bin/tracker_service.py

#!/usr/bin/python

import sys
sys.path.insert(0,'../lib/python')
import bugs
import re
import security_db
from web_support import *

if len(sys.argv) not in (3, 5):
    print "usage: python tracker_service.py SOCKET-PATH DATABASE-PATH"
    print "       python tracker_service.py URL HOST PORT DATABASE-PATH"
    sys.exit(1)
if len(sys.argv) == 3:
    socket_name = sys.argv[1]
    db_name = sys.argv[2]
    webservice_base_class = WebService
else:
    server_base_url = sys.argv[1]
    server_address = sys.argv[2]
    server_port = int(sys.argv[3])
    socket_name = (server_base_url, server_address, server_port)
    db_name = sys.argv[4]
    webservice_base_class = WebServiceHTTP

class BugFilter:
    default_action_list = [('show_high_urgency', 'only high urgencies'),
                           ('show_medium_urgency', 'only medium and high urgencies'),
                           ('show_undetermined_urgency', 'issues that may be vulnerable but need to be checked (shown in purple)'),
                           ('show_unimportant_urgency', 'unimportant issues'),
                           ('show_remote_only', 'only remote vulnerabilities')]
    def __init__(self, params, action_list=None):
        if action_list is None:
            self.action_list = self.default_action_list
        else:
            self.action_list = action_list
        self.params = {}
        for (prop, desc) in self.action_list:
            self.params[prop] = int(params.get(prop, (0,))[0])

    def actions(self, url):
        """Returns a HTML snippet which can be used to change the filter."""

        l = []
        for (prop, desc) in self.action_list:
            if self.params[prop]:
                if self.params['show_medium_urgency'] and prop == 'show_medium_urgency':
                    note = 'Restore lower than medium urgencies'
                elif self.params['show_high_urgency'] and prop == 'show_high_urgency':
                    note = 'Restore lower than high urgencies'
                else:
                    note = 'Hide ' + desc
                l.append(TR(TD(A(url.updateParamsDict({prop : None}), note))))
            else:
                note = 'Show ' + desc
                l.append(TR(TD(A(url.updateParamsDict({prop : '1'}), note))))

        return TABLE(l)

    def urgencyFiltered(self, urg, vuln):
        """Returns True for urgencies that should be filtered."""
        filterlow = self.params['show_medium_urgency'] and \
                    urg in ('low', 'low**', 'unimportant',
                    'undetermined', 'not yet assigned')
        filtermed = self.params['show_high_urgency'] and \
                    urg in ('medium', 'medium**', 'low', 'low**',
                    'unimportant', 'undetermined', 'not yet assigned')
        filterund = not self.params['show_undetermined_urgency'] and vuln == 2
        filteruni = not self.params['show_unimportant_urgency'] \
                    and urg == 'unimportant'
        return filterlow or filtermed or filterund or filteruni

    def remoteFiltered(self, remote):
        """Returns True for only remote flaws if filtered."""
        return remote is not None and self.params['show_remote_only'] \
               and not remote

class BugFilterNoDSA(BugFilter):
    def __init__(self, params):
        BugFilter.__init__(self, params, self.default_action_list
            + [('show_nodsa', 'issues that are not severe enough to warrant a DSA')])

    def nodsaFiltered(self, nodsa):
        """Returns True for no DSA issues if filtered."""
        return nodsa and not self.params['show_nodsa']

class TrackerService(webservice_base_class):
    head_contents = compose(STYLE(
        """h1 { font-size : 144%; }
h2 { font-size : 120%; }
h3 { font-size : 100%; }

table { padding-left : 1.5em }
td, th { text-align : left; 
         padding-left : 0.25em;
         padding-right : 0.25em; }
td { vertical-align: baseline }
span.red { color: red; }
span.purple { color: purple; }
span.dangerous { color: rgb(191,127,0); }
"""), SCRIPT('''var old_query_value = "";

function selectSearch() {
  document.searchForm.query.focus();
}

function onSearch(query) {
  if (old_query_value == "") {
    if (query.length > 5) {
      old_query_value = query;
      document.searchForm.submit();
    } else {
      old_query_value = query;
    }
  }
}
''')).toHTML()
    
    nvd_text =  P('''If a "**" is included, the urgency field was automatically
        assigned by the NVD (National Vulnerability Database). Note that this
        rating is automatically derived from a set of known factors about the
        issue (such as access complexity, confidentiality impact, exploitability,
        remediation level, and others). Human intervention is involved in
        determining the values of these factors, but the rating itself comes
        from a fully automated formula.''')

    def __init__(self, socket_name, db_name):
        webservice_base_class.__init__(self, socket_name)
        self.db = security_db.DB(db_name)
        self.register('', self.page_home)
        self.register('*', self.page_object)
        self.register('redirect/*', self.page_redirect)
        self.register('source-package/*', self.page_source_package)
        self.register('binary-package/*', self.page_binary_package)
        self.register('status/release/oldstable',
                      self.page_status_release_oldstable)
        self.register('status/release/stable', self.page_status_release_stable)
        self.register('status/release/stable-backports', 
                      self.page_status_release_stable_backports)
        self.register('status/release/oldstable-backports', 
                      self.page_status_release_oldstable_backports)
        self.register('status/release/testing',
                      self.page_status_release_testing)
        self.register('status/release/unstable',
                      self.page_status_release_unstable)
        self.register('status/dtsa-candidates',
                      self.page_status_dtsa_candidates)
        self.register('status/todo', self.page_status_todo)
        self.register('status/undetermined', self.page_status_undetermined)
        self.register('status/unimportant', self.page_status_unimportant)
        self.register('status/itp', self.page_status_itp)
        self.register('data/unknown-packages', self.page_data_unknown_packages)
        self.register('data/missing-epochs', self.page_data_missing_epochs)
        self.register('data/latently-vulnerable',
                      self.page_data_latently_vulnerable)
        self.register('data/releases', self.page_data_releases)
        self.register('data/funny-versions', self.page_data_funny_versions)
        self.register('data/fake-names', self.page_data_fake_names)
        self.register('data/pts/1', self.page_data_pts)
        self.register('debsecan/**', self.page_debsecan)
        self.register('data/report', self.page_report)

    def page_home(self, path, params, url):
        query = params.get('query', ('',))[0]
        if query:
            if '/' in query:
                return self.page_not_found(url, query)
            else:
                return RedirectResult(url.scriptRelativeFull(query))
        
        return self.create_page(
            url, 'Security Bug Tracker',
            [P(
            """The data in this tracker comes solely from the bug database maintained
by Debian's security team located in the testing-security Subversion """,
            A("http://svn.debian.org/wsvn/secure-testing/data/", "repository"),
            """.  The data represented here is derived from: """,
            A("http://www.debian.org/security/#DSAS", "DSAs"),
            """ issued by the Security Team; issues tracked in the """,
            A("http://cve.mitre.org/cve/", "CVE database"),
            """, issues tracked in the """,
            A("http://nvd.nist.gov/", "National Vulnerability Database"),
            """ (NVD), maintained by NIST; and security issues
discovered in Debian packages as reported in the BTS."""),
             P("""All external data (including Debian bug reports and official Debian
security advisories) must be added to this database before it appears
here. Please help us keep this information up-to-date by """,
               A(url.scriptRelative("data/report"), "reporting"),
               """ any discrepancies or change of states that you are
aware of and/or help us improve the quality of this information by """,
               A(url.scriptRelative("data/report"), "participating"),
               "."),
            make_menu(
            url.scriptRelative,
            ('status/release/unstable',
             'Vulnerable packages in the unstable suite'),
            ('status/release/testing',
             'Vulnerable packages in the testing suite'),
            ('status/release/stable',
             'Vulnerable packages in the stable suite'),
            ('status/release/oldstable',
             'Vulnerable packages in the old stable suite'),
            ('status/release/stable-backports',
             'Vulnerable packages in backports for stable'),
            ('status/release/oldstable-backports',
             'Vulnerable packages in backports for oldstable'),
            ('status/dtsa-candidates', "Candidates for DTSAs"),
            ('status/todo', 'TODO items'),
            ('status/undetermined', 'Packages that may be vulnerable but need to be checked (undetermined issues)'),
            ('status/unimportant', 'Packages that have open unimportant issues'),
            ('status/itp', 'ITPs with potential security issues'),
            ('data/unknown-packages',
             'Packages names not found in the archive'),
            ('data/fake-names', 'Tracked issues without a CVE name'),
            ('data/missing-epochs',
             'Package versions which might lack an epoch'),
            ('data/latently-vulnerable',
             'Packages which are latently vulnerable in unstable'),
            ('data/funny-versions',
             'Packages with strange version numbers'),
            ('data/releases',
             'Covered Debian releases and architectures (slow)'),
            self.make_search_button(url)),
             P("""(You can enter CVE names, Debian bug numbers and package
names in the search forms.)"""),

             H2("External interfaces"),
             P("""If you want to automatically open a relevant web page for
some object, use the """,
               CODE(str(url.scriptRelative("redirect/")), EM("object")),
               """ URL.  If no information is contained in this database,
the browser is automatically redirected to the corresponding external
data source.""")],
            search_in_page=True)

    def page_object(self, path, params, url):
        obj = path[0]
        return self.page_object_or_redirect(url, obj, False)

    def page_redirect(self, path, params, url):
        if path == ():
            obj = ''
        else:
            obj = path[0]
        return self.page_object_or_redirect(url, obj, True)

    def page_object_or_redirect(self, url, obj, redirect):
        c = self.db.cursor()

        if not obj:
            # Redirect to start page.
            return RedirectResult(url.scriptRelativeFull(""))

        # Attempt to decode a bug number.  TEMP-nnn bugs (but not
        # TEMP-nnn-mmm bugs) are treated as bug references, too.
        bugnumber = 0
        fake_bug = False
        try:
            if obj[0:5] == 'FAKE-' or obj[0:5] == 'TEMP-':
                bugnumber = int(obj[5:])
                fake_bug = True
            else:
                bugnumber = int(obj)
        except ValueError:
            pass
        if bugnumber:
            buglist = list(self.db.getBugsFromDebianBug(c, bugnumber))
            if buglist:
                return self.page_debian_bug(url, bugnumber, buglist, fake_bug)
            if redirect:
                return RedirectResult(self.url_debian_bug(url, str(bugnumber)),
                                      permanent=False)

        if 'A' <= obj[0] <= 'Z':
            # Bug names start with a capital letter.
            return self.page_bug(url, obj, redirect)

        if self.db.isSourcePackage(c, obj):
            return RedirectResult(self.url_source_package(url, obj, full=True))
        if  self.db.isBinaryPackage(c, obj):
            return RedirectResult(self.url_binary_package(url ,obj, full=True))

        return self.page_not_found(url, obj)

    def page_bug(self, url, name, redirect):
        # FIXME: Normalize CAN-* to CVE-* when redirecting.  Too many
        # people still use CAN.
        if redirect and name[0:4] == 'CAN-':
            name = 'CVE-' + name[4:]

        cursor = self.db.cursor()
        try:
            bug = bugs.BugFromDB(cursor, name)
        except ValueError:
            if redirect:
                if name[0:4] == 'CVE-':
                    return RedirectResult(self.url_cve(url, name),
                                          permanent=False)
            return self.page_not_found(url, name)
        if bug.name <> name or redirect:
            # Show the normalized bug name in the browser address bar.
            return RedirectResult(url.scriptRelativeFull(bug.name))

        page = []

        def gen_header():
            yield B("Name"), bug.name

            source = bug.name.split('-')[0]
            if source == 'CVE':
                source_xref = compose(self.make_cve_ref(url, bug.name, 'CVE'),
                                      " (at ",
                                      self.make_nvd_ref(url, bug.name,
                                                        'NVD'),
                                      "; ",
                                      self.make_rhbug_ref(url, bug.name,
                                                        'RH'),
                                      ")")
            elif source == 'DSA':
                source_xref = self.make_dsa_ref(url, bug.name, 'Debian')
            elif source == 'DTSA':
                source_xref = 'Debian Testing Security Team'
            elif source == 'TEMP':
                source_xref = (
        'Automatically generated temporary name.  Not for external reference.')
            else:
                source_xref = None

            if source_xref:
                yield B("Source"), source_xref
        
            nvd = self.db.getNVD(cursor, bug.name)

            if nvd and nvd.cve_desc:
                yield B("Description"), nvd.cve_desc
            elif bug.description:
                yield B("Description"), bug.description

            xref = list(self.db.getBugXrefs(cursor, bug.name))
            if xref:
                yield B("References"), self.make_xref_list(url, xref)

            if nvd:
                nvd_range = nvd.rangeString()
                if nvd.severity:
                    nvd_severity = nvd.severity.lower()
                    if nvd_range:
                        nvd_severity = "%s (attack range: %s)" \
                                       % (nvd_severity, nvd_range)
                    yield B("NVD severity"), nvd_severity
            
            debian_bugs = bug.getDebianBugs(cursor)
            if debian_bugs:
                yield (B("Debian Bugs"),
                       self.make_debian_bug_list(url, debian_bugs))

            if not bug.not_for_us:
                for (release, status, reason) in bug.getStatus(cursor):
                    if status == 'undetermined':
                        reason = self.make_purple(reason)
                    elif status <> 'fixed':
                        reason = self.make_red(reason)
                    yield B('Debian/%s' % release), reason

        page.append(make_table(gen_header()))

        if bug.notes:
            page.append(H2("Vulnerable and fixed packages"))

            def gen_source():
                old_pkg = ''
                for (package, release, version, vulnerable) \
                        in self.db.getSourcePackages(cursor, bug.name):
                    if package == old_pkg:
                        package = ''
                    else:
                        old_pkg = package
                        package = compose(
                            self.make_source_package_ref(url, package),
                            " (", self.make_pts_ref(url, package, 'PTS'), ")")
                    if vulnerable == 1:
                        vuln = self.make_red('vulnerable')
                        version = self.make_red(version)
                    elif vulnerable == 2:
                        vuln = self.make_purple('undetermined')
                        version = self.make_purple(version)
                    else:
                        vuln = 'fixed'

                    yield package, ', '.join(release), version, vuln

            page.append(make_table(gen_source(),
    caption=("Source Package", "Release", "Version", "Status"),
    introduction=P('The table below lists information on source packages.')))

            def gen_binary():
                old_pkg = ''
                for (packages, releases, version, archs, vulnerable) \
                    in self.db.getBinaryPackages(cursor, bug.name):
                    pkg = ', '.join(packages)
                    if pkg == old_pkg:
                        packages = ''
                    else:
                        old_pkg = pkg
                        packages = self.make_binary_packages_ref(url, packages)

                    if vulnerable == 1:
                        vuln = self.make_red('vulnerable')
                        version = self.make_red(version)
                    elif vulnerable == 2:
                        vuln = self.make_purple('undetermined')
                        version = self.make_purple(version)
                    else:
                        vuln = 'fixed'
                    yield (packages,
                           ', '.join(releases),
                           version, vuln,
                           ', '.join(archs))

            page.append(make_table(gen_binary(),
        caption=("Binary Package", "Release", "Version", "Status",
                 "Architecures"),
        introduction=P("The next table lists affected binary packages.")))

            def gen_data():
                notes_sorted = bug.notes[:]
                notes_sorted.sort(lambda a, b: cmp(a.package, b.package))
                for n in notes_sorted:
                    if n.release:
                        rel = str(n.release)
                    else:
                        rel = '(unstable)'
                    urgency = str(n.urgency)
                    if n.fixed_version:
                        ver = str(n.fixed_version)
                        if ver == '0':
                            ver = '(not affected)'
                            urgency = ''
                    else:
                        ver = self.make_red('(unfixed)')
                    if urgency == 'not yet assigned':
                        urgency = ''

                    pkg = n.package
                    pkg_kind = n.package_kind
                    if pkg_kind == 'source':
                        pkg = self.make_source_package_ref(url, pkg)
                    elif pkg_kind == 'binary':
                        pkg = self.make_binary_package_ref(url, pkg)
                    elif pkg_kind == 'itp':
                        pkg_kind = 'ITP'
                        rel = ''
                        ver = ''
                        urgency = ''

                    bugs = n.bugs
                    bugs.sort()
                    bugs = make_list(
                        map(lambda x: self.make_debian_bug(url, x), bugs))
                    if n.bug_origin:
                        origin = self.make_xref(url, n.bug_origin)
                    else:
                        origin = ''
                    yield (pkg, pkg_kind, rel, ver, urgency, origin, bugs)

            page.append(
                make_table(gen_data(),
                    caption=("Package", "Type", "Release", "Fixed Version",
                             "Urgency", "Origin", "Debian Bugs"),
                    introduction=P("The information above is based on the following data on fixed versions.")))

        if bug.comments:
            page.append(H2("Notes"))
            def gen_comments():
                for (t, c) in bug.comments:
                    yield c
            page.append(make_pre(gen_comments()))

        return self.create_page(url, bug.name, page)

    def page_debian_bug(self, url, bugnumber, buglist, fake_bug):
        if fake_bug:
            new_buglist = []
            for b in buglist:
                (bug_name, urgency, description) = b
                if bug_name[0:5] == 'FAKE-' or bug_name[0:5] == 'TEMP-':
                    new_buglist.append(b)
            if len(new_buglist) > 0:
                # Only replace the bug list if there are still fake
                # bug reports.
                buglist = new_buglist

        if len(buglist) == 1:
            # Single issue, redirect.
            return RedirectResult(url.scriptRelativeFull(buglist[0][0]))

        def gen():
            for (name, urgency, description) in buglist:
                if urgency == "unknown":
                    urgency = ""
                yield self.make_xref(url, name), urgency, description

        if fake_bug:
            intro = """The URL you used contained a non-stable name
based on a Debian bug number.  This name cannot be mapped to a specific
issue. """
        else:
            intro = ""

        return self.create_page(
            url, "Information related to Debian bug #%d" % bugnumber,
            [P(intro + "The following issues reference to Debian bug ",
               self.make_debian_bug(url, bugnumber), ":"),
             make_table(gen(),
                        caption=("Name", "Urgency", "Description"))])

    def page_not_found(self, url, query):
        return self.create_page(url, 'Not found',
                                [P('Your query ',
                                   CODE(query),
                                   ' matched no results.')],
                                status=404)

    def page_report(self, path, params, url):
        return self.create_page(
            url, 'Reporting discrepancies in the data',
            [P("""The data in this tracker is always in flux, as bugs are fixed and new
issues disclosed, the data contained herein is updated. We strive to
maintain complete and accurate state information, and appreciate any
updates in status, information or new issues."""),
             P("There are three ways that you can report updates to this information:"),
             make_numbered_list(
            [P("""IRC: We can be found at """,
               CODE("irc.oftc.net"),
               ", ",
               CODE("#debian-security"),
    """. If you have information to report, please go ahead and join
the channel and tell us.  Please feel free to state the issue,
regardless if there is someone who has acknowledged you. Many of us
idle on this channel and may not be around when you join, but we read
the backlog and will see what you have said. If you require a
response, do not forget to let us know how to get a hold of you."""),
             P("Mailing list: Our mailing list is: ",
               A("mailto:debian-security-tracker@lists.debian.org",
                 "debian-security-tracker@lists.debian.org")),
             P("""Helping out: We welcome people who wish to join us in tracking
issues. The process is designed to be easy to learn and participate,
please read our """,
               A("http://svn.debian.org/wsvn/secure-testing/doc/narrative_introduction?op=file&rev=0&sc=0",
                 "Introduction"),
               """ to get familiar with how things work.  Join us on
our mailing list, and on IRC and request to be added to the Alioth """,
               A("http://alioth.debian.org/projects/secure-testing/", "project"),
               """. We are really quite friendly. If you have a
question about how things work, don't be afraid to ask, we would like
to improve our documentation and procedures, so feedback is welcome.""")])])

    def page_source_package(self, path, params, url):
        pkg = path[0]
        
        def gen_versions():
            for (releases, version) in self.db.getSourcePackageVersions(
                self.db.cursor(), pkg):
                yield ', '.join(releases), version
        def gen_binary():
            for (packages, releases, archs, version) \
                    in self.db.getBinaryPackagesForSource(
                self.db.cursor(), pkg):
                yield (self.make_binary_packages_ref(url, packages),
                       ', '.join(releases), version, ', '.join(archs))
        def gen_bug_list(lst):
            for (bug, description) in lst:
                yield self.make_xref(url, bug), description
                
        return self.create_page(
            url, "Information on source package " + pkg,
            [make_menu(lambda x: x,
                       (self.url_pts(url, pkg),
                        pkg + ' in the Package Tracking System'),
                       (self.url_debian_bug_pkg(url, pkg),
                        pkg + ' in the Bug Tracking System'),
                       (self.url_testing_status(url, pkg),
                        pkg + ' in the testing migration checker')),
             H2("Available versions"),
             make_table(gen_versions(), caption=("Release", "Version")),
             
             H2("Available binary packages"),
             make_table(gen_binary(),
            caption=('Package', 'Release', 'Version', 'Architectures'),
            replacement="""No binary packages are recorded in this database.
This probably means that the package is architecture-specific, and the
architecture is currently not tracked."""),

             H2("Open issues"),
             make_table(gen_bug_list(self.db.getBugsForSourcePackage
                                     (self.db.cursor(), pkg, True)),
                        caption=('Bug', 'Description'),
                        replacement='No known open issues.'),

             H2("Resolved issues"),
             make_table(gen_bug_list(self.db.getBugsForSourcePackage
                                     (self.db.cursor(), pkg, False)),
                        caption=('Bug', 'Description'),
                        replacement='No known resolved issues.')])

    def page_binary_package(self, path, params, url):
        pkg = path[0]

        def gen_versions():
            for (releases, source, version, archs) \
                    in self.db.getBinaryPackageVersions(self.db.cursor(), pkg):
                yield (', '.join(releases),
                       self.make_source_package_ref(url, source),
                       version, ', '.join(archs))
        def gen_bug_list(lst):
            for (bug, description) in lst:
                yield self.make_xref(url, bug), description

        return self.create_page(
            url, "Information on binary package " + pkg,
            [make_menu(lambda x: x,
                       (self.url_debian_bug_pkg(url, pkg),
                        pkg + ' in the Bug Tracking System')),
             H2("Available versions"),
             make_table(gen_versions(),
                caption=("Release", "Source", "Version", "Architectures")),
             
             H2("Open issues"),
             make_table(gen_bug_list(self.db.getBugsForBinaryPackage
                                     (self.db.cursor(), pkg, True)),
                        caption=('Bug', 'Description'),
                        replacement='No known open issues.'),

             H2("Resolved issues"),
             make_table(gen_bug_list(self.db.getBugsForBinaryPackage
                                     (self.db.cursor(), pkg, False)),
                        caption=('Bug', 'Description'),
                        replacement='No known resolved issues.'),

             H2("Non-issues"),
                make_table(gen_bug_list(self.db.getNonBugsForBinaryPackage
                                        (self.db.cursor(), pkg)),
                    caption=('Bug', 'Description'),
                    replacement="""No known issues which do not affect
this package, but still reference it.""")])

    def page_status_release_stable_oldstable(self, release, params, url):
        assert release in ('stable', 'oldstable')
        
        bf = BugFilterNoDSA(params)
        
        def gen():
            old_pkg_name = ''
            for (pkg_name, bug_name, archive, urgency, vulnerable, remote, no_dsa) in \
                    self.db.cursor().execute(
                """SELECT package, bug, section, urgency, vulnerable, remote, no_dsa
                FROM %s_status""" % release):
                if bf.urgencyFiltered(urgency, vulnerable):
                    continue
                if bf.remoteFiltered(remote):
                    continue
                if bf.nodsaFiltered(no_dsa):
                    continue

                if pkg_name == old_pkg_name:
                    pkg_name = ''
                else:
                    old_pkg_name = pkg_name
                    if archive <> 'main':
                        pkg_name = "%s (%s)" % (pkg_name, archive)

                if remote is None:
                    remote = ''
                elif remote:
                    remote = 'yes'
                else:
                    remote = 'no'

                if urgency.startswith('high'):
                    urgency = self.make_red(urgency)
                elif vulnerable == 2:
                    urgency = self.make_purple(urgency)
                else:
                    if no_dsa:
                        urgency = urgency + '*'

                yield pkg_name, self.make_xref(url, bug_name), urgency, remote

        return self.create_page(
            url, 'Vulnerable source packages in the %s suite' % release,
            [bf.actions(url), BR(),
             make_table(gen(), caption=("Package", "Bug", "Urgency", "Remote")),
             P('''If a "*" is included in the urgency field, no DSA is planned
                  for this vulnerability.'''),
             self.nvd_text])

    def page_status_release_stable(self, path, params, url):
        return self.page_status_release_stable_oldstable('stable', params, url)
    def page_status_release_oldstable(self, path, params, url):
        return self.page_status_release_stable_oldstable('oldstable',
                                                         params, url)
            
    def page_status_release_testing(self, path, params, url):
        bf = BugFilterNoDSA(params)

        def gen():
            old_pkg_name = ''
            for (pkg_name, bug_name, archive, urgency, vulnerable,
                 sid_vulnerable, ts_fixed, remote, no_dsa) \
                 in self.db.cursor().execute(
                """SELECT package, bug, section, urgency, vulnerable,
                unstable_vulnerable, testing_security_fixed, remote, no_dsa
                FROM testing_status"""):
                if bf.urgencyFiltered(urgency, vulnerable):
                    continue
                if bf.remoteFiltered(remote):
                    continue
                if bf.nodsaFiltered(no_dsa):
                    continue

                if pkg_name == old_pkg_name:
                    pkg_name = ''
                else:
                    old_pkg_name = pkg_name
                    if archive <> 'main':
                        pkg_name = "%s (%s)" % (pkg_name, archive)

                if remote is None:
                    remote = ''
                elif remote:
                    remote = 'yes'
                else:
                    remote = 'no'

                if ts_fixed:
                    status = 'fixed in testing-security'
                else:
                    if sid_vulnerable:
                        status = self.make_red('unstable is vulnerable')
                    else:
                        status = self.make_dangerous('fixed in unstable')

                if urgency.startswith('high'):
                    urgency = self.make_red(urgency)
                elif vulnerable == 2:
                    urgency = self.make_purple(urgency)

                yield (pkg_name, self.make_xref(url, bug_name),
                       urgency, remote, status)

        return self.create_page(
            url, 'Vulnerable source packages in the testing suite',
            [make_menu(url.scriptRelative,
                       ("status/dtsa-candidates", "Candidates for DTSAs")),
             bf.actions(url), BR(),
             make_table(gen(), caption=("Package", "Bug", "Urgency", "Remote")),
             self.nvd_text])

    def page_status_release_unstable_like(self, path, params, url,
                                          rel, title):
        bf = BugFilter(params)

        def gen():
            old_pkg_name = ''
            for (pkg_name, bug_name, section, urgency, vulnerable, remote) \
                    in self.db.cursor().execute(
                """SELECT DISTINCT sp.name, st.bug_name,
                sp.archive, st.urgency, st.vulnerable,
                (SELECT range_remote FROM nvd_data
                 WHERE cve_name = st.bug_name)
                FROM source_package_status AS st, source_packages AS sp
                WHERE st.vulnerable AND sp.rowid = st.package
                AND sp.release = ?  AND sp.subrelease = ''
                ORDER BY sp.name, st.bug_name""", (rel,)):
                if bf.urgencyFiltered(urgency, vulnerable):
                    continue
                if bf.remoteFiltered(remote):
                    continue

                if pkg_name == old_pkg_name:
                    pkg_name = ''
                else:
                    old_pkg_name = pkg_name
                    if section <> 'main':
                        pkg_name = "%s (%s)" % (pkg_name, section)
                    else:
                        pkg_name = self.make_xref(url, pkg_name)

                if remote is None:
                    remote = ''
                elif remote:
                    remote = 'yes'
                else:
                    remote = 'no'

                if urgency.startswith('high'):
                    urgency = self.make_red(urgency)
                elif vulnerable == 2:
                    urgency = self.make_purple(urgency)

                yield pkg_name, self.make_xref(url, bug_name), urgency, remote

        return self.create_page(
            url, title,
            [P("""Note that the list below is based on source packages.
            This means that packages are not listed here once a new,
            fixed source version has been uploaded to the archive, even
            if there are still some vulnerably binary packages present
            in the archive."""),
             bf.actions(url), BR(),
             make_table(gen(), caption=('Package', 'Bug', 'Urgency', 'Remote')),
             self.nvd_text])

    def page_status_release_unstable(self, path, params, url):
        return self.page_status_release_unstable_like(
            path, params, url,
            title='Vulnerable source packages in the unstable suite',
            rel='sid')

    def page_status_release_stable_backports(self, path, params, url):
        return self.page_status_release_unstable_like(
            path, params, url,
            title='Vulnerable source packages among backports for stable',
            rel='lenny-backports')

    def page_status_release_oldstable_backports(self, path, params, url):
        return self.page_status_release_unstable_like(
            path, params, url,
            title='Vulnerable source packages among backports for oldstable',
            rel='etch-backports')

    def page_status_dtsa_candidates(self, path, params, url):
        bf = BugFilter(params)

        def gen():
            old_pkg_name = ''
            for (pkg_name, bug_name, archive, urgency, vulnerable,
                 stable_later, remote) \
                    in self.db.cursor().execute(
                """SELECT package, bug, section, urgency, vulnerable,
                (SELECT testing.version_id < stable.version_id
                 FROM source_packages AS testing, source_packages AS stable
                 WHERE testing.name = testing_status.package
                 AND testing.release = 'squeeze'
                 AND testing.subrelease = ''
                 AND testing.archive = testing_status.section
                 AND stable.name = testing_status.package
                 AND stable.release = 'lenny'
                 AND stable.subrelease = 'security'
                 AND stable.archive = testing_status.section),
                (SELECT range_remote FROM nvd_data
                 WHERE cve_name = bug)
                FROM testing_status
                WHERE (NOT unstable_vulnerable)
                AND (NOT testing_security_fixed)"""):
                if bf.urgencyFiltered(urgency, vulnerable):
                    continue
                if bf.remoteFiltered(remote):
                    continue

                if pkg_name == old_pkg_name:
                    pkg_name = ''
                    migration = ''
                else:
                    old_pkg_name = pkg_name
                    migration = A(self.url_testing_status(url, pkg_name),
                                  "check")
                    if archive <> 'main':
                        pkg_name = "%s (%s)" % (pkg_name, archive)
                    else:
                        pkg_name = self.make_source_package_ref(url, pkg_name)

                if remote is None:
                    remote = ''
                elif remote:
                    remote = 'yes'
                else:
                    remote = 'no'

                if urgency.starstwith('high'):
                    urgency = self.make_red(urgency)
                elif vulnerable == 2:
                    urgency = self.make_purple(urgency)

                if stable_later:
                    notes = "(fixed in stable?)"
                else:
                    notes = ''

                yield (pkg_name, migration, self.make_xref(url, bug_name),
                       urgency, remote, notes)

        return self.create_page(
            url, "Candidates for DTSAs",
            [P("""The table below lists packages which are fixed
in unstable, but unfixed in testing.  Use the testing migration
checker to find out why they have not entered testing yet."""),
             make_menu(url.scriptRelative,
                       ("status/release/testing",
                        "List of vulnerable packages in testing")),
             bf.actions(url), BR(),
             make_table(gen(),
                        caption=("Package", "Migration", "Bug", "Urgency",
                                 "Remote"))])

    def page_status_todo(self, path, params, url):
        hide_check = params.get('hide_check', False)
        if hide_check:
            flags = A(url.updateParamsDict({'hide_check' : None}),
                      'Show "check" TODOs')
        else:
            flags = A(url.updateParamsDict({'hide_check' : '1'}),
                  'Hide "check" TODOs')
            
        def gen():
            for (bug, description) in self.db.getTODOs(hide_check=hide_check):
                yield self.make_xref(url, bug), description
        return self.create_page(
            url, "Bugs with TODO items",
            [P(flags), make_table(gen(), caption=("Bug", "Description"))])

    def page_status_undetermined(self, path, params, url):
        def gen():
            outrel = []
            old_bug = ''
            old_pkg = ''
            old_dsc = ''
            last_displayed = ''
            releases = ('sid', 'squeeze', 'lenny', 'etch')
            for (pkg_name, bug_name, release, desc) in self.db.cursor().execute(
                    """SELECT DISTINCT sp.name, st.bug_name, sp.release,
                    bugs.description
                    FROM source_package_status AS st, source_packages AS sp, bugs
                    WHERE st.vulnerable == 2 AND sp.rowid = st.package
                    AND ( sp.release = ? OR sp.release = ? OR sp.release = ?
                    OR sp.release = ? )
                    AND sp.subrelease = '' AND st.bug_name == bugs.name
                    ORDER BY sp.name, st.bug_name""", releases):

                if old_bug == '':
                    old_bug = bug_name
                    old_pkg = pkg_name
                    old_dsc = desc
                elif old_bug != bug_name:
                    if old_pkg == last_displayed:
                        to_display = ''
                    else:
                        to_display = old_pkg
                    yield to_display, self.make_xref(url, old_bug), old_dsc, ', '.join(outrel)
                    last_displayed = old_pkg
                    old_bug = bug_name
                    old_pkg = pkg_name
                    old_dsc = desc
                    outrel = []
                outrel.append( release )
            yield old_pkg, self.make_xref(url, old_bug), old_dsc, ', '.join(outrel)

        return self.create_page(url, 'Packages that may be vulnerable but need to be checked      (undetermined issues)',
            [P("""This page lists packages that may or may not be affected
            by known issues.  This means that some additional work needs to
            be done to determined whether the package is actually
            vulnerable or not.  This list is a good area for new
            contributors to make quick and meaningful contributions."""),
            make_table(gen(), caption=('Package', 'Bug', 'Description', 'Releases'))])

    def page_status_unimportant(self, path, params, url):
        def gen():
            outrel = []
            old_bug = ''
            old_pkg = ''
            old_dsc = ''
            old_name = ''
            last_displayed = ''
            releases = ('sid', 'squeeze', 'lenny', 'etch')
            for (pkg_name, bug_name, release, desc) in self.db.cursor().execute(
                    """SELECT DISTINCT sp.name, st.bug_name, sp.release,
                    bugs.description
                    FROM source_package_status AS st, source_packages AS sp, bugs
                    WHERE st.vulnerable > 0 AND sp.rowid = st.package
                    AND ( sp.release = ? OR sp.release = ? OR sp.release = ?
                    OR sp.release = ? ) AND st.urgency == 'unimportant'
                    AND sp.subrelease = '' AND st.bug_name == bugs.name
                    ORDER BY sp.name, st.bug_name""", releases):

                if old_bug == '':
                    old_bug = bug_name
                    old_pkg = pkg_name
                    old_dsc = desc
                elif old_bug != bug_name:
                    if old_pkg == last_displayed:
                        to_display = ''
                    else:
                        to_display = old_pkg
                    yield to_display, self.make_xref(url, old_bug), old_dsc, ', '.join(outrel)
                    last_displayed = old_pkg
                    old_bug = bug_name
                    old_pkg = pkg_name
                    old_dsc = desc
                    outrel = []
                outrel.append( release )
            yield old_pkg, self.make_xref(url, old_bug), old_dsc, ', '.join(outrel)

        return self.create_page(url, 'Packages that have open unimportant issues',
            [P("""This page lists packages that are affected by issues
            that are considered unimportant from a security perspective.
            These issues are thought to be unexploitable or uneffective
            in most situations (for example, browser denial-of-services)."""),
            make_table(gen(), caption=('Package', 'Bug', 'Description', 'Releases'))])

    def page_status_itp(self, path, params, url):
        def gen():
            old_pkg = ''
            for pkg, bugs, debian_bugs in self.db.getITPs(self.db.cursor()):
                if pkg == old_pkg:
                    pkg = ''
                else:
                    old_pkg = pkg
                yield (pkg, self.make_xref_list(url, bugs),
                       self.make_debian_bug_list(url, debian_bugs))
        return self.create_page(
            url, "ITPs with potential security issues",
            [make_table(gen(), caption=("Package", "Issue", "Debian Bugs"),
                        replacement="No ITP bugs are currently known.")])

    def page_data_unknown_packages(self, path, params, url):
        def gen():
            for name, bugs in self.db.getUnknownPackages(self.db.cursor()):
                yield name, self.make_xref_list(url, bugs)
        return self.create_page(
            url, "Unknown packages",
            [P("""Sometimes, a package referenced in a bug report
cannot be found in the database.  This can be the result of a spelling
error, or a historic entry refers to a
package which is no longer in the archive."""),
             make_table(gen(), caption=("Package", "Bugs"),
        replacement="No unknown packages are referenced in the database.")])

    def page_data_missing_epochs(self, path, params, url):
        def gen():
            old_bug = ''
            old_pkg = ''
            for bug, pkg, ver1, ver2 in self.db.cursor().execute(
                """SELECT DISTINCT bug_name, n.package,
                n.fixed_version, sp.version
                FROM package_notes AS n, source_packages AS sp
                WHERE n.package_kind = 'source'
                AND n.fixed_version NOT LIKE '%:%'
                AND n.fixed_version <> '0'
                AND n.bug_origin = ''
                AND sp.name = n.package
                AND sp.version LIKE '%:%'
                ORDER BY bug_name, package"""):
                if bug == old_bug:
                    bug = ''
                else:
                    old_bug = bug
                    old_pkg = ''
                    bug = self.make_xref(url, bug)
                if pkg == old_pkg:
                    pkg = ''
                else:
                    old_pkg = pkg
                    pkg = self.make_source_package_ref(url, pkg)
                yield bug, pkg, ver1, ver2

        return self.create_page(
            url, "Missing epochs in package versions",
            [make_table(gen(),
                caption=("Bug", "Package", "Version 1", "Version 2"),
                replacement="No source package version with missing epochs.")])

    def page_data_latently_vulnerable(self, path, params, url):
        def gen():
            for pkg, bugs in self.db.cursor().execute(
                """SELECT package, string_set(bug_name)
                FROM package_notes AS p1
                WHERE release <> ''
                AND (bug_name LIKE 'CVE-%' OR bug_name LIKE 'TEMP-%')
                AND NOT EXISTS (SELECT 1 FROM package_notes AS p2
                                WHERE p2.bug_name = p1.bug_name
                                AND p2.package = p1.package
                                AND release = '')
                AND EXISTS (SELECT 1 FROM source_packages
                           WHERE name = p1.package AND release = 'sid')
                GROUP BY package
                ORDER BY package"""):
                pkg = self.make_source_package_ref(url, pkg)
                bugs = bugs.split(',')
                yield pkg, self.make_xref_list(url, bugs)

        def gen_unimportant():
            for pkg, bugs in self.db.cursor().execute(
                """SELECT package, string_set(bug_name)
                FROM package_notes AS p1
                WHERE release <> ''
                AND urgency <> 'unimportant'
                AND (bug_name LIKE 'CVE-%' OR bug_name LIKE 'TEMP-%')
                AND EXISTS (SELECT 1 FROM package_notes AS p2
                                WHERE p2.bug_name = p1.bug_name
                                AND p2.package = p1.package
                                AND release = '')
                AND NOT EXISTS (SELECT 1 FROM package_notes AS p2
                                WHERE p2.bug_name = p1.bug_name
                                AND p2.package = p1.package
                                AND urgency <> 'unimportant'
                                AND release = '')
                AND EXISTS (SELECT 1 FROM source_packages
                           WHERE name = p1.package AND release = 'sid')
                GROUP BY package
                ORDER BY package"""):
                pkg = self.make_source_package_ref(url, pkg)
                bugs = bugs.split(',')
                yield pkg, self.make_xref_list(url, bugs)

        return self.create_page(
            url, "Latently vulnerable packages in unstable",
            [P(
"""A package is latently vulnerable in unstable if it is vulnerable in
any release, and there is no package note for the same vulnerability
and package in unstable (and the package is still available in
unstable, of course)."""),
             make_table(gen(),
                caption=("Package", "Bugs"),
                replacement="No latently vulnerable packages were found."),
             P(
"""The next table lists issues which are marked unimportant for
unstable, but for which release-specific annotations exist which are
not unimportant."""),
             make_table(gen_unimportant(),
                caption=("Package", "Bugs"),
                replacement=
    "No packages with unimportant latent vulnerabilities were found."),
            ])

    def page_data_releases(self, path, params, url):
        def gen():
            for (rel, subrel, archive, sources, archs) \
                    in self.db.availableReleases():
                if sources:
                    sources = 'yes'
                else:
                    sources = 'no'
                yield rel, subrel, archive, sources, make_list(archs)
        return self.create_page(
            url, "Available releases",
            [P("""The security issue database is checked against
the Debian releases listed in the table below."""),
             make_table(gen(),
                        caption=("Release", "Subrelease", "Archive",
                                 "Sources", "Architectures"))])

    def page_data_funny_versions(self, path, params, url):
        def gen():
            for name, release, archive, version, source_version \
                in self.db.getFunnyPackageVersions():
                yield name, release, archive, source_version, version

        return self.create_page(
            url, "Version conflicts between source/binary packages",
            [P("""The table below lists source packages
            which have a binary package of the same name, but with a different
            version.  This means that extra care is necessary to determine
            the version of a package which has been fixed.  (Note that
            the bug tracker prefers source versions to binary versions
            in this case.)"""),
             make_table(gen(),
                        caption=("Package",
                                 "Release", 
                                 "Archive",
                                 "Source Version",
                                 "Binary Version")),
             P("""Technically speaking, these version numbering is fine,
but it makes version-based bug tracking quite difficult for these packages."""),
             P("""There are many binary packages which are built from source
             packages with different version numbering schemes.  However, as
             long as none of the binary packages carries the same name as the
             source package, most confusion is avoided or can be easily
             explained.""")])

    def page_data_fake_names(self, path, params, url):
        def gen():
            for (bug, description) in self.db.getFakeBugs():
                yield self.make_xref(url, bug), description
        return self.create_page(
            url, "Automatically generated issue names",
            [P("""Some issues have not been assigned CVE names, but are still
tracked by this database.  In this case, the system automatically assigns
a unique name.  These names are not stable and can change when the database
is updated, so they should not be used in external references."""),
             P('''The automatically generated names come in two flavors:
the first kind starts with the string "''', CODE("TEMP-000000-"),
               '''".  This means that no Debian bug has been assigned to this
issue (or a bug has been created and is not recorded in this database).
In the second kind of names, there is a Debian bug for the issue, and the "''',
               CODE("000000"), '''"part of the name is replaced with the
Debian bug number.'''),
             make_table(gen(),
                        caption=("Bug", "Description"))])

    def page_data_pts(self, path, params, url):
        data = []
        for pkg, bugs in self.db.cursor().execute(
                """SELECT package, COUNT(DISTINCT bug) FROM
                (SELECT package, bug FROM stable_status
                 UNION ALL SELECT package, bug FROM oldstable_status
                 UNION ALL SELECT DISTINCT sp.name, st.bug_name
                   FROM source_package_status AS st, source_packages AS sp
                   WHERE st.vulnerable AND st.urgency <> 'unimportant'
                   AND sp.rowid = st.package AND sp.release = 'sid'
                   AND sp.subrelease = '') x
                GROUP BY package ORDER BY package"""):
            data.append(pkg)
            data.append(':')
            data.append(str(bugs))
            data.append('\n')
        return BinaryResult(''.join(data))

    def page_debsecan(self, path, params, url):
        obj = '/'.join(path)
        data = self.db.getDebsecan(obj)
        if data:
            return BinaryResult(data)
        else:
            return self.create_page(
                url, "Object not found",
                [P("The requested debsecan object has not been found.")],
                status=404)

    def create_page(self, url, title, body, search_in_page=False, status=200):
        append = body.append
        append(HR())
        if not search_in_page:
            append(self.make_search_button(url))
        append(P(A(url.scriptRelative(""), "Home"),
                    " - ", A(url.absolute("http://secure-testing.debian.net/"),
                             "Testing Security Team"),
                    " - ", A(url.absolute("http://www.debian.org/security/"),
                             "Debian Security"),
                    " - ", A(url.absolute
                             ("http://www.enyo.de/fw/impressum.html"),
                             "Imprint")))
        if search_in_page:
            on_load = "selectSearch()"
        else:
            on_load = None
        return HTMLResult(self.add_title(title, body,
                                         head_contents=self.head_contents,
                                         body_attribs={'onload': on_load}),
                          doctype=self.html_dtd(),
                          status=status)

    def make_search_button(self, url):
        return FORM("Search for package or bug name: ",
                    INPUT(type='text', name='query',
                          onkeyup="onSearch(this.value)",
                          onmousemove="onSearch(this.value)"),
                    INPUT(type='submit', value='Go'),
                    ' ',
                    A(url.scriptRelative("data/report"), "Reporting problems"),
                    method='get',
                    action=url.scriptRelative(''))

    def url_cve(self, url, name):
        return url.absolute("http://cve.mitre.org/cgi-bin/cvename.cgi",
                            name=name)
    def url_nvd(self, url, name):
        return url.absolute("http://web.nvd.nist.gov/view/vuln/detail",
                            vulnId=name)
    def url_rhbug(self, url, name):
        return url.absolute("https://bugzilla.redhat.com/show_bug.cgi",
                            id=name)
    
    def url_dsa(self, url, dsa, re_dsa=re.compile(r'^DSA-(\d+)(?:-\d+)?$')):
        match = re_dsa.match(dsa)
        if match:
            # We must determine the year because there is no generic URL.
            (number,) = match.groups()
            for (date,) in self.db.cursor().execute(
                "SELECT release_date FROM bugs WHERE name = ?", (dsa,)):
                (y, m, d) = date.split('-')
                return url.absolute("http://www.debian.org/security/%d/dsa-%d"
                                    % (int(y), int(number)))
        return None

    def url_debian_bug(self, url, debian):
        return url.absolute("http://bugs.debian.org/cgi-bin/bugreport.cgi",
                            bug=str(debian))
    def url_debian_bug_pkg(self, url, debian):
        return url.absolute("http://bugs.debian.org/cgi-bin/pkgreport.cgi",
                            pkg=debian)
    def url_pts(self, url, package):
        return url.absolute("http://packages.qa.debian.org/common/index.html",
                            src=package)
    def url_testing_status(self, url, package):
        return url.absolute("http://release.debian.org/migration/testing.pl",
                            package=package)
    def url_source_package(self, url, package, full=False):
        if full:
            return url.scriptRelativeFull("source-package/" + package)
        else:
            return url.scriptRelative("source-package/" + package)
    def url_binary_package(self, url, package, full=False):
        if full:
            return url.scriptRelativeFull("binary-package/" + package)
        else:
            return url.scriptRelative("binary-package/" + package)

    def make_xref(self, url, name):
        return A(url.scriptRelative(name), name)

    def make_xref_list(self, url, lst, separator=', '):
        return make_list(map(lambda x: self.make_xref(url, x), lst), separator)

    def make_debian_bug(self, url, debian):
        return A(self.url_debian_bug(url, debian), str(debian))
    def make_debian_bug_list(self, url, lst):
        return make_list(map(lambda x: self.make_debian_bug(url, x), lst))

    def make_cve_ref(self, url, cve, name=None):
        if name is None:
            name = cve
        return A(self.url_cve(url, cve), name)
    
    def make_nvd_ref(self, url, cve, name=None):
        if name is None:
            name = cve
        return A(self.url_nvd(url, cve), name)
    
    def make_rhbug_ref(self, url, cve, name=None):
        if name is None:
            name = cve
        return A(self.url_rhbug(url, cve), name)

    def make_dsa_ref(self, url, dsa, name=None):
        if name is None:
            name = dsa
        u = self.url_dsa(url, dsa)
        if u:
            return A(u, name)
        else:
            return name

    def make_pts_ref(self, url, pkg, name=None):
        if name is None:
            name = pkg
        return A(self.url_pts(url, pkg), name)

    def make_source_package_ref(self, url, pkg, title=None):
        if title is None:
            title = pkg
        return A(self.url_source_package(url, pkg), title)
    def make_binary_package_ref(self, url, pkg, title=None):
        if title is None:
            title = pkg
        return A(self.url_binary_package(url, pkg), title)
    def make_binary_packages_ref(self, url, lst):
        assert type(lst) <> types.StringType
        return make_list(map(lambda x: self.make_binary_package_ref(url, x),
                             lst))

    def make_red(self, contents):
        return SPAN(contents, _class="red")

    def make_purple(self, contents):
        return SPAN(contents, _class="purple")
                    
    def make_dangerous(self, contents):
        return SPAN(contents, _class="dangerous")

    def pre_dispatch(self):
        self.db.refresh()

TrackerService(socket_name, db_name).run()