"""Minimal HTTP server that reads/writes Thread credentials via DBUS. The Thread Border Router daemon must be built with its DBUS interface. This script must run in the same DBUS domain as the TBR daemon, which usually means the same machine. The script exposes the following HTTP endpoints: - GET /node to retrieve human-readable information on the credentials - GET /node/dataset/active to retrieve credentials to the pending dataset in a JSON structure that contains the hex-encoded credentials - PUT /node/dataset/pending to set credentials from the active dataset in JSON format that contains the hex-encoded credentials The script usually needs to be run as a privileged user. """ import http.server import json import socketserver import sys import dbus def call_dbus_method(interface, method_name, *arguments): """OTBR DBUS interface for method call.""" bus = dbus.SystemBus() obj = bus.get_object( 'io.openthread.BorderRouter.wpan0', '/io/openthread/BorderRouter/wpan0' ) iface = dbus.Interface(obj, interface) method = getattr(iface, method_name) res = method(*arguments) return res def get_dbus_property(property_name): """OTBR DBUS interface for reading properties.""" return call_dbus_method( 'org.freedesktop.DBus.Properties', 'Get', 'io.openthread.BorderRouter', property_name, ) def set_dbus_property(property_name, property_value): """OTBR DBUS interface for setting properties.""" return call_dbus_method( 'org.freedesktop.DBus.Properties', 'Set', 'io.openthread.BorderRouter', property_name, property_value, ) class MyHandler(http.server.SimpleHTTPRequestHandler): """Handles the HTTP requests for GET and PUT on different endpoints.""" def do_get_readable(self): """Returns credentials in human-readable format.""" self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() # prints human-readable credentials network_key = '' for el in get_dbus_property('NetworkKey'): network_key += f'{el:02x}' self.wfile.write( 'My HTTP Server\n' '

Thread Credentials

\n' f'NetworkName: {get_dbus_property("NetworkName")}
\n' f'PanId: 0x{get_dbus_property("PanId"):x}
\n' f'ExtPanId: {get_dbus_property("ExtPanId"):x}
\n' 'PSKc: not available
\n' f'NetworkKey: {network_key}
\n' f'Channel: {get_dbus_property("Channel")}
\n' ''.encode('utf8') ) def do_get_creds(self): """Returns credentials in hex-encoded TLV.""" # prints TLV credentials self.send_response(200) self.send_header('Content-type', 'text/plain') self.end_headers() active_dataset_tlvs = '' for el in get_dbus_property('ActiveDatasetTlvs'): active_dataset_tlvs += f'{el:02X}' self.wfile.write(f'{active_dataset_tlvs}'.encode('utf8')) def do_GET(self): """Handles GET requests.""" print(self.path) if self.path == '/node/dataset/active': self.do_get_creds() elif self.path == '/node': self.do_get_readable() def do_PUT(self): # pylint: disable=invalid-name """Handles PUT requests.""" if self.path == '/node/dataset/pending': content_length = int(self.headers['Content-Length']) # decodes payload post_data = self.rfile.read(content_length).decode('utf8') payload = json.loads(post_data) active_dataset = [ dbus.Byte(b) for b in bytes.fromhex(payload['ActiveDataset']) ] active_dataset = dbus.Array( active_dataset, signature=dbus.Signature('y'), variant_level=1 ) # delay via dbus interface is 300s, set at # https://github.com/openthread/ot-br-posix/blob/84c6aff23814bd67d4866d27379b39f8a0df91b5/src/utils/thread_helper.cpp#L650 # The DBUS message will be rejected by OTBR agent if either: # - the same credentials are already active # - there's a set of pending credentials executing its delay period call_dbus_method( 'io.openthread.BorderRouter', 'AttachAllNodesTo', active_dataset ) # sends 200 OK response self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write( 'HTTP Server

Credentials' ' committed

'.encode('utf8') ) if __name__ == '__main__': PORT = int(sys.argv[1]) Handler = MyHandler httpd = socketserver.TCPServer(('', PORT), Handler) print('serving at port', PORT) httpd.serve_forever()