path: root/lib/python/sectracker/parsers.py

# sectracker.parsers -- various text file parsers
# Copyright (C) 2010 Florian Weimer <fw@deneb.enyo.de>
# 
# 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

import re

import debian_support
import sectracker.regexpcase as _regexpcase
from sectracker.xcollections import namedtuple as _namedtuple
import sectracker.xpickle as _xpickle
import sectracker.diagnostics

FORMAT = "4"

def _sortedtuple(seq):
    l = list(seq)
    l.sort()
    return tuple(l)

@_xpickle.loader("BINARY" + FORMAT)
def binarypackages(name, f):
    """Returns a sequence of binary package names"""
    obj = set(v for p in debian_support.PackageFile(name, f)
            for k, v in p if k == "Package")
    obj = list(obj)
    obj.sort()
    return tuple(obj)

SourcePackage = _namedtuple("SourcePackage", "name version binary")

@_xpickle.loader("SOURCE" + FORMAT)
def sourcepackages(name, f):
    """Returns a dictionary of source package objects"""
    data = {}
    for p in debian_support.PackageFile(name, f):
        pkg_name = pkg_version = pkg_binary = None
        for name, contents in p:
            if name == "Package":
                pkg_name = intern(contents)
            elif name == "Version":
                pkg_version = contents
            elif name == "Binary":
                pkg_binary = _sortedtuple(contents.replace(",", " ")
                                          .strip().split())
        if pkg_name is None:
            raise SyntaxError("package record does not contain package name")
        if pkg_version is None:
            raise SyntaxError("package record for %s does not contain version"
                              % pkg_name)
        if pkg_binary is None:
            raise SyntaxError("package record lacks Binary field")

        if pkg_name in data:
            oversion = debian_support.Version(data[pkg_name].version)
            if oversion >= debian_support.Version(pkg_version):
                continue
        data[pkg_name] = SourcePackage(pkg_name, pkg_version, pkg_binary)
    return data

FlagAnnotation = _namedtuple("FlagAnnotation", "line type")
StringAnnotation = _namedtuple("StringAnnotation",
                                           "line type description")
XrefAnnotation = _namedtuple("XrefAnnotation", "line type bugs")
PackageAnnotation = _namedtuple(
    "PackageAnnotation",
    "line type release package kind version description "
    + "urgency debian_bugs bug_filed")

def _annotationdispatcher():
    # Parser for inner annotations, like (bug #1345; low)
    urgencies=set("unimportant low medium high".split())
    @_regexpcase.rule('(bug filed|%s)' % '|'.join(urgencies))
    def innerflag(groups, diag, flags, bugs):
        f = groups[0]
        if f in flags:
            diag.error("duplicate flag: " + repr(f))
        else:
            flags.add(f)
    @_regexpcase.rule(r'bug #(\d+)')
    def innerbug(groups, diag, flags, bugs):
        no = int(groups[0])
        if no in bugs:
            diag.error("duplicate bug number: " + groups[0])
        else:
            bugs.add(no)
    def innerdefault(text, diag, flags, bugs):
        diag.error("invalid inner annotation: " + repr(text))
    innerdispatch = _regexpcase.RegexpCase((innerflag, innerbug),
                                           default=innerdefault)

    def parseinner(diag, inner):
        if not inner:
            return (None, (), False)
        flags = set()
        bugs = set()
        for innerann in inner.split(";"):
            innerdispatch(innerann.strip(), diag, flags, bugs)

        urgency = urgencies.intersection(flags)
        if urgency:
            if len(urgency) > 1:
                diag.error("multiple urgencies: " + ", ".join(urgency))
            else:
                urgency = urgency.pop()
        else:
            urgency = None

        bug_filed = "bug filed" in flags 
        if bugs and bug_filed:
            diag.error("'bug filed' and bug numbers listed")
            bug_filed = False

        return (urgency, _sortedtuple(bugs), bug_filed)

    # Parsers for indented annotations (NOT-FOR-US:, " - foo <unfixed>" etc.)

    @_regexpcase.rule(r'(?:\[([a-z]+)\]\s)?-\s([A-Za-z0-9:.+-]+)\s*'
                      + r'(?:\s([A-Za-z0-9:.+~-]+)\s*)?(?:\s\((.*)\))?')
    def package_version(groups, diag):
        release, package, version, inner = groups
        inner = parseinner(diag, inner)
        if version is None:
            kind = "unfixed"
        else:
            kind = "fixed"
        return PackageAnnotation(
            *((diag.line(), "package", release, package, kind,
               version, None) + inner))

    pseudo_freetext = "no-dsa not-affected end-of-life ignored postponed".split()
    pseudo_struct = set("unfixed removed itp undetermined".split())
    @_regexpcase.rule(r'(?:\[([a-z]+)\]\s)?-\s([A-Za-z0-9:.+-]+)'
                      + r'\s+<([a-z-]+)>\s*(?:\s\((.*)\))?')
    def package_pseudo(groups, diag):
        release, package, version, inner = groups
        if version in pseudo_freetext:
            return PackageAnnotation(
                diag.line(), "package", release, package, version,
                None, inner, None, (), False)
        elif version in pseudo_struct:
            inner = parseinner(diag, inner)
            if version == "itp" and not inner[1]:
                diag.error("<itp> needs Debian bug reference")
            return PackageAnnotation(
                *((diag.line(), "package", release, package, version,
                   None, None) + inner))
        else:
            diag.error("invalid pseudo-version: " + repr(version))
            return None

    @_regexpcase.rule(r'\{(.*)\}')
    def xref(groups, diag):
        x = _sortedtuple(groups[0].strip().split())
        if x:
            return XrefAnnotation(diag.line(), "xref", x)
        else:
            diag.error("empty cross-reference")
            return None
        
    return _regexpcase.RegexpCase(
        ((r'(RESERVED|REJECTED)',
          lambda groups, diag: FlagAnnotation(diag.line(), groups[0])),
         (r'(NOT-FOR-US|NOTE|TODO):\s+(\S.*)',
          lambda groups, diag: StringAnnotation(diag.line(), *groups)),
         package_version, package_pseudo, xref),
        prefix=r"\s+", suffix=r"\s*",
        default=lambda text, diag: diag.error("invalid annotation"))
