~snappy-debug-developers/snappy-hub/snappy-debug

#!/usr/bin/env python3
#  Copyright (C) 2015-2019 Canonical Ltd.
#
#  This script is distributed under the terms and conditions of the GNU General
#  Public License, Version 3 or later. See http://www.gnu.org/copyleft/gpl.html
#  for details.

import codecs
import io
import optparse
import os
import re
import subprocess
import sys
import time
import LibAppArmor

DEBUGGING = False

sc_arch = dict()

#
# Helpers
#
os_release = "16"
os_is_core = False
restricted_interfaces = [
    "classic-support",
    "core-support",
    "docker-support",
    "greengrass-support",
    "kernel-module-control",
    "kubernetes-support",
    "lxd-support",
    "multipass-support",
    "snapd-control",
]
deprecated_interfaces = ["pulseaudio"]

# FIXME: detect this, don't hardcode these
sc_arg_socket = [
    ("account-control", "NETLINK_AUDIT"),
    ("audio-playback", "NETLINK_KOBJECT_UEVENT"),
    ("bluetooth-control", "AF_{ALG,BLUETOOTH}"),
    ("firewall-control", "NETLINK_{FIREWALL,IP6_FW,NETFILTER,NF_LOG,ROUTE}"),
    ("hardware-observe", "NETLINK_{GENERIC,KOBJECT_UEVENT}"),
    ("netlink-audit", "NETLINK_AUDIT"),
    ("netlink-connector", "NETLINK_CONNECTOR"),
    ("network", "AF_INET{,6}, AF_CONN, NETLINK_ROUTE"),
    ("network-bind", "AF_INET{,6}, NETLINK_ROUTE"),
    (
        "network-control",
        "AF_{APPLETALK,BRIDGE,INET,INET6,IPX,PACKET,PPPOX,SNA}, NETLINK_{DNRTMSG,FIB_LOOKUP,GENERIC,INET_DIAG,ISCSI,KOBJECT_UEVENT,RDMA,ROUTE,XFRM}",
    ),  # noqa
    (
        "network-observe",
        "SOCK_RAW, AF_INET{,6}), NETLINK_{GENERIC,INET_DIAG,KOBJECT_UEVENT,ROUTE}",
    ),  # noqa
    ("raw-usb", "NETLINK_KOBJECT_UEVENT"),
    ("time-control", "NETLINK_AUDIT"),
    ("unity7", "NETLINK_KOBJECT_UEVENT"),
    ("upower-observe", "NETLINK_KOBJECT_UEVENT"),
    ("x11", "NETLINK_KOBJECT_UEVENT"),
]

sc_arg_setns = [("network-control", "CLONE_NEWNET")]

sc_arg_quotactl = [("mount-observe", "Q_{GETQUOTA,GETINFO,GETFMT,XGETQUOTA,XGETQSTAT}")]


# Ubuntu Core only has releases of form '16', '18', etc
def detect_os_release():
    global os_release
    global os_is_core

    if "SNAPPY_DEBUG_OSTYPE" in os.environ:
        if os.environ["SNAPPY_DEBUG_OSTYPE"] == "core":
            os_is_core = True
        else:
            os_is_core = False
    elif os.path.isdir("/writable/lost+found"):
        # this is cheating for not having os-release connected. uc16 only
        os_is_core = True
    elif os.path.isfile("/var/lib/snapd/hostfs/usr/share/snappy/dpkg.list"):
        # this is cheating for not having os-release connected. Only core
        # devices should have this in hostfs since it isn't part of the deb and
        # only part of the image generation.
        os_is_core = True

    return

    # TODO: need to adjust for base snaps, etc
    osr = "/var/lib/snapd/hostfs/etc/os-release"
    if not os.path.isfile(osr):
        return

    try:
        fd = open_file_read(osr)
    except Exception:
        return

    for line in fd.readlines():
        line = line.strip()
        if re.search(r"VERSION_ID=[0-9]+(\.[0-9]+)?$", line):
            version = line.strip().split("=")[1].split(".")[0]
            try:
                int(version)
                os_release = version
            except Exception:
                pass
        if line == "ID=ubuntu-core":
            os_is_core = True

    return


def seccomp_architecture(audit_entry):
    global sc_arch

    if audit_entry not in sc_arch:
        if "SNAPPY_DEBUG_AUDIT_ARCH" in os.environ:
            exe = os.environ["SNAPPY_DEBUG_AUDIT_ARCH"]
        else:
            if "SNAP" not in os.environ:
                error("Could not find 'SNAP' in environment")
            exe = os.path.join(os.environ["SNAP"], "bin", "audit-arch")
            if not os.path.exists(exe):  # try the one in our build dir
                exe = os.path.join(os.environ["SNAP"], "src", "audit-arch")
        output = subprocess.check_output(
            [exe, audit_entry], universal_newlines=True
        ).splitlines()
        if len(output) != 2:
            sc_arch[audit_entry] = "unknown"
        else:
            sc_arch[audit_entry] = output[1].split()[1]

    return sc_arch[audit_entry]


# This is python3-specfic
def _print(s, output=sys.stdout):
    writer = output.buffer
    writer.write(bytes(s.encode("utf-8", "replace")))
    output.flush()


def debug(out):
    """Print debug message"""
    if DEBUGGING:
        try:
            _print("DEBUG: %s\n" % (out), sys.stderr)
        except IOError:
            pass


def info(out):
    """Print warning"""
    try:
        _print("INFO: %s\n" % (out), sys.stderr)
    except IOError:
        pass


def warn(out):
    """Print warning"""
    try:
        _print("WARN: %s\n" % (out), sys.stderr)
    except IOError:
        pass


def error(out, do_exit=True):
    """Print error and optionally exit"""
    try:
        _print("ERROR: %s\n" % (out), sys.stderr)
    except IOError:
        pass

    if do_exit:
        sys.exit(1)


def open_file_read(path):
    """Open specified file read-only"""
    try:
        orig = codecs.open(path, "r", "UTF-8", errors="replace")
    except Exception:
        raise

    return orig


def _deprecated(iface):
    if iface in deprecated_interfaces:
        return "%s (deprecated)" % iface
    return iface


