import json import logging import requests import urllib3 from odoo.exceptions import ValidationError, UserError _logger = logging.getLogger(__name__) class UnifiApiClient: """Custom UniFi API Client implementation.""" def __init__(self, host: str, username: str, password: str, port: int = 8443, is_udm: bool = False, verify_ssl: bool = False, timeout: int = 10): """Initialize the UniFi API client. Args: host: Hostname or IP address of the UniFi Controller username: Username for API authentication password: Password for API authentication port: Port number for the API (default: 8443) is_udm: Whether this is a UDM Pro device verify_ssl: Whether to verify SSL certificates timeout: Request timeout in seconds """ if not host or not username or not password: raise ValidationError("Host, username and password are required") self.host = host self.port = port self.is_udm = is_udm self.verify_ssl = verify_ssl self.timeout = timeout self.site_id = 'default' # Disable SSL warnings if verify_ssl is False if not verify_ssl: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Setup session self.session = requests.Session() self.session.verify = verify_ssl # Base URLs if is_udm: self.base_url = f"https://{host}" self.api_url = f"https://{host}/proxy/network" self.auth_url = f"https://{host}/api/auth/login" else: self.base_url = f"https://{host}:{port}" self.api_url = self.base_url self.auth_url = f"{self.base_url}/api/login" _logger.info(f"Initialized with base_url={self.base_url}, api_url={self.api_url}") try: self._login(username, password) _logger.info("Successfully authenticated with UniFi controller") # Only try to discover sites for non-UDM controllers if not is_udm: try: sites = self.get_sites() _logger.info(f"Available sites: {sites}") if sites: # If 'default' site doesn't exist, use the first available site site_exists = any(site.get('name') == 'default' for site in sites) if not site_exists and sites: self.site_id = sites[0].get('name') _logger.info(f"Using site: {self.site_id}") except Exception as e: _logger.warning(f"Failed to get sites, using default site: {str(e)}") except Exception as e: _logger.error(f"Failed to initialize UniFi controller: {str(e)}") raise ValidationError(f"Failed to initialize UniFi controller: {str(e)}") def _login(self, username: str, password: str): """Authenticate with the UniFi Controller.""" _logger.info(f"Attempting login to {self.auth_url}") headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } # Add CSRF token for UDM Pro if available if self.is_udm and hasattr(self, 'csrf_token'): headers['X-CSRF-Token'] = self.csrf_token data = { 'username': username, 'password': password, 'remember': True } try: _logger.info(f"Making login request to {self.auth_url}") response = self.session.post( self.auth_url, headers=headers, json=data, timeout=self.timeout ) _logger.info(f"Login response status: {response.status_code}") _logger.info(f"Login response headers: {response.headers}") _logger.info(f"Login response content: {response.text}") if not response.ok: _logger.error(f"Login failed with status {response.status_code}") _logger.error(f"Response content: {response.text}") response.raise_for_status() # Store cookies and CSRF token for subsequent requests self.session.cookies.update(response.cookies) if self.is_udm: self.csrf_token = response.headers.get('X-CSRF-Token') _logger.info(f"Stored CSRF token: {self.csrf_token}") _logger.info("Successfully logged in and stored session cookies") except requests.exceptions.RequestException as e: _logger.error(f"Login request failed: {str(e)}") if hasattr(e, 'response') and e.response is not None: _logger.error(f"Error response content: {e.response.text}") raise ValidationError(f"Login failed: {str(e)}") def _api_request(self, method: str, endpoint: str, data: dict = None) -> dict: """Make an API request to the UniFi Controller.""" if self.is_udm: # For UDM Pro, check if endpoint already has proxy/network prefix if endpoint.startswith('proxy/network/'): url = f"{self.base_url}/{endpoint}" else: # If no prefix, use api_url which includes the proxy/network prefix url = f"{self.api_url}/{endpoint.lstrip('/')}" else: url = f"{self.base_url}/{endpoint.lstrip('/')}" _logger.info(f"Making {method} request to {url}") headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } if self.is_udm and hasattr(self, 'csrf_token'): headers['X-CSRF-Token'] = self.csrf_token try: if data and method in ['POST', 'PUT', 'PATCH']: response = self.session.request( method, url, headers=headers, json=data, # Use json parameter for requests that send data timeout=self.timeout ) else: response = self.session.request( method, url, headers=headers, timeout=self.timeout ) _logger.info(f"Response status: {response.status_code}") _logger.info(f"Response headers: {response.headers}") _logger.info(f"Response content: {response.text}") response.raise_for_status() if response.text: return response.json() return None except requests.exceptions.RequestException as e: _logger.error(f"API request failed with status {e.response.status_code if hasattr(e, 'response') and e.response else 'unknown'}") if hasattr(e, 'response') and e.response is not None: _logger.error(f"Error response content: {e.response.text}") raise def get_sites(self) -> list: """Get list of available sites.""" try: if self.is_udm: return [{'name': 'default'}] response = self._api_request('GET', 'api/self/sites') return response.get('data', []) except Exception as e: _logger.error(f"Failed to get sites: {str(e)}") return [] def get_firewall_rules(self, site_id=None): """Get all firewall rules for a site.""" site = site_id or self.site_id if self.is_udm: # For UDM Pro, use the network proxy endpoint return self._api_request('GET', f'proxy/network/api/s/{site}/firewallrule') return self._api_request('GET', f'api/s/{site}/rest/firewallrule') def get_active_firewall_rules(self, site_id=None): """Get active firewall rules for a site.""" site = site_id or self.site_id return self._api_request('GET', f'api/s/{site}/stat/firewallrule') def get_firewall_groups(self, site_id=None): """Get firewall groups for a site.""" site = site_id or self.site_id return self._api_request('GET', f'api/s/{site}/rest/firewallgroup') def update_firewall_rule(self, rule_id: str, data: dict, site_id=None): """Update a firewall rule.""" site = site_id or self.site_id return self._api_request('PUT', f'api/s/{site}/rest/firewallrule/{rule_id}', data) def toggle_firewall_rule(self, rule_id: str, enabled: bool, site_id=None): """Enable or disable a firewall rule.""" return self.update_firewall_rule(rule_id, {'enabled': enabled}, site_id) def get_firewall_rules(self) -> list: """Get firewall rules from the controller.""" try: if self.is_udm: # Try UDM Pro Network Settings endpoint first try: _logger.info("Trying network settings endpoint") response = self._api_request('GET', 'proxy/network/api/s/default/rest/setting/firewall') _logger.info(f"Network settings response: {response}") if response and isinstance(response, dict): firewall_settings = response.get('data', [{}])[0] if firewall_settings: _logger.info(f"Found firewall settings: {firewall_settings}") rules = [] # Get WAN rules wan_in = firewall_settings.get('wan_in', []) for rule in wan_in: rule['ruleset'] = 'WAN_IN' rules.append(rule) wan_out = firewall_settings.get('wan_out', []) for rule in wan_out: rule['ruleset'] = 'WAN_OUT' rules.append(rule) wan_local = firewall_settings.get('wan_local', []) for rule in wan_local: rule['ruleset'] = 'WAN_LOCAL' rules.append(rule) # Get LAN rules lan_in = firewall_settings.get('lan_in', []) for rule in lan_in: rule['ruleset'] = 'LAN_IN' rules.append(rule) lan_out = firewall_settings.get('lan_out', []) for rule in lan_out: rule['ruleset'] = 'LAN_OUT' rules.append(rule) lan_local = firewall_settings.get('lan_local', []) for rule in lan_local: rule['ruleset'] = 'LAN_LOCAL' rules.append(rule) _logger.info(f"Found {len(rules)} firewall rules") return rules except Exception as e: _logger.warning(f"Failed to get rules from network settings: {str(e)}") # Try site-specific endpoint for non-UDM or as fallback endpoint = f'api/s/{self.site_id}/rest/setting/firewall' _logger.info(f"Trying {endpoint} endpoint") response = self._api_request('GET', endpoint) _logger.info(f"Response from {endpoint}: {response}") if response and isinstance(response, dict): firewall_settings = response.get('data', [{}])[0] if firewall_settings: rules = [] # Get WAN rules wan_in = firewall_settings.get('wan_in', []) for rule in wan_in: rule['ruleset'] = 'WAN_IN' rules.append(rule) wan_out = firewall_settings.get('wan_out', []) for rule in wan_out: rule['ruleset'] = 'WAN_OUT' rules.append(rule) wan_local = firewall_settings.get('wan_local', []) for rule in wan_local: rule['ruleset'] = 'WAN_LOCAL' rules.append(rule) # Get LAN rules lan_in = firewall_settings.get('lan_in', []) for rule in lan_in: rule['ruleset'] = 'LAN_IN' rules.append(rule) lan_out = firewall_settings.get('lan_out', []) for rule in lan_out: rule['ruleset'] = 'LAN_OUT' rules.append(rule) lan_local = firewall_settings.get('lan_local', []) for rule in lan_local: rule['ruleset'] = 'LAN_LOCAL' rules.append(rule) _logger.info(f"Found {len(rules)} firewall rules") return rules _logger.warning("Could not retrieve firewall rules from any known endpoint") return [] except Exception as e: _logger.error(f"Failed to retrieve firewall rules: {str(e)}") raise UserError(f"Failed to retrieve firewall rules: {str(e)}") def create_firewall_group(self, data: dict, site_id=None) -> dict: """Create a firewall group. Args: data: Dictionary containing the firewall group data site_id: Optional site ID. If not provided, uses the default site Returns: dict: Created firewall group data """ try: _logger.info(f"Creating firewall group: {data}") site = site_id or self.site_id if self.is_udm: endpoint = f'proxy/network/api/s/{site}/rest/firewallgroup' else: endpoint = f'api/s/{site}/rest/firewallgroup' return self._api_request('POST', endpoint, data=data) except Exception as e: _logger.error(f"Failed to create firewall group: {str(e)}") raise UserError(f"Failed to create firewall group: {str(e)}") def ensure_default_firewall_groups(self, site_id=None): """Ensure default firewall groups exist. Creates default LAN and WAN groups if they don't exist. """ try: _logger.info("Checking for default firewall groups") groups = self.get_firewall_groups(site_id) if not groups: _logger.info("No firewall groups found, creating defaults") defaults = [ { 'name': 'Default LAN', 'group_type': 'address-group', 'group_members': ['192.168.0.0/16', '172.16.0.0/12', '10.0.0.0/8'], 'group_description': 'Default LAN networks' }, { 'name': 'Default WAN', 'group_type': 'address-group', 'group_members': ['0.0.0.0/0'], 'group_description': 'Default WAN networks' } ] for group in defaults: try: _logger.info(f"Creating firewall group: {group}") self.create_firewall_group(group, site_id) except Exception as e: _logger.warning(f"Failed to create firewall group: {str(e)}") except Exception as e: _logger.error(f"Failed to ensure default firewall groups: {str(e)}") def list_devices(self) -> list: """Get a list of all devices from the UniFi Controller. Returns: list: List of device dictionaries containing device information """ try: _logger.info("Attempting to get all devices") devices = [] # Try primary endpoint first try: if self.is_udm: endpoint = 'proxy/network/api/s/default/stat/device' # UDM Pro endpoint else: endpoint = f'api/s/{self.site_id}/stat/device' _logger.info(f"Getting devices from endpoint: {endpoint}") response = self._api_request('GET', endpoint) if isinstance(response, dict): devices = response.get('data', []) else: devices = response _logger.info(f"Found {len(devices)} devices") except Exception as e: _logger.warning(f"Failed to get devices using primary endpoint: {str(e)}") # Try fallback endpoint for UDM Pro try: if self.is_udm: endpoint = 'proxy/network/v2/api/site/default/devices' # New UDM Pro endpoint else: endpoint = f'api/s/{self.site_id}/stat/device/basic' _logger.info(f"Getting devices from endpoint: {endpoint}") response = self._api_request('GET', endpoint) if isinstance(response, dict): devices = response.get('data', []) else: devices = response _logger.info(f"Found {len(devices)} devices using fallback endpoint") except Exception as e: _logger.warning(f"Failed to get devices using fallback endpoint: {str(e)}") _logger.info(f"Total devices retrieved: {len(devices)}") if devices: _logger.info("Sample device data:") _logger.info(str(devices[0])) return devices except Exception as e: _logger.error(f"Failed to get devices: {str(e)}") raise UserError(f"Failed to get devices: {str(e)}") def list_clients(self) -> list: """Get a list of all clients from the UniFi Controller. Returns: list: List of client dictionaries containing client information """ try: _logger.info("Attempting to get all clients") clients = [] # Try to get all clients at once try: if self.is_udm: endpoint = 'proxy/network/api/s/default/stat/sta' else: endpoint = f'api/s/{self.site_id}/stat/sta' _logger.info(f"Attempting to get all clients using endpoint: {endpoint}") response = self._api_request('GET', endpoint) if isinstance(response, dict): clients = response.get('data', []) else: clients = response _logger.info(f"Retrieved {len(clients)} clients") except Exception as e: _logger.warning(f"Failed to get clients using primary endpoint: {str(e)}") # Fallback to alternative endpoint try: if self.is_udm: endpoint = 'proxy/network/v2/api/site/default/clients' else: endpoint = f'api/s/{self.site_id}/stat/alluser' _logger.info(f"Trying to get clients from endpoint: {endpoint}") response = self._api_request('GET', endpoint) if isinstance(response, dict): clients = response.get('data', []) else: clients = response _logger.info(f"Found {len(clients)} clients using fallback endpoint") except Exception as e: _logger.warning(f"Failed to get clients using fallback endpoint: {str(e)}") _logger.info(f"Total clients retrieved: {len(clients)}") if clients: _logger.info("Sample client data:") _logger.info(str(clients[0])) return clients except Exception as e: _logger.error(f"Failed to get clients: {str(e)}") raise UserError(f"Failed to get clients: {str(e)}") def list_networks(self) -> list: """Get a list of all networks from the UniFi Controller. Returns: list: List of network dictionaries containing network information """ try: _logger.info("Attempting to get all networks") if self.is_udm: endpoint = 'proxy/network/api/s/default/rest/networkconf' else: endpoint = f'api/s/{self.site_id}/rest/networkconf' _logger.info(f"Attempting to get networks using endpoint: {endpoint}") response = self._api_request('GET', endpoint) if isinstance(response, dict): networks = response.get('data', []) else: networks = response _logger.info(f"Retrieved {len(networks)} networks") if networks: _logger.info("Sample network data:") _logger.info(str(networks[0])) return networks except Exception as e: _logger.error(f"Failed to get networks: {str(e)}") raise UserError(f"Failed to get networks: {str(e)}") def list_sites(self) -> list: """Get a list of all sites from the UniFi Controller. Returns: list: List of site dictionaries containing site information """ try: _logger.info("Attempting to get all sites") if self.is_udm: # UDM Pro typically has only one site return [{ 'name': 'default', 'desc': 'Default', '_id': 'default', 'role': 'admin' }] endpoint = 'api/self/sites' _logger.info(f"Attempting to get sites using endpoint: {endpoint}") response = self._api_request('GET', endpoint) if isinstance(response, dict): sites = response.get('data', []) else: sites = response _logger.info(f"Retrieved {len(sites)} sites") if sites: _logger.info("Sample site data:") _logger.info(str(sites[0])) return sites except Exception as e: _logger.error(f"Failed to get sites: {str(e)}") raise UserError(f"Failed to get sites: {str(e)}") def get_wifis(self, site_id=None) -> list: """Get a list of all WiFi networks from the UniFi Controller. Args: site_id: Optional site ID. If not provided, uses the default site. Returns: list: List of WiFi network configurations """ try: _logger.info("Attempting to get WiFi networks") site = site_id or self.site_id # Try primary endpoint first try: if self.is_udm: endpoint = f'proxy/network/api/s/{site}/rest/wlanconf' else: endpoint = f'api/s/{site}/rest/wlanconf' _logger.info(f"Getting WiFi networks from endpoint: {endpoint}") response = self._api_request('GET', endpoint) if isinstance(response, dict): networks = response.get('data', []) else: networks = response or [] _logger.info(f"Found {len(networks)} WiFi networks") if networks: _logger.info("Sample WiFi network data:") _logger.info(str(networks[0])) return networks except Exception as e: _logger.error(f"Failed to get WiFi networks: {str(e)}") raise except Exception as e: _logger.error(f"Failed to get WiFi networks: {str(e)}") raise UserError(f"Failed to get WiFi networks: {str(e)}") def get_firewall_rules(self, site_id=None): """Get all firewall rules for a site.""" site = site_id or self.site_id if self.is_udm: # For UDM Pro, use the network proxy endpoint return self._api_request('GET', f'proxy/network/api/s/{site}/firewallrule') return self._api_request('GET', f'api/s/{site}/rest/firewallrule') def get_port_forward_rules(self, site_id=None): """Get all port forwarding rules for a site.""" site = site_id or self.site_id _logger.info("Trying port forward rules endpoint") if self.is_udm: # For UDM Pro, use the network proxy endpoint return self._api_request('GET', f'proxy/network/api/s/{site}/rest/portforward') return self._api_request('GET', f'api/s/{site}/rest/portforward')