mirror of
https://github.com/borgbackup/borg.git
synced 2026-06-10 17:32:13 -04:00
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
import errno
|
|
import os
|
|
|
|
from ..helpers import Buffer
|
|
|
|
|
|
try:
|
|
ENOATTR = errno.ENOATTR
|
|
except AttributeError:
|
|
# On some platforms, ENOATTR is missing; use ENODATA there.
|
|
ENOATTR = errno.ENODATA
|
|
|
|
|
|
buffer = Buffer(bytearray, limit=2**24)
|
|
|
|
|
|
def split_string0(buf):
|
|
"""Split a list of zero-terminated strings into Python bytes (without terminating zeros)."""
|
|
if isinstance(buf, bytearray):
|
|
buf = bytes(buf) # use a bytes object, so we return a list of bytes objects
|
|
return buf.split(b'\0')[:-1]
|
|
|
|
|
|
def split_lstring(buf):
|
|
"""Split a list of length-prefixed strings into Python bytes (without length prefixes)."""
|
|
result = []
|
|
mv = memoryview(buf)
|
|
while mv:
|
|
length = mv[0]
|
|
result.append(bytes(mv[1:1 + length]))
|
|
mv = mv[1 + length:]
|
|
return result
|
|
|
|
|
|
class BufferTooSmallError(Exception):
|
|
"""The buffer given to an xattr function was too small for the result."""
|
|
|
|
|
|
def _check(rv, path=None, detect_buffer_too_small=False):
|
|
from . import get_errno
|
|
if rv < 0:
|
|
e = get_errno()
|
|
if detect_buffer_too_small and e == errno.ERANGE:
|
|
# listxattr and getxattr indicate with ERANGE that they need a bigger result buffer.
|
|
# setxattr indicates this way that, e.g., an xattr key name is too long or unacceptable.
|
|
raise BufferTooSmallError
|
|
else:
|
|
try:
|
|
msg = os.strerror(e)
|
|
except ValueError:
|
|
msg = ''
|
|
if isinstance(path, int):
|
|
path = '<FD %d>' % path
|
|
raise OSError(e, msg, path)
|
|
if detect_buffer_too_small and rv >= len(buffer):
|
|
# FreeBSD does not error with ERANGE if the buffer is too small;
|
|
# it just fills the buffer, truncates, and returns.
|
|
# Therefore, we play it safe and assume the result is truncated if
|
|
# it happens to be a full buffer.
|
|
raise BufferTooSmallError
|
|
return rv
|
|
|
|
|
|
def _listxattr_inner(func, path):
|
|
assert isinstance(path, (bytes, int))
|
|
size = len(buffer)
|
|
while True:
|
|
buf = buffer.get(size)
|
|
try:
|
|
n = _check(func(path, buf, size), path, detect_buffer_too_small=True)
|
|
except BufferTooSmallError:
|
|
size *= 2
|
|
else:
|
|
return n, buf
|
|
|
|
|
|
def _getxattr_inner(func, path, name):
|
|
assert isinstance(path, (bytes, int))
|
|
assert isinstance(name, bytes)
|
|
size = len(buffer)
|
|
while True:
|
|
buf = buffer.get(size)
|
|
try:
|
|
n = _check(func(path, name, buf, size), path, detect_buffer_too_small=True)
|
|
except BufferTooSmallError:
|
|
size *= 2
|
|
else:
|
|
return n, buf
|
|
|
|
|
|
def _setxattr_inner(func, path, name, value):
|
|
assert isinstance(path, (bytes, int))
|
|
assert isinstance(name, bytes)
|
|
assert isinstance(value, bytes)
|
|
_check(func(path, name, value, len(value)), path, detect_buffer_too_small=False)
|