#!/usr/bin/env python3 # pyright: reportUnusedFunction=false import asyncio import datetime import json import logging from multiprocessing import Queue import os from typing import Any, Callable from urllib.parse import quote_plus as url_quote from functools import wraps import aiohttp from flask import ( Flask, Response, render_template, request, url_for, redirect, make_response, ) from werkzeug.wrappers import Response as WerkzeugResponse from waitress import serve import shared.config import shared.metrics as metrics from shared.helpers import ( natural_join, format_mb, format_text_fragments, format_datetime, format_datetime_network_table, natural_join_categories, ) import shared.policy_summaries as policy_summaries import shared.system_stats as system_stats import shared.networking_helpers as networking_helpers import shared.oui_parser as oui_parser import shared.arp_helpers as arp_helpers from model.database_manager import DatabaseClient import scan.homeassistant logger = logging.getLogger(__name__) ARP_SPOOF_MIN = 3 iot_manufacturers_file = os.path.join( os.path.dirname(os.path.realpath(__file__)), "..", "data", "manufacturers.json", ) with open(iot_manufacturers_file, "r") as file: iot_manufacturers = json.load(file) def create_app( app_root: str, data_dir: str, debug_mode: bool, db_client: DatabaseClient, ha_status: dict[str, Any] | None, ha_client: scan.homeassistant.HomeAssistantClient | None, ) -> Flask: app = Flask(__name__) app.config["APPLICATION_ROOT"] = app_root def require_setup_state( check_study_complete: bool = True, check_setup_activation: bool = True, check_setup_scan: bool = True, check_setup_network: bool = True, check_setup_interview: bool = True, check_setup_complete: bool = True, check_scan: bool = True, ): """ Decorator to check various setup states and redirect/render appropriate responses. Args: check_study_complete: If True, check if study is complete and show completion wall check_setup_activation: If True, check if setup activation is complete check_setup_scan: If True, check if setup scan is complete check_setup_network: If True, check if setup network is complete check_setup_interview: If True, check if setup interview is complete check_setup_complete: If True, check if setup complete is complete """ def decorator(f: Callable[..., Any]) -> Callable[..., Any]: @wraps(f) def decorated_function(*args: Any, **kwargs: Any) -> Any: config = shared.config.items() study_complete = config.get("study_complete") study_complete_reenable = config.get("study_complete_reenable") if ( check_study_complete and study_complete and study_complete_reenable is not True ): # ruff: ignore return render_template( "study_complete_wall.html", no_nav=True, study_complete_reenable=study_complete_reenable, ) if check_setup_activation and not config.get( "setup_activation_complete" ): return redirect(url_for("setup_activation")) if check_setup_scan and not config.get("setup_scan_complete"): return redirect(url_for("setup_scan")) if check_setup_network and not config.get("setup_network_complete"): return redirect(url_for("setup_network")) if check_setup_interview and not config.get("setup_interview_complete"): return redirect(url_for("setup_interview")) if check_setup_complete and not config.get("setup_complete_complete"): return redirect(url_for("setup_complete")) if check_scan and not config.get("first_scan_complete"): return render_template( "waiting.html", no_nav=True, scan_step=config.get("scan_step", 0), ) return f(*args, **kwargs) return decorated_function return decorator @app.after_request def after_request(response: Response) -> Response: response.access_control_allow_origin = "*" return response ### UI pages @app.route("/study_home", methods=["GET"]) @require_setup_state() def study_home(): config = shared.config.items() # reminder banner context # devices_req = db_client.send_request("get_devices") # devices_resp = db_client.wait_for_response(devices_req) # devices = devices_resp.data # interview_reminder_dismissed = config.get("interview_reminder_dismissed", False) # template info for intro sentence # devices_iot, devices_non_iot = [], [] # manufacturers_iot = set() # for dev in devices: # if dev["is_iot"]: # devices_iot.append(dev) # manufacturers_iot.add(dev["preferred_mfg"]) # else: # devices_non_iot.append(dev) installation_id = config.get("installation_id") asyncio.run(metrics.pageload_send("study_home")) return render_template( "study_home.html", debug_mode=debug_mode, user_id=installation_id, study_complete=config.get("study_complete"), # todo list arp_spoof_min=ARP_SPOOF_MIN, arp_spoof_count=config.get("arp_spoof_ever_enabled", 0), # interview_reminder_dismissed=interview_reminder_dismissed, # summary info # devices_iot_count=len(devices_iot), # devices_non_iot_count=len(devices_non_iot), # manufacturers_iot=list(manufacturers_iot), ha_status=ha_status, ) @app.route("/", methods=["GET"]) @require_setup_state() def devices() -> str | WerkzeugResponse: config = shared.config.items() devices_req = db_client.send_request("get_devices") # list files available for LLM summaries llm_summaries_dir = os.path.join( os.path.dirname(os.path.realpath(__file__)), "..", "data", "policies", ) llm_summaries_available = policy_summaries.get_policy_summaries_list( llm_summaries_dir ) devices_resp = db_client.wait_for_response(devices_req) if not devices_resp.success: logger.error(f"Error getting device details for {id}: {devices_resp.error}") return f"Error getting device details: {devices_resp.error}" devices = devices_resp.data for device in devices: # Add LLM summary availability to each device mfg = device.get("preferred_mfg") brand = mfg and mfg.casefold() device["policy_available"] = brand in llm_summaries_available asyncio.run(metrics.pageload_send("devices")) return render_template( "devices.html", user_id=config.get("installation_id"), debug_mode=debug_mode, items=devices, # devices_unknown_count=unknown_count, iot_manufacturers=iot_manufacturers, ) @app.route("/manage_network", methods=["GET"]) @require_setup_state() def manage_network() -> str | WerkzeugResponse: devices_req = db_client.send_request("get_devices") config = shared.config.items() devices_resp = db_client.wait_for_response(devices_req) if not devices_resp.success: logger.error(f"Error getting device details for {id}: {devices_resp.error}") return f"Error getting device details: {devices_resp.error}" devices = [device for device in devices_resp.data if device["ip_addr"]] arp_spoof = config.get("arp_spoof") arp_spoof_mode = config.get("spoof_mode", "continuous") spoofed_devices = [ device for device in devices if device["is_iot"] and device["is_arp_spoofed"] ] asyncio.run(metrics.pageload_send("manage_network")) return render_template( "network/manage_network.html", debug_mode=debug_mode, items=devices, arp_spoof=arp_spoof, spoof_mode=arp_spoof_mode, arp_spoof_min=ARP_SPOOF_MIN, arp_spoof_count=len(spoofed_devices), ) @app.route("/devices/", methods=["GET"]) @require_setup_state() def details_devices(id: int) -> str | WerkzeugResponse: config = shared.config.items() asyncio.run(metrics.pageload_send("device_details")) device_req = db_client.send_request("get_device_details", {"device_id": id}) device_resp = db_client.wait_for_response(device_req) if not device_resp.success: logger.error(f"Error getting device details for {id}: {device_resp.error}") return f"Error getting device details: {device_resp.error}" device = device_resp.data brand = ( device.get("preferred_mfg").casefold() if device.get("preferred_mfg") else None ) alternate_names: list[str | None] = [ device.get("ha_name"), device.get("ssdp_name"), device.get("mdns_hostname").rstrip(".local").replace("-", " ") if device.get("mdns_hostname") else None, device.get("dhcp_hostname").replace("-", " ") if device.get("dhcp_hostname") else None, ] alternate_names = [ name for name in alternate_names if name != device.get("preferred_name") and name is not None ] mfg_data = iot_manufacturers.get(brand, {}) # load llm summary llm_summary: dict[str, Any] | None = None llm_summary_path = os.path.join( os.path.dirname(os.path.realpath(__file__)), "..", "data", "policies", f"{brand}.json", ) llm_summary = policy_summaries.get_policy_summary(llm_summary_path) return render_template( "device_details.html", debug_mode=debug_mode, device=device, alternate_names=alternate_names, policy_summary=llm_summary, arp_spoof=config.get("arp_spoof"), mfg_data=mfg_data, now_plus_5_minutes=datetime.datetime.now() + datetime.timedelta(minutes=5), ) @app.route("/device_edit_modal/", methods=["GET"]) def device_edit_modal(id: int) -> str: device_req = db_client.send_request("get_device_details", {"device_id": id}) device_resp = db_client.wait_for_response(device_req) if not device_resp.success: logger.error( f"Error getting device details for modal {id}: {device_resp.error}" ) return f"Error getting device details: {device_resp.error}" device_data = device_resp.data return render_template( "devices/partials/device_edit.html", device=device_data, ) @app.route("/device_edit/", methods=["POST"]) def device_edit(id: int) -> Response: user_name = request.form.get("name") user_model = request.form.get("model") user_mfg = request.form.get("manufacturer") is_iot_form_status = request.form.get("is_iot") logger.debug(f"editing device {id}") update_req = db_client.send_request( "update_device", { "device_id": id, "user_name": user_name, "user_model": user_model, "user_mfg": user_mfg, "is_iot_form_status": is_iot_form_status, }, ) update_resp = db_client.wait_for_response(update_req) if not update_resp.success: logger.error(f"Error updating device {id}: {update_resp.error}") return Response(f"Error updating device: {update_resp.error}", status=500) logger.debug(f"done editing device {id}") resp = Response() resp.headers["HX-Refresh"] = "true" return resp @app.route("/devices//network_data", methods=["GET"]) def device_network_data(id: int) -> str: how_recent = request.args.get("how_recent", "all") if how_recent not in ["1h", "1d", "all"]: how_recent = "all" if how_recent == "1h": start_ts = datetime.datetime.now() - datetime.timedelta(hours=1) elif how_recent == "1d": start_ts = datetime.datetime.now() - datetime.timedelta(days=1) else: start_ts = None # query database (using db_client) device_req = db_client.send_request("get_device_by_id", {"id": id}) network_entities_req = db_client.send_request( "get_entities_by_device", { "device_id": id, "start_ts": start_ts, }, ) device_resp = db_client.wait_for_response(device_req) network_entities_resp = db_client.wait_for_response(network_entities_req) if not device_resp.success: logger.error(f"Error getting device {id}: {device_resp.error}") return f"Error getting device data {device_resp.error}" if not network_entities_resp.success: logger.error( f"Error getting network entities for device {id}: {network_entities_resp.error}" ) return f"Error getting network entities data {network_entities_resp.error}" device = device_resp.data network_entities = network_entities_resp.data return render_template( "devices/partials/network_table.html", device=device, network_entities=network_entities, how_recent=how_recent, ) @app.route("/setup_activation", methods=["GET"]) def setup_activation() -> str: asyncio.run(metrics.pageload_send("setup_activation")) installation_id = shared.config.get("installation_id") return render_template( "setup/setup_activation.html", debug_mode=debug_mode, no_nav=True, installation_id=installation_id, ha_status=ha_status, ) @app.route("/setup_scan", methods=["GET"]) @require_setup_state( check_setup_activation=True, check_setup_scan=False, check_setup_network=False, check_setup_interview=False, check_setup_complete=False, check_scan=False, ) def setup_scan() -> str | WerkzeugResponse: config = shared.config.items() asyncio.run(metrics.pageload_send("setup_scan")) return render_template( "setup/setup_scan.html", debug_mode=debug_mode, no_nav=True, first_scan_complete=config.get("first_scan_complete"), scan_step=config.get("scan_step", 0), ) @app.route("/setup_network", methods=["GET"]) @require_setup_state( check_setup_activation=True, check_setup_scan=True, check_setup_network=False, check_setup_interview=False, check_setup_complete=False, check_scan=False, ) def setup_network() -> str | WerkzeugResponse: config = shared.config.items() devices_req = db_client.send_request("get_devices") devices_resp = db_client.wait_for_response(devices_req) if not devices_resp.success: logger.error(f"Error getting device data: {devices_resp.error}") return f"Error getting device data: {devices_resp.error}" devices = [device for device in devices_resp.data if device["ip_addr"]] arp_spoof = config.get("arp_spoof") arp_spoof_mode = config.get("spoof_mode", "continuous") spoofed_devices = [ device for device in devices if device["is_iot"] and device["is_arp_spoofed"] ] asyncio.run(metrics.pageload_send("setup_network")) return render_template( "setup/setup_network.html", debug_mode=debug_mode, items=devices, arp_spoof=arp_spoof, spoof_mode=arp_spoof_mode, arp_spoof_min=ARP_SPOOF_MIN, arp_spoof_count=len(spoofed_devices), no_nav=True, ) @app.route("/setup_interview", methods=["GET"]) @require_setup_state( check_setup_activation=True, check_setup_scan=True, check_setup_network=True, check_setup_interview=False, check_setup_complete=False, check_scan=False, ) def setup_interview() -> str: asyncio.run(metrics.pageload_send("setup_interview")) return render_template( "setup/setup_interview.html", debug_mode=debug_mode, no_nav=True, user_id=shared.config.get("installation_id"), ) @app.route("/setup_complete", methods=["GET"]) @require_setup_state( check_setup_activation=True, check_setup_scan=True, check_setup_network=True, check_setup_interview=True, check_setup_complete=False, check_scan=False, ) def setup_complete() -> str: asyncio.run(metrics.pageload_send("setup_complete")) return render_template( "setup/setup_complete.html", debug_mode=debug_mode, no_nav=True, ) @app.route("/debug", methods=["GET"]) def debug() -> str: asyncio.run(metrics.pageload_send("debug")) config = shared.config.items() debug_req = db_client.send_request("get_debug_data") debug_resp = db_client.wait_for_response(debug_req) if not debug_resp.success: logger.error(f"Error getting debug data: {debug_resp.error}") return f"Error getting debug data: {debug_resp.error}" debug_data = debug_resp.data devices_data = debug_data["devices_data"] mdns_data = debug_data["mdns_data"] pendingdevice_data = debug_data["pendingdevice_data"] total_packet_count = debug_data["total_packet_count"] # Get retransmission ratios data for debug display retrans_req = db_client.send_request("get_retransmission_ratios") retrans_resp = db_client.wait_for_response(retrans_req) retransmission_data = [] retransmission_headers = [] if retrans_resp.success and retrans_resp.data: retransmission_data = retrans_resp.data if retransmission_data and len(retransmission_data) > 0: retransmission_headers = list(retransmission_data[0].keys()) else: logger.warning( f"Could not get retransmission data: {retrans_resp.error if retrans_resp else 'No response'}" ) devices_headers = [] mdns_headers = [] pendingdevice_headers = [] if devices_data and len(devices_data) > 0: devices_headers = list(devices_data[0].keys()) if mdns_data and len(mdns_data) > 0: mdns_headers = list(mdns_data[0].keys()) if pendingdevice_data and len(pendingdevice_data) > 0: pendingdevice_headers = list(pendingdevice_data[0].keys()) installation_id = config.get("installation_id") netstat = system_stats.netstat() packets_forwarded_total = int(netstat["Ip"]["ForwDatagrams"]) packets_forwarded_at_db_init = config.get("packets_forwarded_at_db_init") or 0 packets_forwarded = packets_forwarded_total - packets_forwarded_at_db_init loadavg = system_stats.loadavg() memstat = system_stats.memstat() diskstat = system_stats.diskstat(data_dir) route = networking_helpers.get_default_route() return render_template( "debug.html", debug_mode=debug_mode, config=config, devices_headers=devices_headers, devices_data=devices_data, mdns_headers=mdns_headers, mdns_data=mdns_data, pendingdevice_headers=pendingdevice_headers, pendingdevice_data=pendingdevice_data, installation_id=installation_id, packets_forwarded=packets_forwarded, packets_in_db=total_packet_count, netstat=netstat, loadavg=loadavg, memstat=memstat, diskstat=diskstat, route=route, arp_spoof=config.get("arp_spoof"), spoof_mode=config.get("spoof_mode"), arp_spoof_op=config.get("arp_spoof_op"), retransmission_headers=retransmission_headers, retransmission_data=retransmission_data, ) ### API ENDPOINTS # Actions that are triggered by the UI # Return HTML snippets to be inserted into the page @app.route("/actions/set_activation_code", methods=["POST"]) async def set_activation_code() -> str | Response: """ Set the activation code for the device. This is used to identify the device in the cloud. """ activation_code = request.form.get("activation_code", None) if not activation_code or activation_code == "": return render_template( "setup/partials/setup_activation_form.html", invalid="Please enter an activation code.", ) activation_status = await metrics.enroll(activation_code) if activation_status == 201: resp = Response() resp.headers["HX-Redirect"] = url_for( "setup_complete_step", step="activation" ) return resp elif activation_status == 403: return render_template( "setup/partials/setup_activation_form.html", invalid="This activation code is not enrolled in the study. Please return to Qualtrics and report this issue to the researchers.", ) else: return render_template( "setup/partials/setup_activation_form.html", invalid="The activation server was unable to process the request. Please return to Qualtrics and report this issue to the researchers.", ) @app.route("/actions/setup_complete_step/", methods=["GET"]) def setup_complete_step(step: str) -> WerkzeugResponse: """ Complete a step in the setup process. """ if step == "activation": shared.config.set("setup_activation_complete", True) return redirect(url_for("setup_scan")) elif step == "scan": shared.config.set("setup_scan_complete", True) return redirect(url_for("setup_network")) elif step == "network": shared.config.set("setup_network_complete", True) return redirect(url_for("setup_interview")) elif step == "interview": shared.config.set("setup_interview_complete", True) return redirect(url_for("setup_complete")) elif step == "complete": shared.config.set("setup_complete_complete", True) return redirect(url_for("devices")) else: return redirect(url_for("setup_activation")) @app.route("/actions/complete_study", methods=["POST"]) async def complete_study() -> str | Response: """ Tell the server that the user has completed the study, and update the config on our end """ logger.error("complete_study is not implemented") return "complete_study is not implemented" @app.route("/actions/study_complete_reenable", methods=["POST"]) async def study_complete_reenable() -> str | Response: """ Re-enable the study. """ shared.config.set("study_complete_reenable", "pending_restart") resp = Response() resp.headers["HX-Trigger"] = ( '{"showMessage":"IoT Transparency has been re-enabled. To re-enable all functionality, go to Home Assistant Add-on settings, select IoT Transparency, and restart the add-on."}' ) return resp @app.route("/actions/reset_onboarding", methods=["POST"]) def reset_onboarding() -> Response: """ Reset the onboarding process. This is used to allow the user to re-do the onboarding process. """ logger.info("resetting onboarding") shared.config.set("setup_activation_complete", False) shared.config.set("setup_scan_complete", False) shared.config.set("setup_network_complete", False) shared.config.set("setup_interview_complete", False) shared.config.set("setup_complete_complete", False) shared.config.set("interview_reminder_dismissed", False) resp = Response() resp.headers["HX-Redirect"] = url_for("setup_activation") return resp # @app.route("/actions/toggle_arp_spoof_global", methods=["POST"]) # def toggle_arp_spoof_global() -> Response: # """ # Toggle the global ARP spoofing setting. # We refresh the whole page, as opposed to returning an element, as many items on the devices table page need to be changed # """ # # Get the desired state from the request # desired_state = request.form.get("desired_state", "") == "true" # logger.debug(f"setting spoofing to {desired_state}") # shared.config.set("arp_spoof", desired_state) # resp = Response() # resp.headers["HX-Refresh"] = "true" # return resp # @app.route("/actions/arp_spoof_mode", methods=["POST"]) # def arp_spoof_mode(): # mode = request.form.get("mode") # logger.debug(f"setting spoof mode to {mode}") # shared.config.set("spoof_mode", mode) # if mode == "continuous": # return """ # # # """ # else: # return """ # # # """ @app.route("/actions/toggle_arp_spoof/", methods=["POST"]) def toggle_arp_spoof(id: int) -> Response: # Get the desired state from the request desired_state = request.form.get("desired_state", "") == "true" config = shared.config.items() if desired_state: # Enabling ARP spoofing toggle_resp_data = arp_helpers.enable_arp_spoofing_for_device(db_client, id) else: # Disabling ARP spoofing - sends corrective ARP packets toggle_resp_data = arp_helpers.disable_arp_spoofing_for_device( db_client, id ) if not toggle_resp_data.get("success"): error_msg = toggle_resp_data.get("error", "Unknown error") logger.error(f"Error toggling ARP spoof for device {id}: {error_msg}") return Response(f"Error toggling ARP spoof: {error_msg}", status=500) result = toggle_resp_data["data"] enabled_count = result["enabled_count"] # Return the updated button HTML with the correct state if desired_state is True: ever_enabled_count = config.get("arp_spoof_ever_enabled", 0) + 1 shared.config.set("arp_spoof_ever_enabled", ever_enabled_count) resp = make_response( render_template( "partials/arp_spoof_toggle.html", device_id=id, is_active=desired_state, global_enabled=config.get("arp_spoof"), ) ) resp.headers["HX-Trigger-After-Swap"] = json.dumps( {"updateContinueButton": enabled_count} ) return resp @app.route("/actions/disable_all_network_monitoring", methods=["POST"]) def disable_all_network_monitoring() -> Response: """Disable ARP spoofing for all devices and return updated table""" asyncio.run(metrics.pageload_send("disable_arp_spoofing_all")) result = arp_helpers.disable_arp_spoofing_for_all_devices(db_client) if not result.get("success"): logger.error( f"Error disabling all network monitoring: {result.get('error', 'Unknown error')}" ) return Response( f"Error: {result.get('error', 'Unknown error')}", status=500 ) processed = result.get("processed", 0) total = result.get("total", 0) errors = result.get("errors", []) if errors: logger.warning( f"Disabled monitoring for {processed}/{total} devices with {len(errors)} errors" ) else: logger.info( f"Successfully disabled network monitoring for {processed} devices" ) # Get updated device list to refresh the table devices_req = db_client.send_request("get_devices") devices_resp = db_client.wait_for_response(devices_req) if not devices_resp.success: logger.error(f"Error getting updated devices: {devices_resp.error}") return Response( f"Error refreshing device list: {devices_resp.error}", status=500 ) devices = devices_resp.data or [] config = shared.config.items() # Return the updated table HTML return render_template( "network/partials/manage_network_table.html", items=devices, arp_spoof=config.get("arp_spoof", False), ) @app.route("/actions/arp_op_code/", methods=["POST"]) def arp_spoof_op(opcode: int) -> str: # Get the desired state from the request logger.debug(f"setting arp op code to {opcode}") shared.config.set("arp_spoof_op", opcode) return str(opcode) @app.route("/actions/send_log", methods=["POST"]) async def send_log() -> str: _, message = await metrics.log_send( os.path.join(DATA_DIR, "iot-transparency.log") ) return message @app.route("/actions/send_tcpdump/", methods=["POST"]) async def send_tcpdump(ip: str) -> str: logger.info(f"App server received request to capture tcpdump for {ip}") return await metrics.send_tcpdump(ip, DATA_DIR) # @app.route("/actions/dismiss_interview_reminder", methods=["POST"]) # def dismiss_interview_reminder() -> str: # logger.info("Dismissing interview reminder") # shared.config.set("interview_reminder_dismissed", True) # return "" # @app.route("/actions/toggle_interview_reminder_checkbox", methods=["POST"]) # def toggle_interview_reminder_checkbox() -> str: # logger.info("Toggling interview reminder") # old_state = shared.config.get("interview_reminder_dismissed", False) # new_state = not old_state # shared.config.set("interview_reminder_dismissed", new_state) # if new_state: # return "" # else: # return "" @app.route("/actions/retention_cleanup", methods=["POST"]) def retention_cleanup() -> str: logger.info("Running retention cleanup for data older than 1 hour") if DB_QUEUE is None: logger.error("DB_QUEUE is None, cannot run retention cleanup") return "DB_QUEUE is not initialized" DB_QUEUE.put(("retention_cleanup", {"hours": 1})) return "Retention cleanup requested" @app.route("/actions/fix_ha_config", methods=["GET"]) async def fix_ha_config() -> str: """ Fix the Home Assistant configuration. TODO: this does not work, error about async task attached to different loop """ return "Unimplemented" # logger.info("Fixing Home Assistant configuration") # ha_status = None # if ha_client: # _ = await ha_client.supervisor_api_fix_config() # ha_status = await ha_client.supervisor_api_get_info() # return render_template( # "setup/partials/ha_reminders.html", # ha_status=ha_status, # ) return app async def main( data_dir: str, listen_ip: str, port: str, app_root: str, db_queue: "Queue[tuple[str, dict[str, Any]]]", debug_mode: bool, ha_addon: bool, ha_url: str, db_client: DatabaseClient, ) -> None: global DATA_DIR DATA_DIR = data_dir global DB_QUEUE DB_QUEUE = db_queue # initialize ha_client aiohttp_session = aiohttp.ClientSession() ha_client = scan.homeassistant.HomeAssistantClient( is_addon=ha_addon, ha_url=ha_url, session=aiohttp_session, db_queue=db_queue, ) ha_status = await ha_client.supervisor_api_get_info() app = create_app(app_root, data_dir, debug_mode, db_client, ha_status, ha_client) app.jinja_env.filters["natural_join"] = natural_join app.jinja_env.filters["natural_join_categories"] = natural_join_categories app.jinja_env.filters["casefold"] = ( lambda s: s.casefold() if isinstance(s, str) else s ) app.jinja_env.filters["format_mb"] = format_mb app.jinja_env.filters["format_datetime"] = format_datetime app.jinja_env.filters["format_datetime_network_table"] = ( format_datetime_network_table ) app.jinja_env.filters["format_text_fragments"] = format_text_fragments app.jinja_env.filters["urlencode"] = lambda s: url_quote(s, safe="") app.jinja_env.filters["oui_vendor"] = oui_parser.get_vendor serve(app, listen=f"{listen_ip}:{port}", url_prefix=app_root) def run_main_async(*args: Any, **kwargs: Any) -> None: asyncio.run(main(*args, **kwargs))