def list_to_commas(l):
    return ", ".join([_deprecated(i) for i in l])


def list_to_newlines(l):
    return "\n".join([_deprecated(i) for i in l])


def _escape(s):
    """Python < 3.7 escaped various items that aren't regex special chars"""
    # for now, just bring back '/' for path names
    pat = re.compile(r"\\/")
    return pat.sub("/", re.escape(s))


def check_syslog():
    import syslog

    # LOG_USER | LOG_INFO is what dbus uses to log security denials on the
    # session bus, so seems like a reasonable default
    syslog.openlog(
        ident="snappy-debug",
        logoption=syslog.LOG_PID,
        facility=syslog.LOG_USER | syslog.LOG_INFO,
    )

    tm = time.time()
    check_syslog_msg = "%s: Starting scanlog with --follow" % tm
    syslog.syslog(check_syslog_msg)
    syslog.closelog()

    return (tm, check_syslog_msg)


def _aa_decode(s):
    try:
        decoded = bytearray.fromhex(s).decode()
    except ValueError:
        return s

    # abstract sockets may have trailing NULs
    return decoded.rstrip('\0')


class ScanLogs:
    def __init__(
        self,
        logs,
        snap_name=None,
        follow=False,
        recommend=False,
        display="both",
        from_end=False,
        exclude=None,
    ):
        self.rules = None
        self._get_rules()
        self.recommend = recommend
        self.display = display
        self.snap_name = snap_name
        self.exclude_re = re.compile(r"%s" % exclude)
        self.scan_log(logs, snap_name, follow, from_end)

    def _get_iface_dir(self):
        if "SNAP" not in os.environ:
            error("Could not find 'SNAP' in environment")

        global os_is_core
        top = "classic"
        if os_is_core:
            top = "core"

        dir = "%s/policy/%s/%s" % (os.environ["SNAP"], top, os_release)
        if not os.path.isdir(dir):
            error("Could not find '%s'" % dir)

        return dir

    def _get_rules(self):
        d = {}
        for rules_type in ["apparmor", "seccomp"]:
            iface_dir = os.path.join(self._get_iface_dir(), rules_type)
            interfaces = os.listdir(iface_dir)

            d[rules_type] = {}
            for iface in interfaces:
                d[rules_type][iface] = open_file_read(
                    os.path.join(iface_dir, iface)
                ).readlines()

        self.rules = d

    def _interface_exists(self, iface):
        for rules_type in ["apparmor", "seccomp"]:
            if os.path.exists(os.path.join(self._get_iface_dir(), rules_type, iface)):
                return True
        return False

    def _print_entry(self, entry):
        out = ""
        out += "= %s =\n" % entry["type"]
        out += "Time: %s\n" % entry["time"]
        out += "Log: %s\n" % entry["log"]
        if "msg" in entry:
            out += "%s\n" % entry["msg"]
        if self.recommend and len(entry["recommendation"]) > 0:
            out += "Suggestion"
            if len(entry["recommendation"]) > 1:
                out += "s"
            out += ":\n"
            for r in entry["recommendation"]:
                out += "* %s\n" % r

        _print("%s\n" % out)

    def scan_log(self, logs, snap_name, follow, from_end):
        apparmor_re = re.compile("type=1400 ")
        seccomp_re = re.compile("type=1326 ")

        apparmor_audit_re = re.compile(r"audit\[[0-9]+\]: AVC apparmor=")
        seccomp_audit_re = re.compile(r"audit\[[0-9]+\]: SECCOMP ")

        apparmor_dbus_re = re.compile('apparmor="(ALLOWED|DENIED)" operation="dbus')
        log_re = re.compile(
            r"^.*(type=[0-9]+ audit\([0-9]+\.[0-9]+:[0-9]+\):|audit\[[0-9]+\]: (AVC|SECCOMP)) "
        )
        dbus_log_re = re.compile(r'^.*apparmor="(ALLOWED|DENIED)"')
        last_log = ""

        # There are two types of lines:
        #   May 07 13:01:21 host audit[2]: SECCOMP auid=1000 ...
        #   May 07 13:01:21 host kernel: audit: type=1326 audit(1.2:3): auid=1000 ...
        # Currently, when 'follow'ing, we choose those with 'type=N' but
        # allow for scanning with a different regex to support, piping the
        # first form into stdin
        def _scan_line(line, snap_name, aa_re=apparmor_re, sc_re=seccomp_re):
            show = False
            rec = None

            if line.startswith("#"):
                return

            if self.exclude_re.search(line):
                return

            entry = dict()
            entry["raw"] = line.rstrip()
            entry["time"] = line[0:15]
            entry["log"] = log_re.sub("", line.rstrip())

            if self.display != "seccomp" and aa_re.search(line):
                (show, rec, entry["log"], msg) = self.make_apparmor_recommendation(line, snap_name, entry["log"])
                if show:
                    entry["type"] = "AppArmor"
                if msg is not None:
                    entry["msg"] = msg
            elif self.display != "seccomp" and apparmor_dbus_re.search(line):
                (show, rec, msg) = self.make_apparmor_dbus_recommendation(
                    line, snap_name
                )
                entry["log"] = dbus_log_re.sub('apparmor="\\1"', line.rstrip())
                if show:
                    entry["type"] = "AppArmor"
                if msg is not None:
                    entry["msg"] = msg
            elif self.display != "apparmor" and sc_re.search(line):
                (show, rec, entry["log"], msg) = self.make_seccomp_recommendation(
                    line, snap_name, entry["log"]
                )
                if show:
                    entry["type"] = "Seccomp"
                    entry["msg"] = msg

            entry["recommendation"] = rec

            # handle AVC vs audit rules that have the same violation
            nonlocal last_log
            if show:
                if entry["log"] == last_log:
                    return
                last_log = entry["log"]
                self._print_entry(entry)

        # First see if we can read a value in /proc to see if the interface is
        # connected
        try:
            open_file_read("/proc/sys/kernel/printk_ratelimit")
        except PermissionError:
            m = "Please run 'snap connect snappy-debug:log-observe' as root"
            raise ScanLogsException(m)

        if logs == "stdin":
            input = io.TextIOWrapper(
                sys.stdin.buffer, encoding="UTF-8", errors="replace"
            )
            try:
                while True:
                    line = input.readline()
                    if not line.endswith("\n"):  # eof
                        break
                    # type=... style
                    _scan_line(line, snap_name)
                    # audit[] THING style
                    _scan_line(
                        line, snap_name, aa_re=apparmor_audit_re, sc_re=seccomp_audit_re
                    )
            except KeyboardInterrupt:
                _print("\n")
        else:  # syslog
            try:
                log = open_file_read(logs)
            except PermissionError:
                m = "Could not read '%s'." % logs + " Check owner " + "and/or group."
                raise ScanLogsException(m)

            if follow:
                syslog_msg_found = False
                syslog_msg_re = None
                syslog_msg_time = None
                syslog_warned = False

                try:
                    do_scan = not from_end
                    while True:
                        if not syslog_msg_found:
                            if syslog_msg_time is None:
                                (syslog_msg_time, check_syslog_msg) = check_syslog()
                                syslog_msg_re = re.compile(check_syslog_msg)
                                time.sleep(2)

                        line = log.readline()

                        if not syslog_msg_found:
                            if not syslog_warned and time.time() - syslog_msg_time > 10:
                                warn("could not find log mark, is syslog " + "enabled?")
                                syslog_warned = True
                            elif syslog_msg_re is not None and syslog_msg_re.search(
                                line
                            ):
                                debug("syslog seems to work (found syslog " + "mark)")
                                syslog_msg_found = True

                        if line != "":

                            if do_scan:
                                _scan_line(line, snap_name)
                        else:
                            do_scan = True
                            time.sleep(0.5)
                except KeyboardInterrupt:
                    _print("\n")
            else:
                for line in log:
                    _scan_line(line, snap_name)

    def apparmor_ifaces_allowing_rule(self, rule_re):
        # This is where logprof would be handy
        matched_ifaces = []

        interfaces = self.rules["apparmor"].keys()

        for iface in interfaces:
            if iface in restricted_interfaces:
                continue  # don't suggest restricted
            for line in self.rules["apparmor"][iface]:
                # check if the apparmor rule matches our regex
                if rule_re.search(line) and iface not in matched_ifaces:
                    # yuck, but we really don't want to promote browser-support
                    if iface == "browser-support" and (
                        "owner @{PROC}/@{pid}/mounts r," in line
                        or "owner @{PROC}/@{pid}/mountinfo r," in line
                    ):
                        continue
                    matched_ifaces += [iface]

        if not matched_ifaces:
            return None

        return list_to_commas(sorted(matched_ifaces))

    def _aa_file(self, fn):
        """Convert fn to a list of potential matches"""
        res = []

        s = fn
        if s.startswith("/proc/"):
            proc_re = re.compile(r"^/proc/")
            s = proc_re.sub("@{PROC}/", s)

            proc_pid_re = re.compile(r"^@{PROC}/\d+/")
            if proc_pid_re.search(s):
                res.append(proc_pid_re.sub("@{PROC}/@{pid}/", s))
                res.append(proc_pid_re.sub("@{PROC}/*/", s))
            else:
                res.append(s)
        elif s.startswith("/run/") and len(s) > 5:
            res.append("/{,var/}run/%s" % s[5:])
            res.append(s)
        else:
            home_re = re.compile(r"^/home/[^/]+/")
            s = home_re.sub("@{HOME}/", s)
            res.append(s)

        # convert digit to [0-9]*
        fuzzy_num = []
        num_re = re.compile(r"\d+")
        for r in res:
            if num_re.search(r):
                fuzzy_num.append(num_re.sub("[0-9]*", r))
        if len(fuzzy_num) > 0:
            res += fuzzy_num

        return res

    def _format_interface_suggestions(self, ifaces):
        rec = []
        with_gadget_or_os = ["gpio", "hidraw", "serial-port"]
        tmp = []
        urls = {
            "personal-files": "https://forum.snapcraft.io/t/the-personal-files-interface",
            "system-files": "https://forum.snapcraft.io/t/the-system-files-interface",
        }
        if ifaces is not None:
            for iface in ifaces.split(","):
                if iface in with_gadget_or_os:
                    tmp.append("%s (with gadget or core support)" % iface)
                elif iface == "personal-files" or iface == "system-files":
                    tmp.append(
                        "%s (see %s for acceptance criteria)" % (iface, urls[iface])
                    )
                else:
                    tmp.append(iface)

            if len(tmp) > 1:
                rec.append("add one of '%s' to 'plugs'" % ",".join(tmp))
            else:
                rec.append("add '%s' to 'plugs'" % tmp[0])

        return rec

    def _aa_parse_dbus(self, line):
        ignored_attrs = ["pid", "peer_pid"]
        attr_map = {
            "label": "profile",
            "peer_label": "peer_profile",
            "bus": "dbus_bus",
            "path": "dbus_path",
            "interface": "dbus_interface",
            "member": "dbus_member",
            "mask": "denied_mask",
        }

        event = LibAppArmor.aa_log_record()
        for item in line.split():
            if "=" not in item:
                continue
            key, value = item.split("=", 1)
            if key in ignored_attrs:
                continue
            if key in attr_map:
                key = attr_map[key]
            if hasattr(event, key):
                # TODO: handle decode
                setattr(event, key, value.strip('"'))

        return event

    def make_apparmor_dbus_recommendation(self, line, snap_name):
        rec = []
        show = True
        msg = "DBus access"

        event = self._aa_parse_dbus(line)
        if ' apparmor="STATUS" ' in line:
            show = False
        elif (
            snap_name is not None
            and event is not None
            and event.profile is not None
            and not event.profile.startswith("snap.%s." % snap_name)
        ):
            show = False

        # quick exit
        if not show:
            event = None
            return (show, rec, msg)

        # TODO: recommendations. Until we have proper searches, just hardcode
        # some searches we know about
        dbus_rec = {
            "org.gnome.OnlineAccounts.": ["accounts-service"],
            "org.freedesktop.Avahi": ["avahi-control"],
            "org.freedesktop.Avahi.AddressResolver": ["avahi-observe"],
            "org.freedesktop.Avahi.DomainBrowser": ["avahi-observe"],
            "org.freedesktop.Avahi.HostNameResolver": ["avahi-observe"],
            "org.freedesktop.Avahi.RecordBrowser": ["avahi-observe"],
            "org.freedesktop.Avahi.Server": ["avahi-observe"],
            "org.freedesktop.Avahi.ServerBrowser": ["avahi-observe"],
            "org.freedesktop.Avahi.ServerResolver": ["avahi-observe"],
            "org.freedesktop.Avahi.ServerTypeBrowser": ["avahi-observe"],
            "org.bluez.": ["bluez"],
            "org.gnome.evolution.dataserver.Calendar": ["calendar-service"],
            "org.gnome.evolution.dataserver.AddressBook": ["contacts-service"],
            "io.snapcraft.Launcher": ["desktop", "unity7"],
            "com.canonical.SafeLauncher": ["desktop", "unity7"],
            "org.a11y.atspi.": ["desktop-legacy", "unity7"],
            "org.fcitx.Fcitx.": ["desktop-legacy", "unity7"],
            "org.gtk.vfs.MountTracker": ["desktop-legacy"],
            "org.freedesktop.fwupd": ["fwupd"],
            "ca.desrt.dconf.Writer": ["gsettings"],
            "org.freedesktop.hostname1": ["hostname-control"],
            "com.ubuntu.location.Service": ["location-observe", "location-service"],
            "com.ubuntu.location.Service.Provider": ["location-service"],
            "com.ubuntu.location.Service.Session": ["location-observe"],
            "org.maliit.Server.Address": ["maliit"],
            "core.ubuntu.media.Service": ["media-hub"],
            "org.freedesktop.ModemManager": ["modem-manager"],
            "MediaPlayer2": ["mpris"],
            "org.freedesktop.NetworkManager": [
                "network-manager",
                "network-manager-observe",
            ],
            "com.ubuntu.connectivity1.NetworkingStatus": ["network-status"],
            "org.freedesktop.resolve1.Manager": [
                "firewall-control",
                "network-bind",
                "network-control",
                "network",
                "network-observe",
            ],
            "org.ofono.": ["ofono"],
            "com.ubuntu.OnlineAccounts.Manager": ["online-accounts-service"],
            "org.freedesktop.Secret.": ["password-manager-service"],
            "org.cinnamon.ScreenSaver": ["screen-inhibit-control"],
            "org.freedesktop.ScreenSaver": ["screen-inhibit-control"],
            "org.gnome.ScreenSaver": ["screen-inhibit-control"],
            "org.kde.ScreenSaver": ["screen-inhibit-control"],
            "org.gnome.Shell.Screencast": ["screencast-legacy"],
            "org.gnome.Shell.Screenshot": ["screencast-legacy"],
            "org.freedesktop.login1.Manager": ["shutdown"],
            "com.canonical.StorageFramework.Registry": ["storage-framework-service"],
            "com.canonical.Thumbnailer": ["thumbnailer-service"],
            "org.freedesktop.timedate1": [
                "time-control",
                "timeserver-control",
                "timezone-control",
            ],
            "com.canonical.applications.Download": ["ubuntu-download-manager"],
            "org.freedesktop.UDisks2": ["udisks2"],
            "com.canonical.AppMenu.": ["unity7"],
            "com.canonical.dbusmenu": ["unity7"],
            "com.ubuntu.MenuRegistrar": ["unity7"],
            "org.freedesktop.UPower": ["upower-observe"],
        }

        matched_ifaces = []
        if event.operation is not None and event.operation != "dbus_bind":
            for i in dbus_rec:
                dbus_iface = i
                dbus_path = "/%s" % dbus_iface.replace(".", "/")
                ifaces = dbus_rec[i]
                found = False
                if dbus_iface in line:
                    found = True
                if dbus_path in line:
                    found = True

                if found:
                    for iface in ifaces:
                        if (
                            iface in restricted_interfaces
                            or iface not in self.rules["apparmor"]
                        ):
                            continue  # don't suggest restricted/unavailable
                        if iface not in matched_ifaces:
                            matched_ifaces += [iface]

        if len(matched_ifaces) > 0:
            plural = ""
            if len(matched_ifaces) > 1:
                plural = "one of "
            rec.append(
                "try adding %s'%s' to 'plugs'"
                % (plural, list_to_commas(sorted(matched_ifaces)))
            )

        if event.operation is not None and event.operation == "dbus_bind":
            if event.name is not None and event.name.startswith(
                "org.mpris.MediaPlayer2."
            ):
                rec.append(
                    "use 'mpris' slot (https://github.com/snapcore/snapd/wiki/Interfaces#mpris)"
                )  # noqa
            else:
                rec.append(
                    "use 'dbus' slot (https://forum.snapcraft.io/t/the-dbus-interface/2038)"
                )  # noqa

        event = None
        return (show, rec, msg)

    def make_apparmor_recommendation(self, line, snap_name, log):
        def _convert_avc_to_type1400(line):
            """python3-libapparmor doesn't understand AVC log entries so
               convert audit[...]: AVC ... to
               kernel: audit: type=1400 audit(XXX.YYY:ZZZ):
            """
            pat = re.compile(r" audit\[[0-9]+\]: AVC apparmor=")
            return pat.sub(
                " kernel: audit: type=1400 audit(1234.56:78): apparmor=", line
            )

        def _file_in_alt(event, rec, perm):
            """perm_type should be one of the permission bits. Eg r, w, x, etc"""
            bn = os.path.basename(event.name)
            # don't suggest the binaries snapd wraps
            if bn not in ["xdg-open", "xdg-settings"]:
                rec.append("adjust snap to ship '%s'" % bn)
                rec.append(
                    "adjust program to use relative paths if the "
                    "snap already ships '%s'" % bn
                )
            exec_re = re.compile(
                r"^\s*%s\s+\w*%s*\s*,\s*(#.*)?$" % (_escape(event.name), perm)
            )
            ifaces = self.apparmor_ifaces_allowing_rule(exec_re)
            if ifaces is not None:
                rec.append("add one of '%s' to 'plugs'" % ifaces)
            # try to match our alternation. This is where we need logprof
            alt_exec_re = re.compile(r"^(/usr)?/s?bin/%s" % _escape(bn))
            if alt_exec_re.search(_escape(event.name)):
                exec_re = re.compile(
                    r"^\s*/\{,usr/\}(\{,s\})?bin/%s\s+\w*%s*\s*,\s*(#.*)?"
                    % (_escape(bn), perm)
                )
                ifaces = self.apparmor_ifaces_allowing_rule(exec_re)
                if ifaces is not None:
                    rec.append("add one of '%s' to 'plugs'" % ifaces)

        rec = []
        show = True
        msg = None

        if ": AVC apparmor=" in line:
            line = _convert_avc_to_type1400(line)

        event = LibAppArmor.parse_record(line)
        if ' apparmor="STATUS" ' in line:
            show = False
        elif (
            snap_name is not None
            and event is not None
            and event.profile is not None
            and not event.profile.startswith("snap.%s." % snap_name)
        ):
            show = False

        # quick exit
        if not show:
            LibAppArmor.free_record(event)
            return (show, rec, log, msg)

        if (
            event.operation is not None
            and event.operation == "capable"
            and event.name is not None
        ):
            # capability rules
            rule_re = re.compile(
                r"^\s*capability\s+%s\s*,\s*(#.*)?$" % _escape(event.name)
            )

            msg = "Capability: %s" % event.name
            rec.append(
                "adjust program to not require 'CAP_%s' (see 'man 7 "
                "capabilities')" % (event.name.upper())
            )
            if event.name != "mac_admin":  # policy is pointless with this
                if event.name == "sys_module":  # policy is pointless with this
                    rec.append("configure modules on the system instead of " "via snap")
                else:
                    ifaces = self.apparmor_ifaces_allowing_rule(rule_re)
                    # don't search for ptrace rules since we have a bunch of
                    # special ones
                    if ifaces is not None and event.name != "sys_ptrace":
                        rec.append("add one of '%s' to 'plugs'" % ifaces)
                    if event.name == "net_admin":
                        rec.append(
                            "do nothing if using systemd utility (eg, "
                            "timedatectl): https://forum.snapcraft.io/t/managing-time-date-and-timezone-in-ubuntu-core/408/44"
                        )  # noqa
                        rec.append("do nothing " "(https://launchpad.net/bugs/1465724)")
                    else:
                        rec.append("do nothing if program otherwise works " "properly")
        elif (
            event.operation is not None
            and event.operation == "exec"
            and event.name is not None
        ):
            msg = "File: %s (exec)" % (event.name)
            # exec rules
            if event.name.startswith("/snap/bin/"):
                rec.append(
                    "adjust program to execute binaries directly from "
                    "$SNAP instead of /snap/bin"
                )
            else:
                _file_in_alt(event, rec, "x")
        elif (
            event.name is not None
            and event.name.startswith("/")
            and event.operation is not None
            and event.operation != "mount"
        ):
            # file rules

            # FIXME: the mask check is too simplistic since we only look at
            # mask[0] for permission, but we might have r, w, m, rw, rwm, wm, m
            # in any order. We need logprof...
            mask = "write"
            if event.denied_mask == "r":
                mask = "read"
            if event.denied_mask is not None and "m" in event.denied_mask:
                mask = "mmap"

            msg = "File: %s (%s)" % (event.name, mask)

            nameservice_files = ["/etc/passwd", "/etc/group", "/etc/nsswitch.conf"]
            layouts = ["/etc/", "/run/", "/var/lib/"]
            bin_dirs = [
                "/bin",
                "/sbin",
                "/usr/bin",
                "/usr/sbin",
                "/usr/local/bin",
                "/usr/local/sbin",
            ]

            # With some file paths we don't want to suggest anything since
            # recommendations might be confusing to the user.
            no_recommendations_file = ["/run/snapd.socket"]

            if event.name.startswith("/dev/shm/") or event.name.startswith("/run/shm/"):
                if os.path.basename(event.name).startswith("lttng-"):
                    rec.append(
                        "lttng access not currently supported with " + "confined snaps"
                    )
                elif os.path.basename(event.name).startswith("sem."):
                    url = "https://forum.snapcraft.io/t/python-multiprocessing-sem-open-blocked-in-strict-mode/962/10"  # noqa
                    rec.append(
                        "adjust program to use snap-specific "
                        + "semaphore with sem_open() (%s)" % url
                    )
                else:
                    rec.append(
                        "adjust program to create files and "
                        + "directories in %s/snap.$SNAP_NAME.*" % event.name[:8]
                    )
                    rec.append(
                        "try the snapcraft preload plugin: "
                        + "https://github.com/sergiusens/snapcraft-preload"
                    )  # noqa
            elif event.name.startswith("/snap/bin/"):
                rec.append(
                    "adjust program to access files in $SNAP " "instead of /snap/bin"
                )
            elif event.name.startswith("/snap/") and mask == "write":
                rec.append("adjust program to not write to $SNAP")
            elif re.search(r"^/(root|home/[^/]+)/[^\.]", event.name) and not re.search(
                r"^/(root|home/[^/]+)/snap/", event.name
            ):
                rec += self._format_interface_suggestions("home")
            elif re.search(
                r"^/(root|home/[^/]+)/\.config/dconf/", event.name
            ) and self._interface_exists("gsettings"):
                rec += self._format_interface_suggestions("gsettings")
            elif event.name.startswith("/var/tmp/"):
                rec.append("adjust program to use TMPDIR or /tmp")
            elif mask == "write" and event.name.startswith("/var/log/"):
                rec.append("adjust program to write log files to " "$SNAP_DATA")
            elif event.name.startswith("/etc/wgetrc") and mask == "read":
                rec.append("adjust program to use " + "SYSTEM_WGETRC=$SNAP/etc/wgetrc")
            elif (
                event.name in ["/usr/share/misc/magic", "/usr/share/file/magic"]
                and mask == "read"
            ):
                rec.append(
                    "adjust program to use " + "MAGIC=$SNAP/usr/share/misc/magic"
                )
            elif (
                mask == "read"
                and event.operation is not None
                and event.operation == "file_mprotect"
                and event.name == "/usr/bin/snap-confine"
            ):
                rec.append(
                    "adjust program to execute binaries directly from "
                    "$SNAP instead of /snap/bin"
                )
            else:
                suggested_ifaces = []
                for fn in [event.name] + self._aa_file(event.name):
                    file_re = re.compile(
                        r"^\s*(owner +)?%s\s+\w*%s\w*\s*,\s*(#.*)?$"
                        % (_escape(fn), _escape(mask[0]))
                    )
                    ifaces = self.apparmor_ifaces_allowing_rule(file_re)

                    # we really need logprof... Check if any parent dirs have
                    # globs
                    level = 0
                    parent = os.path.dirname(fn)
                    while parent != "":
                        if parent == "/":
                            parent = ""  # we'll add this back in dir_re

                        glob_str = r"(\*\*|{,\*\*})"
                        if level == 0:
                            glob_str = r"(\*\*?|{,\*\*?})"

                        dir_re = re.compile(
                            r"^\s*(owner +)?%s/%s\s+\w*%s\w*\s*,\s*(# .*)?$"
                            % (_escape(parent), glob_str, _escape(mask[0]))
                        )

                        dir_ifaces = self.apparmor_ifaces_allowing_rule(dir_re)
                        if dir_ifaces is not None:
                            if dir_ifaces not in suggested_ifaces:
                                suggested_ifaces.append(dir_ifaces)

                        parent = os.path.dirname(parent)
                        level += 1

                    if ifaces is not None:
                        if ifaces not in suggested_ifaces:
                            suggested_ifaces.append(ifaces)

                gpio_re = re.compile(r"^/sys/devices/.*/gpio/")
                if event.name.startswith("/sys/class/gpio") or gpio_re.search(
                    event.name
                ):
                    if "gpio" not in suggested_ifaces:
                        suggested_ifaces.append("gpio")
                elif event.name.startswith("/dev/"):
                    if event.name.startswith("/dev/hidraw"):
                        if "hiddraw" not in suggested_ifaces:
                            suggested_ifaces.append("hidraw")
                    elif event.name.startswith("/dev/tty"):
                        serial_re = re.compile(
                            r"^/dev/tty(mxc|USB|ACM|AMA|XRUSB|S|O)[0-9]+$"
                        )
                        if (
                            serial_re.search(event.name)
                            and "serial-port" not in suggested_ifaces
                        ):
                            suggested_ifaces.append("serial-port")
                        rawusb_re = re.compile(r"^/dev/tty(USB|ACM)[0-9]+$")
                        if (
                            rawusb_re.search(event.name)
                            and "raw-usb" not in suggested_ifaces
                        ):
                            suggested_ifaces.append("raw-usb")
                        console_re = re.compile(r"^/dev/tty[0-9]+$")
                        if console_re.search(event.name):
                            rec.append("adjust program to not access '%s'" % fn)
                elif event.name.startswith("/proc/") or event.name.startswith("/sys/"):
                    for fn in self._aa_file(event.name):
                        if fn.startswith("@{PROC}/*/"):
                            # prefer @{PROC}/@{pid}/ to @{PROC}/*/
                            continue
                        rec.append("adjust program to not access '%s'" % fn)

                        if (
                            fn == "@{PROC}/@{pid}/oom_adj"
                            or fn == "@{PROC}/@{pid}/oom_score_adj"
                        ) and mask == "write":
                            rec.append(
                                "do nothing if program otherwise works " "properly"
                            )
                        elif event.name == "/proc/1/environ":
                            rec.append(
                                "do nothing if using systemd utility "
                                "(eg, timedatectl): https://forum.snapcraft.io/t/managing-time-date-and-timezone-in-ubuntu-core/408/44"
                            )  # noqa
                            rec.append(
                                "do nothing if program otherwise " "works properly"
                            )
                elif (
                    "/run/shm/snap." not in event.name
                    and "/run/snap." not in event.name
                    and (
                        event.name.startswith("/var/run/")
                        or event.name.startswith("/run/")
                    )
                ):
                    rec.append("adjust program to use $SNAP_DATA")
                    rec.append("adjust program to use " + "/run/shm/snap.$SNAP_NAME.*")
                    rec.append("adjust program to use " + "/run/snap.$SNAP_NAME.*")
                else:
                    if mask == "write":
                        rec.append(
                            "adjust program to write to $SNAP_DATA, "
                            "$SNAP_COMMON, $SNAP_USER_DATA or "
                            "$SNAP_USER_COMMON"
                        )
                    elif (
                        mask == "read"
                        and os.path.dirname(event.name.rstrip("/")) in bin_dirs
                    ):
                        bn = os.path.basename(event.name)
                        # don't suggest the binaries snapd wraps
                        if bn not in ["xdg-open", "xdg-settings"]:
                            _file_in_alt(event, rec, "r")
                    elif mask == "read":
                        rec.append(
                            "adjust program to read necessary files "
                            "from $SNAP, $SNAP_DATA, $SNAP_COMMON, "
                            "$SNAP_USER_DATA or $SNAP_USER_COMMON"
                        )

                if event.name.startswith("/dev/") and mask == "mmap":
                    url = (
                        "https://forum.snapcraft.io/t/snap-and-executable-stacks/1812"
                    )  # noqa
                    rec.append(
                        "verify program isn't using an executable " + "stack: %s" % url
                    )

                if event.name in nameservice_files:
                    ns_re = re.compile(
                        r"^\s*#include\s+<abstractions/nameservice>\s*(#.*)?$"
                    )
                    ifaces = self.apparmor_ifaces_allowing_rule(ns_re)
                    if ifaces is not None:
                        if ifaces not in suggested_ifaces:
                            suggested_ifaces.append(ifaces)
                elif list(filter(lambda x: event.name.startswith(x), layouts)):
                    url = "https://forum.snapcraft.io/t/snap-layouts/7207"  # noqa
                    rec.append("adjust snap to use snap layouts (%s)" % url)

                if len(suggested_ifaces) == 0:
                    if re.search(r"^/etc/[^\.]", event.name):
                        rec += self._format_interface_suggestions("system-files")
                    if re.search(
                        r"^/(root|home/[^/]+)/\.", event.name
                    ) and not re.search(
                        r"^/(root|home/[^/]+)/\.config/dconf/", event.name
                    ):
                        rec += self._format_interface_suggestions("personal-files")
                for c in sorted(suggested_ifaces):
                    rec += self._format_interface_suggestions(c)

            if event.name in no_recommendations_file:
                rec = []

        elif event.net_family is not None and event.net_family == "unix":
            # Libapparmor doesn't handle unix rules yet, so fake it
            addr_re = re.compile(r'(.*\s+addr=")([^\s]+)(")(.*)')
            if addr_re.search(line):
                addr = addr_re.sub("\\2", line.rstrip())
                profile_prefix = "snap.<your snap name>"
                if event.profile is not None and event.profile.startswith("snap."):
                    profile_prefix = "snap.%s" % event.profile.split('.', 2)[1]

                # decode abstract sockets
                if addr.startswith('@'):
                    decoded = "@%s" % _aa_decode(addr[1:])
                    if decoded != addr:
                        log = addr_re.sub("\\1\\2\\3(%s)\\4" % decoded, log)
                        addr = decoded

                if not addr.startswith("@%s." % profile_prefix):
                    rec.append("adjust '%s' to start with '%s.' (eg, '@%s.%s')" % (addr, profile_prefix, profile_prefix, addr[1:]))
                    rec.append("use 'listen-stream: @%s.%s' for a socket-activated daemon" % (profile_prefix, addr[1:]))
                # FIXME: unix rules can span multiple lines in the snapd
                # policy
                unix_re = re.compile(
                    r'^\s*unix\s+.*\s+addr=[\'"]?%s[\'"]?[\s,]?' % _escape(addr)
                )
                ifaces = self.apparmor_ifaces_allowing_rule(unix_re)
                if ifaces is not None:
                    if len(ifaces.split(",")) > 1:
                        rec.append("add one of '%s' to 'plugs'" % ifaces)
                    else:
                        rec.append("add '%s' to 'plugs'" % ifaces)
        elif event.signal is not None:
            msg = "Signal: %s (%s)" % (event.signal, event.denied_mask)
            # NOTE: for now, setup a very restrictive regex so we just get the
            # one interface (process-control)
            signal_re = re.compile(r"^\s*signal \(send\),\s+")
            ifaces = self.apparmor_ifaces_allowing_rule(signal_re)
            if ifaces is not None:
                if len(ifaces.split(",")) > 1:
                    rec.append("add one of '%s' to 'plugs'" % ifaces)
                else:
                    rec.append("add '%s' to 'plugs'" % ifaces)
            rec.append("adjust program to only send signals to itself")
        elif event.operation is not None and event.operation == "ptrace":
            msg = "Ptrace: peer=%s (%s)" % (event.peer, event.denied_mask)
            if event.denied_mask == "read":
                # NOTE: for now, setup a very restrictive regex so we just get
                # the one interface (system-observe)
                ptrace_re = re.compile(r"^\s*ptrace \(read\),\s+")
                ifaces = self.apparmor_ifaces_allowing_rule(ptrace_re)
                if ifaces is not None:
                    if len(ifaces.split(",")) > 1:
                        rec.append("add one of '%s' to 'plugs'" % ifaces)
                    else:
                        rec.append("add '%s' to 'plugs'" % ifaces)
            elif event.denied_mask == "trace":
                rec.append("adjust program to not trace processes")
            rec.append("do nothing if program otherwise works properly")
        elif event.net_family is not None:
            net_re = re.compile(
                r"^\s*network\s+%s\s+%s\s*,\s*(# .*)?$"
                % (_escape(event.net_family), _escape(event.net_sock_type))
            )
            ifaces = self.apparmor_ifaces_allowing_rule(net_re)
            if ifaces is not None:
                if len(ifaces.split(",")) > 1:
                    rec.append("add one of '%s' to 'plugs'" % ifaces)
                else:
                    rec.append("add '%s' to 'plugs'" % ifaces)

            if event.net_family == "netlink" and event.net_sock_type == "dgram":
                rec.append("do nothing (https://launchpad.net/bugs/1499897)")

        LibAppArmor.free_record(event)

        return (show, rec, log, msg)

    def seccomp_ifaces_allowing_syscall(self, syscall):
        global os_release
        matched_ifaces = []

        syscall_re = re.compile(r"^\s*%s\s*(#.*)?$" % _escape(syscall))
        interfaces = self.rules["seccomp"].keys()
        for iface in interfaces:
            if iface in restricted_interfaces:
                continue  # don't suggest restricted

            # don't confuse people with this one
            if syscall == "bind":
                iface = "network-bind"
                if iface not in matched_ifaces:
                    matched_ifaces += [iface]
                continue

            for line in self.rules["seccomp"][iface]:
                if syscall_re.search(line) and iface not in matched_ifaces:
                    matched_ifaces += [iface]

        if not matched_ifaces:
            return None

        return list_to_commas(sorted(matched_ifaces))

    def make_seccomp_recommendation(self, line, snap_name, log):
        show = True
        rec = []
        msg = None

        syscall_re = re.compile(r"syscall=(\d+)")
        setuid_re = re.compile(r"^set((re|res)?[ug]id)")
        arch_re = re.compile(r" arch=([0-9a-f]+) ")

        syscall_key = syscall_re.search(line)
        arch_key = arch_re.search(line)
        if syscall_key and arch_key:
            num = syscall_key.groups()[0]
            arch = arch_key.groups()[0]

            output = subprocess.check_output(
                ["scmp_sys_resolver", "-a", seccomp_architecture(arch), num],
                universal_newlines=True,
            )
            syscall = output.rstrip()
            log = syscall_re.sub("\\1(%s)" % syscall, log)
            msg = "Syscall: %s" % syscall

            if "chown" in syscall:
                rec.append(
                    "don't copy ownership of files (eg, use 'cp -r "
                    "--preserve=mode' instead of 'cp -a')"
                )
                rec.append(
                    "try the snapcraft preload plugin: "
                    + "https://github.com/sergiusens/snapcraft-preload"
                )
                rec.append("adjust program to not use '%s'" % syscall)
                rec.append(
                    "ignore the denial if the program otherwise works "
                    "correctly (unconditial chown is often just noise)"
                )
            elif setuid_re.search(syscall):
                rec.append(
                    "adjust program to not use '%s' until per-snap "
                    "user/groups are supported "
                    "(https://launchpad.net/bugs/1446748)" % syscall
                )
            elif "setgroups" in syscall:
                url = (
                    "https://forum.snapcraft.io/t/seccomp-filtering-for-setgroups/2109"
                )  # noqa
                rec.append("adjust program to not use '%s' (%s)" % (syscall, url))
            elif "_module" in syscall:
                rec.append("configure modules on the system instead of via " "snap")
            elif syscall == "socketcall":
                rec.append(
                    "install with --devmode on x86 until system is "
                    "updated (https://launchpad.net/bugs/1576066)"
                )
            elif syscall == "socket":
                for i in sc_arg_socket:
                    if i[0] in self.rules["seccomp"]:
                        rec.append("add %s (if using %s)" % (i[0], i[1]))
            elif syscall == "setns":
                for i in sc_arg_setns:
                    if i[0] in self.rules["seccomp"]:
                        rec.append("add %s (if using %s)" % (i[0], i[1]))
            elif syscall == "setpriority":
                rec.append(
                    "ignore the denial if the program otherwise works "
                    "correctly (unconditial setpriority is often just "
                    "noise)"
                )
            elif syscall == "quotactl":
                for i in sc_arg_quotactl:
                    if i[0] in self.rules["seccomp"]:
                        rec.append("add %s (if using %s)" % (i[0], i[1]))
            else:
                ifaces = self.seccomp_ifaces_allowing_syscall(syscall)
                rec += self._format_interface_suggestions(ifaces)

        return (show, rec, log, msg)


