~apport-hackers/apport/main

"""Crash database implementation for Launchpad."""

# Copyright (C) 2007 - 2009 Canonical Ltd.

# Authors: Martin Pitt <martin.pitt@ubuntu.com>

#          Markus Korn <thekorn@gmx.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.  See http://www.gnu.org/copyleft/gpl.html for

# the full text of the license.

import atexit

import email

import gzip

import http.client

import io

import os.path

import re

import shutil

import sys

import tempfile

import time

import urllib.parse

import urllib.request

from httplib2 import FailedToDecompressContent

try:

    from launchpadlib.errors import HTTPError, RestfulError

    from launchpadlib.launchpad import Launchpad

except ImportError:

    # if launchpadlib is not available, only client-side reporting will work

    Launchpad = None

import apport

import apport.crashdb

default_credentials_path = os.path.expanduser(

    "~/.cache/apport/launchpad.credentials"

)

def filter_filename(attachments):

    for attachment in attachments:

        try:

            f = attachment.data.open()

        except (HTTPError, FailedToDecompressContent):

            apport.error("Broken attachment on bug, ignoring")

            continue

        name = f.filename

        if name.endswith(".txt") or name.endswith(".gz"):

            yield f

def id_set(tasks):

    # same as set(int(i.bug.id) for i in tasks) but faster

    return set(int(i.self_link.split("/").pop()) for i in tasks)

