tools/dist/security/parser.py - trunk - niceSOFT/subversion

tools/dist/security/parser.py
287 строк8 KB
Daniel Shahaf
* tools/dist/security/parser.py
24 сен 2018, 17:22
24 сен 2018, 17:2226c46b6
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

"""
Parser for CVE/CAN advisories and patches.
"""

from __future__ import absolute_import


import os
import re
import ast
import base64
import quopri


class Notification(object):
    """
    The complete security notification, containing multiple advisories.
    """

    class Metadata(object):
        """
        The metadata for one advisory, with the following properties:
            tracking_id - the CVE/CAN number
            title       - a short description of the issue
            culprit     - server, client or both
            advisory    - a Text object with the text of the advisory
            patches     - a list of Patch objects, sorted in descending
                          order by the base version
        """

        CULPRIT_SERVER = 'server'
        CULPRIT_CLIENT = 'client'

        # For compatibility, 'client' and 'server' may be specified either with
        # or without a tuple.
        __CULPRITS = (
            CULPRIT_SERVER,
            CULPRIT_CLIENT,
            (CULPRIT_SERVER,)
            (CULPRIT_CLIENT,)
            (CULPRIT_SERVER, CULPRIT_CLIENT),
            (CULPRIT_CLIENT, CULPRIT_SERVER),
        )

        def __init__(self, basedir, tracking_id,
                     title, culprit, advisory, patches):
            if culprit not in self.__CULPRITS:
                raise ValueError('Culprit should be one of: '
                                 + ', '.join(repr(x) for x in self.__CULPRITS))
            if not isinstance(culprit, tuple):
                culprit = (culprit,)

            self.__tracking_id = tracking_id
            self.__title = title
            self.__culprit = frozenset(culprit)
            self.__advisory = Text(os.path.join(basedir, advisory))
            self.__patches = []
            for base_version, patchfile in patches.items():
                patch = Patch(base_version, os.path.join(basedir, patchfile))
                self.__patches.append(patch)
            self.__patches.sort(reverse=True, key=lambda x: x.base_version_key)

        @property
        def tracking_id(self):
            return self.__tracking_id

        @property
        def title(self):
            return self.__title

        @property
        def culprit(self):
            return self.__culprit

        @property
        def advisory(self):
            return self.__advisory

        @property
        def patches(self):
            return tuple(self.__patches)


    def __init__(self, rootdir, *tracking_ids):
        """
        Create the security notification for all TRACKING_IDS.
        The advisories and patches for each tracking ID must be
        in the appropreiately named subdirectory of ROOTDIR.

        The notification text assumes that RELEASE_VERSIONS will
        be published on RELEASE_DATE and that the tarballs are
        available in DIST_REVISION of the dist repository.
        """

        assert(len(tracking_ids) > 0)
        self.__advisories = []
        for tid in tracking_ids:
            self.__advisories.append(self.__parse_advisory(rootdir, tid))

    def __iter__(self):
        return self.__advisories.__iter__()

    def __len__(self):
        return len(self.__advisories)

    def __parse_advisory(self, rootdir, tracking_id):
        """
        Parse a single advisory named TRACKING_ID in ROOTDIR.
        """

        basedir = os.path.join(rootdir, tracking_id)
        with open(os.path.join(basedir, 'metadata'), 'rt') as md:
            metadata = ast.literal_eval(md.read())

        return self.Metadata(basedir, tracking_id,
                             metadata['title'],
                             metadata['culprit'],
                             metadata['advisory'],
                             metadata['patches'])

    def base_version_keys(self):
        """
        Return the set of base-version keys of all the patches.
        """

        base_version_keys = set()
        for metadata in self:
            for patch in metadata.patches:
                base_version_keys.add(patch.base_version_key)
        return base_version_keys


class __Part(object):
    def __init__(self, path, text=None):
        """
        Create a text object with contents from the file at PATH.
        If self.TEXTMODE is True, strip whitespace from the end of
        all lines and strip empty lines from the end of the file.

        Alternatively, if PATH is None, set the contents to TEXT,
        which must be convertible to bytes.
        """

        assert (path is None) is not (text is None)
        if path:
            self.__text = self.__load_file(path)
        else:
            self.__text = bytes(text)

    def __load_file(self, path):
        with open(path, 'rb') as src:
            if not self.TEXTMODE:
                return src.read()

            text = []
            for line in src:
                text.append(line.rstrip() + b'\n')

            # Strip trailing empty lines in text mode
            while len(text) and not text[-1]:
                del text[-1]
            return b''.join(text)

    @property
    def text(self):
        """
        Return the raw contents.
        """

        return self.__text.decode('UTF-8')

    @property
    def quoted_printable(self):
        """
        Return contents encoded as quoted-printable.
        """

        return quopri.encodestring(self.__text).decode('ascii')

    BASE64_LINE_LENGTH = 64

    @property
    def base64(self):
        """
        Return multi-line Base64-encoded contents with the lenght
        of the lines limited to BASE64_LINE_LENGTH.
        """

        text = []
        data = base64.standard_b64encode(self.__text)
        start = 0
        end = self.BASE64_LINE_LENGTH
        while end < len(data):
            text.append(data[start:end] + b'\n')
            start += self.BASE64_LINE_LENGTH
            end += self.BASE64_LINE_LENGTH
        if start < len(data):
            text.append(data[start:] + b'\n')
        return b''.join(text).decode('ascii')


class Text(__Part):
    """
    In-memory container for the text of the advisory.
    """

    TEXTMODE = True


class Patch(__Part):
    """
    In-memory container for patches.
    """

    TEXTMODE = False

    def __init__(self, base_version, path):
        super(Patch, self).__init__(path)
        self.__base_version = base_version
        self.__base_version_key = self.split_version(base_version)

    @property
    def base_version(self):
        return self.__base_version

    @property
    def base_version_key(self):
        return self.__base_version_key

    @property
    def quoted_printable(self):
        raise NotImplementedError('Quoted-printable patches? Really?')


    __SPLIT_VERSION_RX = re.compile(r'^(\d+)(?:\.(\d+))?(?:\.(\d+))?(.+)?$')

    @classmethod
    def split_version(cls, version):
        """
        Splits a version number in the form n.n.n-tag into a tuple
        of its components.
        """
        def splitv(version):
            for s in cls.__SPLIT_VERSION_RX.match(version).groups():
                if s is None:
                    continue
                try:
                    n = int(s)
                except ValueError:
                    n = s
                yield n
        return tuple(splitv(version))

    @classmethod
    def join_version(cls, version_tuple):
        """
        Joins a version number tuple returned by Patch.split_version
        into a string.
        """

        def joinv(version_tuple):
            prev = None
            for n in version_tuple:
                if isinstance(n, int) and prev is not None:
                    yield '.'
                prev = n
                yield str(n)
        return ''.join(joinv(version_tuple))