"
+ },
+ {
+ "equals": "watched-changed",
+ "replacement": "
"
+ },
+ {
+ "equals": "new",
+ "replacement": "
"
+ },
+ {
+ "equals": "missing-in-last-scan",
+ "replacement": "
"
+ }
+ ],
+ "localized": [
+ "name"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Status"
+ }
+ ]
+ }
+ ]
+}
\ No newline at end of file
diff --git a/front/plugins/freebox/freebox.py b/front/plugins/freebox/freebox.py
new file mode 100644
index 00000000..593345be
--- /dev/null
+++ b/front/plugins/freebox/freebox.py
@@ -0,0 +1,168 @@
+#!/usr/bin/env python
+
+import os
+import pathlib
+import sys
+import json
+import sqlite3
+from pytz import timezone
+import asyncio
+from datetime import datetime
+from pathlib import Path
+from typing import cast
+import socket
+import aiofreepybox
+from aiofreepybox import Freepybox
+from aiofreepybox.api.lan import Lan
+from aiofreepybox.exceptions import NotOpenError, AuthorizationError
+
+# Define the installation path and extend the system path for plugin imports
+INSTALL_PATH = "/app"
+sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
+
+from plugin_helper import Plugin_Object, Plugin_Objects, decodeBase64
+from plugin_utils import get_plugins_configs
+from logger import mylog
+from const import pluginsPath, fullDbPath
+from helper import timeNowTZ, get_setting_value
+from notification import write_notification
+import conf
+
+# Make sure the TIMEZONE for logging is correct
+conf.tz = timezone(get_setting_value("TIMEZONE"))
+
+# Define the current path and log file paths
+CUR_PATH = str(Path(__file__).parent.resolve())
+LOG_FILE = os.path.join(CUR_PATH, "script.log")
+RESULT_FILE = os.path.join(CUR_PATH, "last_result.log")
+
+# Initialize the Plugin obj output file
+plugin_objects = Plugin_Objects(RESULT_FILE)
+
+pluginName = "FREEBOX"
+
+device_type_map = {
+ "workstation": "PC",
+ "laptop": "Laptop",
+ "smartphone": "Smartphone",
+ "tablet": "Tablet",
+ "printer": "Printer",
+ "vg_console": "Game Console",
+ "television": "SmartTV",
+ "nas": "NAS",
+ "ip_camera": "IP Camera",
+ "ip_phone": "Phone",
+ "freebox_player": "TV Decoder",
+ "freebox_hd": "TV Decoder",
+ "freebox_crystal": "TV Decoder",
+ "freebox_mini": "TV Decoder",
+ "freebox_delta": "Gateway",
+ "freebox_one": "Gateway",
+ "freebox_wifi": "Gateway",
+ "freebox_pop": "AP",
+ "networking_device": "Router",
+ "multimedia_device": "TV Decoder",
+ "car": "House Appliance",
+ "other": "(Unknown)",
+}
+
+
+def map_device_type(type: str):
+ return device_type_map[type]
+
+
+async def get_device_data(api_version: int, api_address: str, api_port: int):
+ # ensure existence of db path
+ data_dir = Path("/app/config/freeboxdb")
+ data_dir.mkdir(parents=True, exist_ok=True)
+
+ # Instantiate Freepybox class using default application descriptor
+ # and custom token_file location
+ fbx = Freepybox(
+ app_desc={
+ "app_id": "netalertx",
+ "app_name": "NetAlertX",
+ "app_version": aiofreepybox.__version__,
+ "device_name": socket.gethostname(),
+ },
+ api_version="v" + str(api_version),
+ data_dir=data_dir,
+ )
+
+ # Connect to the freebox
+ # Be ready to authorize the application on the Freebox if you run this
+ # for the first time
+ try:
+ await fbx.open(host=api_address, port=api_port)
+ except NotOpenError as e:
+ mylog("verbose", [f"[{pluginName}] Error connecting to freebox: {e}"])
+ except AuthorizationError as e:
+ mylog("verbose", [f"[{pluginName}] Auth error: {str(e)}"])
+
+ # get also info of the freebox itself
+ config = await fbx.system.get_config()
+ freebox = await cast(Lan, fbx.lan).get_config()
+ hosts = await cast(Lan, fbx.lan).get_hosts_list()
+ assert config is not None
+ assert freebox is not None
+ freebox["mac"] = config["mac"]
+ freebox["operator"] = config["model_info"]["net_operator"]
+
+ # Close the freebox session
+ await fbx.close()
+
+ return freebox, hosts
+
+
+def main():
+ mylog("verbose", [f"[{pluginName}] In script"])
+
+ # Retrieve configuration settings
+ api_settings = {
+ "api_address": get_setting_value("FREEBOX_address"),
+ "api_version": get_setting_value("FREEBOX_api_version"),
+ "api_port": get_setting_value("FREEBOX_api_port"),
+ }
+
+ mylog("verbose", [f"[{pluginName}] Settings: {api_settings}"])
+
+ # retrieve data
+ loop = asyncio.new_event_loop()
+ freebox, hosts = loop.run_until_complete(get_device_data(**api_settings))
+ loop.close()
+
+ mylog("verbose", [freebox])
+ mylog("verbose", [hosts])
+
+ plugin_objects.add_object(
+ primaryId=freebox["mac"],
+ secondaryId=freebox["ip"],
+ watched1=freebox["name"],
+ watched2=freebox["operator"],
+ watched3="Gateway",
+ watched4=datetime.now,
+ extra="",
+ foreignKey=freebox["mac"],
+ )
+ for host in hosts:
+ for ip in [ip for ip in host["l3connectivities"] if ip["reachable"]]:
+ mac: str = host["l2ident"]["id"]
+ plugin_objects.add_object(
+ primaryId=mac,
+ secondaryId=ip["addr"],
+ watched1=host["primary_name"],
+ watched2=host["vendor_name"] if host["vendor_name"] else "(unknown)",
+ watched3=map_device_type(host["host_type"]),
+ watched4=datetime.fromtimestamp(ip["last_time_reachable"]),
+ extra="",
+ foreignKey=mac,
+ )
+
+ # commit result
+ plugin_objects.write_result_file()
+
+ return 0
+
+
+if __name__ == "__main__":
+ main()