class CrashDatabase(apport.crashdb.CrashDatabase):

    """Launchpad implementation of crash database interface."""

    def __init__(self, auth, options):

        """Initialize Launchpad crash database.

        You need to specify a launchpadlib-style credentials file to

        access launchpad. If you supply None, it will use

        default_credentials_path (~/.cache/apport/launchpad.credentials).

        Recognized options are:

        - distro: Name of the distribution in Launchpad

        - project: Name of the project in Launchpad

          (Note that exactly one of "distro" or "project" must be given.)

        - launchpad_instance: If set, this uses the given launchpad instance

          instead of production (optional). This can be overriden or set by

          $APPORT_LAUNCHPAD_INSTANCE environment. For example: "qastaging" or

          "staging".

        - cache_dir: Path to a permanent cache directory; by default it uses a

          temporary one. (optional). This can be overridden or set by

          $APPORT_LAUNCHPAD_CACHE environment.

        - escalation_subscription: This subscribes the given person or team to

          a bug once it gets the 10th duplicate.

        - escalation_tag: This adds the given tag to a bug once it gets more

          than 10 duplicates.

        - initial_subscriber: The Launchpad user which gets subscribed to newly

          filed bugs (default: "apport"). It should be a bot user which the

          crash-digger instance runs as, as this will get to see all bug

          details immediately.

        - triaging_team: The Launchpad user/team which gets subscribed after

          updating a crash report bug by the retracer (default:

          "ubuntu-crashes-universe")

        - architecture: If set, this sets and watches out for needs-*-retrace

          tags of this architecture. This is useful when being used with

          apport-retrace and crash-digger to process crash reports of foreign

          architectures. Defaults to system architecture.

        """

        if os.getenv("APPORT_LAUNCHPAD_INSTANCE"):

            options["launchpad_instance"] = os.getenv(

                "APPORT_LAUNCHPAD_INSTANCE"

            )

        if not auth:

            lp_instance = options.get("launchpad_instance")

            if lp_instance:

                auth = (

                    default_credentials_path

                    + "."

                    + lp_instance.split("://", 1)[-1]

                )

            else:

                auth = default_credentials_path

        apport.crashdb.CrashDatabase.__init__(self, auth, options)

        self.distro = options.get("distro")

        if self.distro:

            assert (

                "project" not in options

            ), 'Must not set both "project" and "distro" option'

        else:

            assert (

                "project" in options

            ), 'Need to have either "project" or "distro" option'

        if "architecture" in options:

            self.arch_tag = "need-%s-retrace" % options["architecture"]

        else:

            self.arch_tag = (

                "need-%s-retrace" % apport.packaging.get_system_architecture()

            )

        self.options = options

        self.auth = auth

        assert self.auth

        self.__launchpad = None

        self.__lp_distro = None

        self.__lpcache = os.getenv(

            "APPORT_LAUNCHPAD_CACHE", options.get("cache_dir")

        )

        if not self.__lpcache:

            # use a temporary dir

            self.__lpcache = tempfile.mkdtemp(prefix="launchpadlib.cache.")

            atexit.register(shutil.rmtree, self.__lpcache, ignore_errors=True)

    @property

    def launchpad(self):

        """Return Launchpad instance."""

        if self.__launchpad:

            return self.__launchpad

        if Launchpad is None:

            sys.stderr.write(

                "ERROR: The launchpadlib Python %s module is not installed."

                " Please install the python3-launchpadlib package!\n"

                % sys.version[0]

            )

            sys.exit(1)

        if self.options.get("launchpad_instance"):

            launchpad_instance = self.options.get("launchpad_instance")

        else:

            launchpad_instance = "production"

        auth_dir = os.path.dirname(self.auth)

        if auth_dir and not os.path.isdir(auth_dir):

            os.makedirs(auth_dir)

        try:

            self.__launchpad = Launchpad.login_with(

                "apport-collect",

                launchpad_instance,

                launchpadlib_dir=self.__lpcache,

                allow_access_levels=["WRITE_PRIVATE"],

                credentials_file=self.auth,

                version="1.0",

            )

        except (RestfulError, OSError, ValueError) as error:

            apport.error(

                "connecting to Launchpad failed: %s\n"

                'You can reset the credentials by removing the file "%s"',

                getattr(error, "content", str(error)),

                self.auth,

            )

            sys.exit(99)  # transient error

        return self.__launchpad

    def _get_distro_tasks(self, tasks):

        if not self.distro:

            return

        for t in tasks:

            if t.bug_target_name.lower() == self.distro or re.match(

                r"^.+\(%s.*\)$" % self.distro, t.bug_target_name.lower()

            ):

                yield t

    @property

    def lp_distro(self):

        if self.__lp_distro is None:

            if self.distro:

                self.__lp_distro = self.launchpad.distributions[self.distro]

            elif "project" in self.options:

                self.__lp_distro = self.launchpad.projects[

                    self.options["project"]

                ]

            else:

                raise SystemError(

                    "distro or project needs to be specified"

                    " in crashdb options"

                )

        return self.__lp_distro

    def upload(self, report, progress_callback=None):

        """Upload given problem report return a handle for it.

        This should happen noninteractively.

        If the implementation supports it, and a function progress_callback is

        passed, that is called repeatedly with two arguments: the number of

        bytes already sent, and the total number of bytes to send. This can be

        used to provide a proper upload progress indication on frontends.

        """

        assert self.accepts(report)

        blob_file = self._generate_upload_blob(report)

        ticket = upload_blob(

            blob_file, progress_callback, hostname=self.get_hostname()

        )

        blob_file.close()

        assert ticket

        return ticket

    def get_hostname(self):

        """Return the hostname for the Launchpad instance."""

        launchpad_instance = self.options.get("launchpad_instance")

        if launchpad_instance:

            if launchpad_instance == "qastaging":

                hostname = "qastaging.launchpad.net"

            elif launchpad_instance == "staging":

                hostname = "staging.launchpad.net"

            else:

                hostname = "launchpad.dev"

        else:

            hostname = "launchpad.net"

        return hostname

    def get_comment_url(self, report, handle):

        """Return an URL that should be opened after report has been uploaded

        and upload() returned handle.

        Should return None if no URL should be opened (anonymous filing without

        user comments); in that case this function should do whichever

        interactive steps it wants to perform."""

        args = {}

        title = report.get("Title", report.standard_title())

        if title:

            args["field.title"] = title

        hostname = self.get_hostname()

        if "SnapSource" in report:

            project = report["SnapSource"]

        else:

            project = self.options.get("project")

        if not project:

            if "SourcePackage" in report:

                return "https://bugs.%s/%s/+source/%s/+filebug/%s?%s" % (

                    hostname,

                    self.distro,

                    report["SourcePackage"],

                    handle,

                    urllib.parse.urlencode(args),

                )

            else:

                return "https://bugs.%s/%s/+filebug/%s?%s" % (

                    hostname,

                    self.distro,

                    handle,

                    urllib.parse.urlencode(args),

                )

        else:

            return "https://bugs.%s/%s/+filebug/%s?%s" % (

                hostname,

                project,

                handle,

                urllib.parse.urlencode(args),

            )

    def get_id_url(self, report, crash_id):

        """Return URL for a given report ID.

        The report is passed in case building the URL needs additional

        information from it, such as the SourcePackage name.

        Return None if URL is not available or cannot be determined.

        """

        return "https://bugs.launchpad.net/bugs/" + str(crash_id)

    def download(self, crash_id):

        """Download the problem report from given ID and return a Report."""

        report = apport.Report()

        b = self.launchpad.bugs[crash_id]

        # parse out fields from summary

        m = re.search(r"(ProblemType:.*)$", b.description, re.S)

        if not m:

            m = re.search(r"^--- \r?$[\r\n]*(.*)", b.description, re.M | re.S)

        assert m, "bug description must contain standard apport format data"

        description = (

            m.group(1)

            .encode("UTF-8")

            .replace(b"\xc2\xa0", b" ")

            .replace(b"\r\n", b"\n")

        )

        if b"\n\n" in description:

            # this often happens, remove all empty lines between top and

            # 'Uname'

            if b"Uname:" in description:

                # this will take care of bugs like LP #315728 where stuff

                # is added after the apport data

                (part1, part2) = description.split(b"Uname:", 1)

                description = (

                    part1.replace(b"\n\n", b"\n")

                    + b"Uname:"

                    + part2.split(b"\n\n", 1)[0]

                )

            else:

                # just parse out the Apport block; e. g. LP #269539

                description = description.split(b"\n\n", 1)[0]

        report.load(io.BytesIO(description))

        if "Date" not in report:

            # We had not submitted this field for a while, claiming it

            # redundant. But it is indeed required for up-to-the-minute

            # comparison with log files, etc. For backwards compatibility with

            # those reported bugs, read the creation date

            try:

                report["Date"] = b.date_created.ctime()

            except AttributeError:

                # support older wadllib API which returned strings

                report["Date"] = b.date_created

        if "ProblemType" not in report:

            if "apport-bug" in b.tags:

                report["ProblemType"] = "Bug"

            elif "apport-crash" in b.tags:

                report["ProblemType"] = "Crash"

            elif "apport-kernelcrash" in b.tags:

                report["ProblemType"] = "KernelCrash"

            elif "apport-package" in b.tags:

                report["ProblemType"] = "Package"

            else:

                raise ValueError(

                    "cannot determine ProblemType from tags: " + str(b.tags)

                )

        report["Tags"] = " ".join(b.tags)

        if "Title" in report:

            report["OriginalTitle"] = report["Title"]

        report["Title"] = b.title

        for attachment in filter_filename(b.attachments):

            key, ext = os.path.splitext(attachment.filename)

            # ignore attachments with invalid keys

            try:

                report[key] = ""

            except (AssertionError, TypeError, ValueError):

                continue

            if ext == ".txt":

                report[key] = attachment.read()

                try:

                    report[key] = report[key].decode("UTF-8")

                except UnicodeDecodeError:

                    pass

            elif ext == ".gz":

                try:

                    report[key] = gzip.GzipFile(fileobj=attachment).read()

                except OSError as error:

                    # some attachments are only called .gz, but are

                    # uncompressed (LP #574360)

                    if "Not a gzip" not in str(error):

                        raise

                    attachment.seek(0)

                    report[key] = attachment.read()

            else:

                raise Exception(

                    "Unknown attachment type: " + attachment.filename

                )

        return report

    def update(

        self,

        crash_id,

        report,

        comment,

        change_description=False,

        attachment_comment=None,

        key_filter=None,

    ):

        """Update the given report ID with all data from report.

        This creates a text comment with the "short" data (see

        ProblemReport.write_mime()), and creates attachments for all the

        bulk/binary data.

        If change_description is True, and the crash db implementation supports

        it, the short data will be put into the description instead (like in a

        new bug).

        comment will be added to the "short" data. If attachment_comment is

        given, it will be added to the attachment uploads.

        If key_filter is a list or set, then only those keys will be added.

        """

        bug = self.launchpad.bugs[crash_id]

        # TODO: raise an error if key_filter is not a list or set

        if key_filter:

            skip_keys = set(report.keys()) - set(key_filter)

        else:

            skip_keys = None

        # we want to reuse the knowledge of write_mime() with all its

        # different input types and output formatting; however, we have to

        # dissect the mime ourselves, since we can't just upload it as a blob

        with tempfile.TemporaryFile() as mime:

            report.write_mime(mime, skip_keys=skip_keys)

            mime.flush()

            mime.seek(0)

            msg = email.message_from_binary_file(mime)

            msg_iter = msg.walk()

            # first part is the multipart container

            part = next(msg_iter)

            assert part.is_multipart()

            # second part should be an inline text/plain attachments with

            # all short fields

            part = next(msg_iter)

            assert not part.is_multipart()

            assert part.get_content_type() == "text/plain"

            if not key_filter:

                # when we update a complete report, we are updating

                # an existing bug with apport-collect

                x = bug.tags[:]  # LP#254901 workaround

                x.append("apport-collected")

                # add any tags (like the release) to the bug

                if "Tags" in report:

                    x += self._filter_tag_names(report["Tags"]).split()

                bug.tags = x

                bug.lp_save()

                # fresh bug object, LP#336866 workaround

                bug = self.launchpad.bugs[crash_id]

            # short text data

            text = part.get_payload(decode=True).decode("UTF-8", "replace")

            # text can be empty if you are only adding an attachment to a bug

            if text:

                if change_description:

                    bug.description = bug.description + "\n--- \n" + text

                    bug.lp_save()

                else:

                    if not comment:

                        comment = bug.title

                    bug.newMessage(content=text, subject=comment)

            # other parts are the attachments:

            for part in msg_iter:

                bug.addAttachment(

                    comment=attachment_comment or "",

                    description=part.get_filename(),

                    content_type=None,

                    data=part.get_payload(decode=True),

                    filename=part.get_filename(),

                    is_patch=False,

                )

    def update_traces(self, crash_id, report, comment=""):

        """Update the given report ID for retracing results.

        This updates Stacktrace, ThreadStacktrace, StacktraceTop,

        and StacktraceSource. You can also supply an additional comment.

        """

        apport.crashdb.CrashDatabase.update_traces(

            self, crash_id, report, comment

        )

        bug = self.launchpad.bugs[crash_id]

        # ensure it's assigned to a package

        if "SourcePackage" in report:

            for task in bug.bug_tasks:

                if task.target.resource_type_link.endswith("#distribution"):

                    task.target = self.lp_distro.getSourcePackage(

                        name=report["SourcePackage"]

                    )

                    task.lp_save()

                    bug = self.launchpad.bugs[crash_id]

                    break

        # remove core dump if stack trace is usable

        if report.has_useful_stacktrace():

            for a in bug.attachments:

                if a.title == "CoreDump.gz":

                    try:

                        a.removeFromBug()

                    except HTTPError:

                        pass  # LP#249950 workaround

            try:

                task = self._get_distro_tasks(bug.bug_tasks)

                task = next(task)

                if task.importance == "Undecided":

                    task.importance = "Medium"

                    task.lp_save()

            except StopIteration:

                pass  # no distro tasks

            # update bug title with retraced function name

            fn = report.stacktrace_top_function()

            if fn:

                m = re.match(

                    r"^(.*crashed with SIG.* in )([^( ]+)(\(\).*$)", bug.title

                )

                if m and m.group(2) != fn:

                    bug.title = m.group(1) + fn + m.group(3)

                    try:

                        bug.lp_save()

                    except HTTPError:

                        pass  # LP#336866 workaround

                    bug = self.launchpad.bugs[crash_id]

        self._subscribe_triaging_team(bug, report)

    def get_distro_release(self, crash_id):

        """Get 'DistroRelease: <release>' from the given report ID and return

        it."""

        bug = self.launchpad.bugs[crash_id]

        m = re.search("DistroRelease: ([-a-zA-Z0-9.+/ ]+)", bug.description)

        if m:

            return m.group(1)

        raise ValueError("URL does not contain DistroRelease: field")

    def get_affected_packages(self, crash_id):

        """Return list of affected source packages for given ID."""

        bug_target_re = re.compile(

            r"/%s/(?:(?P<suite>[^/]+)/)?\+source/(?P<source>[^/]+)$"

            % self.distro

        )

        bug = self.launchpad.bugs[crash_id]

        result = []

        for task in bug.bug_tasks:

            match = bug_target_re.search(task.target.self_link)

            if not match:

                continue

            if task.status in ("Invalid", "Won't Fix", "Fix Released"):

                continue

            result.append(match.group("source"))

        return result

    def is_reporter(self, crash_id):

        """Check whether the user is the reporter of given ID."""

        bug = self.launchpad.bugs[crash_id]

        return bug.owner.name == self.launchpad.me.name

    def can_update(self, crash_id):

        """Check whether the user is eligible to update a report.

        A user should add additional information to an existing ID if (s)he is

        the reporter or subscribed, the bug is open, not a duplicate, etc. The

        exact policy and checks should be done according to the particular

        implementation.

        """

        bug = self.launchpad.bugs[crash_id]

        if bug.duplicate_of:

            return False

        if bug.owner.name == self.launchpad.me.name:

            return True

        # check subscription

        me = self.launchpad.me.self_link

        for sub in bug.subscriptions.entries:

            if sub["person_link"] == me:

                return True

        return False

    def get_unretraced(self):

        """Return an ID set of all crashes which have not been retraced yet and

        which happened on the current host architecture."""

        try:

            bugs = self.lp_distro.searchTasks(

                tags=self.arch_tag, created_since="2011-08-01"

            )

            return id_set(bugs)

        except HTTPError as error:

            apport.error("connecting to Launchpad failed: %s", str(error))

            sys.exit(99)  # transient error

    def get_dup_unchecked(self):

        """Return an ID set of all crashes which have not been checked for

        being a duplicate.

        This is mainly useful for crashes of scripting languages such as

        Python, since they do not need to be retraced. It should not return

        bugs that are covered by get_unretraced()."""

        try:

            bugs = self.lp_distro.searchTasks(

                tags="need-duplicate-check", created_since="2011-08-01"

            )

            return id_set(bugs)

        except HTTPError as error:

            apport.error("connecting to Launchpad failed: %s", str(error))

            sys.exit(99)  # transient error

    def get_unfixed(self):

        """Return an ID set of all crashes which are not yet fixed.

        The list must not contain bugs which were rejected or duplicate.

        This function should make sure that the returned list is correct. If

        there are any errors with connecting to the crash database, it should

        raise an exception (preferably OSError)."""

        bugs = self.lp_distro.searchTasks(tags="apport-crash")