ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

Dealing with invalid symlinks and symlink permissions

Browse source
This commit is contained in:
David Simon 2013-05-09 15:35:05 -04:00
parent 8ef18c2f98
commit 190ce16bab
2 changed files with 23 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
lib/ansible/module_common.py
View file

@ -316,7 +316,7 @@ class AnsibleModule(object):
def user_and_group(self, filename): def user_and_group(self, filename):
filename = os.path.expanduser(filename) filename = os.path.expanduser(filename)
st = os.stat(filename) st = os.lstat(filename)
uid = st.st_uid uid = st.st_uid
gid = st.st_gid gid = st.st_gid
return (uid, gid) return (uid, gid)
@ -370,7 +370,7 @@ class AnsibleModule(object):
return True return True
if orig_uid != uid: if orig_uid != uid:
try: try:
os.chown(path, uid, -1) os.lchown(path, uid, -1)
except OSError: except OSError:
self.fail_json(path=path, msg='chown failed') self.fail_json(path=path, msg='chown failed')
changed = True changed = True
@ -392,7 +392,7 @@ class AnsibleModule(object):
return True return True
if orig_gid != gid: if orig_gid != gid:
try: try:
os.chown(path, -1, gid) os.lchown(path, -1, gid)
except OSError: except OSError:
self.fail_json(path=path, msg='chgrp failed') self.fail_json(path=path, msg='chgrp failed')
changed = True changed = True
@ -409,7 +409,7 @@ class AnsibleModule(object):
except Exception, e: except Exception, e:
self.fail_json(path=path, msg='mode needs to be something octalish', details=str(e)) self.fail_json(path=path, msg='mode needs to be something octalish', details=str(e))
st = os.stat(path) st = os.lstat(path)
prev_mode = stat.S_IMODE(st[stat.ST_MODE]) prev_mode = stat.S_IMODE(st[stat.ST_MODE])
if prev_mode != mode: if prev_mode != mode:
@ -418,11 +418,19 @@ class AnsibleModule(object):
# FIXME: comparison against string above will cause this to be executed # FIXME: comparison against string above will cause this to be executed
# every time # every time
try: try:
if 'lchmod' in dir(os):
os.lchmod(path, mode)
else:
os.chmod(path, mode) os.chmod(path, mode)
except OSError, e:
if e.errno == errno.ENOENT: # Can't set mode on broken symbolic links
pass
else:
raise e
except Exception, e: except Exception, e:
self.fail_json(path=path, msg='chmod failed', details=str(e)) self.fail_json(path=path, msg='chmod failed', details=str(e))
st = os.stat(path) st = os.lstat(path)
new_mode = stat.S_IMODE(st[stat.ST_MODE]) new_mode = stat.S_IMODE(st[stat.ST_MODE])
if new_mode != prev_mode: if new_mode != prev_mode:
@ -483,7 +491,7 @@ class AnsibleModule(object):
group = str(gid) group = str(gid)
kwargs['owner'] = user kwargs['owner'] = user
kwargs['group'] = group kwargs['group'] = group
st = os.stat(path) st = os.lstat(path)
kwargs['mode'] = oct(stat.S_IMODE(st[stat.ST_MODE])) kwargs['mode'] = oct(stat.S_IMODE(st[stat.ST_MODE]))
# secontext not yet supported # secontext not yet supported
if os.path.islink(path): if os.path.islink(path):

8
test/TestRunner.py
View file

@ -230,6 +230,14 @@ class TestRunner(unittest.TestCase):
assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed'] assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
assert not os.path.exists(filedemo) assert not os.path.exists(filedemo)
assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed'] assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
# Make sure that we can deal safely with bad symlinks
os.symlink('/tmp/non_existent_target', filedemo)
assert self._run('file', ['dest=' + tmp_dir, 'state=directory recurse=yes mode=701'])['changed']
assert not self._run('file', ['dest=' + tmp_dir, 'state=directory', 'recurse=yes', 'owner=' + str(os.getuid())])['changed']
assert os.path.islink(filedemo)
assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
assert not os.path.exists(filedemo)
os.rmdir(tmp_dir) os.rmdir(tmp_dir)
def test_large_output(self): def test_large_output(self):