#
# End helpers
#


class ScanLogsException(Exception):
    """This class represents ScanLogs exceptions"""

    def __init__(self, value):
        self.value = value

    def __str__(self):
        return repr(self.value)


def main():
    global DEBUGGING

    if "SNAP" in os.environ:
        os.chdir(os.environ["SNAP"])

    parser = optparse.OptionParser()
    parser.add_option(
        "-d",
        "--debug",
        help="Show debugging output",
        action="store_true",
        default=False,
    )
    parser.add_option(
        "--policy-version",
        dest="policy_version",
        help="Use policy version",
        metavar="version",
        default=None,
    )
    parser.add_option(
        "--log-file",
        dest="log_file",
        help="Use non-default log file",
        metavar="FILE",
        default=None,
    )
    parser.add_option(
        "-f",
        "--follow",
        dest="follow",
        help="Follow specified file",
        action="store_true",
        default=False,
    )
    parser.add_option(
        "-r",
        "--recommend",
        dest="recommend",
        help="Suggest a recommendation it possible",
        action="store_true",
        default=False,
    )
    parser.add_option(
        "--only-apparmor",
        dest="only_apparmor",
        help="Only show apparmor denials",
        action="store_true",
        default=False,
    )
    parser.add_option(
        "--only-seccomp",
        dest="only_seccomp",
        help="Only show seccomp denials",
        action="store_true",
        default=False,
    )
    parser.add_option(
        "--only-snap",
        dest="only_snap",
        help="Only show denials for specified snap",
        metavar="NAME",
        default=None,
    )
    parser.add_option(
        "--only-new",
        dest="only_new",
        help="Only show entries since start",
        action="store_true",
        default=False,
    )
    parser.add_option(
        "--exclude",
        dest="exclude",
        help="Exclude lines matching PATTERN (python3 raw "
        + "string. Ie: re.compile(r'PATTERN'))",
        metavar="PATTERN",
        default=None,
    )

    (opt, args) = parser.parse_args()

    if opt.debug:
        DEBUGGING = True

    global os_release
    detect_os_release()
    if opt.policy_version:
        os_release = opt.policy_version

    jctl_msg = (
        "$ sudo journalctl --output=short --follow --all | " + "sudo snappy-debug"
    )

    logs = "/var/log/syslog"
    if not sys.stdin.isatty():
        logs = "stdin"
    elif opt.log_file:
        logs = opt.log_file
        debug("logs: %s" % opt.log_file)

    if logs != "stdin":
        if not os.path.exists(logs):
            if os_is_core:
                # uc18 doesn't ship /var/log/syslog, so provide some help
                error(
                    "'%s' does not exist. Redirect journalctl instead. " "Eg:" % logs,
                    do_exit=False,
                )
                error(jctl_msg)
            else:
                error(
                    "'%s' does not exist. Aborting. Please choose another "
                    "file with\n       --log-file or try redirecting "
                    "journalctl. Eg:" % logs,
                    do_exit=False,
                )
                error(jctl_msg)
        elif opt.follow:
            if os_is_core:
                info(
                    "Detected Ubuntu Core. For best results, redirect "
                    "journalctl. Eg:"
                )
            else:
                info("Following '%s'. If have dropped messages, use:" % logs)
            info(jctl_msg)

    prev_rate = None
    try:
        output = subprocess.check_output(
            ["sysctl", "kernel.printk_ratelimit"], universal_newlines=True
        )
        prev_rate = output.rstrip().split()[-1]
        if opt.follow and os.geteuid() == 0:
            # Turn off kernel rate limiting if we are debugging with this tool
            subprocess.call(["sysctl", "-w", "kernel.printk_ratelimit=0"])
    except Exception:
        warn("Could not set kernel rate limiting")

    display = "both"
    if opt.only_seccomp:
        display = "seccomp"
    elif opt.only_apparmor:
        display = "apparmor"

    try:
        ScanLogs(
            logs,
            follow=opt.follow,
            recommend=opt.recommend,
            display=display,
            snap_name=opt.only_snap,
            from_end=opt.only_new,
            exclude=opt.exclude,
        )
    except ScanLogsException as e:
        _print("%s\n" % (str(e).strip('"')), sys.stderr)
        sys.exit(1)
    except Exception:
        raise
    if opt.follow and prev_rate is not None and os.geteuid() == 0:
        try:
            subprocess.call(["sysctl", "-w", "kernel.printk_ratelimit=%s" % prev_rate])
        except Exception:
            warn("Could not reset kernel rate limiting")


if __name__ == "__main__":
    sys.exit(main())