From patchwork Tue Dec 19 18:37:02 2023 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Denis Kenzior X-Patchwork-Id: 13498853 Received: from mail-io1-f49.google.com (mail-io1-f49.google.com [209.85.166.49]) (using TLSv1.2 with cipher ECDHE-RSA-AES128-GCM-SHA256 (128/128 bits)) (No client certificate requested) by smtp.subspace.kernel.org (Postfix) with ESMTPS id 151D3381AD for ; Tue, 19 Dec 2023 18:41:10 +0000 (UTC) Authentication-Results: smtp.subspace.kernel.org; dmarc=pass (p=none dis=none) header.from=gmail.com Authentication-Results: smtp.subspace.kernel.org; spf=pass smtp.mailfrom=gmail.com Authentication-Results: smtp.subspace.kernel.org; dkim=pass (2048-bit key) header.d=gmail.com header.i=@gmail.com header.b="ZwXgaDZQ" Received: by mail-io1-f49.google.com with SMTP id ca18e2360f4ac-7b3708b3eacso186434239f.2 for ; Tue, 19 Dec 2023 10:41:10 -0800 (PST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20230601; t=1703011270; x=1703616070; darn=lists.linux.dev; h=content-transfer-encoding:mime-version:references:in-reply-to :message-id:date:subject:cc:to:from:from:to:cc:subject:date :message-id:reply-to; bh=UPASXJ0yU69si0HPT8jG4GVlI9QDdKyX0KfLIH7ipFc=; b=ZwXgaDZQarMpjYyT9MUFI6p77rKcB3rSvfZ3GYB2RVrASrMKiKUVSswl0IF/KDqqqF RoIrFCL2iSuXaSwSShWX2bNjP1uR8SFFVoPu/fRPvYEWE8Rm1cwoBM1EyJpLV9W3gUu9 HEn5HXJIbxzYSMYMuY3FDM1cBBly9R3DE7NgtSMEBN+Hcjc5jx0faj5kg4HdTzCnERPy 9Jvyo6IoHOuavn4oNkDKWrN7GZDpOQWBcq9BJZC+fq2SarnGTR9vki4pGCTOBTBLM4LP Kc3y9N8H2KySsuLumJNNCdMPmGf9nVXMeOfdiQUYg8VUV2FVafuGkvN/QXcjmtThNqky BRXA== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20230601; t=1703011270; x=1703616070; h=content-transfer-encoding:mime-version:references:in-reply-to :message-id:date:subject:cc:to:from:x-gm-message-state:from:to:cc :subject:date:message-id:reply-to; bh=UPASXJ0yU69si0HPT8jG4GVlI9QDdKyX0KfLIH7ipFc=; b=kMDjaoA1d5s8FGlg3KjvLX517K3HyRxDTJljrTwE9efD+NMenNWhfaZs2PCHmxjMxw i8zL4zqs9BvgpH+UbteR1ztGD2cVrl5lS/Nun5JePhzDibIYoUVylxlTPT/ljR0qns+x TXKYoTnRzjQQ1wKDDID4ls9uV/IDQ8l2CqwlPuD5Z1y9snGD++h4uf00f1Q20BhIg6sC RoQ/i6C970smTKEOIeqp+DLFdWGlXhk6d5tgKl/B7Lz7d7iKPOwM1wDLCQ7WHQQRLzGk JP8Q31GBw8IF7RYlclT+3oIUY+hLH98sVSGDBnXZCGKX5hxNSbb5IAIQJS49x1BPYH4Q 07aQ== X-Gm-Message-State: AOJu0YwIZBjg+RI2NXjCpbR4RHST1BBlr2za7ZqOAFwzq3xn5NjNfWqy cPQeAVuCs0Fct+ggv8jRZy+CfRtqrvE= X-Google-Smtp-Source: AGHT+IH4t4JiWtnzdXVozMWRQ0uyZHt29+irUNCP9stbi1sVncbiLiQkjhgiCipBx/8x9Od4yH+0qg== X-Received: by 2002:a6b:a16:0:b0:7b7:cd8:7a6c with SMTP id z22-20020a6b0a16000000b007b70cd87a6cmr19504775ioi.35.1703011270001; Tue, 19 Dec 2023 10:41:10 -0800 (PST) Received: from localhost.localdomain ([136.33.23.24]) by smtp.gmail.com with ESMTPSA id co13-20020a0566383e0d00b0046b406d9d95sm1549213jab.38.2023.12.19.10.41.07 (version=TLS1_3 cipher=TLS_AES_256_GCM_SHA384 bits=256/256); Tue, 19 Dec 2023 10:41:07 -0800 (PST) From: Denis Kenzior To: ofono@lists.linux.dev Cc: Denis Kenzior Subject: [PATCH v2 05/15] tools: Add provision.py Date: Tue, 19 Dec 2023 12:37:02 -0600 Message-ID: <20231219184016.420116-5-denkenz@gmail.com> X-Mailer: git-send-email 2.43.0 In-Reply-To: <20231219184016.420116-1-denkenz@gmail.com> References: <20231219184016.420116-1-denkenz@gmail.com> Precedence: bulk X-Mailing-List: ofono@lists.linux.dev List-Id: List-Subscribe: List-Unsubscribe: MIME-Version: 1.0 Introduce a new tool that will convert the intermediate JSON format (documented in doc/provision.rst) to the binary provisioning format, which will be used by ofonod's default provisioning plugin for automatically setting up carrier specific settings. The tool also supports import of and conversion of 'mobile-broadband-provider-info' XML files to the new intermediate JSON file format. This is accomplished using: % tools/provisiontool mbpi-convert --outfile=provision.json Conversion of JSON intermediate format to binary format is accomplished using: % tools/provisiontool generate --infile=provision.json By default, the output will be placed in the same directory in the file 'provision.db'. Alternatively, the output file can be specified using the --outfile option. Finally, the tool supports a simple selftest method, which can be invoked as follows: % tools/provisiontool selftest --- tools/provisiontool | 727 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 727 insertions(+) create mode 100755 tools/provisiontool v2 - Changed the name to tools/provisiontool instead of tools/provision.py diff --git a/tools/provisiontool b/tools/provisiontool new file mode 100755 index 00000000..79273277 --- /dev/null +++ b/tools/provisiontool @@ -0,0 +1,727 @@ +#!/usr/bin/python3 +# +# oFono - Open Source Telephony +# Copyright (C) 2023 Cruise, LLC +# +# SPDX-License-Identifier: GPL-2.0-only +import xml.etree.ElementTree as ET +import sys +import json +import bisect +from argparse import ArgumentParser, FileType +from pathlib import Path +import random +import struct +import ctypes + +class ProviderInfo: + sort_order_map = { v : pos for pos, v in + enumerate( ['name', + 'apn', + 'type', + 'protocol', + 'mmsc', + 'mmsproxy', + 'authentication', + 'username', + 'password'] ) } + + @classmethod + def rawimport(cls, entry): + if 'name' not in entry: + raise SystemExit('No name for entry: ' + str(entry)) + + info = ProviderInfo(entry['name']) + + for networkid in entry.get('ids', []): + if not info.add_id(networkid): + raise SystemExit('Invalid network id: ' + str(networkid)) + + if 'spn' in entry: + if not info.set_spn(entry['spn']): + raise SystemExit('Invalid spn: ' + str(spn)) + + for apn in entry.get('apns', []): + if not info.add_context(apn): + raise SystemExit('Invalid apn: ' + str(apn)) + + if not info.is_valid(): + raise SystemExit('Invalid entry: ' + str(entry)) + + return info + + def __init__(self, name): + self.context_list = [] + self.mccmnc_list = [] + self.name = name + self.spn = None + + @staticmethod + def is_valid_id(id_string, expected_lengths): + """ + Check if the identifier string is valid. + + Parameters: + - id_string: The id string to check. + - expected_lengths (tuple): A tuple representing the valid range of + lengths. + + Returns: + - bool: True if the MCC string is valid, False otherwise. + """ + if not id_string.isdigit(): + return False + + if len(id_string) not in expected_lengths: + return False + + if int(id_string) == 0: + return False + + return True + + def add_mccmnc(self, mcc, mnc): + if not self.is_valid_id(mcc, (3,)) or not self.is_valid_id(mnc, (2, 3)): + return False + + bisect.insort(self.mccmnc_list, mcc + mnc) + return True + + def add_id(self, mccmnc): + if not self.is_valid_id(mccmnc, (5,6)): + return False + + if int(mccmnc[:3]) == 0 or int(mccmnc[3:]) == 0: + return False + + bisect.insort(self.mccmnc_list, mccmnc) + return True + + def set_spn(self, spn): + if len(spn) == 0 or len(spn) > 254: + return False + + self.spn = spn + return True + + def add_context(self, info): + info = dict(sorted(info.items(), + key = lambda pair: self.sort_order_map[pair[0]])) + self.context_list.append(info) + + return True + + def is_valid(self): + return len(self.context_list) and len(self.mccmnc_list) + + def __str__(self): + s = 'Provider \'' + self.name + '\'' + + if (self.spn != None): + s += ' [SPN:\'' + self.spn + '\']' + + s+= ' ' + str(self.mccmnc_list) + '\n' + + for context in self.context_list: + s += '\t' + str(context) + '\n' + + return s + +class MobileBroadbandProviderInfo: + usage_to_type = { 'internet' : ['internet'], + 'mms' : ['mms'], + 'wap' : ['wap'], + 'mms-internet-hipri' : ['internet', 'mms'], + 'mms-internet-hipri-fota' : ['internet','mms'], + } + @classmethod + def type_from_usage(cls, usage): + return cls.usage_to_type[usage] + + def __init__(self, xml_path): + self.tree = ET.parse(xml_path) + + def parse(self, xml_path): + providers = [] + + try: + tree = ET.parse(xml_path) + root = tree.getroot() + + for provider in root.findall('.//provider'): + name = provider.find('name') + if name is None or not name.text: + continue; + + info = ProviderInfo(name.text) + + for networkid in provider.findall('gsm/network-id'): + info.add_mccmnc(networkid.get('mcc'), networkid.get('mnc')) + + for apn in provider.findall('gsm/apn'): + context = {} + + context['apn'] = apn.get('value') + if context['apn'] == None: + continue + + # Usage is missing for some APNs, skip such contexts for now + usage = apn.find('usage') + if usage is None or usage.get('type') is None: + continue; + + context['type'] = self.type_from_usage(usage.get('type')) + if context['type'] == None: + sys.stderr.write("Unable to convert type: %s\n" % + usage.get('type')) + continue + + if 'mms' in context['type']: + mmsc = apn.find('mmsc') + + # Ignore MMS contexts with no MMSC since it is needed + # to send messages + if mmsc is None or not mmsc.text: + continue + + context['mmsc'] = mmsc.text + + mmsproxy = apn.find('mmsproxy') + if mmsproxy is not None and mmsproxy.text: + context['mmsproxy'] = mmsproxy.text + + username = apn.find('username') + if username is not None and username.text: + context['username'] = username.text + + password = apn.find('password') + if password is not None and password.text: + context['password'] = password.text + + authentication = apn.find('authentication') + if authentication is not None: + context['authentication'] = authentication.get('method') + + context_name = apn.find('name') + if context_name != None: + context['name'] = context_name.text + + info.add_context(context) + + if info.is_valid(): + providers.append(info) + + except ET.ParseError as e: + print(f"Error parsing XML: {e}") + + return providers + +class ProvisionContext(ctypes.LittleEndianStructure): + _pack_ = 1 + _fields_ = [ + ('type', ctypes.c_uint32), + ('protocol', ctypes.c_uint32), + ('authentication', ctypes.c_uint32), + ('reserved', ctypes.c_uint32), + ('name_offset', ctypes.c_uint64), + ('apn_offset', ctypes.c_uint64), + ('username_offset', ctypes.c_uint64), + ('password_offset', ctypes.c_uint64), + ('mmsproxy_offset', ctypes.c_uint64), + ('mmsc_offset', ctypes.c_uint64) + ] + + authentication_dict = { 'chap' : 0, 'pap' : 1, 'none' : 2 } + protocol_dict = { 'ipv4' : 0, 'ipv6' : 1, 'ipv4v6' : 2 } + attrs = ['name', 'apn', 'username', 'password', 'mmsproxy', 'mmsc'] + + @classmethod + def type_to_context_type(cls, types): + r = 0 + + for t in types: + if t == 'internet': + r |= 0x0001 + elif t == 'mms': + r |= 0x0002 + elif t == 'wap': + r |= 0x0004 + elif t == 'ims': + r |= 0x0008 + elif t == 'supl': + r |= 0x0010 + elif t == 'ia': + r |= 0x0020 + + return r + + def __init__(self, apn, strings): + self.type = self.type_to_context_type(apn['type']) + self.protocol = self.protocol_dict[apn.get('protocol', 'ipv4v6')] + self.authentication = self.authentication_dict[apn.get('authentication', + 'chap')] + + for s in self.attrs: + offset = strings.add_string(apn.get(s, None)) + setattr(self, s + '_offset', offset) + +class ProvisionData(ctypes.LittleEndianStructure): + _pack_ = 1 + _fields_ = [ + ('spn_offset', ctypes.c_uint64), + ('context_offset', ctypes.c_uint64) + ] + + def __init__(self, spn, offset, strings): + self.spn_offset = strings.add_string(spn) + self.context_offset = offset + +class ProvisionNode(ctypes.LittleEndianStructure): + _pack_ = 1 + _fields_ = [ + ('bit_offsets', ctypes.c_uint64 * 2), + ('mccmnc', ctypes.c_uint32), + ('diff', ctypes.c_int32), + ('provision_data_count', ctypes.c_uint64) + ] + + style = "bold" + fmt_connection = '\t"%s/%d" -> "%s/%d"[color="#%06x"];\n' + fmt_declaration = '\t"%s/%d"[style=%s, color="#%06x"];\n' + red = 0xff0000 + green = 0x00ff00 + + def __init__(self, key, diff): + self.bit = [None, None] + self.key = key + self.diff = diff + self.entries = {} + self.node_offset = 0 + + def choose(self, key): + return (key >> (31 - self.diff)) & 1 + + def print_graphviz(self, f): + f.write(self.fmt_declaration % (format(self.key, '032b'), + self.diff, self.style, + random.randint(0, 0x00ffffff))) + f.write(self.fmt_connection % (format(self.key, '032b'), self.diff, + format(self.bit[0].key, '032b'), + self.bit[0].diff, self.red)) + f.write(self.fmt_connection % (format(self.key, '032b'), self.diff, + format(self.bit[1].key, '032b'), + self.bit[1].diff, self.green)) + + if (self.diff < self.bit[0].diff): + self.bit[0].print_graphviz(f) + + if (self.diff < self.bit[1].diff): + self.bit[1].print_graphviz(f) + + def __str__(self): + s = format(self.key, '032b') + '/' + str(self.diff) + return s + +class MccMncTree: + @staticmethod + def clz(v): + count = 32 + while count and v: + v = v >> 1 + count = count - 1 + + return count + + @staticmethod + def diff(key1, key2): + xor = key1 ^ key2; + return MccMncTree.clz(xor) + + def __init__(self): + self.root = ProvisionNode(key = 0, diff = -1) + self.root.bit[0] = self.root + self.root.bit[1] = self.root + self.n_nodes = 1 + + def print_graphviz(self): + f = open("step%d.dot" % self.n_nodes, "w") + # Use 'dot -Tx11' to visualize + f.write('digraph trie {\n') + self.root.print_graphviz(f) + f.write('}\n') + f.close() + + def find_closest(self, key): + parent = self.root + child = self.root.bit[0] + + while parent.diff < child.diff: + parent = child + child = child.bit[child.choose(key)] + + return child + + def find(self, key): + found = self.find_closest(key) + if found.key == key: + return found + + return None + + def insert(self, key, attr, value): + node = self.find_closest(key); + if node.key == key: + node.entries[attr] = value + return + + bit = self.diff(node.key, key) + parent = self.root + child = self.root.bit[0] + + while (parent.diff < child.diff) and (child.diff < bit): + parent = child + child = child.bit[child.choose(key)] + + node = ProvisionNode(key, bit) + bit = node.choose(key) + node.bit[bit] = node + node.bit[not bit] = child + + node.entries[attr] = value + + if parent == self.root: + self.root.bit[0] = node + else: + bit = parent.choose(key) + parent.bit[bit] = node + + self.n_nodes += 1 + + def traverse_recursive(self, node, bit, visitor): + if node == self.root: + return + + if node.diff <= bit: + visitor.visit(node) + return + + self.traverse_recursive(node.bit[0], node.diff, visitor) + self.traverse_recursive(node.bit[1], node.diff, visitor) + + def traverse(self, visitor): + self.traverse_recursive(self.root.bit[0], -1, visitor) + +class StringAccumulator: + def __init__(self): + self.data = bytearray(b'\x00') # So offsets are never 0 used for NULL + self.offsets = {} + + def add_string(self, s): + if s is None: + return 0 + + if s in self.offsets: + return self.offsets[s] + + offset = len(self.data) + self.data.extend(s.encode('utf-8')) + self.data.append(0) + self.offsets[s] = offset + + return offset + + def get_bytes(self): + return self.data + +class ProvisionDatabase(ctypes.LittleEndianStructure): + _pack_ = 1 + _fields_ = [ + ('version', ctypes.c_uint64), + ('file_size', ctypes.c_uint64), + ('header_size', ctypes.c_uint64), + ('node_struct_size', ctypes.c_uint64), + ('provision_data_struct_size', ctypes.c_uint64), + ('context_struct_size', ctypes.c_uint64), + ('nodes_offset', ctypes.c_uint64), + ('nodes_size', ctypes.c_uint64), + ('contexts_offset', ctypes.c_uint64), + ('contexts_size', ctypes.c_uint64), + ('strings_offset', ctypes.c_uint64), + ('strings_size', ctypes.c_uint64) + ] + + class CalculateNodeOffsetVisitor: + def __init__(self): + self.current_offset = 0 + def visit(self, node): + node.node_offset = self.current_offset + + # Node data is followed by at least one ProvisionData object, with + # the only exception being root, which has no data by definition + self.current_offset += ctypes.sizeof(ProvisionNode) + self.current_offset += (ctypes.sizeof(ProvisionData) * + len(node.entries)) + + class SerializeVisitor: + def __init__(self, buffer): + self.buffer = buffer + + def visit(self, node): + # Node doesn't quite fit the C structure definition, so do this + # manually by using struct.pack + self.buffer.extend(struct.pack('