| Server IP : 92.204.138.22 / Your IP : 18.216.3.58 Web Server : Apache System : Linux ns1009439.ip-92-204-138.us 4.18.0-553.8.1.el8_10.x86_64 #1 SMP Tue Jul 2 07:26:33 EDT 2024 x86_64 User : internationaljou ( 1019) PHP Version : 7.4.33 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /lib/python3.6/site-packages/setroubleshoot/ |
Upload File : |
from __future__ import absolute_import
import six
from six.moves import range
import sys
# Authors: John Dennis <jdennis@redhat.com>
# Thomas Liu <tliu@redhat.com
# Copyright (C) 2007-2010 Red Hat, Inc.
#
# 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.
#
# 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.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#
__all__ = ['derive_record_format',
'parse_audit_record_text',
'AvcContext',
'AVC',
'AuditEventID',
'AuditEvent',
'AuditRecord',
'AuditRecordReader',
]
import six
import audit
import struct
import os
import errno
import re
import selinux
import base64
from types import *
import selinux.audit2why as audit2why
from setroubleshoot.util import *
from setroubleshoot.html_util import *
from setroubleshoot.xml_serialize import *
from sepolicy import *
O_ACCMODE = 0o0000003
cmp = lambda x, y: (x > y) - (x < y)
#-----------------------------------------------------------------------------
standard_directories = get_standard_directories()
#-----------------------------------------------------------------------------
def audit_record_from_text(text):
parse_succeeded, record_type, event_id, body_text = parse_audit_record_text(text)
audit_record = AuditRecord(record_type, event_id, body_text)
return audit_record
#-----------------------------------------------------------------------------
def derive_record_format(socket_path):
if re.search('/audispd_events$', socket_path):
return AuditRecordReader.TEXT_FORMAT
if re.search('/audit_events$', socket_path):
return AuditRecordReader.BINARY_FORMAT
return AuditRecordReader.TEXT_FORMAT # assume new format
# regular expression to find message like this:
# msg=audit(1152828325.857:123085): avc: denied { append } for pid=14205 ...
# Note, messages arriving directly from the audit system omit
# 'msg=', but messages in log files prepend 'msg='
# group 1 is the optional "node=XXX "
# group 2 is the node if node=XXX is present
# group 3 is the optional "type=XXX "
# group 4 is the type if type=XXX is present
# group 5 is the optional 'msg='
# group 6 is the complete event id
# group 7 is the seconds component of the timestamp
# group 8 is the millisconds component of the timestamp
# group 9 is the timestamp unique number
# group 10 is the body of the message appearing after the event id
audit_input_re = re.compile(r'(node=(\S+)\s+)?(type=(\S+)\s+)?(msg=)?audit\(((\d+)\.(\d+):(\d+))\):\s*(.*)')
def parse_audit_record_text(input):
parse_succeeded = False
host = None
record_type = None
event_id = None
body_text = None
match = audit_input_re.search(input)
if match is not None:
parse_succeeded = True
if match.group(2):
host = match.group(2)
if match.group(4):
record_type = match.group(4)
if match.group(6):
seconds = int(match.group(7))
milli = int(match.group(8))
serial = int(match.group(9))
event_id = AuditEventID(seconds, milli, serial, host)
body_text = match.group(10)
return (parse_succeeded, record_type, event_id, body_text)
audit_binary_input_re = re.compile(r'audit\(((\d+)\.(\d+):(\d+))\):\s*(.*)')
def parse_audit_binary_text(input):
parse_succeeded = False
event_id = None
body_text = None
match = audit_binary_input_re.search(input)
if match is not None:
parse_succeeded = True
if match.group(1):
seconds = int(match.group(2))
milli = int(match.group(3))
serial = int(match.group(4))
event_id = AuditEventID(seconds, milli, serial)
body_text = match.group(5)
return (parse_succeeded, event_id, body_text)
#------------------------------------------------------------------------
import string
def printable(s):
if s:
filtered_path = [x for x in s if x in string.printable]
if filtered_path == s:
return True
return False
class AvcContext(XmlSerialize):
_xml_info = {
'user': {'XMLForm': 'attribute'},
'role': {'XMLForm': 'attribute'},
'type': {'XMLForm': 'attribute'},
'mls': {'XMLForm': 'attribute'},
}
def __init__(self, data):
super(AvcContext, self).__init__()
if isinstance(data, six.string_types):
fields = data.split(':')
if len(fields) >= 3:
self.user = fields[0]
self.role = fields[1]
self.type = fields[2]
if len(fields) > 3:
self.mls = ':'.join(fields[3:])
else:
self.mls = 's0'
def __str__(self):
return '%s:%s:%s:%s' % (self.user, self.role, self.type, self.mls)
def format(self):
# FIXME, what does selinux_raw_to_trans_context() do and why do we need it?
(rc, trans) = selinux.selinux_raw_to_trans_context(str(self))
return trans
def __ne__(self, other):
return not self.__eq__(other)
def __eq__(self, other):
for name in list(self._xml_info.keys()):
if getattr(self, name) != getattr(other, name):
return False
return True
#-----------------------------------------------------------------------------
class AuditEventID(XmlSerialize):
_xml_info = {
'seconds': {'XMLForm': 'attribute', 'import_typecast': int},
'milli': {'XMLForm': 'attribute', 'import_typecast': int},
'serial': {'XMLForm': 'attribute', 'import_typecast': int},
'host': {'XMLForm': 'attribute'},
}
def __init__(self, seconds, milli, serial, host=None):
super(AuditEventID, self).__init__()
self.seconds = seconds
self.milli = milli
self.serial = serial
if host is not None:
self.host = host
def __eq__(self, other):
if self.host != other.host:
return False
if self.seconds != other.seconds:
return False
if self.milli != other.milli:
return False
if self.serial != other.serial:
return False
return True
def __lt__(self, other):
if self.host != other.host:
raise ValueError("cannot compare two %s objects whose host values differ (%s!=%s)"
% (self.__class__.__name__, self.host, other.host))
if self.seconds != other.seconds:
return self.seconds < other.seconds
if self.milli != other.milli:
return self.milli < other.milli
return self.serial < other.serial
def copy(self):
import copy
return copy.copy(self)
time = property(lambda self: float(self.sec) + self.milli / 1000.0)
def __str__(self):
return "audit(%d.%d:%d)" % (self.seconds, self.milli, self.serial)
def is_valid(self):
if self.seconds is None:
return False
if self.milli is None:
return False
if self.serial is None:
return False
return True
#-----------------------------------------------------------------------------
class AuditRecord(XmlSerialize):
_xml_info = {
'record_type': {'XMLForm': 'attribute', },
'event_id': {'XMLForm': 'element', 'import_typecast': AuditEventID},
'body_text': {'XMLForm': 'element'},
'line_number': {'XMLForm': 'attribute', 'import_typecast': int},
}
binary_version = 0
binary_header_format = "iiii"
binary_header_size = struct.calcsize(binary_header_format)
key_value_pair_re = re.compile(r"([^ \t]+)\s*=\s*([^ \t]+)")
avc_re = re.compile(r"avc:\s+([^\s]+)\s+{([^}]+)}\s+for\s+")
exec_arg_re = re.compile(r'^a\d+$')
def __init__(self, record_type, event_id, body_text, fields=None, line_number=None):
super(AuditRecord, self).__init__()
# Header
self.record_type = record_type
self.event_id = event_id
self.body_text = body_text
self.fields = fields
self.line_number = line_number
self._init_postprocess()
def _init_postprocess(self):
if getattr(self, 'fields', None) is None:
self.set_fields_from_text(self.body_text)
if self.record_type in ['AVC', 'USER_AVC', "1400", "1107"]:
if 'seresult' not in self.fields:
match = AuditRecord.avc_re.search(self.body_text)
if match:
seresult = match.group(1)
self.fields['seresult'] = seresult
seperms = match.group(2)
self.fields['seperms'] = seperms.split()
def __str__(self):
return self.to_host_text()
def audispd_rectify(self):
self.line_number = None
if self.event_id.host is None:
self.event_id.host = get_hostname()
def is_valid(self):
if not self.event_id.is_valid():
return False
if self.record_type is None:
return False
if self.message is None:
return False
return True
def decode_fields(self):
encoded_fields = ['acct', 'cmd', 'comm', 'cwd', 'data', 'dir', 'exe',
'file', 'host', 'key', 'msg', 'name', 'new', 'ocomm'
'old', 'path', 'watch']
for field in encoded_fields:
if field in self.fields:
if self.record_type == 'AVC' and field == 'saddr':
continue
value = self.fields[field]
decoded_value = audit_msg_decode(value)
self.fields[field] = decoded_value
if self.record_type == 'EXECVE':
for field, value in list(self.fields.items()):
if self.exec_arg_re.search(field):
value = self.fields[field]
decoded_value = audit_msg_decode(value)
self.fields[field] = decoded_value
def translate_hex(self, path):
try:
if sys.version_info[0] < 3:
# Produces normal string instead of unicode string which is not
# accepted by libselinux functions.
# This means that len(path) will return inaccurate results when
# the string contains special characters. Also individual bytes
# of path may not be printable.
return path.decode('hex')
else:
# produces str in python 3 and unicode string in python 2
return bytearray.fromhex(path).decode('utf-8')
except:
return path
def set_fields_from_text(self, body_text):
self.fields_ord = []
self.fields = {}
for match in AuditRecord.key_value_pair_re.finditer(body_text):
key = match.group(1)
value = match.group(2)
value = value.strip('"')
try:
if key == "arch":
i = audit.audit_elf_to_machine(int(value, 16))
value = audit.audit_machine_to_name(i)
if key in ["name", "path", "comm", "cmd", "exe", "cwd"]:
# audit uses " to distinguish plain text from hex in listed keys
if not match.group(2).startswith('"'):
value = self.translate_hex(value)
if key == "exit":
try:
value = errno.errorcode[abs(int(value))]
except:
pass
if key == "syscall":
syscall_name = audit.audit_syscall_to_name(int(value), audit.audit_detect_machine())
if syscall_name:
value = syscall_name
except ValueError:
pass
self.fields[key] = value
self.fields_ord.append(key)
def get_field(self, name):
return self.fields.get(name)
def get_binary_header(self, msg):
msg_length = len(msg)
return struct.pack(AuditRecord.binary_header_format, AuditRecord.binary_version,
AuditRecord.binary_header_size, self.record_type, msg_length)
def fields_to_text(self):
if self.fields is None:
return ''
if self.record_type == 'AVC':
buf = "type=%s msg=%s: avc: denied { %s } " % (self.record_type, self.event_id, ' '.join(self.access))
else:
buf = "type=%s msg=%s: " % (self.record_type, self.event_id)
buf += ' '.join(["%s=%s" % (k, self.fields[k]) for k in self.fields_ord]) + "\n"
return buf
def to_text(self):
return "type=%s msg=%s: %s\n" % (self.record_type, self.event_id, self.body_text)
def to_host_text(self):
if self.event_id.host is not None:
return "node=%s type=%s msg=%s: %s\n" % \
(self.event_id.host, self.record_type, self.event_id, self.body_text)
else:
return self.to_text()
def to_binary(self):
record = "%s: %s" % (self.event_id, self.body_text)
return self.get_binary_header(record) + record
#-----------------------------------------------------------------------------
class AuditRecordReader:
BINARY_FORMAT = 1
TEXT_FORMAT = 2
def __init__(self, record_format):
self.record_format = record_format
self._input_buffer = ''
self.line_number = 0
if self.record_format == self.TEXT_FORMAT:
self.feed = self.feed_text
elif self.record_format == self.BINARY_FORMAT:
self.feed = self.feed_binary
else:
raise ValueError("unknown record format (%s) in %s" % (record_format, self.__class__.__name__))
def feed_binary(self, new_data):
if len(new_data) <= 0:
return
self._input_buffer += new_data
# Now process as much of the buffer as we can, iterating over complete
# messages.
while True:
# To read a complete message there must be a complete header and
# all the data the header specified via the header.length
if len(self._input_buffer) < AuditRecord.binary_header_size:
return
binary_version, binary_header_size, record_type, msg_length = \
struct.unpack(AuditRecord.binary_header_format,
self._input_buffer[0:AuditRecord.binary_header_size])
total_len = AuditRecord.binary_header_size + msg_length
if len(self._input_buffer) < total_len:
return
text = self._input_buffer[AuditRecord.binary_header_size:total_len]
parse_succeeded, event_id, body_text = parse_audit_binary_text(text)
self._input_buffer = self._input_buffer[total_len:]
if parse_succeeded:
yield (audit.audit_msg_type_to_name(record_type), event_id, body_text, None, 0)
return
def feed_text(self, new_data):
if len(new_data) <= 0:
return
self._input_buffer += new_data
# Now process as much of the buffer as we can, iterating over complete
# messages.
# To read a complete message we must see a line ending
start = 0
end = self._input_buffer.find('\n', start)
while end >= 0:
self.line_number += 1
end += 1 # include newline
line = self._input_buffer[start:end]
parse_succeeded, record_type, event_id, body_text = parse_audit_record_text(line)
if parse_succeeded:
yield (record_type, event_id, body_text, None, self.line_number)
start = end
end = self._input_buffer.find('\n', start)
self._input_buffer = self._input_buffer[start:]
return
#-----------------------------------------------------------------------------
class AuditEvent(XmlSerialize):
_xml_info = {
'records': {'XMLForm': 'element', 'list': 'audit_record', 'import_typecast': AuditRecord, },
'event_id': {'XMLForm': 'element', 'import_typecast': AuditEventID},
}
def __init__(self):
super(AuditEvent, self).__init__()
self.event_id = None
self.records = []
self.record_types = {}
self.timestamp = None
def _init_postprocess(self):
if getattr(self, 'record_types', None) is None:
self.record_types = {}
for record in self.records:
self.process_record(record)
def __str__(self):
line_numbers = self.line_numbers
line_numbers.sort()
return "%s: is_avc=%s, is_granted=%s: line_numbers=[%s]\n%s" % \
(self.event_id, self.is_avc(), self.is_granted(),
",".join([str(x) for x in line_numbers]),
"\n".join([" %s" % record for record in self.records]))
def format(self, separator='\n'):
return separator.join([str(record) for record in self.records])
def num_records(self):
return len(self.records)
line_numbers = property(lambda self: [record.line_number for record in self.records if record.line_number])
def add_record(self, record):
self.records.append(record)
self.process_record(record)
def process_record(self, record):
if self.event_id is None:
self.event_id = record.event_id.copy()
self.timestamp = float(self.event_id.seconds) + (self.event_id.milli / 1000.0)
else:
if not self.event_id == record.event_id:
raise ValueError("cannot add audit record to audit event, event_id mismatch %s != %s" %
(self.event_id, record.event_id))
record_list = self.record_types.setdefault(record.record_type, [])
record_list.append(record)
def get_field(self, name, record_type=None):
'''Return list of (value, record_type) tuples.
In other words return the value matching name for every record_type.
If record_type is not specified then all records are searched.
Note: it is possible to have more than one record of a given type
thus it is always possible to have multiple values returned.'''
items = []
if record_type is None:
records = self.records
else:
records = self.get_records_of_type(record_type)
for record in records:
value = record.fields.get(name)
if value is None:
continue
items.append((value, record.type))
return items
def get_record_of_type(self, type):
record = None
records = self.record_types.get(type)
if records:
record = records[0]
return record
def get_records_of_type(self, type):
return self.record_types.get(type, [])
def get_avc_record(self):
for record_type in ['AVC', 'USER_AVC', "1400", "1107"]:
record = self.get_record_of_type(record_type)
if (record):
return record
def is_avc(self):
return self.get_avc_record() is not None
def is_granted(self):
avc_record = self.get_avc_record()
if avc_record is None:
return False
seresult = avc_record.fields['seresult']
if seresult == 'denied':
return False
if seresult == 'granted':
return True
log.avc.warn("unknown value for seresult ('%s')", seresult)
return False
#------------------------------------------------------------------------------
class AVC:
# These are the perm sets from the reference policy for file, dirs, and filesystems.
# They are here to be used below in the access matching functions.
stat_file_perms = ['getattr']
x_file_perms = ['getattr', 'execute']
r_file_perms = ['open', 'read', 'getattr', 'lock', 'ioctl']
rx_file_perms = ['open', 'read', 'getattr', 'lock', 'execute', 'ioctl']
ra_file_perms = ['open', 'ioctl', 'read', 'getattr', 'lock', 'append']
link_file_perms = ['getattr', 'link', 'unlink', 'rename']
create_lnk_perms = ['create', 'read', 'getattr', 'setattr', 'link', 'unlink', 'rename']
create_file_perms = ['open', 'create', 'ioctl', 'read', 'getattr', 'lock', 'write', 'setattr', 'append',
'link', 'unlink', 'rename']
r_dir_perms = ['open', 'read', 'getattr', 'lock', 'search', 'ioctl']
rw_dir_perms = ['open', 'read', 'getattr', 'lock', 'search', 'ioctl', 'add_name', 'remove_name', 'write']
ra_dir_perms = ['open', 'read', 'getattr', 'lock', 'search', 'ioctl', 'add_name', 'write']
create_dir_perms = ['open', 'create', 'read', 'getattr', 'lock', 'setattr', 'ioctl', 'link', 'unlink',
'rename', 'search', 'add_name', 'remove_name', 'reparent', 'write', 'rmdir']
mount_fs_perms = ['mount', 'remount', 'unmount', 'getattr']
search_dir_perms = ['getattr', 'search']
getattr_dir_perms = ['getattr']
setattr_dir_perms = ['setattr']
list_dir_perms = ['open', 'getattr', 'search', 'read', 'lock', 'ioctl']
add_entry_dir_perms = ['open', 'getattr', 'search', 'lock', 'ioctl', 'write', 'add_name']
del_entry_dir_perms = ['open', 'getattr', 'search', 'lock', 'ioctl', 'write', 'remove_name']
manage_dir_perms = ['open', 'create', 'getattr', 'setattr', 'read', 'write', 'link', 'unlink', 'rename',
'search', 'add_name', 'remove_name', 'reparent', 'rmdir', 'lock', 'ioctl']
getattr_file_perms = ['getattr']
setattr_file_perms = ['setattr']
read_file_perms = ['open', 'getattr', 'read', 'lock', 'ioctl']
append_file_perms = ['open', 'getattr', 'append', 'lock', 'ioctl']
write_file_perms = ['open', 'getattr', 'write', 'append', 'lock', 'ioctl']
rw_file_perms = ['open', 'getattr', 'read', 'write', 'append', 'ioctl', 'lock']
delete_file_perms = ['getattr', 'unlink']
manage_file_perms = ['open', 'create', 'getattr', 'setattr', 'read', 'write', 'append', 'rename', 'link',
'unlink', 'ioctl', 'lock']
pipe_instance_path_re = re.compile(r'^(\w+):\[([^\]]*)\]')
proc_pid_instance_re = re.compile(r'^(/proc/)(\d+)(.*)')
def __init__(self, audit_event, query_environment=True):
self.audit_event = audit_event
self.query_environment = query_environment
# if audit_event.timestamp is None:
# self.audit_event.timestamp = TimeStamp()
self.template_substitutions = {}
self.tpath = None
self.spath = None
self.source = None
self.source_pkg = None
self.access = None
self.scontext = None
self.tcontext = None
self.tclass = None
self.port = None
self.src_rpms = []
self.tgt_rpms = []
self.host = None
self.pid = None
self.kmod = None
self.syscall = None
self.why = None
self.bools = []
self.derive_avc_info_from_audit_event()
def __str__(self):
return self.format_avc()
def format_avc(self):
text = ''
text += 'scontext=%s ' % self.scontext
text += 'tcontext=%s ' % self.tcontext
text += 'access=%s ' % self.access
text += 'tclass=%s ' % self.tclass
text += 'tpath=%s ' % self.tpath
return text
# Below are helper functions to get values that might be
# stored in one or more fields in an AVC.
def has_any_access_in(self, access_list):
'Returns true if the AVC contains _any_ of the permissions in the access list.'
if self.access is None:
return False
for a in self.access:
if a in access_list:
return True
return False
def all_accesses_are_in(self, access_list):
"""Returns true if _every_ access in the AVC matches at
least one of the permissions in the access list."""
if self.access is None:
return False
for a in self.access:
if a not in access_list:
return False
return True
def allowed_target_types(self):
all_types = get_all_file_types() + get_all_port_types()
all_types.sort()
all_attributes = get_all_attributes()
all_attributes.sort()
allowed_types = []
wtypes = [x[TARGET] for x in [y for y in search([ALLOW], {SOURCE: self.scontext.type, CLASS: self.tclass, PERMS: self.access}) if y["enabled"]]]
types = wtypes
for t in types:
if t in all_attributes:
wtypes.extend(next(info(ATTRIBUTE, t))["types"])
for t in wtypes:
if t in all_types:
if t not in allowed_types:
allowed_types.append(t)
allowed_types.sort()
return allowed_types
def open_with_write(self):
if self.has_any_access_in(['open']):
try:
if self.a1 and (int(self.a1) & O_ACCMODE) != os.O_RDONLY:
return True
except:
pass
return False
def __typeMatch(self, context, type_list):
for type in type_list:
if re.match(type, context.type):
return True
return False
def matches_source_types(self, type_list):
"""Returns true if the type in the source context of the
avc regular expression matches any of the types in the type list."""
if self.scontext is None:
return False
return self.__typeMatch(self.scontext, type_list)
def matches_target_types(self, type_list):
"""Returns true if the type in the target context of the
avc regular expression matches any of the types in the type list."""
if self.tcontext is None:
return False
return self.__typeMatch(self.tcontext, type_list)
def has_tclass_in(self, tclass_list):
if self.tclass is None:
return False
return self.tclass in tclass_list
def update(self):
self.derive_environmental_info()
self.update_derived_template_substitutions()
def path_is_not_standard_directory(self):
if self.tpath is None:
return True
return self.tpath not in standard_directories
def _set_tpath(self):
'''Derive the target path.
If path information is available the avc record will have a path field
and no name field because the path field is more specific and supercedes
name. The name field is typically the directory entry.
For some special files the kernel embeds instance information
into the file name. For example 'pipe:[1234]' or 'socket:[1234]'
where the number inside the brackets is the inode number. The proc
pseudo file system has the process pid embedded in the name, for
example '/proc/1234/mem'. These numbers are ephemeral and do not
contribute meaningful information for our reports. Plus we may use
the path information to decide if an alert is identical to a
previous alert, we coalesce them if they are. The presence of an
instance specific number in the path confuses this comparison.
For these reasons we strip any instance information out of the
path,
Example input and output:
pipe:[1234] --> pipe
socket:[1234] --> socket
/proc/1234/fd --> /proc/<pid>/fd
./foo --> ./foo
/etc/sysconfig --> /etc/sysconfig
'''
path = None
name = self.avc_record.get_field('name')
# First try to get the path from the AVC record, new kernel
# versions put it there rather than in AVC_PATH
path = self.avc_record.get_field('path')
inodestr = self.avc_record.get_field("ino")
if path is None:
# No path field in AVC record, try to get path from PATH records
avc_path_records = self.audit_event.get_records_of_type('PATH')
for avc_path_record in avc_path_records:
record_type = avc_path_record.get_field('nametype')
#Avoid PARENT records if possible
if path and record_type == "PARENT":
continue
path = avc_path_record.get_field('name')
# If there is more non-PARENT PATH records, use the one matching
# the name from AVC record... otherwise use the last one
if path and name and record_type != "PARENT" and path.endswith(name):
break
if path is None:
# No path field, so try and use the name field instead
name = self.avc_record.get_field('name')
if name is not None:
# Use the class to be smart about formatting the name
tclass = self.avc_record.get_field('tclass')
if tclass == 'file':
# file name is not a full path so make it appear relative
path = '%s' % name
elif tclass == 'dir':
# directory component is not a full path so make it appear
# relative, but only if it's not the root
if name == '/':
path = name
else:
path = '%s' % name
else:
# just use the bare name
path = name
if path is not None:
if path == "/" and inodestr:
matches = []
try:
dev_rdev = 0
dev = self.avc_record.get_field('dev')
if os.path.exists("/dev/" + dev):
dev_rdev = os.lstat("/dev/" + dev).st_rdev
ino = int(inodestr)
fd = open("/proc/mounts", "r")
for i in fd.read().split("\n"):
x = i.split()
if len(x) and x[1][0] == '/':
try:
if (dev_rdev == 0 or os.stat(x[0]).st_rdev == dev_rdev) and int(os.lstat(x[1]).st_ino) == ino:
matches.append(x[:3])
except OSError:
continue
fd.close()
if len(matches) == 1:
path = matches[0][1]
elif len(matches) > 1:
for i in matches:
if i[0] == ("/dev/%s" % dev) or i[2] == dev:
path = i[1]
break
else:
try:
if dev_rdev != 0 and os.lstat(i[0]).st_rdev == dev_rdev:
path = i[1]
break
except OSError:
pass
except TypeError:
path = "unknown mountpoint"
pass
except OSError:
path = "unknown mountpoint"
pass
else:
if path.startswith("/") == False and inodestr:
import subprocess
command = ["locate", "-b", r"\%s" % path]
try:
output = subprocess.check_output(command,
stderr=subprocess.STDOUT,
universal_newlines=True)
ino = int(inodestr)
for file in output.split("\n"):
try:
if int(os.lstat(file).st_ino) == ino:
path = file
break
except:
pass
except:
pass
if path is not None:
if path.startswith('/'):
# Fully qualified path
# map /proc/1234/ to /proc/<pid>, replacing numeric pid with <pid>
path = self.proc_pid_instance_re.sub(r'\1<pid>\3', path)
else:
# map pipe:[1234] to pipe, stripping out inode instance (e.g. [1234])
# applies to socket as well
match = self.pipe_instance_path_re.search(path)
if match:
path = self.tclass
# abstract socket paths start with '\0'
if path[0] == '\0':
path = "@" + path.strip('\0')
self.tpath = path
if self.tpath is None:
if self.tclass == "filesystem":
self.tpath = ""
elif self.tclass == "udp_socket" or self.tclass == "tcp_socket":
self.tpath = _("port %s") % self.port
else:
self.tpath = _("Unknown")
def derive_avc_info_from_audit_event(self):
self.tpath = None
self.spath = None
self.source = None
self.a1 = None
self.success = False
self.syscall_paths = []
exe = comm = arch = syscall = None
self.avc_record = self.audit_event.get_avc_record()
syscall_record = self.audit_event.get_record_of_type('SYSCALL')
self.access = self.avc_record.get_field('seperms')
if not isinstance(self.scontext, AvcContext):
self.scontext = AvcContext(self.avc_record.get_field('scontext'))
if not isinstance(self.tcontext, AvcContext):
self.tcontext = AvcContext(self.avc_record.get_field('tcontext'))
self.tclass = self.avc_record.get_field('tclass')
if self.avc_record.get_field('dest') is None:
self.port = self.avc_record.get_field('src')
else:
self.port = self.avc_record.get_field('dest')
self._set_tpath()
self.kmod = self.avc_record.get_field('kmod')
self.pid = self.avc_record.get_field('pid')
# exe, cwd, name, path, key, dir, comm, ocomm, key_desc
if syscall_record:
exe = syscall_record.get_field('exe')
comm = syscall_record.get_field('comm')
self.syscall = syscall_record.get_field('syscall')
self.success = (syscall_record.get_field('success') == "yes")
self.a1 = syscall_record.get_field('a1')
if comm is None:
comm = self.avc_record.get_field('comm')
if exe is None:
exe = self.avc_record.get_field('exe')
self.spath = exe
if comm:
self.source = comm
elif exe:
self.source = self.spath
if not self.spath:
self.spath = self.source
if not self.spath:
self.spath = self.scontext.type
cwd_record = self.audit_event.get_record_of_type('CWD')
if cwd_record:
cwd = cwd_record.get_field('cwd')
else:
cwd = None
path_records = self.audit_event.get_records_of_type('PATH')
for path_record in path_records:
path = path_record.get_field('name')
if os.path.isabs(path) or not cwd:
self.syscall_paths.append(path)
else:
self.syscall_paths.append(os.path.join(cwd, path))
self.src_rpms = []
self.tgt_rpms = []
self.host = self.audit_event.event_id.host
self.why, bools = audit2why.analyze(str(self.scontext), str(self.tcontext), str(self.tclass), self.access)
if self.why == audit2why.ALLOW:
raise ValueError(_("%s \n**** Recorded AVC is allowed in current policy ****\n") % self.avc_record)
if self.why == audit2why.DONTAUDIT:
raise ValueError(_("%s \n**** Recorded AVC is dontaudited in current policy. 'semodule -B' will turn on dontaudit rules ****\n") % self.avc_record)
if self.why == audit2why.NOPOLICY:
raise ValueError(_("Must call policy_init first"))
if self.why == audit2why.BADTCON:
raise ValueError(_("%s \n**** Invalid AVC: bad target context ****\n") % self.avc_record)
if self.why == audit2why.BADSCON:
raise ValueError(_("%s \n**** Invalid AVC: bad source context ****\n") % self.avc_record)
if self.why == audit2why.BADTCLASS:
raise ValueError(_("%s \n**** Invalid AVC: bad type class ****\n") % self.avc_record)
if self.why == audit2why.BADPERM:
raise ValueError(_("%s \n**** Invalid AVC: bad permission ****\n") % self.avc_record)
if self.why == audit2why.BADCOMPUTE:
raise ValueError(_("Error during access vector computation"))
if self.why == audit2why.BOOLEAN:
self.bools = bools
def derive_environmental_info(self):
if self.query_environment:
if self.spath:
self.source_pkg = get_package_nvr_by_file_path(self.spath)
if self.source_pkg:
self.src_rpms.append(self.source_pkg)
if self.tpath:
rpm = get_package_nvr_by_file_path(self.tpath)
if rpm:
self.tgt_rpms.append(rpm)
def set_alt_path(self, path):
if self.tpath is None:
self.tpath = path
def set_template_substitutions(self, **kwds):
for key, value in list(kwds.items()):
if value:
self.template_substitutions[key] = value
def update_derived_template_substitutions(self):
self.template_substitutions["SOURCE_TYPE"] = escape_html(self.scontext.type)
self.template_substitutions["TARGET_TYPE"] = escape_html(self.tcontext.type)
self.template_substitutions["SOURCE"] = escape_html(self.source)
self.template_substitutions["SOURCE_PATH"] = escape_html(self.spath)
if self.spath:
self.template_substitutions["FIX_SOURCE_PATH"] = re.sub(" ", ".", escape_html(self.spath))
self.template_substitutions["TARGET_PATH"] = escape_html(self.tpath)
if self.tpath:
self.template_substitutions["FIX_TARGET_PATH"] = re.sub(" ", ".", escape_html(self.tpath))
if self.tpath is None:
self.template_substitutions["TARGET_DIR"] = None
else:
if self.tclass == 'dir':
self.template_substitutions["TARGET_DIR"] = escape_html(self.tpath)
elif self.tclass == 'file':
self.template_substitutions["TARGET_DIR"] = escape_html(os.path.dirname(self.tpath))
else:
self.template_substitutions["TARGET_DIR"] = None
self.template_substitutions["TARGET_CLASS"] = escape_html(self.tclass)
if self.access is None:
self.template_substitutions["ACCESS"] = None
else:
self.template_substitutions["ACCESS"] = escape_html(' '.join(self.access))
self.template_substitutions["SOURCE_PACKAGE"] = escape_html(self.source_pkg)
self.template_substitutions["PORT_NUMBER"] = escape_html(self.port)
def validate_template_substitutions(self):
# validate, replace any None values with friendly string
for key, value in list(self.template_substitutions.items()):
if value is None:
self.template_substitutions[key] = escape_html(default_text(value))