Merge pull request #8125 from ThomasWaldmann/acl-error-handling-1.4

improve acl_get / acl_set error handling (1.4-maint)
This commit is contained in:
TW 2024-03-30 17:56:16 +01:00 committed by GitHub
commit 5ad7364209
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 192 additions and 98 deletions

View File

@ -902,7 +902,11 @@ Utilization of max. archive size: {csize_max:.0%}
if not symlink:
os.chmod(path, item.mode)
if not self.noacls:
acl_set(path, item, self.numeric_ids, fd=fd)
try:
acl_set(path, item, self.numeric_ids, fd=fd)
except OSError as e:
if e.errno not in (errno.ENOTSUP, ):
raise
if not self.noxattrs:
# chown removes Linux capabilities, so set the extended attributes at the end, after chown, since they include
# the Linux capabilities in the "security.capability" attribute.
@ -1165,7 +1169,11 @@ class MetadataCollector:
attrs['xattrs'] = StableDict(xattrs)
if not self.noacls:
with backup_io('extended stat (ACLs)'):
acl_get(path, attrs, st, self.numeric_ids, fd=fd)
try:
acl_get(path, attrs, st, self.numeric_ids, fd=fd)
except OSError as e:
if e.errno not in (errno.ENOTSUP, ):
raise
return attrs
def stat_attrs(self, st, path, fd=None):

View File

@ -1,6 +1,7 @@
import os
from libc.stdint cimport uint32_t
from libc cimport errno
from .posix import user2uid, group2gid
from ..helpers import safe_decode, safe_encode
@ -121,10 +122,13 @@ def acl_get(path, item, st, numeric_ids=False, fd=None):
else:
acl = acl_get_link_np(path, ACL_TYPE_EXTENDED)
if acl == NULL:
return
if errno.errno == errno.ENOENT:
# macOS weirdness: if a file has no ACLs, it sets errno to ENOENT. :-(
return
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
text = acl_to_text(acl, NULL)
if text == NULL:
return
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
if numeric_ids:
item['acl_extended'] = _remove_non_numeric_identifier(text)
else:
@ -139,17 +143,19 @@ def acl_set(path, item, numeric_ids=False, fd=None):
acl_text = item.get('acl_extended')
if acl_text is not None:
try:
if isinstance(path, str):
path = os.fsencode(path)
if numeric_ids:
acl = acl_from_text(acl_text)
else:
acl = acl_from_text(<bytes>_remove_numeric_id_if_possible(acl_text))
if acl == NULL:
return
if isinstance(path, str):
path = os.fsencode(path)
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
if fd is not None:
acl_set_fd_np(fd, acl, ACL_TYPE_EXTENDED)
if acl_set_fd_np(fd, acl, ACL_TYPE_EXTENDED) == -1:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
else:
acl_set_link_np(path, ACL_TYPE_EXTENDED, acl)
if acl_set_link_np(path, ACL_TYPE_EXTENDED, acl) == -1:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
finally:
acl_free(acl)

View File

