a partir do código de btrfs-snapshots-diff.py [ github.com] foi capaz de fazer um script adaptado às minhas necessidades. Eu posso usá-lo desta maneira para obter o uuid s:
with BtrfsStream('snapshot_0-1.data') as btrfs_stream:
print(btrfs_stream.get_send_command())
# ('BTRFS_SEND_C_SNAPSHOT',
# (UUID('01234567-89ab-cdef-0123-456789abcdef'),
# UUID('fedcba98-7654-3210-fedc-ba9876543210')))
com a classe BtrfsStream como abaixo.
Eu fiz algumas modificações no código original:
- python 3 (em vez de python 2)
- iterando sobre o arquivo em vez de ler tudo na memória
- adicionou recursos
contextmanagerpara usá-lo emwithsatement
o código usado é:
from struct import unpack
import io
from uuid import UUID
class BtrfsStream:
# From btrfs/send.h
send_cmds = (
'BTRFS_SEND_C_UNSPEC BTRFS_SEND_C_SUBVOL BTRFS_SEND_C_SNAPSHOT '
'BTRFS_SEND_C_MKFILE BTRFS_SEND_C_MKDIR BTRFS_SEND_C_MKNOD '
'BTRFS_SEND_C_MKFIFO BTRFS_SEND_C_MKSOCK BTRFS_SEND_C_SYMLINK '
'BTRFS_SEND_C_RENAME BTRFS_SEND_C_LINK BTRFS_SEND_C_UNLINK '
'BTRFS_SEND_C_RMDIR BTRFS_SEND_C_SET_XATTR BTRFS_SEND_C_REMOVE_XATTR '
'BTRFS_SEND_C_WRITE BTRFS_SEND_C_CLONE BTRFS_SEND_C_TRUNCATE '
'BTRFS_SEND_C_CHMOD BTRFS_SEND_C_CHOWN BTRFS_SEND_C_UTIMES '
'BTRFS_SEND_C_END BTRFS_SEND_C_UPDATE_EXTENT').split()
send_attrs = (
'BTRFS_SEND_A_UNSPEC BTRFS_SEND_A_UUID BTRFS_SEND_A_CTRANSID '
'BTRFS_SEND_A_INO BTRFS_SEND_A_SIZE BTRFS_SEND_A_MODE '
'BTRFS_SEND_A_UID BTRFS_SEND_A_GID BTRFS_SEND_A_RDEV '
'BTRFS_SEND_A_CTIME BTRFS_SEND_A_MTIME BTRFS_SEND_A_ATIME '
'BTRFS_SEND_A_OTIME BTRFS_SEND_A_XATTR_NAME '
'BTRFS_SEND_A_XATTR_DATA BTRFS_SEND_A_PATH BTRFS_SEND_A_PATH_TO '
'BTRFS_SEND_A_PATH_LINK BTRFS_SEND_A_FILE_OFFSET BTRFS_SEND_A_DATA '
'BTRFS_SEND_A_CLONE_UUID BTRFS_SEND_A_CLONE_CTRANSID '
'BTRFS_SEND_A_CLONE_PATH BTRFS_SEND_A_CLONE_OFFSET '
'BTRFS_SEND_A_CLONE_LEN').split()
# From btrfs/ioctl.h:#define BTRFS_UUID_SIZE 16
BTRFS_UUID_SIZE = 16
HEADER_SIZE = 17
# Headers length
l_head = 10
l_tlv = 4
def __init__(self, path):
'''
'''
self.path = path
self._stream = None
def __enter__(self):
'''
enter for context manager
'''
self._stream = open(self.path, 'rb')
self._read_header()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
'''
exit for context manager
'''
self._stream.close()
def _read_header(self):
'''
read the header
'''
header = self.read(BtrfsStream.HEADER_SIZE, assert_lengh=False)
if len(header) < BtrfsStream.HEADER_SIZE:
raise IOError('Invalid stream length\n')
magic, null, self.version = unpack('<12scI', header)
if magic != b'btrfs-stream':
raise IOError('Not a Btrfs stream!')
def seek(self, offset, whence=io.SEEK_SET):
'''
seek to a given point
'''
self._stream.seek(offset)
def tell(self):
'''
tell where we are
'''
return self._stream.tell()
def read(self, n_bytes, assert_lengh=True):
'''
try to read n_bytes
'''
tell_before = self.tell()
btes = self._stream.read(n_bytes)
if assert_lengh is True and len(btes) != n_bytes:
msg = ('could only read {} instead of {} at offset {}'
).format(len(btes), n_bytes, tell_before)
raise IOError(msg)
return btes
def tlv_get(self, attr_type):
attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
if self.send_attrs[attr] != attr_type:
raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
ret, = unpack('<H', self.read(2))
return ret
def _tlv_get_string(self, attr_type):
attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
if self.send_attrs[attr] != attr_type:
raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
ret, = unpack('<%ds' % l_attr, self.read(l_attr))
return ret
def _tlv_get_u64(self, attr_type):
attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
if self.send_attrs[attr] != attr_type:
raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
ret, = unpack('<Q', self.read(l_attr))
return ret
def _tlv_get_uuid(self, attr_type):
attr, l_attr = unpack('<HH', self.read(BtrfsStream.l_tlv))
if self.send_attrs[attr] != attr_type:
raise ValueError('Unexpected attribute %s' % self.send_attrs[attr])
return UUID(bytes=self.read(l_attr))
def get_send_command(self):
'''
search uuids only.
'''
# start at the right point in the file
self.seek(BtrfsStream.HEADER_SIZE)
while True:
l_cmd, cmd, crc = unpack('<IHI', self.read(BtrfsStream.l_head))
tell_before_cmd = self.tell()
try:
command = self.send_cmds[cmd]
except:
raise ValueError('Unkown command %d' % cmd)
if command == 'BTRFS_SEND_C_SNAPSHOT':
self._tlv_get_string('BTRFS_SEND_A_PATH')
uuid = self._tlv_get_uuid('BTRFS_SEND_A_UUID')
self._tlv_get_u64('BTRFS_SEND_A_CTRANSID')
clone_uuid = self._tlv_get_uuid('BTRFS_SEND_A_CLONE_UUID')
return 'BTRFS_SEND_C_SNAPSHOT', (uuid, clone_uuid)
elif command == 'BTRFS_SEND_C_SUBVOL':
self._tlv_get_string('BTRFS_SEND_A_PATH')
uuid = self._tlv_get_uuid('BTRFS_SEND_A_UUID')
return 'BTRFS_SEND_C_SUBVOL', (uuid, )
elif command == 'BTRFS_SEND_C_CLONE':
self._tlv_get_string('BTRFS_SEND_A_PATH')
self._tlv_get_u64('BTRFS_SEND_A_FILE_OFFSET')
self._tlv_get_u64('BTRFS_SEND_A_CLONE_LEN')
clone_uuid = self._tlv_get_uuid('BTRFS_SEND_A_CLONE_UUID')
return 'BTRFS_SEND_C_CLONE', (clone_uuid, )
elif command == 'BTRFS_SEND_C_END':
return
self.seek(tell_before_cmd + l_cmd)