def _AddDiversions(self, chroot_path):
for source, dest in self._DIVERSIONS.items():
self._ExecChroot(
chroot_path,
'dpkg-divert',
'--local',
'--rename',
'--add',
source)
self._ExecChroot(
chroot_path,
'ln',
'--symbolic',
'--force',
dest,
source)
with open(os.path.join(chroot_path, 'usr', 'sbin', 'policy-rc.d'), 'w') as fh:
fh.write('#!/bin/sh\n')
fh.write('exit 101\n')
os.fchmod(fh.fileno(), 0o744)
python类fchmod()的实例源码
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_data(fpath, data, modified, raise_err=True, tmp_dir=None):
"""Safely write data to file path.
"""
tmp_dir = tmp_dir or os.path.dirname(fpath)
with tempfile.NamedTemporaryFile(dir=tmp_dir,
delete=False,
prefix='.tmp') as temp:
if data:
temp.write(data)
os.fchmod(temp.fileno(), 0o644)
os.utime(temp.name, (modified, modified))
try:
os.rename(temp.name, fpath)
except OSError:
_LOGGER.error('Unable to rename: %s => %s', temp.name, fpath,
exc_info=True)
if raise_err:
raise
def create_script(filename, templatename, mode=EXEC_MODE, **kwargs):
"""This Creates a file from a JINJA template.
The templates exist in our lib/python/treadmill/templates directory.
:param ``str`` filename:
Name of the file to generate.
:param ``str`` templatename:
The name of the template file.
:param ``int`` mode:
The mode for the file (Defaults to +x).
:param ``dict`` kwargs:
key/value passed into the template.
"""
filepath = os.path.dirname(filename)
with tempfile.NamedTemporaryFile(dir=filepath, delete=False) as f:
for data in generate_template(templatename, **kwargs):
f.write(data)
if os.name == 'posix':
# cast to int required in order for default EXEC_MODE to work
os.fchmod(f.fileno(), int(mode))
os.rename(f.name, filename)
def set_list_write(filename, entries):
"""Write a list of values to a file. One per line.
:param ``str`` filename:
Name of the file to read.
:param ``set`` entries:
Set of values to write into ``filename``. Value can be unicode.
"""
values = {
entry.encode(encoding='utf8', errors='replace')
for entry in entries
}
with io.open(filename, 'wb') as f:
f.writelines(values)
if os.name == 'posix':
os.fchmod(f.fileno(), 0o644)
def script_write(filename, script):
"""Write a script to a file.
Proper execute permissions will be set.
:param ``str`` filename:
File to write to.
:param ``script:
String or iterable returning strings. Can be unicode.
"""
if isinstance(script, six.string_types):
# If the script is fully provided in a string, wrap it in a StringIO
if hasattr(script, 'decode'):
script = io.StringIO(script.decode())
else:
script = io.StringIO(script)
with io.open(filename, 'wb') as f:
for chunk in script:
# The value must be properly encoded
data = chunk.encode(encoding='utf8', errors='replace')
f.write(data)
if os.name == 'posix':
os.fchmod(f.fileno(), 0o755)
def safe_replacement(path, *open_args, mode=None, **open_kws):
path = str(path)
if mode is None:
try: mode = stat.S_IMODE(os.lstat(path).st_mode)
except OSError: pass
open_kws.update( delete=False,
dir=os.path.dirname(path), prefix=os.path.basename(path)+'.' )
if not open_args: open_kws['mode'] = 'w'
with tempfile.NamedTemporaryFile(*open_args, **open_kws) as tmp:
try:
if mode is not None: os.fchmod(tmp.fileno(), mode)
yield tmp
if not tmp.closed: tmp.flush()
os.rename(tmp.name, path)
finally:
try: os.unlink(tmp.name)
except OSError: pass
def submit(config, user, run_id, pids):
"""
Submits pipeline defined by 'config' as user 'user'.
Dumps the config in a temp. file that is removed after succesful completion.
Returns exit code, stdout, and stderr.
"""
pids[run_id] = mp.current_process().pid
(fd, tmp_cfg) = tempfile.mkstemp(prefix='pypers_', suffix='.cfg', text=True)
os.fchmod(fd, 0644)
with os.fdopen(fd, 'w') as fh:
json.dump(config, fh)
cmd = [which('np_submit.py'), '-i', tmp_cfg]
(ec, err, out) = run_as(cmd=cmd, user=user)
if ec == 0:
os.unlink(tmp_cfg)
return (err, out)
else:
raise Exception('Unable to execute cmd %s:\n%s\n%s' % (cmd, err, out))
def write_homepage(self, status_info):
"""
Re-writes the site homepage using the provided statistics in the
homepage template (which is effectively a simple Python format string).
:param tuple status_info:
A namedtuple containing statistics obtained by :class:`BigBrother`.
"""
self.logger.info('writing homepage')
with tempfile.NamedTemporaryFile(mode='w', dir=str(self.output_path),
delete=False) as index:
try:
index.file.write(self.homepage_template.format(**status_info))
except BaseException:
index.delete = True
raise
else:
os.fchmod(index.file.fileno(), 0o664)
os.replace(index.name, str(self.output_path / 'index.html'))
def _create_tempfile(self, pathname): # pragma: no cover
path = self.join(pathname)
dirname = os.path.dirname(path)
if not os.path.exists(dirname):
tracing.trace('os.makedirs(%s)' % dirname)
try:
os.makedirs(dirname, mode=obnamlib.NEW_DIR_MODE)
except OSError as e: # pragma: no cover
# This avoids a race condition: another Obnam process
# may have created the directory between our check and
# creation attempt. If so, we ignore it. As long as
# the directory exists, all's good.
if e.errno != errno.EEXIST:
raise
fd, tempname = tempfile.mkstemp(dir=dirname)
os.fchmod(fd, obnamlib.NEW_FILE_MODE)
os.close(fd)
f = self.open(tempname, 'wb')
f.close()
return tempname
def handle_begin_put(req_id, path, mode):
prev_umask = os.umask(0o077)
try:
if path is None:
f = tempfile.NamedTemporaryFile(delete=False)
path = wpath = f.name
else:
path = force_str(path)
if os.path.isdir(path):
raise IOError('%s is a directory' % path)
wpath = path + '~chopsticks-tmp'
f = open(wpath, 'wb')
finally:
os.umask(prev_umask)
os.fchmod(f.fileno(), mode)
active_puts[req_id] = (f, wpath, path, sha1())
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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 store_context(self, file_name, config_data):
if not os.path.isabs(file_name):
file_name = os.path.join(hookenv.charm_dir(), file_name)
with open(file_name, 'w') as file_stream:
os.fchmod(file_stream.fileno(), 0o600)
yaml.dump(config_data, file_stream)
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)
