1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
community.general/library/file
Michael DeHaan 7e9e29011e Add encoding lines to python modules such that they can take unicode options if they are fed them, since the
AnsibleModule stuff no longer base64 encodes for simplicity and speed reasons.
2012-08-02 21:29:10 -04:00

354 lines
11 KiB
Python
Executable file

#!/usr/bin/python
# -*- coding: utf-8 -*-
# (c) 2012, Michael DeHaan <michael.dehaan@gmail.com>
#
# This file is part of Ansible
#
# Ansible 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 3 of the License, or
# (at your option) any later version.
#
# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
import shutil
import stat
import grp
import pwd
try:
import selinux
HAVE_SELINUX=True
except ImportError:
HAVE_SELINUX=False
def add_path_info(kwargs):
path = kwargs['path']
if os.path.exists(path):
(user, group) = user_and_group(path)
kwargs['owner'] = user
kwargs['group'] = group
st = os.stat(path)
kwargs['mode'] = oct(stat.S_IMODE(st[stat.ST_MODE]))
# secontext not yet supported
if os.path.islink(path):
kwargs['state'] = 'link'
elif os.path.isfile(path):
kwargs['state'] = 'file'
else:
kwargs['state'] = 'directory'
if HAVE_SELINUX and selinux_enabled():
kwargs['secontext'] = ':'.join(selinux_context(path))
else:
kwargs['state'] = 'absent'
return kwargs
def module_exit_json(**kwargs):
add_path_info(kwargs)
module.exit_json(**kwargs)
def module_fail_json(**kwargs):
add_path_info(kwargs)
module.fail_json(**kwargs)
# Detect whether using selinux that is MLS-aware.
# While this means you can set the level/range with
# selinux.lsetfilecon(), it may or may not mean that you
# will get the selevel as part of the context returned
# by selinux.lgetfilecon().
def selinux_mls_enabled():
if not HAVE_SELINUX:
return False
if selinux.is_selinux_mls_enabled() == 1:
return True
else:
return False
def selinux_enabled():
if not HAVE_SELINUX:
return False
if selinux.is_selinux_enabled() == 1:
return True
else:
return False
# Determine whether we need a placeholder for selevel/mls
def selinux_initial_context():
context = [None, None, None]
if selinux_mls_enabled():
context.append(None)
return context
# If selinux fails to find a default, return an array of None
def selinux_default_context(path, mode=0):
context = selinux_initial_context()
if not HAVE_SELINUX or not selinux_enabled():
return context
try:
ret = selinux.matchpathcon(path, mode)
except OSError:
return context
if ret[0] == -1:
return context
context = ret[1].split(':')
return context
def selinux_context(path):
context = selinux_initial_context()
if not HAVE_SELINUX or not selinux_enabled():
return context
try:
ret = selinux.lgetfilecon(path)
except:
module_fail_json(path=path, msg='failed to retrieve selinux context')
if ret[0] == -1:
return context
context = ret[1].split(':')
return context
# ===========================================
# ===========================================
# support functions
def user_and_group(filename):
st = os.stat(filename)
uid = st.st_uid
gid = st.st_gid
try:
user = pwd.getpwuid(uid)[0]
except KeyError:
user = str(uid)
try:
group = grp.getgrgid(gid)[0]
except KeyError:
group = str(gid)
return (user, group)
def set_context_if_different(path, context, changed):
if not HAVE_SELINUX or not selinux_enabled():
return changed
cur_context = selinux_context(path)
new_context = list(cur_context)
# Iterate over the current context instead of the
# argument context, which may have selevel.
for i in range(len(cur_context)):
if context[i] is not None and context[i] != cur_context[i]:
new_context[i] = context[i]
if cur_context != new_context:
try:
rc = selinux.lsetfilecon(path, ':'.join(new_context))
except OSError:
module_fail_json(path=path, msg='invalid selinux context')
if rc != 0:
module_fail_json(path=path, msg='set selinux context failed')
changed = True
return changed
def set_owner_if_different(path, owner, changed):
if owner is None:
return changed
user, group = user_and_group(path)
if owner != user:
rc = os.system("/bin/chown -R %s %s 2>/dev/null" % (owner, path))
if rc != 0:
module_fail_json(path=path, msg='chown failed')
return True
return changed
def set_group_if_different(path, group, changed):
if group is None:
return changed
old_user, old_group = user_and_group(path)
if old_group != group:
rc = os.system("/bin/chgrp -R %s %s" % (group, path))
if rc != 0:
module_fail_json(path=path, msg='chgrp failed')
return True
return changed
def set_mode_if_different(path, mode, changed):
if mode is None:
return changed
try:
# FIXME: support English modes
mode = int(mode, 8)
except Exception, e:
module_fail_json(path=path, msg='mode needs to be something octalish', details=str(e))
st = os.stat(path)
prev_mode = stat.S_IMODE(st[stat.ST_MODE])
if prev_mode != mode:
# FIXME: comparison against string above will cause this to be executed
# every time
try:
os.chmod(path, mode)
except Exception, e:
module_fail_json(path=path, msg='chmod failed', details=str(e))
st = os.stat(path)
new_mode = stat.S_IMODE(st[stat.ST_MODE])
if new_mode != prev_mode:
return True
return changed
def rmtree_error(func, path, exc_info):
module_fail_json(path=path, msg='failed to remove directory')
def main():
global module
module = AnsibleModule(
check_invalid_arguments = False,
argument_spec = dict(
state = dict(choices=['file','directory','link','absent'], default='file'),
path = dict(aliases=['dest', 'name'], required=True),
src = dict(),
mode = dict(),
owner = dict(),
group = dict(),
seuser = dict(),
serole = dict(),
selevel = dict(),
setype = dict(),
)
)
params = module.params
state = params['state']
path = os.path.expanduser(params['path'])
src = params.get('src', None)
if src:
src = os.path.expanduser(src)
mode = params.get('mode', None)
owner = params.get('owner', None)
group = params.get('group', None)
# selinux related options
seuser = params.get('seuser', None)
serole = params.get('serole', None)
setype = params.get('setype', None)
selevel = params.get('serange', 's0')
secontext = [seuser, serole, setype]
if selinux_mls_enabled():
secontext.append(selevel)
default_secontext = selinux_default_context(path)
for i in range(len(default_secontext)):
if i is not None and secontext[i] == '_default':
secontext[i] = default_secontext[i]
if state == 'link' and (src is None or path is None):
module_fail_json(msg='src and dest are required for "link" state')
elif path is None:
module_fail_json(msg='path is required')
changed = False
prev_state = 'absent'
if os.path.lexists(path):
if os.path.islink(path):
prev_state = 'link'
elif os.path.isfile(path):
prev_state = 'file'
else:
prev_state = 'directory'
if prev_state != 'absent' and state == 'absent':
try:
if prev_state == 'directory':
if os.path.islink(path):
os.unlink(path)
else:
shutil.rmtree(path, ignore_errors=False, onerror=rmtree_error)
else:
os.unlink(path)
except Exception, e:
module_fail_json(path=path, msg=str(e))
module_exit_json(path=path, changed=True)
if prev_state != 'absent' and prev_state != state:
module_fail_json(path=path, msg='refusing to convert between %s and %s' % (prev_state, state))
if prev_state == 'absent' and state == 'absent':
module_exit_json(path=path, changed=False)
if state == 'file':
if prev_state == 'absent':
module_fail_json(path=path, msg='file does not exist, use copy or template module to create')
# set modes owners and context as needed
changed = set_context_if_different(path, secontext, changed)
changed = set_owner_if_different(path, owner, changed)
changed = set_group_if_different(path, group, changed)
changed = set_mode_if_different(path, mode, changed)
module_exit_json(path=path, changed=changed)
elif state == 'directory':
if prev_state == 'absent':
os.makedirs(path)
changed = True
# set modes owners and context as needed
changed = set_context_if_different(path, secontext, changed)
changed = set_owner_if_different(path, owner, changed)
changed = set_group_if_different(path, group, changed)
changed = set_mode_if_different(path, mode, changed)
module_exit_json(path=path, changed=changed)
elif state == 'link':
if os.path.isabs(src):
abs_src = src
else:
abs_src = os.path.join(os.path.dirname(dest), src)
if not os.path.exists(abs_src):
module_fail_json(path=path, src=src, msg='src file does not exist')
if prev_state == 'absent':
os.symlink(src, path)
changed = True
elif prev_state == 'link':
old_src = os.readlink(path)
if not os.path.isabs(old_src):
old_src = os.path.join(os.path.dirname(path), old_src)
if old_src != src:
os.unlink(path)
os.symlink(src, path)
else:
module_fail_json(dest=path, src=src, msg='unexpected position reached')
# set modes owners and context as needed
changed = set_context_if_different(path, secontext, changed)
changed = set_owner_if_different(path, owner, changed)
changed = set_group_if_different(path, group, changed)
changed = set_mode_if_different(path, mode, changed)
module.exit_json(dest=path, src=src, changed=changed)
module_fail_json(path=path, msg='unexpected position reached')
# this is magic, see lib/ansible/module_common.py
#<<INCLUDE_ANSIBLE_MODULE_COMMON>>
main()