Use of constant for 644 permission variable (#82762)

Signed-off-by: Abhijeet Kasurde <akasurde@redhat.com>
pull/80787/head^2
Abhijeet Kasurde 2 months ago committed by GitHub
parent cdd303101e
commit dd44449b6e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -124,6 +124,7 @@ from ansible.galaxy.dependency_resolution.dataclasses import (
)
from ansible.galaxy.dependency_resolution.versioning import meets_requirements
from ansible.plugins.loader import get_all_plugin_loaders
from ansible.module_utils.common.file import S_IRWU_RG_RO, S_IRWXU_RXG_RXO, S_IXANY
from ansible.module_utils.common.text.converters import to_bytes, to_native, to_text
from ansible.module_utils.common.collections import is_sequence
from ansible.module_utils.common.yaml import yaml_dump
@ -1317,7 +1318,7 @@ def _build_collection_tar(
tar_info = tarfile.TarInfo(name)
tar_info.size = len(b)
tar_info.mtime = int(time.time())
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tar_file.addfile(tarinfo=tar_info, fileobj=b_io)
for file_info in file_manifest['files']: # type: ignore[union-attr]
@ -1331,7 +1332,7 @@ def _build_collection_tar(
def reset_stat(tarinfo):
if tarinfo.type != tarfile.SYMTYPE:
existing_is_exec = tarinfo.mode & stat.S_IXUSR
tarinfo.mode = 0o0755 if existing_is_exec or tarinfo.isdir() else 0o0644
tarinfo.mode = S_IRWXU_RXG_RXO if existing_is_exec or tarinfo.isdir() else S_IRWU_RG_RO
tarinfo.uid = tarinfo.gid = 0
tarinfo.uname = tarinfo.gname = ''
@ -1373,7 +1374,7 @@ def _build_collection_dir(b_collection_path, b_collection_output, collection_man
This should follow the same pattern as _build_collection_tar.
"""
os.makedirs(b_collection_output, mode=0o0755)
os.makedirs(b_collection_output, mode=S_IRWXU_RXG_RXO)
files_manifest_json = to_bytes(json.dumps(file_manifest, indent=True), errors='surrogate_or_strict')
collection_manifest['file_manifest_file']['chksum_sha256'] = secure_hash_s(files_manifest_json, hash_func=sha256)
@ -1385,7 +1386,7 @@ def _build_collection_dir(b_collection_path, b_collection_output, collection_man
with open(b_path, 'wb') as file_obj, BytesIO(b) as b_io:
shutil.copyfileobj(b_io, file_obj)
os.chmod(b_path, 0o0644)
os.chmod(b_path, S_IRWU_RG_RO)
base_directories = []
for file_info in sorted(file_manifest['files'], key=lambda x: x['name']):
@ -1396,11 +1397,11 @@ def _build_collection_dir(b_collection_path, b_collection_output, collection_man
dest_file = os.path.join(b_collection_output, to_bytes(file_info['name'], errors='surrogate_or_strict'))
existing_is_exec = os.stat(src_file, follow_symlinks=False).st_mode & stat.S_IXUSR
mode = 0o0755 if existing_is_exec else 0o0644
mode = S_IRWXU_RXG_RXO if existing_is_exec else S_IRWU_RG_RO
# ensure symlinks to dirs are not translated to empty dirs
if os.path.isdir(src_file) and not os.path.islink(src_file):
mode = 0o0755
mode = S_IRWXU_RXG_RXO
base_directories.append(src_file)
os.mkdir(dest_file, mode)
else:
@ -1549,10 +1550,10 @@ def write_source_metadata(collection, b_collection_path, artifacts_manager):
shutil.rmtree(b_info_dir)
try:
os.mkdir(b_info_dir, mode=0o0755)
os.mkdir(b_info_dir, mode=S_IRWXU_RXG_RXO)
with open(b_info_dest, mode='w+b') as fd:
fd.write(b_yaml_source_data)
os.chmod(b_info_dest, 0o0644)
os.chmod(b_info_dest, S_IRWU_RG_RO)
except Exception:
# Ensure we don't leave the dir behind in case of a failure.
if os.path.isdir(b_info_dir):
@ -1681,7 +1682,7 @@ def _extract_tar_dir(tar, dirname, b_dest):
b_parent_path = os.path.dirname(b_dir_path)
try:
os.makedirs(b_parent_path, mode=0o0755)
os.makedirs(b_parent_path, mode=S_IRWXU_RXG_RXO)
except OSError as e:
if e.errno != errno.EEXIST:
raise
@ -1696,7 +1697,7 @@ def _extract_tar_dir(tar, dirname, b_dest):
else:
if not os.path.isdir(b_dir_path):
os.mkdir(b_dir_path, 0o0755)
os.mkdir(b_dir_path, S_IRWXU_RXG_RXO)
def _extract_tar_file(tar, filename, b_dest, b_temp_path, expected_hash=None):
@ -1722,7 +1723,7 @@ def _extract_tar_file(tar, filename, b_dest, b_temp_path, expected_hash=None):
if not os.path.exists(b_parent_dir):
# Seems like Galaxy does not validate if all file entries have a corresponding dir ftype entry. This check
# makes sure we create the parent directory even if it wasn't set in the metadata.
os.makedirs(b_parent_dir, mode=0o0755)
os.makedirs(b_parent_dir, mode=S_IRWXU_RXG_RXO)
if tar_member.type == tarfile.SYMTYPE:
b_link_path = to_bytes(tar_member.linkname, errors='surrogate_or_strict')
@ -1737,9 +1738,9 @@ def _extract_tar_file(tar, filename, b_dest, b_temp_path, expected_hash=None):
# Default to rw-r--r-- and only add execute if the tar file has execute.
tar_member = tar.getmember(to_native(filename, errors='surrogate_or_strict'))
new_mode = 0o644
new_mode = S_IRWU_RG_RO
if stat.S_IMODE(tar_member.mode) & stat.S_IXUSR:
new_mode |= 0o0111
new_mode |= S_IXANY
os.chmod(b_dest_filepath, new_mode)

@ -120,12 +120,13 @@ from ansible.module_utils.common.locale import get_best_parsable_locale
from ansible.module_utils.common.process import get_bin_path
from ansible.module_utils.common.file import (
_PERM_BITS as PERM_BITS,
_EXEC_PERM_BITS as EXEC_PERM_BITS,
_DEFAULT_PERM as DEFAULT_PERM,
is_executable,
format_attributes,
get_flags_from_attributes,
FILE_ATTRIBUTES,
S_IXANY,
S_IRWU_RWG_RWO,
)
from ansible.module_utils.common.sys_info import (
get_distribution,
@ -1021,7 +1022,7 @@ class AnsibleModule(object):
if prev_mode is None:
prev_mode = stat.S_IMODE(path_stat.st_mode)
is_directory = stat.S_ISDIR(path_stat.st_mode)
has_x_permissions = (prev_mode & EXEC_PERM_BITS) > 0
has_x_permissions = (prev_mode & S_IXANY) > 0
apply_X_permission = is_directory or has_x_permissions
# Get the umask, if the 'user' part is empty, the effect is as if (a) were
@ -1694,7 +1695,7 @@ class AnsibleModule(object):
# based on the current value of umask
umask = os.umask(0)
os.umask(umask)
os.chmod(b_dest, DEFAULT_PERM & ~umask)
os.chmod(b_dest, S_IRWU_RWG_RWO & ~umask)
try:
os.chown(b_dest, os.geteuid(), os.getegid())
except OSError:

@ -44,9 +44,15 @@ USERS_RE = re.compile(r'[^ugo]')
PERMS_RE = re.compile(r'[^rwxXstugo]')
_PERM_BITS = 0o7777 # file mode permission bits
_EXEC_PERM_BITS = 0o0111 # execute permission bits
_DEFAULT_PERM = 0o0666 # default file permission bits
S_IRANY = 0o0444 # read by user, group, others
S_IWANY = 0o0222 # write by user, group, others
S_IXANY = 0o0111 # execute by user, group, others
S_IRWU_RWG_RWO = S_IRANY | S_IWANY # read, write by user, group, others
S_IRWU_RG_RO = S_IRANY | stat.S_IWUSR # read by user, group, others and write only by user
S_IRWXU_RXG_RXO = S_IRANY | S_IXANY | stat.S_IWUSR # read, execute by user, group, others and write only by user
_PERM_BITS = 0o7777 # file mode permission bits
_EXEC_PERM_BITS = S_IXANY # execute permission bits
_DEFAULT_PERM = S_IRWU_RWG_RWO # default file permission bits
def is_executable(path):

@ -372,6 +372,7 @@ import tempfile
import time
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.file import S_IRWXU_RXG_RXO
from ansible.module_utils.common.locale import get_best_parsable_locale
from ansible.module_utils.common.respawn import has_respawned, probe_interpreters_for_module, respawn_module
from ansible.module_utils.common.text.converters import to_native, to_text
@ -448,7 +449,7 @@ class PolicyRcD(object):
with open('/usr/sbin/policy-rc.d', 'w') as policy_rc_d:
policy_rc_d.write('#!/bin/sh\nexit %d\n' % self.m.params['policy_rc_d'])
os.chmod('/usr/sbin/policy-rc.d', 0o0755)
os.chmod('/usr/sbin/policy-rc.d', S_IRWXU_RXG_RXO)
except Exception:
self.m.fail_json(msg="Failed to create or chmod /usr/sbin/policy-rc.d")

@ -180,6 +180,7 @@ import random
import time
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.file import S_IRWU_RG_RO as DEFAULT_SOURCES_PERM
from ansible.module_utils.common.respawn import has_respawned, probe_interpreters_for_module, respawn_module
from ansible.module_utils.common.text.converters import to_native
from ansible.module_utils.urls import fetch_url
@ -200,7 +201,6 @@ except ImportError:
HAVE_PYTHON_APT = False
APT_KEY_DIRS = ['/etc/apt/keyrings', '/etc/apt/trusted.gpg.d', '/usr/share/keyrings']
DEFAULT_SOURCES_PERM = 0o0644
VALID_SOURCE_TYPES = ('deb', 'deb-src')

@ -214,6 +214,7 @@ import sys
import tempfile
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.file import S_IRWU_RWG_RWO
from ansible.module_utils.common.text.converters import to_bytes, to_native
from ansible.module_utils.six.moves import shlex_quote
@ -307,7 +308,7 @@ class CronTab(object):
fileh = open(self.b_cron_file, 'wb')
else:
filed, path = tempfile.mkstemp(prefix='crontab')
os.chmod(path, int('0644', 8))
os.chmod(path, S_IRWU_RWG_RWO)
fileh = os.fdopen(filed, 'wb')
fileh.write(to_bytes(self.render()))

@ -236,6 +236,7 @@ import traceback
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.collections import is_sequence
from ansible.module_utils.common.file import S_IRWXU_RXG_RXO, S_IRWU_RG_RO
from ansible.module_utils.common.text.converters import to_bytes
from ansible.module_utils.common.text.converters import to_native
from ansible.module_utils.six import raise_from # type: ignore[attr-defined]
@ -259,7 +260,7 @@ def ensure_keyrings_dir(module):
changed = False
if not os.path.isdir(KEYRINGS_DIR):
if not module.check_mode:
os.mkdir(KEYRINGS_DIR, 0o755)
os.mkdir(KEYRINGS_DIR, S_IRWXU_RXG_RXO)
changed |= True
changed |= module.set_fs_attributes_if_different(
@ -353,7 +354,7 @@ def write_signed_by_key(module, v, slug):
module.atomic_move(tmpfile, filename)
changed |= True
changed |= module.set_mode_if_different(filename, 0o0644, False)
changed |= module.set_mode_if_different(filename, S_IRWU_RG_RO, False)
return changed, filename, None

@ -11,6 +11,7 @@ import json
import os
import sys
import tarfile
from ansible.module_utils.common.file import S_IRWXU_RXG_RXO
manifest = {
'collection_info': {
@ -46,7 +47,7 @@ files = {
def add_file(tar_file, filename, b_content, update_files=True):
tar_info = tarfile.TarInfo(filename)
tar_info.size = len(b_content)
tar_info.mode = 0o0755
tar_info.mode = S_IRWXU_RXG_RXO
tar_file.addfile(tarinfo=tar_info, fileobj=io.BytesIO(b_content))
if update_files:

@ -37,6 +37,7 @@ from ansible.cli.galaxy import GalaxyCLI
from ansible.galaxy import collection
from ansible.galaxy.api import GalaxyAPI
from ansible.errors import AnsibleError
from ansible.module_utils.common.file import S_IRWU_RG_RO, S_IRWXU_RXG_RXO
from ansible.module_utils.common.text.converters import to_bytes, to_native, to_text
from ansible.utils import context_objects as co
from ansible.utils.display import Display
@ -658,9 +659,9 @@ def test_collection_build(collection_artifact):
assert member.uid == 0
assert member.uname == ''
if member.isdir() or member.name == 'runme.sh':
assert member.mode == 0o0755
assert member.mode == S_IRWXU_RXG_RXO
else:
assert member.mode == 0o0644
assert member.mode == S_IRWU_RG_RO
manifest_file = tar.extractfile(tar_members[0])
try:

@ -22,6 +22,7 @@ from ansible.errors import AnsibleError
from ansible.galaxy import api as galaxy_api
from ansible.galaxy.api import CollectionVersionMetadata, GalaxyAPI, GalaxyError
from ansible.galaxy.token import BasicAuthToken, GalaxyToken, KeycloakToken
from ansible.module_utils.common.file import S_IRWU_RG_RO
from ansible.module_utils.common.text.converters import to_native, to_text
import urllib.error
from ansible.utils import context_objects as co
@ -47,7 +48,7 @@ def collection_artifact(tmp_path_factory):
b_io = BytesIO(b"\x00\x01\x02\x03")
tar_info = tarfile.TarInfo('test')
tar_info.size = 4
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
yield tar_path

@ -23,6 +23,7 @@ from ansible.cli.galaxy import GalaxyCLI
from ansible.errors import AnsibleError
from ansible.galaxy import api, collection, token
from ansible.module_utils.common.text.converters import to_bytes, to_native, to_text
from ansible.module_utils.common.file import S_IRWU_RG_RO
import builtins
from ansible.utils import context_objects as co
from ansible.utils.display import Display
@ -78,7 +79,7 @@ def collection_artifact(monkeypatch, tmp_path_factory):
b_io = BytesIO(b"\x00\x01\x02\x03")
tar_info = tarfile.TarInfo('test')
tar_info.size = 4
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
return input_file, mock_open
@ -106,14 +107,14 @@ def tmp_tarfile(tmp_path_factory, manifest_info):
b_io = BytesIO(data)
tar_info = tarfile.TarInfo(filename)
tar_info.size = len(data)
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
b_data = to_bytes(json.dumps(manifest_info, indent=True), errors='surrogate_or_strict')
b_io = BytesIO(b_data)
tar_info = tarfile.TarInfo('MANIFEST.json')
tar_info.size = len(b_data)
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
sha256_hash = sha256()
@ -962,7 +963,7 @@ def test_extract_tar_file_outside_dir(tmp_path_factory):
b_io = BytesIO(data)
tar_info = tarfile.TarInfo(tar_filename)
tar_info.size = len(data)
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
expected = re.escape("Cannot extract tar entry '%s' as it will be placed outside the collection directory"

@ -24,6 +24,7 @@ from ansible.cli.galaxy import GalaxyCLI
from ansible.errors import AnsibleError
from ansible.galaxy import collection, api, dependency_resolution
from ansible.galaxy.dependency_resolution.dataclasses import Candidate, Requirement
from ansible.module_utils.common.file import S_IRWU_RG_RO, S_IRWXU_RXG_RXO
from ansible.module_utils.common.text.converters import to_bytes, to_native, to_text
from ansible.module_utils.common.process import get_bin_path
from ansible.utils import context_objects as co
@ -345,7 +346,7 @@ def test_build_requirement_from_tar_no_manifest(tmp_path_factory):
b_io = BytesIO(json_data)
tar_info = tarfile.TarInfo('FILES.json')
tar_info.size = len(json_data)
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False)
@ -369,7 +370,7 @@ def test_build_requirement_from_tar_no_files(tmp_path_factory):
b_io = BytesIO(json_data)
tar_info = tarfile.TarInfo('MANIFEST.json')
tar_info.size = len(json_data)
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False)
@ -387,7 +388,7 @@ def test_build_requirement_from_tar_invalid_manifest(tmp_path_factory):
b_io = BytesIO(json_data)
tar_info = tarfile.TarInfo('MANIFEST.json')
tar_info.size = len(json_data)
tar_info.mode = 0o0644
tar_info.mode = S_IRWU_RG_RO
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False)
@ -784,9 +785,9 @@ def test_install_collection(collection_artifact, monkeypatch):
assert actual_files == [b'FILES.json', b'MANIFEST.json', b'README.md', b'docs', b'playbooks', b'plugins', b'roles',
b'runme.sh']
assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'plugins')).st_mode) == 0o0755
assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'README.md')).st_mode) == 0o0644
assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'runme.sh')).st_mode) == 0o0755
assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'plugins')).st_mode) == S_IRWXU_RXG_RXO
assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'README.md')).st_mode) == S_IRWU_RG_RO
assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'runme.sh')).st_mode) == S_IRWXU_RXG_RXO
assert mock_display.call_count == 2
assert mock_display.mock_calls[0][1][0] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" \

@ -14,6 +14,7 @@ from itertools import product
import pytest
from ansible.module_utils import basic
from ansible.module_utils.common.file import S_IRWU_RG_RO
@pytest.fixture
@ -63,7 +64,7 @@ def atomic_mocks(mocker, monkeypatch):
@pytest.fixture
def fake_stat(mocker):
stat1 = mocker.MagicMock()
stat1.st_mode = 0o0644
stat1.st_mode = S_IRWU_RG_RO
stat1.st_uid = 0
stat1.st_gid = 0
stat1.st_flags = 0

Loading…
Cancel
Save