diff options
| -rwxr-xr-x | bitbake/lib/bb/acl.py | 215 | ||||
| -rwxr-xr-x | bitbake/lib/bb/xattr.py | 126 |
2 files changed, 341 insertions, 0 deletions
diff --git a/bitbake/lib/bb/acl.py b/bitbake/lib/bb/acl.py new file mode 100755 index 0000000000..0f41b275cf --- /dev/null +++ b/bitbake/lib/bb/acl.py | |||
| @@ -0,0 +1,215 @@ | |||
| 1 | #! /usr/bin/env python3 | ||
| 2 | # | ||
| 3 | # Copyright 2023 by Garmin Ltd. or its subsidiaries | ||
| 4 | # | ||
| 5 | # SPDX-License-Identifier: MIT | ||
| 6 | |||
| 7 | |||
| 8 | import sys | ||
| 9 | import ctypes | ||
| 10 | import os | ||
| 11 | import errno | ||
| 12 | import pwd | ||
| 13 | import grp | ||
| 14 | |||
| 15 | libacl = ctypes.CDLL("libacl.so.1", use_errno=True) | ||
| 16 | |||
| 17 | |||
| 18 | ACL_TYPE_ACCESS = 0x8000 | ||
| 19 | ACL_TYPE_DEFAULT = 0x4000 | ||
| 20 | |||
| 21 | ACL_FIRST_ENTRY = 0 | ||
| 22 | ACL_NEXT_ENTRY = 1 | ||
| 23 | |||
| 24 | ACL_UNDEFINED_TAG = 0x00 | ||
| 25 | ACL_USER_OBJ = 0x01 | ||
| 26 | ACL_USER = 0x02 | ||
| 27 | ACL_GROUP_OBJ = 0x04 | ||
| 28 | ACL_GROUP = 0x08 | ||
| 29 | ACL_MASK = 0x10 | ||
| 30 | ACL_OTHER = 0x20 | ||
| 31 | |||
| 32 | ACL_READ = 0x04 | ||
| 33 | ACL_WRITE = 0x02 | ||
| 34 | ACL_EXECUTE = 0x01 | ||
| 35 | |||
| 36 | acl_t = ctypes.c_void_p | ||
| 37 | acl_entry_t = ctypes.c_void_p | ||
| 38 | acl_permset_t = ctypes.c_void_p | ||
| 39 | acl_perm_t = ctypes.c_uint | ||
| 40 | |||
| 41 | acl_tag_t = ctypes.c_int | ||
| 42 | |||
| 43 | libacl.acl_free.argtypes = [acl_t] | ||
| 44 | |||
| 45 | |||
| 46 | def acl_free(acl): | ||
| 47 | libacl.acl_free(acl) | ||
| 48 | |||
| 49 | |||
| 50 | libacl.acl_get_file.restype = acl_t | ||
| 51 | libacl.acl_get_file.argtypes = [ctypes.c_char_p, ctypes.c_uint] | ||
| 52 | |||
| 53 | |||
| 54 | def acl_get_file(path, typ): | ||
| 55 | acl = libacl.acl_get_file(os.fsencode(path), typ) | ||
| 56 | if acl is None: | ||
| 57 | err = ctypes.get_errno() | ||
| 58 | raise OSError(err, os.strerror(err), str(path)) | ||
| 59 | |||
| 60 | return acl | ||
| 61 | |||
| 62 | |||
| 63 | libacl.acl_get_entry.argtypes = [acl_t, ctypes.c_int, ctypes.c_void_p] | ||
| 64 | |||
| 65 | |||
| 66 | def acl_get_entry(acl, entry_id): | ||
| 67 | entry = acl_entry_t() | ||
| 68 | ret = libacl.acl_get_entry(acl, entry_id, ctypes.byref(entry)) | ||
| 69 | if ret < 0: | ||
| 70 | err = ctypes.get_errno() | ||
| 71 | raise OSError(err, os.strerror(err)) | ||
| 72 | |||
| 73 | if ret == 0: | ||
| 74 | return None | ||
| 75 | |||
| 76 | return entry | ||
| 77 | |||
| 78 | |||
| 79 | libacl.acl_get_tag_type.argtypes = [acl_entry_t, ctypes.c_void_p] | ||
| 80 | |||
| 81 | |||
| 82 | def acl_get_tag_type(entry_d): | ||
| 83 | tag = acl_tag_t() | ||
| 84 | ret = libacl.acl_get_tag_type(entry_d, ctypes.byref(tag)) | ||
| 85 | if ret < 0: | ||
| 86 | err = ctypes.get_errno() | ||
| 87 | raise OSError(err, os.strerror(err)) | ||
| 88 | return tag.value | ||
| 89 | |||
| 90 | |||
| 91 | libacl.acl_get_qualifier.restype = ctypes.c_void_p | ||
| 92 | libacl.acl_get_qualifier.argtypes = [acl_entry_t] | ||
| 93 | |||
| 94 | |||
| 95 | def acl_get_qualifier(entry_d): | ||
| 96 | ret = libacl.acl_get_qualifier(entry_d) | ||
| 97 | if ret is None: | ||
| 98 | err = ctypes.get_errno() | ||
| 99 | raise OSError(err, os.strerror(err)) | ||
| 100 | return ctypes.c_void_p(ret) | ||
| 101 | |||
| 102 | |||
| 103 | libacl.acl_get_permset.argtypes = [acl_entry_t, ctypes.c_void_p] | ||
| 104 | |||
| 105 | |||
| 106 | def acl_get_permset(entry_d): | ||
| 107 | permset = acl_permset_t() | ||
| 108 | ret = libacl.acl_get_permset(entry_d, ctypes.byref(permset)) | ||
| 109 | if ret < 0: | ||
| 110 | err = ctypes.get_errno() | ||
| 111 | raise OSError(err, os.strerror(err)) | ||
| 112 | |||
| 113 | return permset | ||
| 114 | |||
| 115 | |||
| 116 | libacl.acl_get_perm.argtypes = [acl_permset_t, acl_perm_t] | ||
| 117 | |||
| 118 | |||
| 119 | def acl_get_perm(permset_d, perm): | ||
| 120 | ret = libacl.acl_get_perm(permset_d, perm) | ||
| 121 | if ret < 0: | ||
| 122 | err = ctypes.get_errno() | ||
| 123 | raise OSError(err, os.strerror(err)) | ||
| 124 | return bool(ret) | ||
| 125 | |||
| 126 | |||
| 127 | class Entry(object): | ||
| 128 | def __init__(self, tag, qualifier, mode): | ||
| 129 | self.tag = tag | ||
| 130 | self.qualifier = qualifier | ||
| 131 | self.mode = mode | ||
| 132 | |||
| 133 | def __str__(self): | ||
| 134 | typ = "" | ||
| 135 | qual = "" | ||
| 136 | if self.tag == ACL_USER: | ||
| 137 | typ = "user" | ||
| 138 | qual = pwd.getpwuid(self.qualifier).pw_name | ||
| 139 | elif self.tag == ACL_GROUP: | ||
| 140 | typ = "group" | ||
| 141 | qual = grp.getgrgid(self.qualifier).gr_name | ||
| 142 | elif self.tag == ACL_USER_OBJ: | ||
| 143 | typ = "user" | ||
| 144 | elif self.tag == ACL_GROUP_OBJ: | ||
| 145 | typ = "group" | ||
| 146 | elif self.tag == ACL_MASK: | ||
| 147 | typ = "mask" | ||
| 148 | elif self.tag == ACL_OTHER: | ||
| 149 | typ = "other" | ||
| 150 | |||
| 151 | r = "r" if self.mode & ACL_READ else "-" | ||
| 152 | w = "w" if self.mode & ACL_WRITE else "-" | ||
| 153 | x = "x" if self.mode & ACL_EXECUTE else "-" | ||
| 154 | |||
| 155 | return f"{typ}:{qual}:{r}{w}{x}" | ||
| 156 | |||
| 157 | |||
| 158 | class ACL(object): | ||
| 159 | def __init__(self, acl): | ||
| 160 | self.acl = acl | ||
| 161 | |||
| 162 | def __del__(self): | ||
| 163 | acl_free(self.acl) | ||
| 164 | |||
| 165 | def entries(self): | ||
| 166 | entry_id = ACL_FIRST_ENTRY | ||
| 167 | while True: | ||
| 168 | entry = acl_get_entry(self.acl, entry_id) | ||
| 169 | if entry is None: | ||
| 170 | break | ||
| 171 | |||
| 172 | permset = acl_get_permset(entry) | ||
| 173 | |||
| 174 | mode = 0 | ||
| 175 | for m in (ACL_READ, ACL_WRITE, ACL_EXECUTE): | ||
| 176 | if acl_get_perm(permset, m): | ||
| 177 | mode |= m | ||
| 178 | |||
| 179 | qualifier = None | ||
| 180 | tag = acl_get_tag_type(entry) | ||
| 181 | |||
| 182 | if tag == ACL_USER or tag == ACL_GROUP: | ||
| 183 | qual = acl_get_qualifier(entry) | ||
| 184 | qualifier = ctypes.cast(qual, ctypes.POINTER(ctypes.c_int))[0] | ||
| 185 | |||
| 186 | yield Entry(tag, qualifier, mode) | ||
| 187 | |||
| 188 | entry_id = ACL_NEXT_ENTRY | ||
| 189 | |||
| 190 | @classmethod | ||
| 191 | def from_path(cls, path, typ): | ||
| 192 | acl = acl_get_file(path, typ) | ||
| 193 | return cls(acl) | ||
| 194 | |||
| 195 | |||
| 196 | def main(): | ||
| 197 | import argparse | ||
| 198 | import pwd | ||
| 199 | import grp | ||
| 200 | from pathlib import Path | ||
| 201 | |||
| 202 | parser = argparse.ArgumentParser() | ||
| 203 | parser.add_argument("path", help="File Path", type=Path) | ||
| 204 | |||
| 205 | args = parser.parse_args() | ||
| 206 | |||
| 207 | acl = ACL.from_path(args.path, ACL_TYPE_ACCESS) | ||
| 208 | for entry in acl.entries(): | ||
| 209 | print(str(entry)) | ||
| 210 | |||
| 211 | return 0 | ||
| 212 | |||
| 213 | |||
| 214 | if __name__ == "__main__": | ||
| 215 | sys.exit(main()) | ||
diff --git a/bitbake/lib/bb/xattr.py b/bitbake/lib/bb/xattr.py new file mode 100755 index 0000000000..7b634944a4 --- /dev/null +++ b/bitbake/lib/bb/xattr.py | |||
| @@ -0,0 +1,126 @@ | |||
| 1 | #! /usr/bin/env python3 | ||
| 2 | # | ||
| 3 | # Copyright 2023 by Garmin Ltd. or its subsidiaries | ||
| 4 | # | ||
| 5 | # SPDX-License-Identifier: MIT | ||
| 6 | |||
| 7 | import sys | ||
| 8 | import ctypes | ||
| 9 | import os | ||
| 10 | import errno | ||
| 11 | |||
| 12 | libc = ctypes.CDLL("libc.so.6", use_errno=True) | ||
| 13 | fsencoding = sys.getfilesystemencoding() | ||
| 14 | |||
| 15 | |||
| 16 | libc.listxattr.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_size_t] | ||
| 17 | libc.llistxattr.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_size_t] | ||
| 18 | |||
| 19 | |||
| 20 | def listxattr(path, follow=True): | ||
| 21 | func = libc.listxattr if follow else libc.llistxattr | ||
| 22 | |||
| 23 | os_path = os.fsencode(path) | ||
| 24 | |||
| 25 | while True: | ||
| 26 | length = func(os_path, None, 0) | ||
| 27 | |||
| 28 | if length < 0: | ||
| 29 | err = ctypes.get_errno() | ||
| 30 | raise OSError(err, os.strerror(err), str(path)) | ||
| 31 | |||
| 32 | if length == 0: | ||
| 33 | return [] | ||
| 34 | |||
| 35 | arr = ctypes.create_string_buffer(length) | ||
| 36 | |||
| 37 | read_length = func(os_path, arr, length) | ||
| 38 | if read_length != length: | ||
| 39 | # Race! | ||
| 40 | continue | ||
| 41 | |||
| 42 | return [a.decode(fsencoding) for a in arr.raw.split(b"\x00") if a] | ||
| 43 | |||
| 44 | |||
| 45 | libc.getxattr.argtypes = [ | ||
| 46 | ctypes.c_char_p, | ||
| 47 | ctypes.c_char_p, | ||
| 48 | ctypes.c_char_p, | ||
| 49 | ctypes.c_size_t, | ||
| 50 | ] | ||
| 51 | libc.lgetxattr.argtypes = [ | ||
| 52 | ctypes.c_char_p, | ||
| 53 | ctypes.c_char_p, | ||
| 54 | ctypes.c_char_p, | ||
| 55 | ctypes.c_size_t, | ||
| 56 | ] | ||
| 57 | |||
| 58 | |||
| 59 | def getxattr(path, name, follow=True): | ||
| 60 | func = libc.getxattr if follow else libc.lgetxattr | ||
| 61 | |||
| 62 | os_path = os.fsencode(path) | ||
| 63 | os_name = os.fsencode(name) | ||
| 64 | |||
| 65 | while True: | ||
| 66 | length = func(os_path, os_name, None, 0) | ||
| 67 | |||
| 68 | if length < 0: | ||
| 69 | err = ctypes.get_errno() | ||
| 70 | if err == errno.ENODATA: | ||
| 71 | return None | ||
| 72 | raise OSError(err, os.strerror(err), str(path)) | ||
| 73 | |||
| 74 | if length == 0: | ||
| 75 | return "" | ||
| 76 | |||
| 77 | arr = ctypes.create_string_buffer(length) | ||
| 78 | |||
| 79 | read_length = func(os_path, os_name, arr, length) | ||
| 80 | if read_length != length: | ||
| 81 | # Race! | ||
| 82 | continue | ||
| 83 | |||
| 84 | return arr.raw | ||
| 85 | |||
| 86 | |||
| 87 | def get_all_xattr(path, follow=True): | ||
| 88 | attrs = {} | ||
| 89 | |||
| 90 | names = listxattr(path, follow) | ||
| 91 | |||
| 92 | for name in names: | ||
| 93 | value = getxattr(path, name, follow) | ||
| 94 | if value is None: | ||
| 95 | # This can happen if a value is erased after listxattr is called, | ||
| 96 | # so ignore it | ||
| 97 | continue | ||
| 98 | attrs[name] = value | ||
| 99 | |||
| 100 | return attrs | ||
| 101 | |||
| 102 | |||
| 103 | def main(): | ||
| 104 | import argparse | ||
| 105 | from pathlib import Path | ||
| 106 | |||
| 107 | parser = argparse.ArgumentParser() | ||
| 108 | parser.add_argument("path", help="File Path", type=Path) | ||
| 109 | |||
| 110 | args = parser.parse_args() | ||
| 111 | |||
| 112 | attrs = get_all_xattr(args.path) | ||
| 113 | |||
| 114 | for name, value in attrs.items(): | ||
| 115 | try: | ||
| 116 | value = value.decode(fsencoding) | ||
| 117 | except UnicodeDecodeError: | ||
| 118 | pass | ||
| 119 | |||
| 120 | print(f"{name} = {value}") | ||
| 121 | |||
| 122 | return 0 | ||
| 123 | |||
| 124 | |||
| 125 | if __name__ == "__main__": | ||
| 126 | sys.exit(main()) | ||
