'Gadget',

'GadgetSubprocessManager',

'ConfigFunctionBase',

'ConfigFunctionFFS',

'ConfigFunctionFFSSubprocess',

_ = func # Silence pylint

_ = args # Silence pylint

if result < 0:

"""

usb_gadget_path = '/sys/kernel/config/usb_gadget/'

class_udc_path = '/sys/class/udc/'

__function_list = ()

def __init__( # pylint: disable=too-many-arguments

self,

config_list,

idVendor=None,

idProduct=None,

lang_dict=(),

bcdDevice=None,

bcdUSB=None,

bDeviceClass=None,

bDeviceSubClass=None,

bDeviceProtocol=None,

name=None,

udc=None,

os_desc=None,

):

"""

if udc is None:

try:

udc_list = os.listdir(self.class_udc_path)

except FileNotFoundError:

udc_list = ()

try:

udc, = udc_list

except ValueError:

raise ValueError(

'More than one UDC available'

if udc_list else

'No UDC available'

) from None

udc = os.path.basename(udc)

elif not os.path.exists(os.path.join(self.class_udc_path, udc)):

raise ValueError('No such UDC')

self.__udc = udc

self.__config_list = list(

(

{

'function_list': tuple(config_dict['function_list']),

'attribute_dict': {

attribute_name: cast(

config_dict[attribute_name],

).encode('ascii')

for attribute_name, cast in (

('bmAttributes', hex),

('MaxPower', lambda x: '%i' % (x, )),

)

if config_dict.get(attribute_name) is not None

},

'lang_dict': {

hex(lang): {

message_name: message_dict[message_name].encode('utf-8')

for message_name in (

'configuration',

)

if message_dict.get(message_name) is not None

}

for lang, message_dict in config_dict.get('lang_dict', {}).items()

},

}

for config_dict in config_list

),

)

self.__lang_dict = {

hex(lang): {

message_name: message_dict[message_name].encode('utf-8')

for message_name in (

'serialnumber',

'product',

'manufacturer',

)

if message_dict.get(message_name) is not None

}

for lang, message_dict in dict(lang_dict).items()

}

self.__attribute_dict = {

name: hex(value).encode('ascii')

for name, value in {

'idVendor': idVendor,

'idProduct': idProduct,

'bcdDevice': bcdDevice,

'bcdUSB': bcdUSB,

'bDeviceProtocol': bDeviceProtocol,

'bDeviceClass': bDeviceClass,

'bDeviceSubClass': bDeviceSubClass,

}.items()

if value is not None

}

self.__os_desc = (

()

if os_desc is None else

(

('b_vendor_code', os_desc['b_vendor_code'].to_bytes(1, 'big')),

('qw_sign', os_desc['qw_sign'].encode('utf-8')),

('use', b'1'),

)

)

self.__name = name

self.__real_name = None # chosen on __enter__

self.__dir_list = []

self.__link_list = []

self.__udc_path = None

def isUDCRegistered(self):

"""

with open(self.__udc_path, 'rb') as udc:

return bool(udc.read())

def __writeAttributeDict(self, base, attribute_dict):

for attribute_name, attribute_value in attribute_dict.items():

with open(os.path.join(base, attribute_name), 'wb') as attribute_file:

attribute_file.write(attribute_value)

def __writeLangDict(self, base, lang_dict):

result = []

for lang, message_dict in lang_dict.items():

lang_path = os.path.join(base, 'strings', lang)

result.append(lang_path)

os.mkdir(lang_path)

self.__writeAttributeDict(lang_path, message_dict)

return result

def __enter__(self):

"""

try:

self.__enter()

except Exception:

self.__unenter()

raise

return self

def __enter(self): # pylint: disable=too-many-branches, too-many-statements

dir_list = self.__dir_list

link_list = self.__link_list

def symlink(source, destination): # pylint: disable=missing-docstring

os.symlink(source, destination)

link_list.append(destination)

def mkdir(path): # pylint: disable=missing-docstring

os.mkdir(path)

dir_list.append(path)

name = self.__name

try:

if name is None:

