def _test_all_chown_common(self, chown_func, first_param):
"""Common code for chown, fchown and lchown tests."""
if os.getuid() == 0:
try:
# Many linux distros have a nfsnobody user as MAX_UID-2
# that makes a good test case for signedness issues.
# http://bugs.python.org/issue1747858
# This part of the test only runs when run as root.
# Only scary people run their tests as root.
ent = pwd.getpwnam('nfsnobody')
chown_func(first_param, ent.pw_uid, ent.pw_gid)
except KeyError:
pass
else:
# non-root cannot chown to root, raises OSError
self.assertRaises(OSError, chown_func,
first_param, 0, 0)
# test a successful chown call
chown_func(first_param, os.getuid(), os.getgid())
python类fchown()的实例源码
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def writefile(path, data, owner=None, group=None, mode=None):
if isinstance(data, bytes):
openmode = 'wb'
else:
openmode = 'w'
if owner is not None:
uid = pwd.getpwnam(owner).pw_uid
else:
uid = -1
if group is not None:
gid = grp.getgrnam(group).gr_gid
else:
gid = -1
with open(path, openmode) as f:
os.fchown(
f.fileno(),
uid,
gid
)
if mode is not None:
os.fchmod(f.fileno(), mode)
return f.write(data)
def write(self, path=None):
"""Writes the ticket to /var/spool/ticket/<princ>."""
def _write_temp(tkt_file):
# Write the file
tkt_file.write(self.ticket)
# Set the owner
if self.uid is not None:
os.fchown(tkt_file.fileno(), self.uid, -1)
# TODO: Should we enforce the mode too?
tkt_file.flush()
if path is None:
path = self.tkt_path
try:
fs.write_safe(
path,
_write_temp,
prefix='.tmp' + self.princ
)
except (IOError, OSError):
_LOGGER.exception('Error writing ticket file: %s', path)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms))
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
target.write(content)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def restore(cls, zip_archive, **kwargs):
fd, path = tempfile.mkstemp()
try:
uid = pwd.getpwnam("postgres").pw_uid
os.fchown(fd, uid, -1)
with os.fdopen(fd, 'w') as tmp:
shutil.copyfileobj(zip_archive.open('postgresql.backup'), tmp)
tmp.seek(0)
pg_restore(path)
finally:
os.remove(path)
return path
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)
def write_file(path, content, owner='root', group='root', perms=0o444):
"""Create or overwrite a file with the contents of a byte string."""
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
# lets see if we can grab the file and compare the context, to avoid doing
# a write.
existing_content = None
existing_uid, existing_gid = None, None
try:
with open(path, 'rb') as target:
existing_content = target.read()
stat = os.stat(path)
existing_uid, existing_gid = stat.st_uid, stat.st_gid
except:
pass
if content != existing_content:
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms),
level=DEBUG)
with open(path, 'wb') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
if six.PY3 and isinstance(content, six.string_types):
content = content.encode('UTF-8')
target.write(content)
return
# the contents were the same, but we might still need to change the
# ownership.
if existing_uid != uid:
log("Changing uid on already existing content: {} -> {}"
.format(existing_uid, uid), level=DEBUG)
os.chown(path, uid, -1)
if existing_gid != gid:
log("Changing gid on already existing content: {} -> {}"
.format(existing_gid, gid), level=DEBUG)
os.chown(path, -1, gid)