apparmor/utils/aa-unconfined

#! /usr/bin/python3
# ----------------------------------------------------------------------
#    Copyright (C) 2013 Kshitij Gupta <kgupta8592@gmail.com>
#    Copyright (C) 2016 Canonical, Ltd.
#
#    This program is free software; you can redistribute it and/or
#    modify it under the terms of version 2 of the GNU General Public
#    License as published by the Free Software Foundation.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
# ----------------------------------------------------------------------
import argparse
import os
import re
import subprocess
import sys

import apparmor.aa as aa
import apparmor.ui as ui
import apparmor.common

# setup exception handling
from apparmor.fail import enable_aa_exception_handler
enable_aa_exception_handler()

# setup module translations
from apparmor.translations import init_translation
_ = init_translation()

parser = argparse.ArgumentParser(description=_("Lists unconfined processes having tcp or udp ports"))
parser.add_argument("--paranoid", action="store_true", help=_("scan all processes from /proc"))
bin_group = parser.add_mutually_exclusive_group()
bin_group.add_argument("--with-ss", action='store_true', help=_("use ss(8) to find listening processes (default)"))
bin_group.add_argument("--with-netstat", action='store_true', help=_("use netstat(8) to find listening processes"))
args = parser.parse_args()

paranoid = args.paranoid

aa_mountpoint = aa.check_for_apparmor()
if not aa_mountpoint:
    raise aa.AppArmorException(_("It seems AppArmor was not started. Please enable AppArmor and try again."))


def get_all_pids():
    '''Return a set of all pids via walking /proc'''
    return set(filter(lambda x: re.search(r"^\d+$", x), aa.get_subdirectories("/proc")))


def get_pids_ss():
    '''Get a set of pids listening on network sockets via ss(8)'''
    regex_lines = re.compile(r"^(tcp|udp|raw|p_dgr)\s.+\s+users:(?P<users>\(\(.*\)\))$")
    regex_users_pids = re.compile(r'(\("[^"]+",(pid=)?(\d+),[^)]+\))')

    pids = set()
    my_env = os.environ.copy()
    my_env['LANG'] = 'C'
    my_env['PATH'] = '/bin:/usr/bin:/sbin:/usr/sbin'
    for family in ['inet', 'inet6', 'link']:
        cmd = ['ss', '-nlp', '--family', family]
        if sys.version_info < (3, 0):
            output = subprocess.check_output(cmd, shell=False, env=my_env).split("\n")
        else:
            # Python3 needs to translate a stream of bytes to string with specified encoding
            output = str(subprocess.check_output(cmd, shell=False, env=my_env), encoding='utf8').split("\n")

        for line in output:
            match = regex_lines.search(line.strip())
            if match:
                users = match.group('users')
                for (_, _, pid) in regex_users_pids.findall(users):
                    pids.add(pid)
    return pids


def get_pids_netstat():
    '''Get a set of pids listening on network sockets via netstat(8)'''
    regex_tcp_udp = re.compile(r"^(tcp|udp|raw)6?\s+\d+\s+\d+\s+\S+\:(\d+)\s+\S+\:(\*|\d+)\s+(LISTEN|\d+|\s+)\s+(?P<pid>\d+)\/(\S+)")

    cmd = ['netstat', '-nlp', '--protocol', 'inet,inet6']
    my_env = os.environ.copy()
    my_env['LANG'] = 'C'
    my_env['PATH'] = '/bin:/usr/bin:/sbin:/usr/sbin'
    if sys.version_info < (3, 0):
        output = subprocess.check_output(cmd, shell=False, env=my_env).split("\n")
    else:
        # Python3 needs to translate a stream of bytes to string with specified encoding
        output = str(subprocess.check_output(cmd, shell=False, env=my_env), encoding='utf8').split("\n")

    pids = set()
    for line in output:
        match = regex_tcp_udp.search(line)
        if match:
            pids.add(match.group('pid'))
    return pids


pids = set()
if paranoid:
    pids = get_all_pids()
elif args.with_ss or (not args.with_netstat and (os.path.exists('/bin/ss') or os.path.exists('/usr/bin/ss'))):
    pids = get_pids_ss()
else:
    pids = get_pids_netstat()

for pid in sorted(map(int, pids)):
    try:
        prog = os.readlink("/proc/%s/exe"%pid)
    except OSError:
        continue
    attr = None
    if os.path.exists("/proc/%s/attr/current"%pid):
        with aa.open_file_read("/proc/%s/attr/current"%pid) as current:
            for line in current:
                line = line.strip()
                if line.endswith(' (complain)', 1) or line.endswith(' (enforce)', 1): # enforce at least one char as profile name
                    attr = line

    cmdline = apparmor.common.cmd(["cat", "/proc/%s/cmdline"%pid])[1]
    pname = cmdline.split("\0")[0]
    if '/' in pname and pname != prog:
        pname = "(%s)"% pname
    else:
        pname = ""
    regex_interpreter = re.compile(r"^(/usr)?/bin/(python|perl|bash|dash|sh)$")
    if not attr:
        if regex_interpreter.search(prog):
            cmdline = re.sub(r"\x00", " ", cmdline)
            cmdline = re.sub(r"\s+$", "", cmdline).strip()

            ui.UI_Info(_("%(pid)s %(program)s (%(commandline)s) not confined") % { 'pid': pid, 'program': prog, 'commandline': cmdline })
        else:
            if pname and pname[-1] == ')':
                pname = ' ' + pname
            ui.UI_Info(_("%(pid)s %(program)s%(pname)s not confined") % { 'pid': pid, 'program': prog, 'pname': pname })
    else:
        if regex_interpreter.search(prog):
            cmdline = re.sub(r"\0", " ", cmdline)
            cmdline = re.sub(r"\s+$", "", cmdline).strip()
            ui.UI_Info(_("%(pid)s %(program)s (%(commandline)s) confined by '%(attribute)s'") % { 'pid': pid, 'program': prog, 'commandline': cmdline, 'attribute': attr })
        else:
            if pname and pname[-1] == ')':
                pname = ' ' + pname
            ui.UI_Info(_("%(pid)s %(program)s%(pname)s confined by '%(attribute)s'") % { 'pid': pid, 'program': prog, 'pname': pname, 'attribute': attr })