_annotationdispatcher = _annotationdispatcher()

List = _namedtuple("List", "list messages")
Bug = _namedtuple("Bug", "file header annotations")
Header = _namedtuple("Header", "line name description")

def _parselist(path, f, parseheader, finish):
    lineno = 0
    headerlineno = None
    bugs = []
    diag = sectracker.diagnostics.Diagnostics()
    header = None
    anns = []
    anns_types = set()
    relpkg = set()

    for line in f.readlines():
        lineno += 1
        diag.setlocation(path, lineno)

        if line[:1] in " \t":
            if header is None:
                diag.error("header expected")
                continue
            ann = _annotationdispatcher(line, diag)
            if ann is not None:
                # Per-annotation checks (spanning multiple annotations)
                anns_types.add(ann.type)
                if ann.type == "package":
                    rp = (ann.release, ann.package)
                    if rp in relpkg:
                        diag.error("duplicate package annotation")
                        ann = None
                    else:
                        relpkg.add(rp)
            if ann is not None:
                anns.append(ann)
        else:
            if header is not None:
                # Per-bug global checks
                if "NOT-FOR-US" in anns_types and "package" in anns_types:
                    diag.error("NOT-FOR-US conflicts with package annotations",
                               line=headerlineno)
                if "REJECTED" in anns_types and "package" in anns_types:
                    diag.warning("REJECTED bug has package annotations",
                                 line=headerlineno)
                bugs.append(finish(header, headerlineno, anns, diag))
                del anns[:]
                anns_types = set()
                relpkg = set()
            headerlineno = lineno
        
            header = parseheader(line)
            if header is None:
                diag.error("malformed header")
                continue

    if header is not None:
        bugs.append(finish(header, headerlineno, anns, diag))
    return List(tuple(bugs), diag.messages())

@_xpickle.loader("CVE" + FORMAT)
def cvelist(path, f):
    re_header = re.compile(r'^(CVE-\d{4}-(?:\d{4,}|XXXX))\s+(.*?)\s*$')
    def parseheader(line):
        match = re_header.match(line)
        if match is None:
            return None
        name, desc = match.groups()
        if desc:
            if desc[0] == '(':
                if desc[-1] != ')':
                    diag.error("error", "missing ')'")
                else:
                    desc = desc[1:-1]
            elif desc[0] == '[':
                if desc[-1] != ']':
                    diag.error("missing ']'")
                else:
                    desc = desc[1:-1]
        return (name, desc)
    def cveuniquename(line, anns):
        bug = 0
        for ann in anns:
            if ann.type == "package" and ann.debian_bugs:
                bug = ann.debian_bugs[0]
                break
        return "TEMP-%07d-%06d" % (bug, line)
    def finish(header, headerlineno, anns, diag):
        name, desc = header
        if name[-1] == "X":
            name1 = cveuniquename(headerlineno, anns)
        else:
            name1 = name
        return Bug(path, Header(headerlineno, name1, desc), tuple(anns))
    return _parselist(path, f, parseheader, finish)

def _checkrelease(anns, diag, kind):
    for ann in anns:
        if ann.type == "package" and ann.release is None:
            diag.error("release annotation required in %s file" % kind,
                       line=ann.line)

@_xpickle.loader("DSA" + FORMAT)
def dsalist(path, f):
    re_header = re.compile(r'^\[(\d\d) ([A-Z][a-z][a-z]) (\d{4})\] '
                            + r'(DSA-\d+(?:-\d+)?)\s+'
                            + r'(.*?)\s*$')
    def parseheader(line):
        match = re_header.match(line)
        if match is None:
            return None
        return match.groups()
    def finish(header, headerlineno, anns, diag):
        d, m, y, name, desc = header
        _checkrelease(anns, diag, "DSA")
        return Bug(path, Header(headerlineno, name, None), tuple(anns))
    return _parselist(path, f, parseheader, finish)

@_xpickle.loader("DTSA" + FORMAT)
def dtsalist(path, f):
    re_header = re.compile(
        r'^\[([A-Z][a-z]{2,}) (\d\d?)(?:st|nd|rd|th), (\d{4})\] '
        + r'(DTSA-\d+-\d+)\s+'
        + r'(.*?)\s*$')
    def parseheader(line):
        match = re_header.match(line)
        if match is None:
            return None
        return match.groups()
    def finish(header, headerlineno, anns, diag):
        d, m, y, name, desc = header
        _checkrelease(anns, diag, "DTSA")
        return Bug(path, Header(headerlineno, name, None), tuple(anns))
    return _parselist(path, f, parseheader, finish)

@_xpickle.loader("DLA" + FORMAT)
def dlalist(path, f):
    re_header = re.compile(r'^\[(\d\d) ([A-Z][a-z][a-z]) (\d{4})\] '
                            + r'(DLA-\d+(?:-\d+)?)\s+'
                            + r'(.*?)\s*$')
    def parseheader(line):
        match = re_header.match(line)
        if match is None:
            return None
        return match.groups()
    def finish(header, headerlineno, anns, diag):
        d, m, y, name, desc = header
        _checkrelease(anns, diag, "DLA")
        return Bug(path, Header(headerlineno, name, None), tuple(anns))
    return _parselist(path, f, parseheader, finish)