name = tempfile.mkdtemp(

prefix='g_',

dir=self.usb_gadget_path,

)

dir_list.append(name)

else:

name = os.path.join(self.usb_gadget_path, name)

mkdir(name)

except OSError as exc:

if exc.errno == errno.ENOENT:

exc.strerror = (

self.usb_gadget_path +

' does not exist, is libcomposite module loaded ?'

)

raise

self.__real_name = name

for key, value in self.__os_desc:

with open(os.path.join(name, key), 'wb') as desc_file:

desc_file.write(value)

dir_list.extend(self.__writeLangDict(name, self.__lang_dict))

self.__writeAttributeDict(name, self.__attribute_dict)

function_list = self.__function_list = []

function_number_iterator = itertools.count()

name_set = set()

functions_root = os.path.join(name, 'functions')

configs_root = os.path.join(name, 'configs')

for configuration_index, configuration_dict in enumerate(

self.__config_list,

1,

):

config_path = os.path.join(configs_root, 'c.%i' % (configuration_index, ))

mkdir(config_path)

dir_list.extend(

self.__writeLangDict(

config_path,

configuration_dict['lang_dict'],

),

)

self.__writeAttributeDict(

config_path,

configuration_dict['attribute_dict'],

)

for function_index, function in enumerate(

configuration_dict['function_list'],

):

function_name = function.name

if function_name is None:

while True:

function_name = 'usb%i' % next(function_number_iterator)

if function_name not in name_set:

break

function.name = function_name

name_set.add(function_name)

function_path = os.path.join(

functions_root,

function.type_name + '.' + function_name,

)

try:

mkdir(function_path)

except OSError as exc:

if exc.errno == errno.ENOENT:

exc.strerror = (

'Cannot create function of type %r, is its module '

'available ?' % (function.type_name, )

)

raise

function_list.append(function)

function.start(path=function_path)

symlink(

function_path,

os.path.join(

config_path,

'function.%i' % (function_index, ),

),

)

for function in function_list:

function.wait()

udc_path = os.path.join(name, 'UDC')

try:

with open(udc_path, 'w', encoding='ascii') as udc:

udc.write(self.__udc)

except (IOError, OSError) as exc:

if exc.errno == 524: # ENOTSUPP, which is not ENOTSUP

exc.strerror = 'UDC cannot allocate this many endpoints'

raise exc from None

self.__udc_path = udc_path

def __exit__(self, exc_type, exc_value, tb):

self.__unenter()

return False

def __unenter(self):

# configfs cleanup is convoluted and rather surprising if it has to

# be done by the user (ex: rmdir on non-empty directories whose content

# refuse to be individually removed). So catch and report (to stderr)

# exceptions which may come from code out of this module, and continue

# the teardown.

# Should the cleanup actually fail, this will give the user the list

# of operations to do and the order to follow.

name = self.__real_name

if not name:

return

udc_path = self.__udc_path

if udc_path:

with open(udc_path, 'w', encoding='ascii') as udc:

udc.write('')

function_list = self.__function_list

for function in function_list:

try:

function.kill()

except Exception: # pylint: disable=broad-except

print(

'Exception caught while killing function %r' % (

function,

),

file=sys.stderr,

)

traceback.print_exc()

for function in function_list:

try:

function.join()

except Exception: # pylint: disable=broad-except

print(

'Exception caught while joining function %r' % (

function,

),

file=sys.stderr,

)

traceback.print_exc()

link_list = self.__link_list

while link_list:

link = link_list.pop()

try:

os.unlink(link)

except OSError as exc:

print(

'Failed to unlink %r: %r' % (link, exc),

file=sys.stderr,

)

dir_list = self.__dir_list

while dir_list:

directory = dir_list.pop()

try:

os.rmdir(directory)

except OSError as exc:

print(

'Failed to rmdir %r: %r' % (directory, exc),

file=sys.stderr,

)

self.__real_name = None

def getFunction(self, configuration_index, function_index):

"""

config_list = self.__config_list

if (

configuration_index < 0 or

not 0 <= function_index < len(

config_list[configuration_index]['function_list']

)

):

raise IndexError

for index in range(configuration_index):

function_index += len(config_list[index]['function_list'])

return self.__function_list[function_index]

def _iterFunctions(self):

"""