@ -1,14 +1,13 @@
import os
import stat
from .posix import posix_acl_use_stored_uid_gid
from ..helpers import safe_encode, safe_decode
from .xattr import _listxattr_inner, _getxattr_inner, _setxattr_inner, split_lstring
API_VERSION = '1.4_01'
from libc cimport errno
cdef extern from "errno.h":
int errno
int EINVAL
API_VERSION = '1.4_01'
cdef extern from "sys/extattr.h":
ssize_t c_extattr_list_file "extattr_list_file" (const char *path, int attrnamespace, void *data, size_t nbytes)
@ -44,10 +43,12 @@ cdef extern from "sys/acl.h":
char *acl_to_text_np(acl_t acl, ssize_t *len, int flags)
int ACL_TEXT_NUMERIC_IDS
int ACL_TEXT_APPEND_ID
int acl_extended_link_np(const char * path) # check also: acl_is_trivial_np
cdef extern from "unistd.h":
long lpathconf(const char *path, int name)
int _PC_ACL_NFS4
int _PC_ACL_EXTENDED
# On FreeBSD, borg currently only deals with the USER namespace as it is unclear
@ -130,13 +131,15 @@ cdef _get_acl(p, type, item, attribute, flags, fd=None):
acl = acl_get_fd_np(fd, type)
else:
acl = acl_get_link_np(p, type)
if acl:
text = acl_to_text_np(acl, NULL, flags)
if text:
item[attribute] = text
acl_free(text)
if acl == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(p))
text = acl_to_text_np(acl, NULL, flags)
if text == NULL:
acl_free(acl)
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(p))
item[attribute] = text
acl_free(text)
acl_free(acl)
def acl_get(path, item, st, numeric_ids=False, fd=None):
"""Saves ACL Entries
@ -144,33 +147,47 @@ def acl_get(path, item, st, numeric_ids=False, fd=None):
If `numeric_ids` is True the user/group field is not preserved only uid/gid
"""
cdef int flags = ACL_TEXT_APPEND_ID
flags |= ACL_TEXT_NUMERIC_IDS if numeric_ids else 0
if isinstance(path, str):
path = os.fsencode(path)
ret = lpathconf(path, _PC_ACL_NFS4)
if ret < 0 and errno == EINVAL:
ret = acl_extended_link_np(path)
if ret < 0:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
if ret == 0:
# there is no ACL defining permissions other than those defined by the traditional file permission bits.
return
flags |= ACL_TEXT_NUMERIC_IDS if numeric_ids else 0
if ret > 0:
ret = lpathconf(path, _PC_ACL_NFS4)
if ret < 0:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
nfs4_acl = ret == 1
if nfs4_acl:
_get_acl(path, ACL_TYPE_NFS4, item, 'acl_nfs4', flags, fd=fd)
else:
_get_acl(path, ACL_TYPE_ACCESS, item, 'acl_access', flags, fd=fd)
_get_acl(path, ACL_TYPE_DEFAULT, item, 'acl_default', flags, fd=fd)
if stat.S_ISDIR(st.st_mode):
_get_acl(path, ACL_TYPE_DEFAULT, item, 'acl_default', flags, fd=fd)
cdef _set_acl(p, type, item, attribute, numeric_ids=False, fd=None):
cdef acl_t acl
text = item.get(attribute)
if text:
if numeric_ids and type == ACL_TYPE_NFS4:
text = _nfs4_use_stored_uid_gid(text)
elif numeric_ids and type in(ACL_TYPE_ACCESS, ACL_TYPE_DEFAULT):
text = posix_acl_use_stored_uid_gid(text)
if numeric_ids:
if type == ACL_TYPE_NFS4:
text = _nfs4_use_stored_uid_gid(text)
elif type in (ACL_TYPE_ACCESS, ACL_TYPE_DEFAULT):
text = posix_acl_use_stored_uid_gid(text)
acl = acl_from_text(<bytes>text)
if acl:
if acl == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(p))
try:
if fd is not None:
acl_set_fd_np(fd, acl, type)
if acl_set_fd_np(fd, acl, type) == -1:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(p))
else:
acl_set_link_np(p, type, acl)
if acl_set_link_np(p, type, acl) == -1:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(p))
finally:
acl_free(acl)
@ -196,6 +213,14 @@ def acl_set(path, item, numeric_ids=False, fd=None):
"""
if isinstance(path, str):
path = os.fsencode(path)
_set_acl(path, ACL_TYPE_NFS4, item, 'acl_nfs4', numeric_ids, fd=fd)
_set_acl(path, ACL_TYPE_ACCESS, item, 'acl_access', numeric_ids, fd=fd)
_set_acl(path, ACL_TYPE_DEFAULT, item, 'acl_default', numeric_ids, fd=fd)
ret = lpathconf(path, _PC_ACL_NFS4)
if ret < 0:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
if ret == 1:
_set_acl(path, ACL_TYPE_NFS4, item, 'acl_nfs4', numeric_ids, fd=fd)
ret = lpathconf(path, _PC_ACL_EXTENDED)
if ret < 0:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
if ret == 1:
_set_acl(path, ACL_TYPE_ACCESS, item, 'acl_access', numeric_ids, fd=fd)
_set_acl(path, ACL_TYPE_DEFAULT, item, 'acl_default', numeric_ids, fd=fd)

View File

@ -52,7 +52,7 @@ cdef extern from "sys/acl.h":
char *acl_to_text(acl_t acl, ssize_t *len)
cdef extern from "acl/libacl.h":
int acl_extended_file(const char *path)
int acl_extended_file_nofollow(const char *path)
int acl_extended_fd(int fd)
cdef extern from "linux/fs.h":
@ -232,15 +232,19 @@ def acl_get(path, item, st, numeric_ids=False, fd=None):
cdef acl_t access_acl = NULL
cdef char *default_text = NULL
cdef char *access_text = NULL
cdef int ret = 0
if stat.S_ISLNK(st.st_mode):
# symlinks can not have ACLs
return
if isinstance(path, str):
path = os.fsencode(path)
if (fd is not None and acl_extended_fd(fd) <= 0
or
fd is None and acl_extended_file(path) <= 0):
if fd is not None:
ret = acl_extended_fd(fd)
else:
ret = acl_extended_file_nofollow(path)
if ret < 0:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
if ret == 0:
# there is no ACL defining permissions other than those defined by the traditional file permission bits.
# note: this should also be the case for symlink fs objects, as they can not have ACLs.
return
if numeric_ids:
converter = acl_numeric_ids
@ -251,22 +255,28 @@ def acl_get(path, item, st, numeric_ids=False, fd=None):
access_acl = acl_get_fd(fd)
else:
access_acl = acl_get_file(path, ACL_TYPE_ACCESS)
if stat.S_ISDIR(st.st_mode):
# only directories can have a default ACL. there is no fd-based api to get it.
default_acl = acl_get_file(path, ACL_TYPE_DEFAULT)
if access_acl:
access_text = acl_to_text(access_acl, NULL)
if access_text:
item['acl_access'] = converter(access_text)
if default_acl:
default_text = acl_to_text(default_acl, NULL)
if default_text:
item['acl_default'] = converter(default_text)
if access_acl == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
access_text = acl_to_text(access_acl, NULL)
if access_text == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
item['acl_access'] = converter(access_text)
finally:
acl_free(default_text)
acl_free(default_acl)
acl_free(access_text)
acl_free(access_acl)
if stat.S_ISDIR(st.st_mode):
# only directories can have a default ACL. there is no fd-based api to get it.
try:
default_acl = acl_get_file(path, ACL_TYPE_DEFAULT)
if default_acl == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
default_text = acl_to_text(default_acl, NULL)
if default_text == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
item['acl_default'] = converter(default_text)
finally:
acl_free(default_text)
acl_free(default_acl)
def acl_set(path, item, numeric_ids=False, fd=None):
@ -277,7 +287,7 @@ def acl_set(path, item, numeric_ids=False, fd=None):
# Linux does not support setting ACLs on symlinks
return
if fd is None and isinstance(path, str):
if isinstance(path, str):
path = os.fsencode(path)
if numeric_ids:
converter = posix_acl_use_stored_uid_gid
@ -287,20 +297,25 @@ def acl_set(path, item, numeric_ids=False, fd=None):
if access_text:
try:
access_acl = acl_from_text(<bytes>converter(access_text))
if access_acl:
if fd is not None:
acl_set_fd(fd, access_acl)
else:
acl_set_file(path, ACL_TYPE_ACCESS, access_acl)
if access_acl == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
if fd is not None:
if acl_set_fd(fd, access_acl) == -1:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
else:
if acl_set_file(path, ACL_TYPE_ACCESS, access_acl) == -1:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
finally:
acl_free(access_acl)
default_text = item.get('acl_default')
if default_text:
try:
default_acl = acl_from_text(<bytes>converter(default_text))
if default_acl:
# only directories can get a default ACL. there is no fd-based api to set it.
acl_set_file(path, ACL_TYPE_DEFAULT, default_acl)
if default_acl == NULL:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
# only directories can get a default ACL. there is no fd-based api to set it.
if acl_set_file(path, ACL_TYPE_DEFAULT, default_acl) == -1:
raise OSError(errno.errno, os.strerror(errno.errno), os.fsdecode(path))
finally:
acl_free(default_acl)

View File

@ -1,6 +1,6 @@
import errno
import functools
import os
import random
import shutil
import sys
import tempfile
@ -13,31 +13,6 @@ from . import BaseTestCase, unopened_tempfile
from .locking import free_pid
ACCESS_ACL = """
user::rw-
user:root:rw-:0
user:9999:r--:9999
group::r--
group:root:r--:0
group:9999:r--:9999
mask::rw-
other::r--
""".strip().encode('ascii')
DEFAULT_ACL = """
user::rw-
user:root:r--:0
user:8888:r--:8888
group::r--
group:root:r--:0
group:8888:r--:8888
mask::rw-
other::r--
""".strip().encode('ascii')
_acls_working = None
def fakeroot_detected():
return 'FAKEROOTKEY' in os.environ
@ -58,20 +33,42 @@ def are_acls_working():
with unopened_tempfile() as filepath:
open(filepath, 'w').close()
try:
access = b'user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:rw-:9999\ngroup:root:rw-:9999\n'
acl = {'acl_access': access}
acl_set(filepath, acl)
if is_darwin:
acl_key = 'acl_extended'
acl_value = b'!#acl 1\nuser:FFFFEEEE-DDDD-CCCC-BBBB-AAAA00000000:root:0:allow:read\n'
elif is_linux:
acl_key = 'acl_access'
acl_value = b'user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:rw-:9999\ngroup:root:rw-:9999\n'
elif is_freebsd:
acl_key = 'acl_access'
acl_value = b'user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:rw-\ngroup:wheel:rw-\n'
else:
return False # ACLs unsupported on this platform.
write_acl = {acl_key: acl_value}
acl_set(filepath, write_acl)
read_acl = {}
acl_get(filepath, read_acl, os.stat(filepath))
read_acl_access = read_acl.get('acl_access', None)
if read_acl_access and b'user::rw-' in read_acl_access:
return True
acl = read_acl.get(acl_key, None)
if acl is not None:
if is_darwin:
check_for = b'root:0:allow:read'
elif is_linux:
check_for = b'user::rw-'
elif is_freebsd:
check_for = b'user::rw-'
else:
return False # ACLs unsupported on this platform.
if check_for in acl:
return True
except PermissionError:
pass
except OSError as e:
if e.errno not in (errno.ENOTSUP, ):
raise
return False
@unittest.skipUnless(sys.platform.startswith('linux'), 'linux only test')
@unittest.skipUnless(is_linux, 'linux only test')
@unittest.skipIf(fakeroot_detected(), 'not compatible with fakeroot')
class PlatformLinuxTestCase(BaseTestCase):
@ -105,6 +102,8 @@ class PlatformLinuxTestCase(BaseTestCase):
@unittest.skipIf(not are_acls_working(), 'ACLs do not work')
def test_default_acl(self):
ACCESS_ACL = b'user::rw-\nuser:root:rw-:0\nuser:9999:r--:9999\ngroup::r--\ngroup:root:r--:0\ngroup:9999:r--:9999\nmask::rw-\nother::r--'
DEFAULT_ACL = b'user::rw-\nuser:root:r--:0\nuser:8888:r--:8888\ngroup::r--\ngroup:root:r--:0\ngroup:8888:r--:8888\nmask::rw-\nother::r--'
self.assert_equal(self.get_acl(self.tmpdir), {})
self.set_acl(self.tmpdir, access=ACCESS_ACL, default=DEFAULT_ACL)
self.assert_equal(self.get_acl(self.tmpdir)['acl_access'], ACCESS_ACL)
@ -149,7 +148,48 @@ class PlatformLinuxTestCase(BaseTestCase):
self.assert_equal(acl_use_local_uid_gid(b'group:root:rw-:0'), b'group:0:rw-')
@unittest.skipUnless(sys.platform.startswith('darwin'), 'macOS only test')
@unittest.skipUnless(is_freebsd, 'freebsd only test')
class PlatformFreeBSDTestCase(BaseTestCase):
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.tmpdir)
def get_acl(self, path, numeric_ids=False):
item = {}
acl_get(path, item, os.stat(path), numeric_ids=numeric_ids)
return item
def set_acl(self, path, access=None, default=None, numeric_ids=False):
item = {'acl_access': access, 'acl_default': default}
acl_set(path, item, numeric_ids=numeric_ids)
@unittest.skipIf(not are_acls_working(), 'ACLs do not work')
def test_access_acl(self):
file = tempfile.NamedTemporaryFile()
self.assert_equal(self.get_acl(file.name), {})
self.set_acl(file.name, access=b'user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:rw-\ngroup:wheel:rw-\n', numeric_ids=False)
self.assert_in(b'user:root:rw-', self.get_acl(file.name)['acl_access'])
self.assert_in(b'group:wheel:rw-', self.get_acl(file.name)['acl_access'])
self.assert_in(b'user:0:rw-', self.get_acl(file.name, numeric_ids=True)['acl_access'])
file2 = tempfile.NamedTemporaryFile()
self.set_acl(file2.name, access=b'user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:rw-\ngroup:wheel:rw-\n', numeric_ids=True)
self.assert_in(b'user::rw-', self.get_acl(file2.name)['acl_access'])
self.assert_in(b'group::r--', self.get_acl(file2.name)['acl_access'])
@unittest.skipIf(not are_acls_working(), 'ACLs do not work')
def test_default_acl(self):
ACCESS_ACL = b'user::rw-\nuser:root:rw-\nuser:9999:r--\ngroup::r--\ngroup:wheel:r--\ngroup:9999:r--\nmask::rw-\nother::r--\n'
DEFAULT_ACL = b'user::rw-\nuser:root:r--\nuser:8888:r--\ngroup::r--\ngroup:wheel:r--\ngroup:8888:r--\nmask::rw-\nother::r--\n'
self.assert_equal(self.get_acl(self.tmpdir), {})
self.set_acl(self.tmpdir, access=ACCESS_ACL, default=DEFAULT_ACL)
self.assert_equal(self.get_acl(self.tmpdir)['acl_access'], ACCESS_ACL)
self.assert_equal(self.get_acl(self.tmpdir)['acl_default'], DEFAULT_ACL)
@unittest.skipUnless(is_darwin, 'macOS only test')
@unittest.skipIf(fakeroot_detected(), 'not compatible with fakeroot')
class PlatformDarwinTestCase(BaseTestCase):
@ -169,7 +209,7 @@ class PlatformDarwinTestCase(BaseTestCase):
acl_set(path, item, numeric_ids=numeric_ids)
@unittest.skipIf(not are_acls_working(), 'ACLs do not work')
def test_access_acl(self):
def test_extended_acl(self):
file = tempfile.NamedTemporaryFile()
file2 = tempfile.NamedTemporaryFile()
self.assert_equal(self.get_acl(file.name), {})