1304 lines
56 KiB
Python
1304 lines
56 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# These imports will work in an Odoo environment, even if your IDE marks them as not found
|
|
# pylint: disable=import-error
|
|
from odoo import models, fields, api, _
|
|
from odoo.exceptions import UserError
|
|
# pylint: enable=import-error
|
|
|
|
import json
|
|
import logging
|
|
import random
|
|
import requests
|
|
import urllib3
|
|
from datetime import datetime, timedelta
|
|
from requests.exceptions import RequestException
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
class UdmSite(models.Model):
|
|
"""Represents a UniFi site managed by one or more UDM Pro"""
|
|
_name = 'udm.site'
|
|
_description = 'UDM Site'
|
|
_order = 'name'
|
|
|
|
name = fields.Char(string='Name', required=True)
|
|
site_id = fields.Char(string='Site ID', help="Site identifier in UniFi (always 'default')", default='default', readonly=True, required=True)
|
|
description = fields.Text(string='Description')
|
|
address = fields.Text(string='Physical Address')
|
|
active = fields.Boolean(string='Active', default=True)
|
|
|
|
# Relations
|
|
configuration_ids = fields.One2many('udm.configuration', 'site_id', string='Configurations')
|
|
dashboard_metric_ids = fields.One2many('udm.dashboard.metric', 'site_id', string='Dashboard Metrics')
|
|
dashboard_stat_ids = fields.One2many('udm.dashboard.stat', 'site_id', string='Statistics')
|
|
|
|
# Counters
|
|
config_count = fields.Integer(compute='_compute_counts', string='Configuration Count')
|
|
device_count = fields.Integer(compute='_compute_device_count', string='Total Devices')
|
|
client_count = fields.Integer(compute='_compute_client_count', string='Connected Clients')
|
|
|
|
@api.depends('network_ids')
|
|
def _compute_counts(self):
|
|
"""Compute the number of networks in this site"""
|
|
for record in self:
|
|
record.config_count = len(record.network_ids)
|
|
|
|
@api.depends('device_ids')
|
|
def _compute_device_count(self):
|
|
"""Compute the total number of devices in this site"""
|
|
for record in self:
|
|
record.device_count = len(record.device_ids)
|
|
|
|
@api.depends('user_ids')
|
|
def _compute_client_count(self):
|
|
"""Compute the number of connected clients"""
|
|
for record in self:
|
|
record.client_count = len(record.user_ids.filtered(lambda u: u.status == 'connected'))
|
|
|
|
def action_view_configurations(self):
|
|
self.ensure_one()
|
|
return {
|
|
'name': _('Configurations'),
|
|
'view_mode': 'tree,form',
|
|
'res_model': 'udm.configuration',
|
|
'domain': [('site_id', '=', self.id)],
|
|
'type': 'ir.actions.act_window',
|
|
}
|
|
|
|
def action_view_dashboard(self):
|
|
self.ensure_one()
|
|
return {
|
|
'name': _('Site Dashboard'),
|
|
'view_mode': 'dashboard,form',
|
|
'res_model': 'udm.site',
|
|
'res_id': self.id,
|
|
'type': 'ir.actions.act_window',
|
|
}
|
|
|
|
def action_refresh_metrics(self):
|
|
"""Refreshes the dashboard metrics and statistics for this site"""
|
|
self.ensure_one()
|
|
configs = self.configuration_ids.filtered(lambda c: c.active)
|
|
if not configs:
|
|
raise UserError(_('No active UDM Pro configuration found for this site'))
|
|
|
|
# Call the UDM Pro API to get real metrics
|
|
try:
|
|
for config in configs:
|
|
config._fetch_metrics()
|
|
except Exception as e:
|
|
raise UserError(_('Failed to fetch metrics: %s') % str(e))
|
|
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'reload',
|
|
}
|
|
|
|
def _generate_sample_metrics(self):
|
|
"""Generates demo metrics for the dashboard"""
|
|
# Delete old metrics
|
|
self.dashboard_metric_ids.unlink()
|
|
|
|
# Types of metrics to generate
|
|
metric_types = [
|
|
'bandwidth_usage', 'cpu_usage', 'memory_usage', 'clients_count',
|
|
'wan_status', 'threat_count', 'device_status'
|
|
]
|
|
|
|
# Generate new metrics
|
|
metrics_vals = []
|
|
now = fields.Datetime.now()
|
|
|
|
for metric_type in metric_types:
|
|
# Generate current value
|
|
current_value = ''
|
|
max_value = ''
|
|
history = ''
|
|
|
|
if metric_type == 'bandwidth_usage':
|
|
current_value = str(random.uniform(5, 200)) # Mbps
|
|
max_value = '1000'
|
|
history = self._generate_history_data(50, 250, 24)
|
|
elif metric_type in ['cpu_usage', 'memory_usage']:
|
|
current_value = str(random.uniform(10, 90)) # Pourcentage
|
|
max_value = '100'
|
|
history = self._generate_history_data(10, 90, 24)
|
|
elif metric_type == 'clients_count':
|
|
current_value = str(random.randint(5, 50))
|
|
max_value = '100'
|
|
history = self._generate_history_data(5, 60, 24, integer=True)
|
|
elif metric_type == 'threat_count':
|
|
current_value = str(random.randint(0, 10))
|
|
max_value = '100'
|
|
history = self._generate_history_data(0, 15, 24, integer=True)
|
|
elif metric_type == 'wan_status':
|
|
current_value = random.choice(['up', 'up']) # Principalement en ligne
|
|
max_value = ''
|
|
history = ''
|
|
elif metric_type == 'device_status':
|
|
total = random.randint(5, 20)
|
|
offline = random.randint(0, 2)
|
|
current_value = f"{total-offline}/{total}"
|
|
max_value = str(total)
|
|
history = ''
|
|
|
|
metrics_vals.append({
|
|
'site_id': self.id,
|
|
'metric_type': metric_type,
|
|
'current_value': current_value,
|
|
'max_value': max_value,
|
|
'history_data': history,
|
|
'last_update': now,
|
|
})
|
|
|
|
# Créer les métriques
|
|
for vals in metrics_vals:
|
|
self.env['udm.dashboard.metric'].create(vals)
|
|
|
|
def _generate_history_data(self, min_val, max_val, points, integer=False):
|
|
"""Generates historical data for graphs"""
|
|
data = []
|
|
for _ in range(points): # Using _ to indicate an unused variable
|
|
if integer:
|
|
value = random.randint(min_val, max_val)
|
|
else:
|
|
value = random.uniform(min_val, max_val)
|
|
value = round(value, 2)
|
|
data.append(value)
|
|
return json.dumps(data)
|
|
|
|
def _generate_sample_statistics(self):
|
|
"""Generates demo statistics for historical data"""
|
|
# Delete old statistics
|
|
self.dashboard_stat_ids.unlink()
|
|
|
|
# Types of statistics to generate
|
|
stat_types = [
|
|
'bandwidth_usage', # Total bandwidth used
|
|
'client_count', # Number of clients over time
|
|
'threat_blocked', # Number of security threats blocked
|
|
'device_uptime' # Device uptime duration
|
|
]
|
|
|
|
# Generate statistics for the last 30 days
|
|
today = fields.Date.today()
|
|
start_date = today - timedelta(days=30)
|
|
|
|
# Generate new statistics
|
|
stats_vals = []
|
|
|
|
for day in range(31): # 31 days of data
|
|
date = start_date + timedelta(days=day)
|
|
|
|
for stat_type in stat_types:
|
|
# Generate value and unit based on type
|
|
stat_value = 0.0
|
|
stat_unit = ''
|
|
|
|
if stat_type == 'bandwidth_usage':
|
|
stat_value = random.uniform(100, 1000) # GB per day
|
|
stat_unit = 'bytes'
|
|
elif stat_type == 'client_count':
|
|
stat_value = random.randint(10, 100)
|
|
stat_unit = 'count'
|
|
elif stat_type == 'threat_blocked':
|
|
stat_value = random.randint(0, 20)
|
|
stat_unit = 'count'
|
|
elif stat_type == 'device_uptime':
|
|
stat_value = random.uniform(20, 24) # Hours per day
|
|
stat_unit = 'hours'
|
|
|
|
# Add daily statistic
|
|
stats_vals.append({
|
|
'site_id': self.id,
|
|
'date': date,
|
|
'stat_type': stat_type,
|
|
'value': stat_value,
|
|
'unit': stat_unit,
|
|
'is_aggregate': False
|
|
})
|
|
|
|
# Add weekly aggregate every 7 days
|
|
if day % 7 == 0:
|
|
stats_vals.append({
|
|
'site_id': self.id,
|
|
'date': date,
|
|
'stat_type': stat_type,
|
|
'value': stat_value * 7, # Simple multiplication for demo
|
|
'unit': stat_unit,
|
|
'is_aggregate': True,
|
|
'aggregate_type': 'sum',
|
|
'time_start': fields.Datetime.to_datetime(date),
|
|
'time_end': fields.Datetime.to_datetime(date + timedelta(days=7))
|
|
})
|
|
|
|
# Create the statistics
|
|
for vals in stats_vals:
|
|
self.env['udm.dashboard.stat'].create(vals)
|
|
|
|
|
|
class UdmPortForward(models.Model):
|
|
"""Port forwarding rules for the UDM Pro"""
|
|
_name = 'udm.port.forward'
|
|
_description = 'UDM Pro Port Forward Rule'
|
|
|
|
name = fields.Char(string='Name', required=True)
|
|
enabled = fields.Boolean(string='Enabled', default=True)
|
|
src_port = fields.Char(string='Source Port')
|
|
dst_port = fields.Char(string='Destination Port')
|
|
protocol = fields.Selection([
|
|
('tcp', 'TCP'),
|
|
('udp', 'UDP'),
|
|
('both', 'TCP & UDP'),
|
|
], string='Protocol', default='tcp')
|
|
dst_address = fields.Char(string='Destination Address')
|
|
raw_data = fields.Text(string='Raw Data')
|
|
|
|
# Relations
|
|
config_id = fields.Many2one('udm.configuration', string='Configuration', ondelete='cascade', required=True)
|
|
|
|
class UdmDnsConfig(models.Model):
|
|
"""DNS configuration for the UDM Pro"""
|
|
_name = 'udm.dns.config'
|
|
_description = 'UDM Pro DNS Configuration'
|
|
|
|
enabled = fields.Boolean(string='Enabled', default=True)
|
|
filters_enabled = fields.Boolean(string='Content Filtering Enabled', default=False)
|
|
custom_dns = fields.Char(string='Custom DNS Servers')
|
|
raw_data = fields.Text(string='Raw Data')
|
|
|
|
# Relations
|
|
config_id = fields.Many2one('udm.configuration', string='Configuration', ondelete='cascade', required=True)
|
|
|
|
class UdmRoutingConfig(models.Model):
|
|
"""Routing configuration for the UDM Pro"""
|
|
_name = 'udm.routing.config'
|
|
_description = 'UDM Pro Routing Configuration'
|
|
|
|
ospf_enabled = fields.Boolean(string='OSPF Enabled', default=False)
|
|
static_routes = fields.Text(string='Static Routes')
|
|
raw_data = fields.Text(string='Raw Data')
|
|
|
|
# Relations
|
|
config_id = fields.Many2one('udm.configuration', string='Configuration', ondelete='cascade', required=True)
|
|
|
|
class UdmConfiguration(models.Model):
|
|
def _get_base_url(self):
|
|
"""Get the base URL for the UDM Pro API"""
|
|
return f'https://{self.host}:{self.port}'
|
|
|
|
def _get_api_url(self, endpoint):
|
|
"""Get the full URL for a UDM Pro API endpoint
|
|
|
|
This method handles the URL generation for different API endpoints.
|
|
Authentication and system endpoints don't need the /proxy/network prefix,
|
|
while all other endpoints require it.
|
|
|
|
Args:
|
|
endpoint (str): API endpoint path (e.g. '/api/auth/login')
|
|
|
|
Returns:
|
|
str: Complete URL for the API endpoint
|
|
"""
|
|
# Add /proxy/network prefix for all endpoints except auth
|
|
if not endpoint.startswith('/api/auth/'):
|
|
endpoint = f'/proxy/network{endpoint}'
|
|
|
|
url = f'{self._get_base_url()}{endpoint}'
|
|
_logger.debug('Generated URL: %s', url)
|
|
return url
|
|
|
|
def _get_api_headers(self, csrf_token=None):
|
|
"""Get headers for UniFi API requests
|
|
|
|
Generates the necessary headers for API requests to the UDM Pro.
|
|
All requests use JSON format and identify themselves as the Odoo UniFi Integration.
|
|
For authenticated endpoints, a CSRF token is included in the headers.
|
|
|
|
Args:
|
|
csrf_token (str, optional): CSRF token required for authenticated requests
|
|
|
|
Returns:
|
|
dict: Dictionary containing the required headers for the API request
|
|
"""
|
|
# Basic headers required for all requests
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'User-Agent': 'Odoo UniFi Integration/1.0'
|
|
}
|
|
|
|
# Add CSRF token for authenticated requests
|
|
if csrf_token:
|
|
headers['X-CSRF-Token'] = csrf_token
|
|
_logger.debug('Adding CSRF token to headers')
|
|
|
|
_logger.debug('Generated API headers: %s', headers)
|
|
return headers
|
|
|
|
def _login(self):
|
|
"""Login to UniFi OS and get authentication cookie and CSRF token
|
|
|
|
This method handles the authentication process with the UDM Pro:
|
|
1. Validates login credentials
|
|
2. Gets CSRF token
|
|
3. Performs login with CSRF token
|
|
4. Handles MFA if required
|
|
|
|
Returns:
|
|
dict: A dictionary containing the session cookies and CSRF token
|
|
|
|
Raises:
|
|
UserError: If authentication fails or MFA is required but not provided
|
|
"""
|
|
_logger.info('Starting connection to UDM Pro')
|
|
_logger.debug('Connection parameters - Host: %s, Port: %s, User: %s',
|
|
self.host, self.port, self.username)
|
|
|
|
# Validate required credentials
|
|
if not self.host or not self.username or not self.password:
|
|
_logger.error('Missing connection information: host=%s, username=%s, password=%s',
|
|
bool(self.host), bool(self.username), bool(self.password))
|
|
raise UserError(_('Please provide host, username and password'))
|
|
|
|
# Disable SSL warnings - UDM Pro often uses self-signed certificates
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
# Create session and disable SSL verification
|
|
session = requests.Session()
|
|
session.verify = False
|
|
|
|
try:
|
|
_logger.info('Attempting to connect to UniFi OS API...')
|
|
|
|
# Prepare login data
|
|
login_data = {
|
|
'username': self.username,
|
|
'password': self.password,
|
|
'rememberMe': True
|
|
}
|
|
|
|
# Add MFA token if present
|
|
if self.mfa_token:
|
|
login_data['token'] = self.mfa_token
|
|
_logger.debug('Adding MFA token to login data')
|
|
|
|
# Set required headers for UniFi OS
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'User-Agent': 'Odoo UniFi Integration/1.0'
|
|
}
|
|
|
|
_logger.debug('Connection details: %s', {
|
|
'url': self._get_api_url('/api/auth/login'),
|
|
'username': self.username,
|
|
'host': self.host,
|
|
'port': self.port
|
|
})
|
|
|
|
# Attempt login
|
|
response = session.post(
|
|
self._get_api_url('/api/auth/login'),
|
|
json=login_data,
|
|
headers=headers,
|
|
verify=False,
|
|
timeout=10
|
|
)
|
|
|
|
# Check response status
|
|
response.raise_for_status()
|
|
|
|
# Check if MFA is required
|
|
if response.status_code == 403 and 'x-factor-required' in response.headers:
|
|
_logger.info('Two-factor authentication required')
|
|
raise UserError(_('Please provide two-factor authentication code'))
|
|
|
|
# Get CSRF token from response headers
|
|
csrf_token = response.headers.get('X-CSRF-Token', '')
|
|
|
|
_logger.info('Successfully connected to UDM Pro')
|
|
_logger.debug('Received cookies: %s', dict(session.cookies))
|
|
|
|
# Return session and CSRF token
|
|
return {
|
|
'session': session,
|
|
'csrf_token': csrf_token
|
|
}
|
|
|
|
except RequestException as e:
|
|
error_msg = str(e)
|
|
if hasattr(e, 'response') and e.response is not None:
|
|
error_msg = e.response.text
|
|
try:
|
|
error_data = json.loads(error_msg)
|
|
if e.response.status_code == 401:
|
|
error_message = error_data.get('error', {}).get('message', '')
|
|
raise UserError(_('Authentication failed: %s\n\nPlease check:\n1. Username and password are correct\n2. User exists in UniFi OS\n3. User has sufficient permissions') % error_message)
|
|
elif e.response.status_code == 403:
|
|
raise UserError(_('Access denied. Please verify your permissions.'))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
_logger.error('Login failed: %s', error_msg)
|
|
raise UserError(_('Connection to UDM Pro failed: %s') % error_msg)
|
|
|
|
def _fetch_metrics(self):
|
|
"""Fetch real-time metrics from the UDM Pro
|
|
|
|
Returns:
|
|
dict: System and network metrics
|
|
|
|
Raises:
|
|
UserError: If connection fails or metrics are inaccessible
|
|
"""
|
|
try:
|
|
# Get authenticated session
|
|
auth = self._login()
|
|
if not auth:
|
|
raise UserError(_('Unable to authenticate with UDM Pro'))
|
|
|
|
session = auth.get('session')
|
|
csrf_token = auth.get('csrf_token')
|
|
if not session or not csrf_token:
|
|
raise UserError(_('Missing session or CSRF token'))
|
|
|
|
# Set headers for API requests
|
|
headers = self._get_api_headers(csrf_token)
|
|
|
|
# Get system statistics
|
|
system_response = session.get(
|
|
self._get_api_url('/api/system/stats'),
|
|
headers=headers
|
|
)
|
|
system_response.raise_for_status()
|
|
system_stats = system_response.json()
|
|
|
|
# Get client information
|
|
clients_response = session.get(
|
|
self._get_api_url('/api/site/default/clients'),
|
|
headers=headers
|
|
)
|
|
clients_response.raise_for_status()
|
|
clients = clients_response.json()
|
|
|
|
# Get network statistics
|
|
network_response = session.get(
|
|
self._get_api_url('/api/site/default/devices'),
|
|
headers=headers
|
|
)
|
|
network_response.raise_for_status()
|
|
network_stats = network_response.json()
|
|
|
|
# Prepare metrics
|
|
metrics_vals = [
|
|
{
|
|
'site_id': self.site_id.id,
|
|
'metric_type': 'cpu_usage',
|
|
'current_value': str(system_stats.get('data', {}).get('cpu', {}).get('usage', 0)),
|
|
'max_value': '100',
|
|
'last_update': fields.Datetime.now(),
|
|
'description': 'CPU Usage'
|
|
},
|
|
{
|
|
'site_id': self.site_id.id,
|
|
'metric_type': 'memory_usage',
|
|
'current_value': str(system_stats.get('data', {}).get('memory', {}).get('used_percentage', 0)),
|
|
'max_value': '100',
|
|
'last_update': fields.Datetime.now(),
|
|
'description': 'Memory Usage'
|
|
},
|
|
{
|
|
'site_id': self.site_id.id,
|
|
'metric_type': 'client_count',
|
|
'current_value': str(len(clients.get('data', []))),
|
|
'max_value': '1000',
|
|
'last_update': fields.Datetime.now(),
|
|
'description': 'Connected Clients'
|
|
},
|
|
{
|
|
'site_id': self.site_id.id,
|
|
'metric_type': 'network_throughput',
|
|
'current_value': str(sum(d.get('tx_bytes', 0) + d.get('rx_bytes', 0)
|
|
for d in network_stats.get('data', []))),
|
|
'max_value': str(1e9), # 1 Gbps
|
|
'last_update': fields.Datetime.now(),
|
|
'description': 'Total Network Throughput'
|
|
}
|
|
]
|
|
|
|
# Get historical statistics
|
|
history_response = session.get(
|
|
self._get_api_url('/api/site/default/statistics/5minutes'),
|
|
headers=headers
|
|
)
|
|
history_response.raise_for_status()
|
|
history_stats = history_response.json()
|
|
|
|
# Create historical statistics
|
|
stats_vals = []
|
|
now = fields.Datetime.now()
|
|
|
|
for stat in history_stats.get('data', []):
|
|
# Get timestamp and ensure it's valid
|
|
try:
|
|
timestamp = fields.Datetime.from_string(stat.get('time')) or now
|
|
except (ValueError, TypeError):
|
|
timestamp = now
|
|
_logger.warning('Invalid timestamp detected in historical statistics')
|
|
|
|
stats_vals.append({
|
|
'site_id': self.site_id.id,
|
|
'timestamp': fields.Datetime.to_string(timestamp),
|
|
'rx_bytes': stat.get('rx_bytes', 0),
|
|
'tx_bytes': stat.get('tx_bytes', 0),
|
|
'num_sta': stat.get('num_sta', 0),
|
|
'cpu_usage': system_stats.get('data', {}).get('cpu', {}).get('usage', 0),
|
|
'memory_usage': system_stats.get('data', {}).get('memory', {}).get('used_percentage', 0)
|
|
})
|
|
|
|
# Update metrics and statistics
|
|
self.env['udm.dashboard.metric'].search([('site_id', '=', self.site_id.id)]).unlink()
|
|
self.env['udm.dashboard.metric'].create(metrics_vals)
|
|
|
|
self.env['udm.dashboard.stat'].search([('site_id', '=', self.site_id.id)]).unlink()
|
|
self.env['udm.dashboard.stat'].create(stats_vals)
|
|
|
|
_logger.info('Metrics and statistics updated successfully')
|
|
|
|
except RequestException as e:
|
|
_logger.error('Error fetching metrics: %s', str(e))
|
|
raise UserError(_('Unable to fetch metrics from UDM Pro: %s') % str(e))
|
|
"""Complete UDM Pro configuration stored in Odoo"""
|
|
_name = 'udm.configuration'
|
|
_description = 'UDM Pro Configuration'
|
|
_order = 'timestamp desc'
|
|
|
|
name = fields.Char(string='Name', compute='_compute_name', store=True)
|
|
timestamp = fields.Datetime(string='Timestamp', default=fields.Datetime.now, required=True)
|
|
raw_data = fields.Text(string='Raw Data', help="Raw configuration data in JSON format")
|
|
active = fields.Boolean(string='Active', default=True, help="Indicates if this configuration is currently active")
|
|
|
|
# UDM Pro Connection
|
|
host = fields.Char(string='Host', help="IP address or hostname of the UDM Pro")
|
|
port = fields.Integer(string='Port', default=443)
|
|
username = fields.Char(string='Username')
|
|
password = fields.Char(string='Password')
|
|
mfa_token = fields.Char(string='MFA Code', help="Two-factor authentication code received by email")
|
|
|
|
# Relations
|
|
site_id = fields.Many2one('udm.site', string='Site', required=True,
|
|
ondelete='restrict',
|
|
help='Site this configuration belongs to')
|
|
system_info_id = fields.Many2one('udm.system.info', string='System Info',
|
|
ondelete='cascade',
|
|
help='System information snapshot')
|
|
network_ids = fields.One2many('udm.network', 'site_id', string='Networks',
|
|
help='Networks in this site')
|
|
vlan_ids = fields.One2many('udm.vlan', 'site_id', string='VLANs',
|
|
help='VLANs in this site')
|
|
device_ids = fields.One2many('udm.device', 'site_id', string='Devices',
|
|
help='Devices in this site')
|
|
user_ids = fields.One2many('udm.user', 'site_id', string='Users',
|
|
help='Users in this site')
|
|
settings_id = fields.Many2one('udm.settings', string='Settings',
|
|
ondelete='cascade',
|
|
help='Site settings')
|
|
firewall_rule_ids = fields.One2many('udm.firewall.rule', 'site_id',
|
|
string='Firewall Rules',
|
|
help='Firewall rules for this site')
|
|
port_forward_ids = fields.One2many('udm.port.forward', 'site_id',
|
|
string='Port Forwards',
|
|
help='Port forwarding rules for this site')
|
|
dns_config_id = fields.Many2one('udm.dns.config', string='DNS Configuration',
|
|
ondelete='cascade',
|
|
help='DNS settings')
|
|
routing_config_id = fields.Many2one('udm.routing.config',
|
|
string='Routing Configuration',
|
|
ondelete='cascade',
|
|
help='Routing settings')
|
|
|
|
# Statistics
|
|
network_count = fields.Integer(compute='_compute_counts', string='Network Count')
|
|
device_count = fields.Integer(compute='_compute_counts', string='Device Count')
|
|
user_count = fields.Integer(compute='_compute_counts', string='User Count')
|
|
firewall_rule_count = fields.Integer(compute='_compute_counts', string='Firewall Rule Count')
|
|
|
|
@api.depends('timestamp', 'system_info_id.hostname')
|
|
def _compute_name(self):
|
|
for record in self:
|
|
hostname = record.system_info_id.hostname or 'Unknown'
|
|
timestamp = record.timestamp.strftime('%Y-%m-%d %H:%M:%S') if record.timestamp else ''
|
|
record.name = f"{hostname} ({timestamp})"
|
|
|
|
@api.depends('network_ids', 'device_ids', 'user_ids', 'firewall_rule_ids')
|
|
def _compute_counts(self):
|
|
for record in self:
|
|
record.network_count = len(record.network_ids)
|
|
record.device_count = len(record.device_ids)
|
|
record.user_count = len(record.user_ids)
|
|
record.firewall_rule_count = len(record.firewall_rule_ids)
|
|
|
|
def action_view_networks(self):
|
|
"""Open the networks view filtered for this site
|
|
|
|
Returns a window action to display the list of networks
|
|
associated with this site.
|
|
"""
|
|
self.ensure_one()
|
|
return {
|
|
'name': _('Networks'),
|
|
'view_mode': 'tree,form',
|
|
'res_model': 'udm.network',
|
|
'domain': [('site_id', '=', self.site_id.id)],
|
|
'type': 'ir.actions.act_window',
|
|
}
|
|
|
|
def action_view_devices(self):
|
|
"""Open the devices view filtered for this site
|
|
|
|
Returns a window action to display the list of devices
|
|
associated with this site.
|
|
"""
|
|
self.ensure_one()
|
|
return {
|
|
'name': _('Devices'),
|
|
'view_mode': 'tree,form',
|
|
'res_model': 'udm.device',
|
|
'domain': [('site_id', '=', self.site_id.id)],
|
|
'type': 'ir.actions.act_window',
|
|
}
|
|
|
|
def action_view_users(self):
|
|
"""Open the users view filtered for this site
|
|
|
|
Returns a window action to display the list of users
|
|
associated with this site.
|
|
"""
|
|
self.ensure_one()
|
|
return {
|
|
'name': _('Users'),
|
|
'view_mode': 'tree,form',
|
|
'res_model': 'udm.user',
|
|
'domain': [('site_id', '=', self.site_id.id)],
|
|
'type': 'ir.actions.act_window',
|
|
}
|
|
|
|
def action_view_firewall_rules(self):
|
|
"""Open the firewall rules view filtered for this site
|
|
|
|
Returns a window action to display the list of firewall rules
|
|
associated with this site.
|
|
"""
|
|
self.ensure_one()
|
|
return {
|
|
'name': _('Firewall Rules'),
|
|
'view_mode': 'tree,form',
|
|
'res_model': 'udm.firewall.rule',
|
|
'domain': [('site_id', '=', self.site_id.id)],
|
|
'type': 'ir.actions.act_window',
|
|
}
|
|
|
|
def action_import_configuration(self):
|
|
"""Import configuration from UDM Pro device.
|
|
|
|
This method is called when the user clicks the Import button in the
|
|
import configuration wizard. It performs the following steps:
|
|
|
|
1. Validates that all required connection information is provided
|
|
2. Connects to the UDM Pro device using the provided credentials
|
|
3. If MFA is required, opens the MFA wizard
|
|
4. Once authenticated, retrieves the current configuration from the device
|
|
5. Imports the configuration into Odoo using import_configuration()
|
|
6. Returns an action to view the imported configuration
|
|
|
|
Returns:
|
|
dict: An action to display the imported configuration or the MFA wizard
|
|
|
|
Raises:
|
|
UserError: If any required connection information is missing
|
|
"""
|
|
self.ensure_one()
|
|
|
|
# Validate that all required connection information is provided
|
|
if not all([self.host, self.port, self.username, self.password]):
|
|
raise UserError(_('Please provide host, username, password and port'))
|
|
|
|
try:
|
|
# Try to connect
|
|
auth = self._login()
|
|
if not auth:
|
|
raise UserError(_('Unable to authenticate with UDM Pro'))
|
|
session = auth.get('session')
|
|
csrf_token = auth.get('csrf_token')
|
|
if not session or not csrf_token:
|
|
raise UserError(_('Missing session or CSRF token'))
|
|
except UserError as e:
|
|
if 'two-factor authentication' in str(e):
|
|
# Open MFA wizard
|
|
return {
|
|
'name': _('Two-Factor Authentication Code'),
|
|
'type': 'ir.actions.act_window',
|
|
'res_model': 'udm.mfa.wizard',
|
|
'view_mode': 'form',
|
|
'target': 'new',
|
|
'context': {
|
|
'default_config_id': self.id
|
|
}
|
|
}
|
|
raise
|
|
|
|
try:
|
|
config_data = {}
|
|
|
|
# Get site context first
|
|
_logger.info('Getting site context...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/self'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
if response.status_code != 200:
|
|
_logger.error('Failed to get site context: %s', response.text)
|
|
raise UserError(_('Failed to access site. Please check your credentials.'))
|
|
|
|
# Fetch system info
|
|
_logger.info('Retrieving system information...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/stat/sysinfo'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
|
|
system_info = response.json()
|
|
_logger.info('System data received: %s', system_info)
|
|
|
|
system_info_data = system_info.get('data')
|
|
if isinstance(system_info_data, list):
|
|
# Si les données sont dans une liste, prenons le premier élément
|
|
system_info_data = system_info_data[0] if system_info_data else {}
|
|
elif not isinstance(system_info_data, dict):
|
|
system_info_data = {}
|
|
|
|
if not system_info_data:
|
|
_logger.error('No system data received in response')
|
|
raise UserError(_('Failed to retrieve system information from UDM Pro'))
|
|
config_data['system_info'] = system_info_data
|
|
|
|
# Fetch networks
|
|
_logger.info('Retrieving networks...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/rest/setting/network'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
networks = response.json()
|
|
networks_data = networks.get('data')
|
|
if isinstance(networks_data, list):
|
|
# Handle list format
|
|
config_data['networks'] = {'networks': networks_data}
|
|
elif isinstance(networks_data, dict):
|
|
# Handle dictionary format
|
|
config_data['networks'] = {'networks': [networks_data]}
|
|
else:
|
|
_logger.error('No network data received in response')
|
|
raise UserError(_('Failed to retrieve networks from UDM Pro'))
|
|
|
|
# Fetch VLANs
|
|
_logger.info('Retrieving VLANs...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/rest/setting/network'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
vlans = response.json()
|
|
vlans_data = vlans.get('data')
|
|
if isinstance(vlans_data, list):
|
|
# Handle list format
|
|
config_data['vlans'] = {'vlans': vlans_data}
|
|
elif isinstance(vlans_data, dict):
|
|
# Handle dictionary format
|
|
config_data['vlans'] = {'vlans': [vlans_data]}
|
|
else:
|
|
_logger.error('No VLAN data received in response')
|
|
raise UserError(_('Failed to retrieve VLANs from UDM Pro'))
|
|
|
|
# Fetch devices
|
|
_logger.info('Retrieving devices...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/stat/device-basic'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
devices = response.json()
|
|
devices_data = devices.get('data')
|
|
if isinstance(devices_data, list):
|
|
# Handle list format
|
|
config_data['devices'] = {'devices': devices_data}
|
|
elif isinstance(devices_data, dict):
|
|
# Handle dictionary format
|
|
config_data['devices'] = {'devices': [devices_data]}
|
|
else:
|
|
_logger.error('No device data received in response')
|
|
raise UserError(_('Failed to retrieve devices from UDM Pro'))
|
|
|
|
# Fetch users
|
|
_logger.info('Retrieving users...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/rest/user'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
users = response.json()
|
|
users_data = users.get('data')
|
|
if isinstance(users_data, list):
|
|
# Handle list format
|
|
config_data['users'] = {'users': users_data}
|
|
elif isinstance(users_data, dict):
|
|
# Handle dictionary format
|
|
config_data['users'] = {'users': [users_data]}
|
|
else:
|
|
_logger.error('No user data received in response')
|
|
raise UserError(_('Failed to retrieve users from UDM Pro'))
|
|
|
|
# Fetch settings
|
|
_logger.info('Retrieving settings...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/rest/setting'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
settings = response.json()
|
|
settings_data = settings.get('data')
|
|
if isinstance(settings_data, list):
|
|
# Handle list format
|
|
config_data['settings'] = {'settings': settings_data}
|
|
elif isinstance(settings_data, dict):
|
|
# Handle dictionary format
|
|
config_data['settings'] = {'settings': [settings_data]}
|
|
else:
|
|
_logger.error('No settings data received in response')
|
|
raise UserError(_('Failed to retrieve settings from UDM Pro'))
|
|
|
|
# Fetch firewall rules
|
|
_logger.info('Retrieving firewall rules...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/rest/firewallrule'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
firewall_rules = response.json()
|
|
firewall_rules_data = firewall_rules.get('data')
|
|
if isinstance(firewall_rules_data, list):
|
|
# Handle list format
|
|
config_data['firewall_rules'] = {'rules': firewall_rules_data}
|
|
elif isinstance(firewall_rules_data, dict):
|
|
# Handle dictionary format
|
|
config_data['firewall_rules'] = {'rules': [firewall_rules_data]}
|
|
else:
|
|
_logger.error('No firewall rule data received in response')
|
|
raise UserError(_('Failed to retrieve firewall rules from UDM Pro'))
|
|
|
|
# Fetch port forwarding rules
|
|
_logger.info('Retrieving port forwarding rules...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/rest/portforward'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
port_forwards = response.json()
|
|
port_forwards_data = port_forwards.get('data')
|
|
if isinstance(port_forwards_data, list):
|
|
# Handle list format
|
|
config_data['port_forwards'] = {'port_forwards': port_forwards_data}
|
|
elif isinstance(port_forwards_data, dict):
|
|
# Handle dictionary format
|
|
config_data['port_forwards'] = {'port_forwards': [port_forwards_data]}
|
|
else:
|
|
_logger.error('No port forwarding data received in response')
|
|
raise UserError(_('Failed to retrieve port forwarding rules from UDM Pro'))
|
|
|
|
# Fetch DNS configuration
|
|
_logger.info('Retrieving DNS configuration...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/rest/setting/dns'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
dns_config = response.json()
|
|
dns_config_data = dns_config.get('data')
|
|
if isinstance(dns_config_data, list):
|
|
# Handle list format
|
|
config_data['dns_config'] = {'dns_config': dns_config_data}
|
|
elif isinstance(dns_config_data, dict):
|
|
# Handle dictionary format
|
|
config_data['dns_config'] = {'dns_config': [dns_config_data]}
|
|
else:
|
|
_logger.error('No DNS configuration data received in response')
|
|
raise UserError(_('Failed to retrieve DNS configuration from UDM Pro'))
|
|
|
|
# Fetch routing configuration
|
|
_logger.info('Retrieving routing configuration...')
|
|
response = session.get(
|
|
self._get_api_url('/api/s/default/stat/routing'),
|
|
headers=self._get_api_headers(csrf_token),
|
|
verify=False
|
|
)
|
|
_logger.info('API response (status: %s): %s', response.status_code, response.text)
|
|
routing_config = response.json()
|
|
routing_config_data = routing_config.get('data')
|
|
if isinstance(routing_config_data, list):
|
|
# Handle list format
|
|
config_data['routing'] = {'routing': routing_config_data}
|
|
elif isinstance(routing_config_data, dict):
|
|
# Handle dictionary format
|
|
config_data['routing'] = {'routing': [routing_config_data]}
|
|
else:
|
|
_logger.error('No routing configuration data received in response')
|
|
raise UserError(_('Failed to retrieve routing configuration from UDM Pro'))
|
|
|
|
except RequestException as e:
|
|
raise UserError(_('Failed to retrieve configuration from UDM Pro: %s') % str(e))
|
|
|
|
# Import the configuration using the model's import method
|
|
config = self.import_configuration(config_data)
|
|
|
|
# Return an action to display the imported configuration
|
|
return {
|
|
'type': 'ir.actions.act_window',
|
|
'name': _('Configuration importée'),
|
|
'res_model': 'udm.configuration',
|
|
'res_id': config.id,
|
|
'view_mode': 'form',
|
|
'target': 'current'
|
|
}
|
|
|
|
@api.model
|
|
def import_configuration(self, config_data):
|
|
"""
|
|
Imports a complete UDM Pro configuration into Odoo
|
|
|
|
Args:
|
|
config_data (dict): Raw configuration data from the API
|
|
|
|
Returns:
|
|
int: ID of the created configuration
|
|
"""
|
|
if not config_data:
|
|
raise UserError(_("Please provide configuration data"))
|
|
|
|
# Create the main configuration
|
|
vals = {
|
|
'timestamp': datetime.now(),
|
|
'raw_data': json.dumps(config_data, indent=2, ensure_ascii=False),
|
|
}
|
|
|
|
config = self.create(vals)
|
|
|
|
# Create system information
|
|
system_info_data = config_data.get('system_info', {}).get('system_info')
|
|
if isinstance(system_info_data, list):
|
|
# Si les données sont dans une liste, prenons le premier élément
|
|
system_info_data = system_info_data[0] if system_info_data else {}
|
|
elif not isinstance(system_info_data, dict):
|
|
system_info_data = {}
|
|
|
|
if system_info_data:
|
|
system_info = self.env['udm.system.info'].create({
|
|
'site_id': config.site_id.id,
|
|
'hostname': system_info_data.get('hostname', ''),
|
|
'version': system_info_data.get('version', ''),
|
|
'model': system_info_data.get('model', ''),
|
|
'uptime': system_info_data.get('uptime', 0),
|
|
'serial': system_info_data.get('serialNumber', ''),
|
|
'mac_address': system_info_data.get('mac', '') or system_info_data.get('macAddress', ''),
|
|
'raw_data': json.dumps(system_info_data, indent=2, ensure_ascii=False),
|
|
})
|
|
config.system_info_id = system_info.id
|
|
|
|
# Create networks
|
|
networks_data = config_data.get('networks', {}).get('networks', [])
|
|
for network_data in networks_data:
|
|
if isinstance(network_data, dict):
|
|
# Create network with all available fields
|
|
self.env['udm.network'].create({
|
|
'site_id': config.site_id.id,
|
|
'name': network_data.get('name', ''),
|
|
'purpose': network_data.get('purpose', 'corporate'),
|
|
'subnet': network_data.get('subnet', ''),
|
|
'vlan_id_number': network_data.get('vlanId'),
|
|
'dhcp_enabled': network_data.get('dhcpEnabled', False),
|
|
'dhcp_start': network_data.get('dhcpStart', ''),
|
|
'dhcp_stop': network_data.get('dhcpStop', ''),
|
|
'domain_name': network_data.get('domainName', ''),
|
|
'raw_data': json.dumps(network_data, indent=2, ensure_ascii=False),
|
|
})
|
|
|
|
# Create VLANs
|
|
vlans_data = config_data.get('networks', {}).get('vlans', [])
|
|
for vlan_data in vlans_data:
|
|
if isinstance(vlan_data, dict):
|
|
# Create VLAN with all available fields
|
|
self.env['udm.vlan'].create({
|
|
'site_id': config.site_id.id,
|
|
'vlan_id': vlan_data.get('id', 0),
|
|
'name': vlan_data.get('name', ''),
|
|
'enabled': vlan_data.get('enabled', True),
|
|
'raw_data': json.dumps(vlan_data, indent=2, ensure_ascii=False)
|
|
})
|
|
|
|
# Create devices
|
|
devices_data = config_data.get('devices', {}).get('devices', [])
|
|
for device_data in devices_data:
|
|
if isinstance(device_data, dict):
|
|
# Determine device type based on model or type
|
|
device_type = device_data.get('type', '')
|
|
if not device_type:
|
|
model = device_data.get('model', '').lower()
|
|
if 'uap' in model or 'ap' in model:
|
|
device_type = 'uap'
|
|
elif 'usw' in model or 'switch' in model:
|
|
device_type = 'usw'
|
|
elif 'ugw' in model or 'gateway' in model:
|
|
device_type = 'ugw'
|
|
elif 'udm' in model or 'dream' in model:
|
|
device_type = 'udm'
|
|
else:
|
|
device_type = 'client'
|
|
|
|
# Create device with all available fields
|
|
self.env['udm.device'].create({
|
|
'site_id': config.site_id.id,
|
|
'name': device_data.get('name', ''),
|
|
'mac_address': device_data.get('mac', ''),
|
|
'ip_address': device_data.get('ip', ''),
|
|
'device_type': device_type,
|
|
'model': device_data.get('model', ''),
|
|
'last_seen': datetime.fromtimestamp(device_data.get('lastSeen', 0)),
|
|
'raw_data': json.dumps(device_data, indent=2, ensure_ascii=False),
|
|
})
|
|
|
|
# Create users
|
|
users_data = config_data.get('users', {}).get('users', [])
|
|
for user_data in users_data:
|
|
if isinstance(user_data, dict):
|
|
self.env['udm.user'].create({
|
|
'site_id': config.site_id.id,
|
|
'name': user_data.get('name', ''),
|
|
'email': user_data.get('email', ''),
|
|
'role': user_data.get('role', ''),
|
|
'enabled': user_data.get('enabled', True),
|
|
'mac_address': user_data.get('mac', ''),
|
|
'ip_address': user_data.get('ip', ''),
|
|
'network_id': self.env['udm.network'].search([('site_id', '=', config.site_id.id), ('name', '=', user_data.get('network', ''))]).id,
|
|
'last_seen': datetime.fromtimestamp(user_data.get('lastSeen', 0)),
|
|
'raw_data': json.dumps(user_data, indent=2, ensure_ascii=False),
|
|
})
|
|
|
|
# Create settings
|
|
settings_data = config_data.get('settings', {})
|
|
if settings_data:
|
|
# Extract system settings
|
|
system_settings = settings_data.get('system', {})
|
|
dns_settings = settings_data.get('dns', {})
|
|
services = settings_data.get('services', {})
|
|
|
|
# Create settings record with all available fields
|
|
settings = self.env['udm.settings'].create({
|
|
'site_id': config.site_id.id,
|
|
'timezone': system_settings.get('timezone', 'America/Montreal'),
|
|
|
|
# Time settings
|
|
'ntp_enabled': system_settings.get('ntp', {}).get('enabled', True),
|
|
'ntp_servers': ','.join(system_settings.get('ntp', {}).get('servers', [])),
|
|
|
|
# DNS settings
|
|
'dns_enabled': dns_settings.get('enabled', True),
|
|
'dns_servers': ','.join(dns_settings.get('servers', [])),
|
|
'dns_forwarding': dns_settings.get('forwarding', {}).get('enabled', True),
|
|
|
|
# Advanced settings
|
|
'upnp_enabled': services.get('upnp', {}).get('enabled', False),
|
|
'mdns_enabled': services.get('mdns', {}).get('enabled', True),
|
|
'igmp_proxy': services.get('igmp_proxy', {}).get('enabled', False),
|
|
|
|
# Raw data for debugging and future reference
|
|
'raw_data': json.dumps(settings_data, indent=2, ensure_ascii=False),
|
|
})
|
|
config.settings_id = settings.id
|
|
|
|
# Create port forwarding rules
|
|
port_forward_data = config_data.get('port_forwards', {}).get('data', [])
|
|
for rule_data in port_forward_data:
|
|
if isinstance(rule_data, dict):
|
|
self.env['udm.port.forward'].create({
|
|
'site_id': config.site_id.id,
|
|
'name': rule_data.get('name', ''),
|
|
'enabled': rule_data.get('enabled', True),
|
|
'src_port': str(rule_data.get('fwd', '')),
|
|
'dst_port': str(rule_data.get('port', '')),
|
|
'protocol': rule_data.get('proto', '').lower(),
|
|
'dst_address': rule_data.get('dst', ''),
|
|
'raw_data': json.dumps(rule_data, indent=2, ensure_ascii=False),
|
|
})
|
|
|
|
# Create DNS configuration
|
|
dns_config_data = config_data.get('dns_config', {}).get('data', {})
|
|
if dns_config_data:
|
|
dns_config = self.env['udm.dns.config'].create({
|
|
'site_id': config.site_id.id, # Lier directement au site
|
|
'enabled': dns_config_data.get('system', {}).get('unifi', {}).get('enabled', True),
|
|
'filters_enabled': dns_config_data.get('system', {}).get('unifi', {}).get('content_filtering_enabled', False),
|
|
'custom_dns': ','.join([str(server) for server in dns_config_data.get('system', {}).get('nameservers', [])]),
|
|
'raw_data': json.dumps(dns_config_data, indent=2, ensure_ascii=False),
|
|
})
|
|
config.dns_config_id = dns_config.id
|
|
|
|
# Create firewall rules
|
|
firewall_rules_data = config_data.get('firewall', {}).get('rules', [])
|
|
for rule_data in firewall_rules_data:
|
|
if isinstance(rule_data, dict):
|
|
# Create firewall rule with all available fields
|
|
self.env['udm.firewall.rule'].create({
|
|
'site_id': config.site_id.id, # Lier directement au site
|
|
'name': rule_data.get('name', ''),
|
|
'description': rule_data.get('description', ''),
|
|
'enabled': rule_data.get('enabled', True),
|
|
'sequence': rule_data.get('sequence', 10),
|
|
'action': rule_data.get('action', 'drop').lower(),
|
|
'protocol': rule_data.get('protocol', 'all').lower(),
|
|
'source': rule_data.get('source', ''),
|
|
'destination': rule_data.get('destination', ''),
|
|
'src_port': rule_data.get('src_port', ''),
|
|
'dst_port': rule_data.get('dst_port', ''),
|
|
'raw_data': json.dumps(rule_data, indent=2, ensure_ascii=False)
|
|
})
|
|
|
|
# Create routing configuration
|
|
routing_config_data = config_data.get('routing', {}).get('data', {})
|
|
if routing_config_data:
|
|
routing_config = self.env['udm.routing.config'].create({
|
|
'site_id': config.site_id.id, # Lier directement au site
|
|
'ospf_enabled': routing_config_data.get('ospf', {}).get('enabled', False),
|
|
'static_routes': ','.join([f"{route.get('network')}/{route.get('prefix')} via {route.get('nexthop')}" for route in routing_config_data.get('static_routes', [])]),
|
|
'raw_data': json.dumps(routing_config_data, indent=2, ensure_ascii=False),
|
|
})
|
|
config.routing_config_id = routing_config.id
|
|
|
|
# Create firewall rules
|
|
firewall_data = config_data.get('firewall_rules', {}).get('rules', [])
|
|
for rule_data in firewall_data:
|
|
if isinstance(rule_data, dict):
|
|
self.env['udm.firewall.rule'].create({
|
|
'site_id': config.site_id.id,
|
|
'name': rule_data.get('name', ''),
|
|
'description': rule_data.get('description', ''),
|
|
'action': rule_data.get('action', 'drop'),
|
|
'protocol': rule_data.get('protocol', ''),
|
|
'source': rule_data.get('source', ''),
|
|
'destination': rule_data.get('destination', ''),
|
|
'enabled': rule_data.get('enabled', True),
|
|
'raw_data': json.dumps(rule_data, indent=2, ensure_ascii=False),
|
|
})
|
|
|
|
# Log successful import
|
|
_logger.info('Successfully imported UDM Pro configuration: %s', config.name)
|
|
|
|
return config
|
|
|
|
def action_compare_with(self):
|
|
"""Opens a wizard to compare this configuration with another"""
|
|
self.ensure_one()
|
|
return {
|
|
'name': _('Compare Configurations'),
|
|
'type': 'ir.actions.act_window',
|
|
'res_model': 'udm.configuration.compare.wizard',
|
|
'view_mode': 'form',
|
|
'target': 'new',
|
|
'context': {
|
|
'default_source_config_id': self.id,
|
|
},
|
|
}
|
|
|
|
def action_duplicate(self):
|
|
"""Duplicates this configuration"""
|
|
self.ensure_one()
|
|
|
|
# Create a new configuration by copying the raw data
|
|
new_config = self.copy({
|
|
'timestamp': datetime.now(),
|
|
'name': _('%s (Duplicate)') % self.name,
|
|
})
|
|
|
|
return {
|
|
'name': _('Duplicated Configuration'),
|
|
'type': 'ir.actions.act_window',
|
|
'res_model': 'udm.configuration',
|
|
'res_id': new_config.id,
|
|
'view_mode': 'form',
|
|
}
|
|
|
|
def action_generate_report(self):
|
|
"""Generates a PDF report of this configuration"""
|
|
self.ensure_one()
|
|
return {
|
|
'type': 'ir.actions.report',
|
|
'report_name': 'unifi_integration.report_udm_configuration',
|
|
'report_type': 'qweb-pdf',
|
|
'res_model': self._name,
|
|
'res_id': self.id,
|
|
}
|
|
|
|
def action_view_dashboard(self):
|
|
"""Displays the dashboard for this configuration's site"""
|
|
self.ensure_one()
|
|
if not self.site_id:
|
|
raise UserError(_('Configuration not linked to a site. Please select a site first.'))
|
|
|
|
return {
|
|
'name': _('Site Dashboard'),
|
|
'view_mode': 'dashboard,form',
|
|
'res_model': 'udm.site',
|
|
'res_id': self.site_id.id,
|
|
'type': 'ir.actions.act_window',
|
|
}
|
|
|
|
def action_update_dashboard_metrics(self):
|
|
"""Updates the dashboard metrics for this configuration's site"""
|
|
self.ensure_one()
|
|
if not self.site_id:
|
|
raise UserError(_('Configuration not linked to a site. Please select a site first.'))
|
|
|
|
# Call the site's metric refresh method
|
|
return self.site_id.action_refresh_metrics()
|
|
|
|
def unlink(self):
|
|
"""Override unlink method to handle configuration deletion properly"""
|
|
for record in self:
|
|
# Suppression des enregistrements liés avec ondelete='cascade'
|
|
if record.system_info_id:
|
|
record.system_info_id.unlink()
|
|
if record.settings_id:
|
|
record.settings_id.unlink()
|
|
if record.dns_config_id:
|
|
record.dns_config_id.unlink()
|
|
if record.routing_config_id:
|
|
record.routing_config_id.unlink()
|
|
|
|
# Suppression des enregistrements One2many
|
|
record.network_ids.unlink()
|
|
record.vlan_ids.unlink()
|
|
record.device_ids.unlink()
|
|
record.user_ids.unlink()
|
|
record.firewall_rule_ids.unlink()
|
|
record.port_forward_ids.unlink()
|
|
|
|
# Appel de la méthode unlink standard
|
|
return super(UdmConfiguration, self).unlink()
|