def __call__(self, parser, namespace, values, option_string=None):

passwd = pwd.getpwnam(values)

namespace.uid = passwd.pw_uid

"""

_ = signal_number # Silence pylint

_ = stack_frame # Silence pylint

"""

@staticmethod

def getArgumentParser(**kw):

"""

parser = argparse.ArgumentParser(

epilog='Requires CAP_SYS_ADMIN in order to mount the required '

'functionfs filesystem, and libcomposite kernel module to be '

'loaded (or built-in).',

**kw

)

parser.add_argument(

'--udc',

help='Name of the UDC to use (default: autodetect)',

)

parser.add_argument(

'--uid',

type=int,

help='User to run function as',

)

parser.add_argument(

'--gid',

type=int,

help='Group to run function as',

)

parser.add_argument(

'--username',

action=_UsernameAction,

help="Run function under this user's uid and gid",

)

return parser

def __init__(self, args, config_list, **kw):

"""

for config in config_list:

config['function_list'] = [

x(uid=args.uid, gid=args.gid)

for x in config['function_list']

]

super().__init__(

udc=args.udc,

config_list=config_list,

**kw

)

def _raiseKeyboardInterruptIfFunctionExited(

self,

signal_number,

stack_frame,

):

"""

_ = signal_number # Silence pylint.

_ = stack_frame # Silence pylint.

if any(

x.getExitStatus() is not None

for x in self._iterFunctions()

):

signal.signal(signal.SIGCHLD, signal.SIG_DFL)

raise KeyboardInterrupt

def __enter__(self):

super().__enter__()

# Note: it is important to set SIGCHLD handler after calling

# super().__enter__(), as otherwise the subprocess would get such

# sighandler.

# XXX: Should ConfigFunctionFFSSubprocess.start reset all sighandlers

# to SIG_DFL ? This seems like a lot of work for (usually) no benefits.

signal.signal(

signal.SIGCHLD,

self._raiseKeyboardInterruptIfFunctionExited,

)

# We are on the same terminal as subprocesses, so we will be getting

# SIGINT at the same time as them. But we need to wait for them to

# cleanup and then we will be notified by SIGCHLD.

signal.signal(signal.SIGINT, signal.SIG_IGN)

return self

def __exit__(self, exc_type, exc_value, tb):

signal.signal(signal.SIGINT, signal.SIG_DFL)

signal.signal(signal.SIGCHLD, signal.SIG_DFL)

result = super().__exit__(exc_type, exc_value, tb)

return result or isinstance(exc_value, KeyboardInterrupt)

def waitForever(self):

"""

while True:

"""

name = None

def __init__(self, name=None):

"""

self.name = name

@property

def type_name(self):

"""

raise NotImplementedError

def start(self, path):

"""

raise NotImplementedError

def wait(self):

"""

def kill(self):

"""

def join(self):

"""

raise NotImplementedError

def getExitStatus(self):

"""

ConfigFunctionBase,

): # pylint: disable=abstract-method

"""

__path = None

def __init__(self, config_dict=(), name=None, uid=None, gid=None):

"""

_ = uid # Silence pylint.

_ = gid # Silence pylint.

self.__config_dict = dict(config_dict)

super().__init__(name=name)

def _getOptionAbsPath(self, path, option_path):

"""

option_abspath = os.path.normpath(os.path.join(path, option_path))

if os.path.commonprefix((path, option_abspath)) != path:

raise ValueError('Invalid option path: %r' % (option_path, ))

return option_abspath

def start(self, path):

"""

self.__path = path

for option_path, option_value in self.__config_dict.items():

with open(

self._getOptionAbsPath(path, option_path),

'w',

encoding='ascii',

) as option_file:

option_file.write(option_value)

def join(self):

"""

def getOption(self, option_path):

"""

with open(

self._getOptionAbsPath(self.__path, option_path),

'r',

encoding='ascii',

) as option_file:

