"""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 ServerCredentials'
' 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()