#!/usr/bin/env python3 r""" See help text for details. """ import json import ssl import sys import requests import websocket from retrying import retry save_path_0 = sys.path[0] del sys.path[0] from gen_arg import * # NOQA from gen_print import * # NOQA from gen_valid import * # NOQA # Restore sys.path[0]. sys.path.insert(0, save_path_0) # Set exit_on_error for gen_valid functions. set_exit_on_error(True) parser = argparse.ArgumentParser( usage="%(prog)s [OPTIONS]", description="%(prog)s will open a websocket session on a remote OpenBMC. " + "When an eSEL is created on that BMC, the monitor will receive " + "notice over websocket that the eSEL was created " + "and it will print a message.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, prefix_chars="-+", ) parser.add_argument( "openbmc_host", default="", help="The BMC host name or IP address." ) parser.add_argument( "--openbmc_username", default="root", help="The userid for the open BMC system.", ) parser.add_argument( "--openbmc_password", default="", help="The password for the open BMC system.", ) parser.add_argument( "--monitor_type", choices=["logging", "dump"], default="logging", help="The type of notifications from websocket to monitor.", ) stock_list = [("test_mode", 0), ("quiet", 0), ("debug", 0)] def exit_function(signal_number=0, frame=None): r""" Execute whenever the program ends normally or with the signals that we catch (i.e. TERM, INT). """ qprint_dashes(width=160) qprint_executing() qprint_pgm_footer() def signal_handler(signal_number, frame): r""" Handle signals. Without a function to catch a SIGTERM or SIGINT, the program would terminate immediately with return code 143 and without calling the exit_function. """ # Our convention is to set up exit_function with atexit.register() so # there is no need to explicitly call exit_function from here. dprint_executing() # Calling exit prevents returning to the code that was running # when the signal was received. exit(0) def validate_parms(): r""" Validate program parameters, etc. """ register_passwords(openbmc_password) valid_value(openbmc_host) valid_value(openbmc_username) valid_value(openbmc_password) global monitoring_uri monitoring_uri = "/xyz/openbmc_project/" + monitor_type gen_post_validation(exit_function, signal_handler) @retry(stop_max_attempt_number=3, wait_fixed=1000) def login(openbmc_host, openbmc_username, openbmc_password): r""" Log into the BMC and return the session object. Description of argument(s): openbmc_host The BMC host name or IP address. openbmc_username The userid for the open BMC system. openbmc_password The password for the open BMC system. """ qprint_executing() http_header = {"Content-Type": "application/json"} session = requests.session() response = session.post( "https://" + openbmc_host + "/login", headers=http_header, json={"data": [openbmc_username, openbmc_password]}, verify=False, timeout=30, ) valid_value(response.status_code, valid_values=[200]) login_response = json.loads(response.text) qprint_var(login_response) valid_value(login_response["status"], valid_values=["ok"]) return session def on_message(websocket_obj, message): """ Websocket message handler. Close the websocket if the message is an eSEL message. Description of argument(s): websocket_obj The websocket established during opne_socket(). message The message sent from the websocket interface. """ qprint_dashes(width=160) qprint_executing() # A typical message: # /xyz/openbmc_project/logging/entry/24","properties":{"Id":24}} # or # /xyz/openbmc_project/dump/entry/1","properties":{"Size":186180}}'). if monitoring_uri + "/entry" in message: if "Id" in message: qprint_timen("eSEL received over websocket interface.") websocket_obj.close() elif "Size" in message: qprint_timen( "Dump notification received over websocket interface." ) websocket_obj.close() def on_error(websocket_obj, wserror): """ Websocket error handler. This routine is called whenever the websocket interfaces wishes to report an issue. Description of argument(s): websocket_obj The websocket established during opne_socket(). wserror The error message sent from the websocket interface. """ # It is normal to receive this message when websocked closes: # 'NoneType' object has no attribute 'connected'. qprint_dashes(width=160) qprint_executing() def on_close(websocket_obj): """ Websocket close event handler. Description of argument(s): websocket_obj The websocket established during opne_socket(). """ qprint_dashes(width=160) qprint_executing() def on_open(websocket_obj): """ Send the filters needed to listen to the logging interface. Description of argument(s): websocket_obj The websocket established during opne_socket(). """ qprint_dashes(width=160) qprint_executing() data = {"paths": [monitoring_uri]} websocket_obj.send(json.dumps(data)) qprint_timen("Registered for websocket monitoring: " + monitoring_uri) def open_socket(openbmc_host, openbmc_username, openbmc_password): """ Open a long-running websocket to the BMC. Description of argument(s): openbmc_host The BMC host name or IP address. openbmc_username The userid for the open BMC system. openbmc_password The Password for the open BMC system. """ websocket.enableTrace(False) qprint_dashes(width=160) qprint_executing() session = login(openbmc_host, openbmc_username, openbmc_password) qprint_timen("Registering websocket handlers.") cookies = session.cookies.get_dict() cookies = sprint_var( cookies, fmt=no_header() | strip_brackets(), col1_width=0, trailing_char="", delim="=", ).replace("\n", ";") # Register the event handlers. When an ESEL is created by the system # under test, the on_message() handler will be called. websocket_obj = websocket.WebSocketApp( "wss://" + openbmc_host + "/subscribe", on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open, cookie=cookies, ) qprint_timen("Completed registering of websocket handlers.") websocket_obj.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) def main(): gen_get_options(parser, stock_list) validate_parms() qprint_pgm_header() qprint_var(monitoring_uri) open_socket(openbmc_host, openbmc_username, openbmc_password) main()