"""

type_name = "ffs"

_mountpoint = None

function = None

def __init__(

self,

name=None,

getFunction=None,

uid=None,

gid=None,

rmode=None,

fmode=None,

mode=None,

no_disconnect=None,

):

"""

super().__init__(name=name)

self._getFunction = getFunction

self._uid = uid

self._gid = gid

self._rmode = rmode

self._fmode = fmode

self._mode = mode

self._no_disconnect = no_disconnect

def getFunction(self, path):

"""

return self._getFunction(path=path)

def _mount(self, path):

"""

function_basename = os.path.basename(path)

_, function_name = function_basename.split('.')

mountpoint = tempfile.mkdtemp(

prefix=function_basename + '_',

)

_mount(

function_name.encode('ascii'),

mountpoint.encode('ascii'),

b'functionfs',

0,

b','.join(

b'%s=%i' % (

key.encode('ascii'),

cast(value),

)

for key, cast, value in (

('uid', int, self._uid),

('gid', int, self._gid),

('rmode', int, self._rmode),

('fmode', int, self._fmode),

('mode', int, self._mode),

('no_disconnect', bool, self._no_disconnect),

)

if value is not None

),

)

self._mountpoint = mountpoint

def start(self, path):

self._mount(path)

self.function = function = self.getFunction(

path=self._mountpoint,

)

# pylint: disable=unnecessary-dunder-call

function.__enter__()

def _umount(self):

"""

mountpoint = self._mountpoint

try:

_umount(mountpoint.encode('ascii'))

except OSError as exc:

# if target is not a mountpoint we can rmdir

if exc.errno != errno.EINVAL:

raise

os.rmdir(mountpoint)

self._mountpoint = None

def join(self):

self.function.__exit__(None, None, None)

"""

__pid = None

__exit_status = None

def __init__(self, *args, **kw):

super().__init__(*args, **kw)

self.__read_pipe, self.__write_pipe = os.pipe()

def start(self, path):

"""

signal.signal(signal.SIGTERM, _raiseKeyboardInterrupt)

self._mount(path)

self.__pid = pid = os.fork()

if pid == 0:

try:

status = 0

os.close(self.__read_pipe)

self.__read_pipe = None

os.chdir('/')

if self._gid is not None:

os.setgid(self._gid)

if self._uid is not None:

os.setuid(self._uid)

self.function = function = self.getFunction(

path=self._mountpoint,

)

with function:

os.write(self.__write_pipe, _READY_MARKER)

self.run()

except SystemExit as exc:

if exc.code is not None:

status = exc.code

if not isinstance(status, int):

print(status, file=sys.stderr)

status = 1

except: # pylint: disable=bare-except

traceback.print_exc()

status = 1

finally:

os.close(self.__write_pipe)

os._exit(status) # pylint: disable=protected-access

# Unreachable

os.close(self.__write_pipe)

self.__write_pipe = None

def wait(self):

"""

with os.fdopen(self.__read_pipe, 'rb', 0) as read_pipef:

read_pipef.read(len(_READY_MARKER))

def kill(self):

"""

try:

os.kill(self.__pid, signal.SIGINT)

except OSError as exc:

# If the child process is not running, then all good.

if exc.errno != errno.ESRCH:

raise

def _updateExitStatus(self, blocking):

if self.__exit_status is not None:

return

try:

wait_id_result = os.waitid(

os.P_PID,

self.__pid,

os.WEXITED | (

0

if blocking else

os.WNOHANG

),

)

except OSError as exc:

if exc.errno != errno.ECHILD:

raise

# Uh oh, child process exited and we did not get to read its exit

# status. Pick the generic error status.

# XXX: warn user ?

self.__exit_status = 1

else:

if wait_id_result is not None:

self.__exit_status = (

wait_id_result.si_status

if wait_id_result.si_code == os.CLD_EXITED else

# Subprocess was killed or dumped core, pick the generic error

# status.

1

)

def join(self):

"""

self._updateExitStatus(blocking=True)

self._umount()

def run(self):

"""

try:

self.function.processEventsForever()

except KeyboardInterrupt:

pass

def getExitStatus(self):

"""

self._updateExitStatus(blocking=False)