Constantin
561 lines
22 KiB
Python
561 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Network Manager for Personal Internet Cell
|
|
Handles DNS, DHCP, and NTP functionality
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import subprocess
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Dict, List, Optional, Tuple, Any
|
|
from base_service_manager import BaseServiceManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class NetworkManager(BaseServiceManager):
|
|
"""Manages network services (DNS, DHCP, NTP)"""
|
|
|
|
def __init__(self, data_dir: str = '/app/data', config_dir: str = '/app/config'):
|
|
super().__init__('network', data_dir, config_dir)
|
|
self.dns_zones_dir = os.path.join(data_dir, 'dns')
|
|
self.dhcp_leases_file = os.path.join(data_dir, 'dhcp', 'leases')
|
|
|
|
# Ensure directories exist
|
|
os.makedirs(self.dns_zones_dir, exist_ok=True)
|
|
os.makedirs(os.path.dirname(self.dhcp_leases_file), exist_ok=True)
|
|
|
|
def update_dns_zone(self, zone_name: str, records: List[Dict]) -> bool:
|
|
"""Update DNS zone file with new records"""
|
|
try:
|
|
zone_file = os.path.join(self.dns_zones_dir, f'{zone_name}.zone')
|
|
|
|
# Create zone file content
|
|
content = self._generate_zone_content(zone_name, records)
|
|
|
|
with open(zone_file, 'w') as f:
|
|
f.write(content)
|
|
|
|
# Reload DNS service
|
|
self._reload_dns_service()
|
|
|
|
logger.info(f"Updated DNS zone {zone_name} with {len(records)} records")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update DNS zone {zone_name}: {e}")
|
|
return False
|
|
|
|
def _generate_zone_content(self, zone_name: str, records: List[Dict]) -> str:
|
|
"""Generate DNS zone file content"""
|
|
timestamp = datetime.now().strftime('%Y%m%d%H')
|
|
|
|
content = f"""$TTL 3600
|
|
@ IN SOA {zone_name}. admin.{zone_name}. (
|
|
{timestamp} ; Serial
|
|
3600 ; Refresh
|
|
1800 ; Retry
|
|
1209600 ; Expire
|
|
3600 ; Minimum TTL
|
|
)
|
|
|
|
; Name servers
|
|
@ IN NS {zone_name}.
|
|
|
|
"""
|
|
|
|
# Add records
|
|
for record in records:
|
|
record_type = record.get('type', 'A')
|
|
name = record.get('name', '')
|
|
value = record.get('value', '')
|
|
ttl = record.get('ttl', '3600')
|
|
|
|
if name and value:
|
|
content += f"{name:<20} {ttl:<8} IN {record_type:<6} {value}\n"
|
|
|
|
return content
|
|
|
|
def add_dns_record(self, zone: str, name: str, record_type: str, value: str, ttl: int = 3600) -> bool:
|
|
"""Add a DNS record to a zone"""
|
|
try:
|
|
# Load existing records
|
|
records = self._load_dns_records(zone)
|
|
|
|
# Add new record
|
|
new_record = {
|
|
'name': name,
|
|
'type': record_type,
|
|
'value': value,
|
|
'ttl': ttl
|
|
}
|
|
|
|
# Remove existing record with same name and type
|
|
records = [r for r in records if not (r['name'] == name and r['type'] == record_type)]
|
|
records.append(new_record)
|
|
|
|
# Update zone
|
|
return self.update_dns_zone(zone, records)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to add DNS record: {e}")
|
|
return False
|
|
|
|
def remove_dns_record(self, zone: str, name: str, record_type: str = 'A') -> bool:
|
|
"""Remove a DNS record from a zone"""
|
|
try:
|
|
# Load existing records
|
|
records = self._load_dns_records(zone)
|
|
|
|
# Remove matching records
|
|
records = [r for r in records if not (r['name'] == name and r['type'] == record_type)]
|
|
|
|
# Update zone
|
|
return self.update_dns_zone(zone, records)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to remove DNS record: {e}")
|
|
return False
|
|
|
|
def _load_dns_records(self, zone: str) -> List[Dict]:
|
|
"""Load DNS records from zone file"""
|
|
zone_file = os.path.join(self.dns_zones_dir, f'{zone}.zone')
|
|
|
|
if not os.path.exists(zone_file):
|
|
return []
|
|
|
|
records = []
|
|
try:
|
|
with open(zone_file, 'r') as f:
|
|
lines = f.readlines()
|
|
|
|
for line in lines:
|
|
line = line.strip()
|
|
if line and not line.startswith(';') and not line.startswith('$'):
|
|
parts = line.split()
|
|
if len(parts) >= 5:
|
|
record_type = parts[3]
|
|
if record_type in ('A', 'CNAME'):
|
|
records.append({
|
|
'name': parts[0],
|
|
'ttl': parts[1],
|
|
'type': record_type,
|
|
'value': parts[4]
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Failed to load DNS records: {e}")
|
|
|
|
return records
|
|
|
|
def get_dhcp_leases(self) -> List[Dict]:
|
|
"""Get current DHCP leases"""
|
|
leases = []
|
|
|
|
try:
|
|
if os.path.exists(self.dhcp_leases_file):
|
|
with open(self.dhcp_leases_file, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line and not line.startswith('#'):
|
|
parts = line.split()
|
|
if len(parts) >= 4:
|
|
leases.append({
|
|
'mac': parts[1],
|
|
'ip': parts[2],
|
|
'hostname': parts[3] if len(parts) > 3 else '',
|
|
'timestamp': parts[0]
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Failed to load DHCP leases: {e}")
|
|
|
|
return leases
|
|
|
|
def add_dhcp_reservation(self, mac: str, ip: str, hostname: str = '') -> bool:
|
|
"""Add a DHCP reservation"""
|
|
try:
|
|
reservation_file = os.path.join(self.config_dir, 'dhcp', 'reservations.conf')
|
|
|
|
# Ensure directory exists
|
|
os.makedirs(os.path.dirname(reservation_file), exist_ok=True)
|
|
|
|
# Add reservation
|
|
with open(reservation_file, 'a') as f:
|
|
f.write(f"dhcp-host={mac},{ip},{hostname}\n")
|
|
|
|
# Reload DHCP service
|
|
self._reload_dhcp_service()
|
|
|
|
logger.info(f"Added DHCP reservation: {mac} -> {ip}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to add DHCP reservation: {e}")
|
|
return False
|
|
|
|
def remove_dhcp_reservation(self, mac: str) -> bool:
|
|
"""Remove a DHCP reservation"""
|
|
try:
|
|
reservation_file = os.path.join(self.config_dir, 'dhcp', 'reservations.conf')
|
|
|
|
if not os.path.exists(reservation_file):
|
|
return True
|
|
|
|
# Read existing reservations
|
|
with open(reservation_file, 'r') as f:
|
|
lines = f.readlines()
|
|
|
|
# Remove matching reservation
|
|
lines = [line for line in lines if not line.startswith(f"dhcp-host={mac},")]
|
|
|
|
# Write back
|
|
with open(reservation_file, 'w') as f:
|
|
f.writelines(lines)
|
|
|
|
# Reload DHCP service
|
|
self._reload_dhcp_service()
|
|
|
|
logger.info(f"Removed DHCP reservation: {mac}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to remove DHCP reservation: {e}")
|
|
return False
|
|
|
|
def get_ntp_status(self) -> Dict:
|
|
"""Get NTP service status"""
|
|
try:
|
|
# Check if NTP service is running
|
|
result = subprocess.run(['docker', 'ps', '--filter', 'name=cell-ntp', '--format', '{{.Names}}'],
|
|
capture_output=True, text=True)
|
|
|
|
is_running = len(result.stdout.strip()) > 0
|
|
|
|
# Get NTP statistics if running
|
|
stats = {}
|
|
if is_running:
|
|
try:
|
|
result = subprocess.run(['docker', 'exec', 'cell-ntp', 'chronyc', 'tracking'],
|
|
capture_output=True, text=True)
|
|
if result.returncode == 0:
|
|
stats['tracking'] = result.stdout
|
|
|
|
result = subprocess.run(['docker', 'exec', 'cell-ntp', 'chronyc', 'sources'],
|
|
capture_output=True, text=True)
|
|
if result.returncode == 0:
|
|
stats['sources'] = result.stdout
|
|
except Exception as e:
|
|
logger.error(f"Failed to get NTP stats: {e}")
|
|
|
|
return {
|
|
'running': is_running,
|
|
'stats': stats
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get NTP status: {e}")
|
|
return {'running': False, 'stats': {}}
|
|
|
|
def _reload_dns_service(self):
|
|
"""Reload DNS service"""
|
|
try:
|
|
subprocess.run(['docker', 'exec', 'cell-dns', 'kill', '-HUP', '1'],
|
|
capture_output=True, timeout=10)
|
|
except Exception as e:
|
|
logger.error(f"Failed to reload DNS service: {e}")
|
|
|
|
def _reload_dhcp_service(self):
|
|
"""Reload DHCP service"""
|
|
try:
|
|
subprocess.run(['docker', 'exec', 'cell-dhcp', 'kill', '-HUP', '1'],
|
|
capture_output=True, timeout=10)
|
|
except Exception as e:
|
|
logger.error(f"Failed to reload DHCP service: {e}")
|
|
|
|
def test_dns_resolution(self, domain: str) -> Dict:
|
|
"""Test DNS resolution for a domain"""
|
|
try:
|
|
result = subprocess.run(['nslookup', domain, '127.0.0.1'],
|
|
capture_output=True, text=True, timeout=10)
|
|
|
|
return {
|
|
'success': result.returncode == 0,
|
|
'output': result.stdout,
|
|
'error': result.stderr
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'output': '',
|
|
'error': str(e)
|
|
}
|
|
|
|
def test_dhcp_functionality(self) -> Dict:
|
|
"""Test DHCP functionality"""
|
|
try:
|
|
# Check if DHCP service is running
|
|
result = subprocess.run(['docker', 'ps', '--filter', 'name=cell-dhcp', '--format', '{{.Names}}'],
|
|
capture_output=True, text=True)
|
|
|
|
is_running = len(result.stdout.strip()) > 0
|
|
|
|
# Get DHCP leases
|
|
leases = self.get_dhcp_leases()
|
|
|
|
return {
|
|
'running': is_running,
|
|
'leases_count': len(leases),
|
|
'leases': leases
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to test DHCP functionality: {e}")
|
|
return {'running': False, 'leases_count': 0, 'leases': []}
|
|
|
|
def test_ntp_functionality(self) -> Dict:
|
|
"""Test NTP functionality"""
|
|
try:
|
|
# Check if NTP service is running
|
|
result = subprocess.run(['docker', 'ps', '--filter', 'name=cell-ntp', '--format', '{{.Names}}'],
|
|
capture_output=True, text=True)
|
|
|
|
is_running = len(result.stdout.strip()) > 0
|
|
|
|
# Test NTP query
|
|
ntp_test = {}
|
|
if is_running:
|
|
try:
|
|
result = subprocess.run(['docker', 'exec', 'cell-ntp', 'chronyc', 'tracking'],
|
|
capture_output=True, text=True, timeout=10)
|
|
ntp_test['tracking'] = result.returncode == 0
|
|
ntp_test['output'] = result.stdout
|
|
except Exception as e:
|
|
ntp_test['tracking'] = False
|
|
ntp_test['error'] = str(e)
|
|
|
|
return {
|
|
'running': is_running,
|
|
'ntp_test': ntp_test
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to test NTP functionality: {e}")
|
|
return {'running': False, 'ntp_test': {}}
|
|
|
|
def get_network_info(self) -> dict:
|
|
"""Return general network info: IP addresses, interfaces, gateway, DNS, etc."""
|
|
try:
|
|
info = {}
|
|
# Get network interfaces
|
|
result = subprocess.run(['ip', '-j', 'addr'], capture_output=True, text=True)
|
|
if result.returncode == 0:
|
|
import json as _json
|
|
info['interfaces'] = _json.loads(result.stdout)
|
|
else:
|
|
info['interfaces'] = []
|
|
# Get default gateway
|
|
result = subprocess.run(['ip', 'route', 'show', 'default'], capture_output=True, text=True)
|
|
if result.returncode == 0:
|
|
info['default_gateway'] = result.stdout.strip()
|
|
else:
|
|
info['default_gateway'] = ''
|
|
# Get DNS servers
|
|
resolv_conf = '/etc/resolv.conf'
|
|
dns_servers = []
|
|
try:
|
|
with open(resolv_conf, 'r') as f:
|
|
for line in f:
|
|
if line.startswith('nameserver'):
|
|
dns_servers.append(line.strip().split()[1])
|
|
except Exception:
|
|
pass
|
|
info['dns_servers'] = dns_servers
|
|
return info
|
|
except Exception as e:
|
|
logger.error(f"Failed to get network info: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def get_dns_status(self) -> dict:
|
|
"""Return DNS service status and summary info."""
|
|
try:
|
|
# Check if DNS service is running
|
|
result = subprocess.run(['docker', 'ps', '--filter', 'name=cell-dns', '--format', '{{.Names}}'], capture_output=True, text=True)
|
|
is_running = len(result.stdout.strip()) > 0
|
|
# Get DNS records count (for all zones)
|
|
records_count = 0
|
|
try:
|
|
for fname in os.listdir(self.dns_zones_dir):
|
|
if fname.endswith('.zone'):
|
|
with open(os.path.join(self.dns_zones_dir, fname), 'r') as f:
|
|
for line in f:
|
|
if line.strip() and not line.startswith(';') and not line.startswith('$'):
|
|
parts = line.split()
|
|
if len(parts) >= 5 and parts[3] in ('A', 'CNAME'):
|
|
records_count += 1
|
|
except Exception:
|
|
pass
|
|
return {'running': is_running, 'records_count': records_count}
|
|
except Exception as e:
|
|
logger.error(f"Failed to get DNS status: {e}")
|
|
return {'running': False, 'records_count': 0, 'error': str(e)}
|
|
|
|
def get_status(self) -> Dict[str, Any]:
|
|
"""Get network service status"""
|
|
try:
|
|
# Check if we're running in Docker environment
|
|
import os
|
|
is_docker = os.path.exists('/.dockerenv') or os.environ.get('DOCKER_CONTAINER') == 'true'
|
|
|
|
if is_docker:
|
|
# Check if network containers are actually running
|
|
dns_running = self._check_dns_container_status()
|
|
dhcp_running = self._check_dhcp_container_status()
|
|
ntp_running = self._check_ntp_container_status()
|
|
all_running = dns_running and dhcp_running and ntp_running
|
|
|
|
status = {
|
|
'dns_running': dns_running,
|
|
'dhcp_running': dhcp_running,
|
|
'ntp_running': ntp_running,
|
|
'running': all_running,
|
|
'status': 'online' if all_running else 'offline',
|
|
'network': {
|
|
'dns_running': dns_running,
|
|
'dhcp_running': dhcp_running,
|
|
'ntp_running': ntp_running,
|
|
'running': all_running,
|
|
'status': 'online' if all_running else 'offline'
|
|
},
|
|
'timestamp': datetime.utcnow().isoformat()
|
|
}
|
|
else:
|
|
# Check actual service status in production
|
|
dns_running = self._check_dns_status()
|
|
dhcp_running = self._check_dhcp_status()
|
|
ntp_running = self._check_ntp_status()
|
|
|
|
status = {
|
|
'dns_running': dns_running,
|
|
'dhcp_running': dhcp_running,
|
|
'ntp_running': ntp_running,
|
|
'running': dns_running and dhcp_running and ntp_running,
|
|
'status': 'online' if (dns_running and dhcp_running and ntp_running) else 'offline',
|
|
'network': {
|
|
'dns_running': dns_running,
|
|
'dhcp_running': dhcp_running,
|
|
'ntp_running': ntp_running,
|
|
'running': dns_running and dhcp_running and ntp_running,
|
|
'status': 'online' if (dns_running and dhcp_running and ntp_running) else 'offline'
|
|
},
|
|
'timestamp': datetime.utcnow().isoformat()
|
|
}
|
|
|
|
return status
|
|
except Exception as e:
|
|
return self.handle_error(e, "get_status")
|
|
|
|
def _check_dns_container_status(self) -> bool:
|
|
"""Check if DNS Docker container is running"""
|
|
try:
|
|
import docker
|
|
client = docker.from_env()
|
|
containers = client.containers.list(filters={'name': 'cell-dns'})
|
|
return len(containers) > 0
|
|
except Exception:
|
|
return False
|
|
|
|
def _check_dhcp_container_status(self) -> bool:
|
|
"""Check if DHCP Docker container is running"""
|
|
try:
|
|
import docker
|
|
client = docker.from_env()
|
|
containers = client.containers.list(filters={'name': 'cell-dhcp'})
|
|
return len(containers) > 0
|
|
except Exception:
|
|
return False
|
|
|
|
def _check_ntp_container_status(self) -> bool:
|
|
"""Check if NTP Docker container is running"""
|
|
try:
|
|
import docker
|
|
client = docker.from_env()
|
|
containers = client.containers.list(filters={'name': 'cell-ntp'})
|
|
return len(containers) > 0
|
|
except Exception:
|
|
return False
|
|
|
|
def test_connectivity(self) -> Dict[str, Any]:
|
|
"""Test network service connectivity"""
|
|
try:
|
|
dns_test = self.test_dns_resolution('google.com')
|
|
dhcp_test = self.test_dhcp_functionality()
|
|
ntp_test = self.test_ntp_functionality()
|
|
|
|
results = {
|
|
'dns_test': dns_test,
|
|
'dhcp_test': dhcp_test,
|
|
'ntp_test': ntp_test,
|
|
'timestamp': datetime.utcnow().isoformat()
|
|
}
|
|
|
|
# Determine overall success
|
|
success = all(
|
|
result.get('success', False)
|
|
for result in [dns_test, dhcp_test, ntp_test]
|
|
)
|
|
results['success'] = success
|
|
|
|
# Add network key for compatibility
|
|
results['network'] = {
|
|
'dns_test': dns_test,
|
|
'dhcp_test': dhcp_test,
|
|
'ntp_test': ntp_test,
|
|
'success': success
|
|
}
|
|
|
|
return results
|
|
except Exception as e:
|
|
return self.handle_error(e, "test_connectivity")
|
|
|
|
def _check_dns_status(self) -> bool:
|
|
"""Check if DNS service is running"""
|
|
try:
|
|
result = subprocess.run(['systemctl', 'is-active', 'coredns'],
|
|
capture_output=True, text=True, timeout=5)
|
|
return result.returncode == 0 and result.stdout.strip() == 'active'
|
|
except Exception:
|
|
# Fallback: check if port 53 is listening
|
|
try:
|
|
result = subprocess.run(['netstat', '-tuln'], capture_output=True, text=True)
|
|
return ':53 ' in result.stdout
|
|
except Exception:
|
|
return False
|
|
|
|
def _check_dhcp_status(self) -> bool:
|
|
"""Check if DHCP service is running"""
|
|
try:
|
|
result = subprocess.run(['systemctl', 'is-active', 'dnsmasq'],
|
|
capture_output=True, text=True, timeout=5)
|
|
return result.returncode == 0 and result.stdout.strip() == 'active'
|
|
except Exception:
|
|
# Fallback: check if port 67 is listening
|
|
try:
|
|
result = subprocess.run(['netstat', '-tuln'], capture_output=True, text=True)
|
|
return ':67 ' in result.stdout
|
|
except Exception:
|
|
return False
|
|
|
|
def _check_ntp_status(self) -> bool:
|
|
"""Check if NTP service is running"""
|
|
try:
|
|
result = subprocess.run(['systemctl', 'is-active', 'chronyd'],
|
|
capture_output=True, text=True, timeout=5)
|
|
return result.returncode == 0 and result.stdout.strip() == 'active'
|
|
except Exception:
|
|
# Fallback: check if port 123 is listening
|
|
try:
|
|
result = subprocess.run(['netstat', '-tuln'], capture_output=True, text=True)
|
|
return ':123 ' in result.stdout
|
|
except Exception:
|
|
return False |