| # | |
| # 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)) |