File: //sbin/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.init_aa()
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(ss='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(netstat='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
def read_proc_current(filename):
attr = None
if os.path.exists(filename):
with apparmor.common.open_file_read(filename) as current:
for line in current:
line = line.strip()
if line.endswith(' (complain)', 1) or line.endswith(' (enforce)', 1) or line.endswith(' (kill)', 1): # enforce at least one char as profile name
# intentionally not checking for '(unconfined)', because $binary confined by $profile (unconfined) would look very confusing
attr = line
return attr
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
if os.path.exists("/proc/%s/attr/apparmor/current" % pid):
attr = read_proc_current("/proc/%s/attr/apparmor/current" % pid)
else:
# fallback to shared attr/current if attr/apparmor/current doesn't exist
attr = read_proc_current("/proc/%s/attr/current" % pid)
pname = None
cmdline = None
with apparmor.common.open_file_read("/proc/%s/cmdline" % pid) as cmd:
cmdline = cmd.readlines()[0]
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})