Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py	0 B	2025-08-24 21:48
📁 __pycache__	-	2026-03-21 15:56
📄 class_loader.py import tuned.logs import os __all__ = ["ClassLoader"] log = tuned.logs.get() class ClassLoader(object): __slots__ = ["_namespace", "_prefix", "_interface"] def _set_loader_parameters(self): """ This method has to be implemented in child class and should set _namespace, _prefix, and _interface member attributes. """ raise NotImplementedError() def __init__(self): super(ClassLoader, self).__init__() self._namespace = None self._prefix = None self._interface = None self._set_loader_parameters() assert type(self._namespace) is str assert type(self._prefix) is str assert type(self._interface) is type and issubclass(self._interface, object) def load_class(self, class_name): assert type(class_name) is str module_name = "%s.%s%s" % (self._namespace, self._prefix, class_name) return self._get_class(module_name) def _get_class(self, module_name): log.debug("loading module %s" % module_name) module = __import__(module_name) path = module_name.split(".") path.pop(0) while len(path) > 0: module = getattr(module, path.pop(0)) for name in module.__dict__: cls = getattr(module, name) if type(cls) is type and issubclass(cls, self._interface): return cls raise ImportError("Cannot find the class %s." % module_name) def load_all_classes(self): package = __import__(self._namespace) basename = self._namespace.split(".")[-1] classes = [] for module_name in os.listdir(getattr(package, basename).__path__[0]): try: module_name = os.path.splitext(module_name)[0] if not module_name.startswith(self._prefix): continue next_class = self._get_class( "%s.%s" % (self._namespace, module_name) ) if next_class not in classes: classes.append(next_class) except ImportError: pass return classes	1.78 KB	2025-08-24 21:48
📄 commands.py import errno import hashlib import tuned.logs import copy import os import shutil import tuned.consts as consts import re from subprocess import * from tuned.exceptions import TunedException log = tuned.logs.get() class commands: def __init__(self, logging = True): self._logging = logging def _error(self, msg): if self._logging: log.error(msg) def _debug(self, msg): if self._logging: log.debug(msg) def get_bool(self, value): v = str(value).upper().strip() return {"Y":"1", "YES":"1", "T":"1", "TRUE":"1", "N":"0", "NO":"0", "F":"0", "FALSE":"0"}.get(v, value) def remove_ws(self, s): return re.sub(r'\s+', ' ', str(s)).strip() def unquote(self, v): return re.sub("^\"(.)\"$", r"\1", v) # escape escape character (by default '\') def escape(self, s, what_escape = "\\", escape_by = "\\"): return s.replace(what_escape, "%s%s" % (escape_by, what_escape)) # clear escape characters (by default '\') def unescape(self, s, escape_char = "\\"): return s.replace(escape_char, "") # add spaces to align s2 to pos, returns resulting string: s1 + spaces + s2 def align_str(self, s1, pos, s2): return s1 + " " (pos - len(s1)) + s2 # convert dictionary 'd' to flat list and return it # it uses sort on the dictionary items to return consistent results # for directories with different inserte/delete history def dict2list(self, d): l = [] if d is not None: for i in sorted(d.items()): l += list(i) return l # Compile regex to speedup multiple_re_replace or re_lookup def re_lookup_compile(self, d): if d is None: return None return re.compile("(%s)" % ")\|(".join(list(d.keys()))) # Do multiple regex replaces in 's' according to lookup table described by # dictionary 'd', e.g.: d = {"re1": "replace1", "re2": "replace2", ...} # r can be regex precompiled by re_lookup_compile for speedup def multiple_re_replace(self, d, s, r = None, flags = 0): if d is None: if r is None: return s else: if len(d) == 0 or s is None: return s if r is None: r = self.re_lookup_compile(d) return r.sub(lambda mo: list(d.values())[mo.lastindex - 1], s, flags) # Do regex lookup on 's' according to lookup table described by # dictionary 'd' and return corresponding value from the dictionary, # e.g.: d = {"re1": val1, "re2": val2, ...} # r can be regex precompiled by re_lookup_compile for speedup def re_lookup(self, d, s, r = None): if len(d) == 0 or s is None: return None if r is None: r = self.re_lookup_compile(d) mo = r.search(s) if mo: return list(d.values())[mo.lastindex - 1] return None def write_to_file(self, f, data, makedir = False, no_error = False, ignore_same = False): """Write data to a file. Parameters: f -- filename where to write data -- data to write makedir -- if True and the path doesn't exist, it will be created no_error -- if True errors are silenced, it can be also list of ignored errnos ignore_same -- if True and the write would not change the file, it is skipped Return: bool -- True on success """ self._debug("Writing to file: '%s' < '%s'" % (f, data)) if makedir: d = os.path.dirname(f) if os.path.isdir(d): makedir = False try: if makedir: os.makedirs(d) if ignore_same and self.read_file(f, no_error=True).strip() == str(data): self._debug("Skipping the write to file '%s', the content would not change" % f) return True fd = open(f, "w") fd.write(str(data)) fd.close() rc = True except (OSError, IOError) as e: rc = False if isinstance(no_error, bool) and not no_error or \ isinstance(no_error, list) and e.errno not in no_error: self._error("Writing to file '%s' error: '%s'" % (f, e)) return rc def read_file(self, f, err_ret = "", no_error = False): old_value = err_ret try: fd = open(f, "r") old_value = fd.read() fd.close() except (OSError,IOError) as e: if not no_error: self._error("Error when reading file '%s': '%s'" % (f, e)) self._debug("Read data from file: '%s' > '%s'" % (f, old_value)) return old_value def rmtree(self, f, no_error = False): self._debug("Removing tree: '%s'" % f) if os.path.exists(f): try: shutil.rmtree(f, no_error) except OSError as error: if not no_error: log.error("cannot remove tree '%s': '%s'" % (f, str(error))) return False return True def unlink(self, f, no_error = False): self._debug("Removing file: '%s'" % f) if os.path.exists(f): try: os.unlink(f) except OSError as error: if not no_error: log.error("cannot remove file '%s': '%s'" % (f, str(error))) return False return True def rename(self, src, dst, no_error = False): self._debug("Renaming file '%s' to '%s'" % (src, dst)) try: os.rename(src, dst) except OSError as error: if not no_error: log.error("cannot rename file '%s' to '%s': '%s'" % (src, dst, str(error))) return False return True def copy(self, src, dst, no_error = False): try: log.debug("copying file '%s' to '%s'" % (src, dst)) shutil.copy(src, dst) return True except IOError as e: if not no_error: log.error("cannot copy file '%s' to '%s': %s" % (src, dst, e)) return False def replace_in_file(self, f, pattern, repl): data = self.read_file(f) if len(data) <= 0: return False; return self.write_to_file(f, re.sub(pattern, repl, data, flags = re.MULTILINE)) # do multiple replaces in file 'f' by using dictionary 'd', # e.g.: d = {"re1": val1, "re2": val2, ...} def multiple_replace_in_file(self, f, d): data = self.read_file(f) if len(data) <= 0: return False; return self.write_to_file(f, self.multiple_re_replace(d, data, flags = re.MULTILINE)) # makes sure that options from 'd' are set to values from 'd' in file 'f', # when needed it edits options or add new options if they don't # exist and 'add' is set to True, 'd' has the following form: # d = {"option_1": value_1, "option_2": value_2, ...} def add_modify_option_in_file(self, f, d, add = True): data = self.read_file(f) for opt in d: o = str(opt) v = str(d[opt]) if re.search(r"\b" + o + r"\s=.$", data, flags = re.MULTILINE) is None: if add: if len(data) > 0 and data[-1] != "\n": data += "\n" data += "%s=\"%s\"\n" % (o, v) else: data = re.sub(r"\b(" + o + r"\s=).$", r"\1" + "\"" + self.escape(v) + "\"", data, flags = re.MULTILINE) return self.write_to_file(f, data) # calcualates md5sum of file 'f' def md5sum(self, f): data = self.read_file(f) return hashlib.md5(str(data).encode("utf-8")).hexdigest() # calcualates sha256sum of file 'f' def sha256sum(self, f): data = self.read_file(f) return hashlib.sha256(str(data).encode("utf-8")).hexdigest() # returns machine ID or empty string "" in case of error def get_machine_id(self, no_error = True): return self.read_file(consts.MACHINE_ID_FILE, no_error).strip() # "no_errors" can be list of return codes not treated as errors, if 0 is in no_errors, it means any error # returns (retcode, out), where retcode is exit code of the executed process or -errno if # OSError or IOError exception happened def execute(self, args, shell = False, cwd = None, env = {}, no_errors = [], return_err = False): retcode = 0 _environment = os.environ.copy() _environment["LC_ALL"] = "C" _environment.update(env) self._debug("Executing %s." % str(args)) out = "" err_msg = None try: proc = Popen(args, stdout = PIPE, stderr = PIPE, \ env = _environment, \ shell = shell, cwd = cwd, \ close_fds = True, \ universal_newlines = True) out, err = proc.communicate() retcode = proc.returncode if retcode and not retcode in no_errors and not 0 in no_errors: err_out = err[:-1] if len(err_out) == 0: err_out = out[:-1] err_msg = "Executing '%s' error: %s" % (' '.join(args), err_out) if not return_err: self._error(err_msg) except (OSError, IOError) as e: retcode = -e.errno if e.errno is not None else -1 if not abs(retcode) in no_errors and not 0 in no_errors: err_msg = "Executing '%s' error: %s" % (' '.join(args), e) if not return_err: self._error(err_msg) if return_err: return retcode, out, err_msg else: return retcode, out # Helper for parsing kernel options like: # [always] never # It will return 'always' def get_active_option(self, options, dosplit = True): m = re.match(r'.\[([^\]]+)\].', options) if m: return m.group(1) if dosplit: return options.split()[0] return options # Checks whether CPU is online def is_cpu_online(self, cpu): scpu = str(cpu) # CPU0 is always online return cpu == "0" or self.read_file("/sys/devices/system/cpu/cpu%s/online" % scpu, err_ret = "1", no_error = True).strip() == "1" # Converts hexadecimal CPU mask to CPU list def hex2cpulist(self, mask): if mask is None: return None mask = str(mask).replace(",", "") try: m = int(mask, 16) except ValueError: log.error("invalid hexadecimal mask '%s'" % str(mask)) return [] return self.bitmask2cpulist(m) # Converts an integer bitmask to a list of cpus (e.g. [0,3,4]) def bitmask2cpulist(self, mask): cpu = 0 cpus = [] while mask > 0: if mask & 1: cpus.append(cpu) mask >>= 1 cpu += 1 return cpus # Unpacks CPU list, i.e. 1-3 will be converted to 1, 2, 3, supports # hexmasks that needs to be prefixed by "0x". Hexmasks can have commas, # which will be removed. If combining hexmasks with CPU list they need # to be separated by ",,", e.g.: 0-3, 0xf,, 6. It also supports negation # cpus by specifying "^" or "!", e.g.: 0-5, ^3, will output the list as # "0,1,2,4,5" (excluding 3). Note: negation supports only cpu numbers. # If "strip_chars" is not None and l is not list, we try strip characters. # It should be string with list of chars that is send to string.strip method # Default is english single and double quotes ("') rhbz#1891036 def cpulist_unpack(self, l, strip_chars='\'"'): rl = [] if l is None: return l ll = l if type(ll) is not list: if strip_chars is not None: ll = str(ll).strip(strip_chars) ll = str(ll).split(",") ll2 = [] negation_list = [] hexmask = False hv = "" # Remove commas from hexmasks for v in ll: sv = str(v) if hexmask: if len(sv) == 0: hexmask = False ll2.append(hv) hv = "" else: hv += sv else: if sv[0:2].lower() == "0x": hexmask = True hv = sv elif sv and (sv[0] == "^" or sv[0] == "!"): nl = sv[1:].split("-") try: if (len(nl) > 1): negation_list += list(range( int(nl[0]), int(nl[1]) + 1 ) ) else: negation_list.append(int(sv[1:])) except ValueError: return [] else: if len(sv) > 0: ll2.append(sv) if len(hv) > 0: ll2.append(hv) for v in ll2: vl = v.split("-") if v[0:2].lower() == "0x": rl += self.hex2cpulist(v) else: try: if len(vl) > 1: rl += list(range(int(vl[0]), int(vl[1]) + 1)) else: rl.append(int(vl[0])) except ValueError: return [] cpu_list = sorted(list(set(rl))) # Remove negated cpus after expanding for cpu in negation_list: if cpu in cpu_list: cpu_list.remove(cpu) return cpu_list # Packs CPU list, i.e. 1, 2, 3 will be converted to 1-3. It unpacks the # CPU list through cpulist_unpack first, so see its description about the # details of the input syntax def cpulist_pack(self, l): l = self.cpulist_unpack(l) if l is None or len(l) == 0: return l i = 0 j = i rl = [] while i + 1 < len(l): if l[i + 1] - l[i] != 1: if j != i: rl.append(str(l[j]) + "-" + str(l[i])) else: rl.append(str(l[i])) j = i + 1 i += 1 if j + 1 < len(l): rl.append(str(l[j]) + "-" + str(l[-1])) else: rl.append(str(l[-1])) return rl # Inverts CPU list (i.e. makes its complement) def cpulist_invert(self, l): cpus = self.cpulist_unpack(l) online = self.cpulist_unpack(self.read_file("/sys/devices/system/cpu/online")) return list(set(online) - set(cpus)) # Converts CPU list to hexadecimal CPU mask def cpulist2hex(self, l): if l is None: return None ul = self.cpulist_unpack(l) if ul is None: return None m = self.cpulist2bitmask(ul) s = "%x" % m ls = len(s) if ls % 8 != 0: ls += 8 - ls % 8 s = s.zfill(ls) return ",".join(s[i:i + 8] for i in range(0, len(s), 8)) def cpulist2bitmask(self, l): m = 0 for v in l: m \|= pow(2, v) return m def cpulist2string(self, l, prefix = ""): return ",".join((prefix + str(v)) for v in l) # Converts string s consisting of "dev1,dev2,dev3,..." to list ["dev1", "dev2, "dev3", ...], # whitespaces are ignored, cpu lists are supported with the prefix "cpulist:", e.g. # "cpulist:0-2,4" is converted to ["cpu0", "cpu1", "cpu2", "cpu4"]. If device name starts # with "cpulist:" write it as "cpulist:cpulist:". Escape commas in name with the "\,". def devstr2devs(self, s): if s[0:8].lower() == "cpulist:": s = s[8:] if s[0:8].lower() != "cpulist:": return [("cpu" + str(v)) for v in self.cpulist_unpack(s)] l = re.split(r"\s(?<!\\),\s", s) return [str(v).replace(r"\,", ",") for v in l] # Do not make balancing on patched Python 2 interpreter (rhbz#1028122). # It means less CPU usage on patched interpreter. On non-patched interpreter # it is not allowed to sleep longer than 50 ms. def wait(self, terminate, time): try: return terminate.wait(time, False) except TypeError as e: return terminate.wait(time) def get_size(self, s): s = str(s).strip().upper() for unit in ["KB", "MB", "GB", ""]: unit_ix = s.rfind(unit) if unit_ix == -1: continue try: val = int(s[:unit_ix]) u = s[unit_ix:] if u == "KB": val = 1024 elif u == "MB": val = 1024 * 1024 elif u == "GB": val = 1024 1024 * 1024 elif u != "": val = None return val except ValueError: return None def get_active_profile(self): profile_name = "" mode = "" try: with open(consts.ACTIVE_PROFILE_FILE, "r") as f: profile_name = f.read().strip() except IOError as e: if e.errno != errno.ENOENT: raise TunedException("Failed to read active profile: %s" % e) except (OSError, EOFError) as e: raise TunedException("Failed to read active profile: %s" % e) try: with open(consts.PROFILE_MODE_FILE, "r") as f: mode = f.read().strip() if mode not in ["", consts.ACTIVE_PROFILE_AUTO, consts.ACTIVE_PROFILE_MANUAL]: raise TunedException("Invalid value in file %s." % consts.PROFILE_MODE_FILE) except IOError as e: if e.errno != errno.ENOENT: raise TunedException("Failed to read profile mode: %s" % e) except (OSError, EOFError) as e: raise TunedException("Failed to read profile mode: %s" % e) if mode == "": manual = None else: manual = mode == consts.ACTIVE_PROFILE_MANUAL if profile_name == "": profile_name = None return (profile_name, manual) def save_active_profile(self, profile_name, manual): try: with open(consts.ACTIVE_PROFILE_FILE, "w") as f: if profile_name is not None: f.write(profile_name + "\n") except (OSError,IOError) as e: raise TunedException("Failed to save active profile: %s" % e.strerror) try: with open(consts.PROFILE_MODE_FILE, "w") as f: mode = consts.ACTIVE_PROFILE_MANUAL if manual else consts.ACTIVE_PROFILE_AUTO f.write(mode + "\n") except (OSError,IOError) as e: raise TunedException("Failed to save profile mode: %s" % e.strerror) def get_post_loaded_profile(self): profile_name = "" try: with open(consts.POST_LOADED_PROFILE_FILE, "r") as f: profile_name = f.read().strip() except IOError as e: if e.errno != errno.ENOENT: raise TunedException("Failed to read the active post-loaded profile: %s" % e) except (OSError, EOFError) as e: raise TunedException("Failed to read the active post-loaded profile: %s" % e) if profile_name == "": profile_name = None return profile_name def save_post_loaded_profile(self, profile_name): try: with open(consts.POST_LOADED_PROFILE_FILE, "w") as f: if profile_name is not None: f.write(profile_name + "\n") except (OSError,IOError) as e: raise TunedException("Failed to save the active post-loaded profile: %s" % e.strerror) # Translates characters in 'text' from 'source_chars' to 'dest_chars' def tr(self, text, source_chars, dest_chars): try: trans = str.maketrans(source_chars, dest_chars) except AttributeError: import string trans = string.maketrans(source_chars, dest_chars) return text.translate(trans) # Checks if name contains only valid characters and has valid length or is empty string or None def is_valid_name(self, name): return not name or (all(c in consts.NAMES_ALLOWED_CHARS for c in name) and len(name) <= consts.NAMES_MAX_LENGTH) def getconf(self, variable): return check_output(["getconf", variable]).decode().strip() # Gets list of available CPUs def get_cpus(self): cpus = self.read_file(consts.SYSFS_CPUS_PRESENT_PATH) # fallback to single core CPU if sysfs is unavailable return self.cpulist_unpack(cpus) if cpus else [ 0 ]	16.84 KB	2025-08-24 21:48
📄 config_parser.py # ConfigParser wrapper providing compatibility layer for python 2.7/3 try: python3 = True import configparser as cp except ImportError: python3 = False import ConfigParser as cp from StringIO import StringIO import re class Error(cp.Error): pass if python3: class ConfigParser(cp.ConfigParser): pass else: class ConfigParser(cp.ConfigParser): def __init__(self, delimiters=None, inline_comment_prefixes=None, strict=False, args, kwargs): delims = "".join(list(delimiters)) # REs taken from the python-2.7 ConfigParser self.OPTCRE = re.compile( r'(?P<option>[^' + delims + r'\s][^' + delims + '])' r'\s(?P<vi>[' + delims + r'])\s' r'(?P<value>.)$' ) self.OPTCRE_NV = re.compile( r'(?P<option>[^' + delims + r'\s][^' + delims + '])' r'\s(?:' r'(?P<vi>[' + delims + r'])\s' r'(?P<value>.))?$' ) cp.ConfigParser.__init__(self, args, *kwargs) self._inline_comment_prefixes = inline_comment_prefixes or [] self._re = re.compile(r"\s+(%s)." % ")\|(".join(list(self._inline_comment_prefixes))) def read_string(self, string, source="<string>"): sfile = StringIO(string) self.read_file(sfile, source) def readfp(self, fp, filename=None): cp.ConfigParser.readfp(self, fp, filename) # remove inline comments all_sections = [self._defaults] all_sections.extend(self._sections.values()) for options in all_sections: for name, val in options.items(): options[name] = self._re.sub("", val) def read_file(self, f, source="<???>"): self.readfp(f, source)	1.53 KB	2025-08-24 21:48
📄 global_config.py import re import tuned.logs from tuned.utils.config_parser import ConfigParser, Error from tuned.exceptions import TunedException import tuned.consts as consts from tuned.utils.commands import commands __all__ = ["GlobalConfig"] log = tuned.logs.get() class GlobalConfig(): def __init__(self,config_file = consts.GLOBAL_CONFIG_FILE): self._cfg = {} self.load_config(file_name=config_file) self._cmd = commands() @staticmethod def get_global_config_spec(): """ Easy validation mimicking configobj Returns two dicts, first with default values (default None) global_default[consts.CFG_SOMETHING] = consts.CFG_DEF_SOMETHING or None second with configobj function for value type (default "get" for string, others eg getboolean, getint) global_function[consts.CFG_SOMETHING] = consts.CFG_FUNC_SOMETHING or get } """ options = [opt for opt in dir(consts) if opt.startswith("CFG_") and not opt.startswith("CFG_FUNC_") and not opt.startswith("CFG_DEF_")] global_default = dict((getattr(consts, opt), getattr(consts, "CFG_DEF_" + opt[4:], None)) for opt in options) global_function = dict((getattr(consts, opt), getattr(consts, "CFG_FUNC_" + opt[4:], "get")) for opt in options) return global_default, global_function def load_config(self, file_name = consts.GLOBAL_CONFIG_FILE): """ Loads global configuration file. """ log.debug("reading and parsing global configuration file '%s'" % file_name) try: config_parser = ConfigParser(delimiters=('='), inline_comment_prefixes=('#'), strict=False) config_parser.optionxform = str with open(file_name) as f: config_parser.read_string("[" + consts.MAGIC_HEADER_NAME + "]\n" + f.read(), file_name) self._cfg, _global_config_func = self.get_global_config_spec() for option in config_parser.options(consts.MAGIC_HEADER_NAME): if option in self._cfg: try: func = getattr(config_parser, _global_config_func[option]) self._cfg[option] = func(consts.MAGIC_HEADER_NAME, option) except Error: raise TunedException("Global TuneD configuration file '%s' is not valid." % file_name) else: log.info("Unknown option '%s' in global config file '%s'." % (option, file_name)) self._cfg[option] = config_parser.get(consts.MAGIC_HEADER_NAME, option, raw=True) except IOError as e: raise TunedException("Global TuneD configuration file '%s' not found." % file_name) except Error as e: raise TunedException("Error parsing global TuneD configuration file '%s'." % file_name) def get(self, key, default = None): return self._cfg.get(key, default) def get_bool(self, key, default = None): if self._cmd.get_bool(self.get(key, default)) == "1": return True return False def get_int(self, key, default = 0): i = self._cfg.get(key, default) if i: if isinstance(i, int): return i else: try: return int(i, 0) except ValueError: log.error("Error parsing integer '%s', using '%d'." %(str(i), default)) return default def get_list(self, key, default = []): value = self._cfg.get(key, default) if isinstance(value, list): return value if value.strip() == "": return [] return [x.strip() for x in re.split(r",\|;", value)] def set(self, key, value): self._cfg[key] = value def get_size(self, key, default = None): val = self.get(key) if val is None: return default ret = self._cmd.get_size(val) if ret is None: log.error("Error parsing value '%s', using '%s'." %(val, default)) return default else: return ret	3.48 KB	2025-08-24 21:48
📄 nettool.py __all__ = ["ethcard"] import tuned.logs from subprocess import * import re log = tuned.logs.get() class Nettool: _advertise_values = { # [ half, full ] 10 : [ 0x001, 0x002 ], 100 : [ 0x004, 0x008 ], 1000 : [ 0x010, 0x020 ], 2500 : [ 0, 0x8000 ], 10000 : [ 0, 0x1000 ], "auto" : 0x03F } _disabled = False def __init__(self, interface): self._interface = interface; self.update() log.debug("%s: speed %s, full duplex %s, autoneg %s, link %s" % (interface, self.speed, self.full_duplex, self.autoneg, self.link)) log.debug("%s: supports: autoneg %s, modes %s" % (interface, self.supported_autoneg, self.supported_modes)) log.debug("%s: advertises: autoneg %s, modes %s" % (interface, self.advertised_autoneg, self.advertised_modes)) # def __del__(self): # if self.supported_autoneg: # self._set_advertise(self._advertise_values["auto"]) def _clean_status(self): self.speed = 0 self.full_duplex = False self.autoneg = False self.link = False self.supported_modes = [] self.supported_autoneg = False self.advertised_modes = [] self.advertised_autoneg = False def _calculate_mode(self, modes): mode = 0; for m in modes: mode += self._advertise_values[m[0]][ 1 if m[1] else 0 ] return mode def _set_autonegotiation(self, enable): if self.autoneg == enable: return True if not self.supported_autoneg: return False return 0 == call(["ethtool", "-s", self._interface, "autoneg", "on" if enable else "off"], close_fds=True) def _set_advertise(self, value): if not self._set_autonegotiation(True): return False return 0 == call(["ethtool", "-s", self._interface, "advertise", "0x%03x" % value], close_fds=True) def get_max_speed(self): max = 0 for mode in self.supported_modes: if mode[0] > max: max = mode[0] if max > 0: return max else: return 1000 def set_max_speed(self): if self._disabled or not self.supported_autoneg: return False #if self._set_advertise(self._calculateMode(self.supported_modes)): if self._set_advertise(self._advertise_values["auto"]): self.update() return True else: return False def set_speed(self, speed): if self._disabled or not self.supported_autoneg: return False mode = 0 for am in self._advertise_values: if am == "auto": continue if am <= speed: mode += self._advertise_values[am][0]; mode += self._advertise_values[am][1]; effective_mode = mode & self._calculate_mode(self.supported_modes) log.debug("%s: set_speed(%d) - effective_mode 0x%03x" % (self._interface, speed, effective_mode)) if self._set_advertise(effective_mode): self.update() return True else: return False def update(self): if self._disabled: return # run ethtool and preprocess output p_ethtool = Popen(["ethtool", self._interface], \ stdout=PIPE, stderr=PIPE, close_fds=True, \ universal_newlines = True) p_filter = Popen(["sed", r"s/^\s//;s/:\s/:\n/g"], \ stdin=p_ethtool.stdout, stdout=PIPE, \ universal_newlines = True, \ close_fds=True) output = p_filter.communicate()[0] errors = p_ethtool.communicate()[1] if errors != "": log.warning("%s: some errors were reported by 'ethtool'" % self._interface) log.debug("%s: %s" % (self._interface, errors.replace("\n", r"\n"))) self._clean_status() self._disabled = True return # parses output - kind of FSM self._clean_status() re_speed = re.compile(r"(\d+)") re_mode = re.compile(r"(\d+)baseT/(Half\|Full)") state = "wait" for line in output.split("\n"): if line.endswith(":"): section = line[:-1] if section == "Speed": state = "speed" elif section == "Duplex": state = "duplex" elif section == "Auto-negotiation": state = "autoneg" elif section == "Link detected": state = "link" elif section == "Supported link modes": state = "supported_modes" elif section == "Supports auto-negotiation": state = "supported_autoneg" elif section == "Advertised link modes": state = "advertised_modes" elif section == "Advertised auto-negotiation": state = "advertised_autoneg" else: state = "wait" del section elif state == "speed": # Try to determine speed. If it fails, assume 1gbit ethernet try: self.speed = re_speed.match(line).group(1) except: self.speed = 1000 state = "wait" elif state == "duplex": self.full_duplex = line == "Full" state = "wait" elif state == "autoneg": self.autoneg = (line == "yes" or line == "on") state = "wait" elif state == "link": self.link = line == "yes" state = "wait" elif state == "supported_modes": # Try to determine supported modes. If it fails, assume 1gibt ethernet fullduplex works try: for m in line.split(): (s, d) = re_mode.match(m).group(1,2) self.supported_modes.append( (int(s), d == "Full") ) del m,s,d except: self.supported_modes.append((1000, True)) elif state == "supported_autoneg": self.supported_autoneg = line == "Yes" state = "wait" elif state == "advertised_modes": # Try to determine advertised modes. If it fails, assume 1gibt ethernet fullduplex works try: if line != "Not reported": for m in line.split(): (s, d) = re_mode.match(m).group(1,2) self.advertised_modes.append( (int(s), d == "Full") ) del m,s,d except: self.advertised_modes.append((1000, True)) elif state == "advertised_autoneg": self.advertised_autoneg = line == "Yes" state = "wait" def ethcard(interface): if not interface in ethcard.list: ethcard.list[interface] = Nettool(interface) return ethcard.list[interface] ethcard.list = {}	5.57 KB	2025-08-24 21:48
📄 polkit.py import dbus import tuned.logs log = tuned.logs.get() class polkit(): def __init__(self): self._bus = dbus.SystemBus() self._proxy = self._bus.get_object('org.freedesktop.PolicyKit1', '/org/freedesktop/PolicyKit1/Authority', follow_name_owner_changes = True) self._authority = dbus.Interface(self._proxy, dbus_interface='org.freedesktop.PolicyKit1.Authority') def check_authorization(self, sender, action_id): """Check authorization, return codes: 1 - authorized 2 - polkit error, but authorized with fallback method 0 - unauthorized -1 - polkit error and unauthorized by the fallback method -2 - polkit error and unable to use the fallback method """ if sender is None or action_id is None: return False details = {} flags = 1 # AllowUserInteraction flag cancellation_id = "" # No cancellation id subject = ("system-bus-name", {"name" : sender}) try: ret = self._authority.CheckAuthorization(subject, action_id, details, flags, cancellation_id)[0] except (dbus.exceptions.DBusException, ValueError) as e: log.error("error querying polkit: %s" % e) # No polkit or polkit error, fallback to always allow root try: uid = self._bus.get_unix_user(sender) except dbus.exceptions.DBusException as e: log.error("error using fallback authorization method: %s" % e) return -2 if uid == 0: return 2 else: return -1 return 1 if ret else 0	1.4 KB	2025-08-24 21:48
📄 profile_recommender.py import os import re import errno import procfs import subprocess from tuned.utils.config_parser import ConfigParser, Error try: import syspurpose.files have_syspurpose = True except: have_syspurpose = False import tuned.consts as consts import tuned.logs from tuned.utils.commands import commands log = tuned.logs.get() class ProfileRecommender: def __init__(self, is_hardcoded = False): self._is_hardcoded = is_hardcoded self._commands = commands() self._chassis_type = None def recommend(self): profile = consts.DEFAULT_PROFILE if self._is_hardcoded: return profile has_root = os.geteuid() == 0 if not has_root: log.warning("Profile recommender is running without root privileges. Profiles with virt recommendation condition will be omitted.") matching = self.process_config(consts.RECOMMEND_CONF_FILE, has_root=has_root) if matching is not None: return matching files = {} for directory in consts.RECOMMEND_DIRECTORIES: contents = [] try: contents = os.listdir(directory) except OSError as e: if e.errno != errno.ENOENT: log.error("error accessing %s: %s" % (directory, e)) for name in contents: path = os.path.join(directory, name) files[name] = path for name in sorted(files.keys()): path = files[name] matching = self.process_config(path, has_root=has_root) if matching is not None: return matching return profile def process_config(self, fname, has_root=True): matching_profile = None syspurpose_error_logged = False try: if not os.path.isfile(fname): return None config = ConfigParser(delimiters=('='), inline_comment_prefixes=('#'), strict=False) config.optionxform = str with open(fname) as f: config.read_file(f, fname) for section in config.sections(): match = True for option in config.options(section): value = config.get(section, option, raw=True) if value == "": value = r"^$" if option == "virt": if not has_root: match = False break if not re.match(value, self._commands.execute(["virt-what"])[1], re.S): match = False elif option == "system": if not re.match(value, self._commands.read_file( consts.SYSTEM_RELEASE_FILE, no_error = True), re.S): match = False elif option[0] == "/": if not os.path.exists(option) or not re.match(value, self._commands.read_file(option), re.S): match = False elif option[0:7] == "process": ps = procfs.pidstats() ps.reload_threads() if len(ps.find_by_regex(re.compile(value))) == 0: match = False elif option == "chassis_type": chassis_type = self._get_chassis_type() if not re.match(value, chassis_type, re.IGNORECASE): match = False elif option == "syspurpose_role": role = "" if have_syspurpose: s = syspurpose.files.SyspurposeStore( syspurpose.files.USER_SYSPURPOSE, raise_on_error = True) try: s.read_file() role = s.contents["role"] except (IOError, OSError, KeyError) as e: if hasattr(e, "errno") and e.errno != errno.ENOENT: log.error("Failed to load the syspurpose\ file: %s" % e) else: if not syspurpose_error_logged: log.error("Failed to process 'syspurpose_role' in '%s'\ , the syspurpose module is not available" % fname) syspurpose_error_logged = True if re.match(value, role, re.IGNORECASE) is None: match = False if match: # remove the ",." suffix r = re.compile(r",[^,]$") matching_profile = r.sub("", section) break except (IOError, OSError, Error) as e: log.error("error processing '%s', %s" % (fname, e)) return matching_profile def _get_chassis_type(self): if self._chassis_type is not None: log.debug("returning cached chassis type '%s'" % self._chassis_type) return self._chassis_type # Check DMI sysfs first # Based on SMBios 3.3.0 specs (https://www.dmtf.org/sites/default/files/standards/documents/DSP0134_3.3.0.pdf) DMI_CHASSIS_TYPES = ["", "Other", "Unknown", "Desktop", "Low Profile Desktop", "Pizza Box", "Mini Tower", "Tower", "Portable", "Laptop", "Notebook", "Hand Held", "Docking Station", "All In One", "Sub Notebook", "Space-saving", "Lunch Box", "Main Server Chassis", "Expansion Chassis", "Sub Chassis", "Bus Expansion Chassis", "Peripheral Chassis", "RAID Chassis", "Rack Mount Chassis", "Sealed-case PC", "Multi-system", "CompactPCI", "AdvancedTCA", "Blade", "Blade Enclosing", "Tablet", "Convertible", "Detachable", "IoT Gateway", "Embedded PC", "Mini PC", "Stick PC"] try: with open('/sys/devices/virtual/dmi/id/chassis_type', 'r') as sysfs_chassis_type: chassis_type_id = int(sysfs_chassis_type.read()) self._chassis_type = DMI_CHASSIS_TYPES[chassis_type_id] except IndexError: log.error("Unknown chassis type id read from dmi sysfs: %d" % chassis_type_id) except (OSError, IOError) as e: log.warning("error accessing dmi sysfs file: %s" % e) if self._chassis_type: log.debug("chassis type - %s" % self._chassis_type) return self._chassis_type # Fallback - try parsing dmidecode output try: p_dmi = subprocess.Popen(['dmidecode', '-s', 'chassis-type'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True) (dmi_output, dmi_error) = p_dmi.communicate() if p_dmi.returncode: log.error("dmidecode finished with error (ret %d): '%s'" % (p_dmi.returncode, dmi_error)) else: self._chassis_type = dmi_output.strip().decode() except (OSError, IOError) as e: log.warning("error executing dmidecode tool : %s" % e) if not self._chassis_type: log.debug("could not determine chassis type.") self._chassis_type = "" else: log.debug("chassis type - %s" % self._chassis_type) return self._chassis_type	5.81 KB	2025-08-24 21:48

Terminal