diff --git a/cell_config.json b/cell_config.json deleted file mode 100644 index 98ba92e..0000000 --- a/cell_config.json +++ /dev/null @@ -1,36 +0,0 @@ -{ - "cell_name": "modified", - "domain": "cell.local", - "ip_range": "10.0.0.0/24", - "network": { - "dns_port": 53, - "dhcp_range": "10.0.0.100-10.0.0.200", - "ntp_servers": ["pool.ntp.org"] - }, - "wireguard": { - "port": 51820, - "private_key": "test_key", - "address": "10.0.0.1/24" - }, - "email": { - "domain": "cell.local", - "smtp_port": 25, - "imap_port": 143 - }, - "calendar": { - "port": 5232, - "data_dir": "/app/data/calendar" - }, - "files": { - "port": 8080, - "data_dir": "/app/data/files" - }, - "routing": { - "nat_enabled": true, - "firewall_enabled": true - }, - "vault": { - "ca_configured": true, - "fernet_configured": true - } -} \ No newline at end of file diff --git a/docs/NETWORK_CONFIGURATION.md b/docs/NETWORK_CONFIGURATION.md deleted file mode 100644 index 741f1db..0000000 --- a/docs/NETWORK_CONFIGURATION.md +++ /dev/null @@ -1,389 +0,0 @@ -# Personal Internet Cell - Network Configuration Guide - -This guide explains how to configure networking for the Personal Internet Cell to provide internet access to WireGuard VPN clients. - -## Table of Contents - -1. [Overview](#overview) -2. [Network Architecture](#network-architecture) -3. [Quick Setup](#quick-setup) -4. [Detailed Configuration](#detailed-configuration) -5. [Troubleshooting](#troubleshooting) -6. [Advanced Configuration](#advanced-configuration) -7. [Security Considerations](#security-considerations) - -## Overview - -The Personal Internet Cell provides a complete VPN solution with internet access. This requires proper configuration of: - -- **IP Forwarding**: Allow traffic to pass through the server -- **NAT (Network Address Translation)**: Translate private IPs to public IPs -- **Routing**: Direct traffic from VPN clients to the internet -- **Firewall Rules**: Control traffic flow and security - -## Network Architecture - -``` -Internet - │ - ▼ -[Host Server] (195.178.106.244) - │ - ├── [Docker Network] (172.20.0.0/16) - │ └── [WireGuard Container] (cell-wireguard) - │ └── [WireGuard Interface] (wg0: 10.0.0.1/24) - │ - └── [VPN Clients] (10.0.0.2-10.0.0.254/24) - └── [Internet Access via NAT] -``` - -### Key Components - -- **Host Interface**: `eth0` (or main network interface) -- **WireGuard Interface**: `wg0` (10.0.0.1/24) -- **Client Network**: `10.0.0.0/24` -- **NAT Translation**: Client IPs → Host IP - -## Quick Setup - -### 1. Run the Network Configuration Script - -```bash -# Make the script executable (if not already done) -chmod +x /opt/pic/scripts/setup-network.sh - -# Run the configuration -sudo /opt/pic/scripts/setup-network.sh setup -``` - -### 2. Verify Configuration - -```bash -# Check status -sudo /opt/pic/scripts/setup-network.sh status - -# Test configuration -sudo /opt/pic/scripts/setup-network.sh test -``` - -### 3. Connect a VPN Client - -Use the generated WireGuard configuration to connect a client. The client should now have internet access. - -## Detailed Configuration - -### IP Forwarding - -IP forwarding allows the server to route packets between different network interfaces. - -**Enable on Host:** -```bash -echo 'net.ipv4.ip_forward = 1' >> /etc/sysctl.conf -sysctl -p -``` - -**Enable in Container:** -```bash -docker exec cell-wireguard sh -c "echo 1 > /proc/sys/net/ipv4/ip_forward" -``` - -### NAT Configuration - -NAT (Network Address Translation) allows VPN clients to access the internet using the server's public IP. - -**Container NAT Rules:** -```bash -# Allow forwarding for WireGuard traffic -iptables -A FORWARD -i wg0 -j ACCEPT -iptables -A FORWARD -o wg0 -j ACCEPT - -# NAT rule for internet access -iptables -t nat -A POSTROUTING -s 10.0.0.0/24 -o eth0 -j MASQUERADE -``` - -**Host NAT Rules:** -```bash -# Allow traffic from WireGuard network -iptables -t nat -A POSTROUTING -s 10.0.0.0/24 -o eth0 -j MASQUERADE -iptables -A FORWARD -i wg0 -j ACCEPT -iptables -A FORWARD -o wg0 -j ACCEPT -``` - -### Routing Configuration - -**WireGuard Interface Setup:** -```bash -# Create WireGuard interface -ip link add dev wg0 type wireguard - -# Set private key -wg set wg0 private-key /path/to/private-key - -# Set listen port -wg set wg0 listen-port 51820 - -# Add IP address -ip addr add 10.0.0.1/24 dev wg0 - -# Bring interface up -ip link set wg0 up - -# Add peers -wg set wg0 peer allowed-ips 10.0.0.2/32 -``` - -## Troubleshooting - -### Common Issues - -#### 1. VPN Connected but No Internet - -**Symptoms:** -- WireGuard shows connected -- Can ping server (10.0.0.1) -- Cannot access internet - -**Solutions:** -```bash -# Check IP forwarding -cat /proc/sys/net/ipv4/ip_forward -# Should return 1 - -# Check NAT rules -iptables -t nat -L POSTROUTING -n -# Should show MASQUERADE rule for 10.0.0.0/24 - -# Check forwarding rules -iptables -L FORWARD -n -# Should show ACCEPT rules for wg0 - -# Restart network configuration -sudo /opt/pic/scripts/setup-network.sh reset -sudo /opt/pic/scripts/setup-network.sh setup -``` - -#### 2. Cannot Connect to VPN - -**Symptoms:** -- WireGuard client cannot connect -- No handshake in server logs - -**Solutions:** -```bash -# Check WireGuard interface -docker exec cell-wireguard wg show - -# Check if port 51820 is open -netstat -ulnp | grep 51820 - -# Check firewall rules -ufw status -iptables -L INPUT -n - -# Check Docker port mapping -docker port cell-wireguard -``` - -#### 3. DNS Issues - -**Symptoms:** -- Can ping IP addresses -- Cannot resolve domain names - -**Solutions:** -```bash -# Check DNS configuration in client config -# Should include: DNS = 8.8.8.8, 1.1.1.1 - -# Test DNS from container -docker exec cell-wireguard nslookup google.com - -# Check if DNS is being blocked -docker exec cell-wireguard iptables -L -n | grep 53 -``` - -### Diagnostic Commands - -```bash -# Check network status -sudo /opt/pic/scripts/setup-network.sh status - -# Test connectivity from container -docker exec cell-wireguard ping -c 3 8.8.8.8 - -# Check routing table -docker exec cell-wireguard ip route show - -# Check interface status -docker exec cell-wireguard ip addr show wg0 - -# Check NAT rules -docker exec cell-wireguard iptables -t nat -L -n - -# Check forwarding rules -docker exec cell-wireguard iptables -L FORWARD -n -``` - -## Advanced Configuration - -### Custom DNS Servers - -To use custom DNS servers, modify the WireGuard client configuration: - -```ini -[Interface] -PrivateKey = -Address = 10.0.0.2/32 -DNS = 1.1.1.1, 1.0.0.1, 8.8.8.8, 8.8.4.4 - -[Peer] -PublicKey = -Endpoint = 195.178.106.244:51820 -AllowedIPs = 0.0.0.0/0 -PersistentKeepalive = 25 -``` - -### Split Tunneling - -To allow only specific traffic through the VPN: - -```ini -[Peer] -AllowedIPs = 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16 -# Only route private networks through VPN -``` - -### Port Forwarding - -To forward specific ports to VPN clients: - -```bash -# Forward port 8080 to client 10.0.0.2 -iptables -t nat -A PREROUTING -p tcp --dport 8080 -j DNAT --to-destination 10.0.0.2:8080 -iptables -A FORWARD -p tcp -d 10.0.0.2 --dport 8080 -j ACCEPT -``` - -### Bandwidth Limiting - -To limit bandwidth for VPN clients: - -```bash -# Install tc (traffic control) -apt-get install iproute2 - -# Limit client 10.0.0.2 to 1Mbps -tc qdisc add dev wg0 root handle 1: htb default 30 -tc class add dev wg0 parent 1: classid 1:1 htb rate 1mbit -tc class add dev wg0 parent 1:1 classid 1:10 htb rate 1mbit ceil 1mbit -tc filter add dev wg0 protocol ip parent 1:0 prio 1 u32 match ip dst 10.0.0.2 flowid 1:10 -``` - -## Security Considerations - -### Firewall Rules - -**Basic Security Rules:** -```bash -# Drop invalid packets -iptables -A INPUT -m conntrack --ctstate INVALID -j DROP - -# Allow established connections -iptables -A INPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT - -# Allow WireGuard traffic -iptables -A INPUT -p udp --dport 51820 -j ACCEPT - -# Allow SSH (if needed) -iptables -A INPUT -p tcp --dport 22 -j ACCEPT - -# Drop everything else -iptables -A INPUT -j DROP -``` - -### Client Isolation - -To prevent clients from communicating with each other: - -```bash -# Block inter-client communication -iptables -A FORWARD -i wg0 -o wg0 -j DROP -``` - -### Logging - -To log VPN traffic: - -```bash -# Log all WireGuard traffic -iptables -A FORWARD -i wg0 -j LOG --log-prefix "WG-FORWARD: " -iptables -A FORWARD -o wg0 -j LOG --log-prefix "WG-FORWARD: " - -# Log NAT traffic -iptables -t nat -A POSTROUTING -s 10.0.0.0/24 -j LOG --log-prefix "WG-NAT: " -``` - -## Monitoring - -### Real-time Monitoring - -```bash -# Monitor WireGuard connections -watch -n 1 "docker exec cell-wireguard wg show" - -# Monitor traffic -watch -n 1 "docker exec cell-wireguard wg show wg0 transfer" - -# Monitor NAT rules -watch -n 1 "iptables -t nat -L POSTROUTING -n -v" -``` - -### Log Analysis - -```bash -# Check system logs -journalctl -u pic-network.service -f - -# Check iptables logs -tail -f /var/log/kern.log | grep WG- - -# Check Docker logs -docker logs cell-wireguard -f -``` - -## Backup and Recovery - -### Backup Configuration - -```bash -# Backup iptables rules -iptables-save > /opt/pic/backups/iptables-backup-$(date +%Y%m%d).rules - -# Backup WireGuard configuration -cp /opt/pic/config/wireguard/wg_confs/wg0.conf /opt/pic/backups/wg0-backup-$(date +%Y%m%d).conf - -# Backup network script -cp /opt/pic/scripts/setup-network.sh /opt/pic/backups/setup-network-backup-$(date +%Y%m%d).sh -``` - -### Restore Configuration - -```bash -# Restore iptables rules -iptables-restore < /opt/pic/backups/iptables-backup-YYYYMMDD.rules - -# Restore WireGuard configuration -cp /opt/pic/backups/wg0-backup-YYYYMMDD.conf /opt/pic/config/wireguard/wg_confs/wg0.conf -docker restart cell-wireguard -``` - -## Support - -If you encounter issues: - -1. Check the troubleshooting section above -2. Run the diagnostic commands -3. Check the logs for error messages -4. Verify your network configuration -5. Test with a simple client configuration - -For additional help, check the main Personal Internet Cell documentation or create an issue in the project repository. diff --git a/fix_imports.py b/fix_imports.py deleted file mode 100644 index 77055c6..0000000 --- a/fix_imports.py +++ /dev/null @@ -1,51 +0,0 @@ -#!/usr/bin/env python3 -""" -Script to fix import statements in test files -""" - -import os -import re -from pathlib import Path - -def fix_imports_in_file(file_path): - """Fix import statements in a test file""" - with open(file_path, 'r', encoding='utf-8') as f: - content = f.read() - - # Fix relative imports to absolute imports from api package - content = re.sub(r'from \.(\w+) import', r'from \1 import', content) - content = re.sub(r'import \.(\w+)', r'import \1', content) - - # Add path setup if not present - if 'sys.path.insert' not in content and 'api_dir' not in content: - path_setup = '''import sys -from pathlib import Path - -# Add api directory to path -api_dir = Path(__file__).parent.parent / 'api' -sys.path.insert(0, str(api_dir)) - -''' - # Insert after the first import line - lines = content.split('\n') - for i, line in enumerate(lines): - if line.startswith('import ') or line.startswith('from '): - lines.insert(i, path_setup.rstrip()) - break - content = '\n'.join(lines) - - with open(file_path, 'w', encoding='utf-8') as f: - f.write(content) - - print(f"Fixed imports in {file_path}") - -def main(): - """Fix all test files""" - tests_dir = Path('tests') - - for test_file in tests_dir.glob('test_*.py'): - if test_file.name not in ['test_cli_tool.py', 'test_peer_registry.py']: # Already fixed - fix_imports_in_file(test_file) - -if __name__ == '__main__': - main() \ No newline at end of file diff --git a/fix_test_imports.py b/fix_test_imports.py deleted file mode 100644 index fa47b4b..0000000 --- a/fix_test_imports.py +++ /dev/null @@ -1,31 +0,0 @@ -#!/usr/bin/env python3 -""" -Fix import statements in test files -""" - -import os -import re -from pathlib import Path - -def fix_imports_in_file(file_path): - """Fix import statements in a test file""" - with open(file_path, 'r', encoding='utf-8') as f: - content = f.read() - - # Replace 'from api.' with 'from .' - content = re.sub(r'from api\.', 'from .', content) - content = re.sub(r'import api\.', 'import .', content) - - with open(file_path, 'w', encoding='utf-8') as f: - f.write(content) - - print(f"Fixed imports in {file_path}") - -def main(): - tests_dir = Path('tests') - - for test_file in tests_dir.glob('test_*.py'): - fix_imports_in_file(test_file) - -if __name__ == '__main__': - main() \ No newline at end of file diff --git a/test_app_endpoints.py b/test_app_endpoints.py deleted file mode 100644 index 0955763..0000000 --- a/test_app_endpoints.py +++ /dev/null @@ -1,559 +0,0 @@ -#!/usr/bin/env python3 -""" -Comprehensive tests for Flask app endpoints -""" - -import unittest -import sys -import os -import tempfile -import shutil -import json -from pathlib import Path -from unittest.mock import patch, MagicMock - -# Add api directory to path -api_dir = Path(__file__).parent / 'api' -sys.path.insert(0, str(api_dir)) - -class TestFlaskAppEndpoints(unittest.TestCase): - def setUp(self): - """Set up test environment""" - # Create temporary directories - self.test_dir = tempfile.mkdtemp() - self.data_dir = os.path.join(self.test_dir, 'data') - self.config_dir = os.path.join(self.test_dir, 'config') - os.makedirs(self.data_dir, exist_ok=True) - os.makedirs(self.config_dir, exist_ok=True) - - # Set environment variables - os.environ['TESTING'] = 'true' - os.environ['LOG_LEVEL'] = 'ERROR' - - # Import and create app - from app import app - self.app = app - self.client = app.test_client() - - # Mock external dependencies - self.patchers = [] - - # Mock subprocess.run - subprocess_patcher = patch('subprocess.run') - self.mock_subprocess = subprocess_patcher.start() - self.mock_subprocess.return_value.returncode = 0 - self.mock_subprocess.return_value.stdout = b"test output" - self.patchers.append(subprocess_patcher) - - # Mock docker - docker_patcher = patch('docker.from_env') - self.mock_docker = docker_patcher.start() - self.mock_docker_client = MagicMock() - self.mock_docker.return_value = self.mock_docker_client - self.patchers.append(docker_patcher) - - # Mock file operations - file_patcher = patch('builtins.open', create=True) - self.mock_file = file_patcher.start() - self.mock_file.return_value.__enter__.return_value.read.return_value = '{}' - self.patchers.append(file_patcher) - - def tearDown(self): - """Clean up test environment""" - shutil.rmtree(self.test_dir) - for patcher in self.patchers: - patcher.stop() - - def test_health_endpoint(self): - """Test /health endpoint""" - response = self.client.get('/health') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_status_endpoint(self): - """Test /api/status endpoint""" - response = self.client.get('/api/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_config_get_endpoint(self): - """Test GET /api/config endpoint""" - response = self.client.get('/api/config') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, dict) - - def test_api_config_put_endpoint(self): - """Test PUT /api/config endpoint""" - test_config = {'test': 'value'} - response = self.client.put('/api/config', - data=json.dumps(test_config), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_config_backup_endpoint(self): - """Test POST /api/config/backup endpoint""" - response = self.client.post('/api/config/backup') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('backup_id', data) - - def test_api_config_backups_endpoint(self): - """Test GET /api/config/backups endpoint""" - response = self.client.get('/api/config/backups') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - def test_api_config_restore_endpoint(self): - """Test POST /api/config/restore/ endpoint""" - response = self.client.post('/api/config/restore/test_backup') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_config_export_endpoint(self): - """Test GET /api/config/export endpoint""" - response = self.client.get('/api/config/export') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, dict) - - def test_api_config_import_endpoint(self): - """Test POST /api/config/import endpoint""" - test_config = {'test': 'value'} - response = self.client.post('/api/config/import', - data=json.dumps(test_config), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_services_bus_status_endpoint(self): - """Test GET /api/services/bus/status endpoint""" - response = self.client.get('/api/services/bus/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('services', data) - - def test_api_services_bus_events_endpoint(self): - """Test GET /api/services/bus/events endpoint""" - response = self.client.get('/api/services/bus/events') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - def test_api_services_bus_start_endpoint(self): - """Test POST /api/services/bus/services//start endpoint""" - response = self.client.post('/api/services/bus/services/test/start') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_services_bus_stop_endpoint(self): - """Test POST /api/services/bus/services//stop endpoint""" - response = self.client.post('/api/services/bus/services/test/stop') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_services_bus_restart_endpoint(self): - """Test POST /api/services/bus/services//restart endpoint""" - response = self.client.post('/api/services/bus/services/test/restart') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_logs_services_endpoint(self): - """Test GET /api/logs/services/ endpoint""" - response = self.client.get('/api/logs/services/test') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - def test_api_logs_search_endpoint(self): - """Test POST /api/logs/search endpoint""" - search_data = {'query': 'test', 'level': 'INFO'} - response = self.client.post('/api/logs/search', - data=json.dumps(search_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - def test_api_logs_export_endpoint(self): - """Test POST /api/logs/export endpoint""" - export_data = {'format': 'json', 'filters': {}} - response = self.client.post('/api/logs/export', - data=json.dumps(export_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('export_path', data) - - def test_api_logs_statistics_endpoint(self): - """Test GET /api/logs/statistics endpoint""" - response = self.client.get('/api/logs/statistics') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('total_entries', data) - - def test_api_logs_rotate_endpoint(self): - """Test POST /api/logs/rotate endpoint""" - response = self.client.post('/api/logs/rotate') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_dns_records_endpoints(self): - """Test DNS records endpoints""" - # GET - response = self.client.get('/api/dns/records') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST - record_data = {'name': 'test.example.com', 'type': 'A', 'value': '192.168.1.1'} - response = self.client.post('/api/dns/records', - data=json.dumps(record_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # DELETE - response = self.client.delete('/api/dns/records', - data=json.dumps(record_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_dhcp_endpoints(self): - """Test DHCP endpoints""" - # GET leases - response = self.client.get('/api/dhcp/leases') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST reservation - reservation_data = {'mac': '00:11:22:33:44:55', 'ip': '192.168.1.100'} - response = self.client.post('/api/dhcp/reservations', - data=json.dumps(reservation_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # DELETE reservation - response = self.client.delete('/api/dhcp/reservations', - data=json.dumps(reservation_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_ntp_status_endpoint(self): - """Test GET /api/ntp/status endpoint""" - response = self.client.get('/api/ntp/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_network_info_endpoint(self): - """Test GET /api/network/info endpoint""" - response = self.client.get('/api/network/info') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('interfaces', data) - - def test_api_dns_status_endpoint(self): - """Test GET /api/dns/status endpoint""" - response = self.client.get('/api/dns/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_network_test_endpoint(self): - """Test POST /api/network/test endpoint""" - test_data = {'target': '8.8.8.8', 'type': 'ping'} - response = self.client.post('/api/network/test', - data=json.dumps(test_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_wireguard_endpoints(self): - """Test WireGuard endpoints""" - # GET keys - response = self.client.get('/api/wireguard/keys') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('public_key', data) - - # POST generate peer keys - response = self.client.post('/api/wireguard/keys/peer') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('public_key', data) - - # GET config - response = self.client.get('/api/wireguard/config') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('config', data) - - # GET peers - response = self.client.get('/api/wireguard/peers') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST add peer - peer_data = {'peer': 'test_peer', 'ip': '10.0.0.1', 'public_key': 'test_key'} - response = self.client.post('/api/wireguard/peers', - data=json.dumps(peer_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # DELETE remove peer - response = self.client.delete('/api/wireguard/peers', - data=json.dumps({'peer': 'test_peer'}), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # GET status - response = self.client.get('/api/wireguard/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_peers_endpoints(self): - """Test peers endpoints""" - # GET peers - response = self.client.get('/api/peers') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST add peer - peer_data = {'peer': 'test_peer', 'ip': '10.0.0.1'} - response = self.client.post('/api/peers', - data=json.dumps(peer_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # DELETE remove peer - response = self.client.delete('/api/peers/test_peer') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - def test_api_email_endpoints(self): - """Test email endpoints""" - # GET users - response = self.client.get('/api/email/users') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST create user - user_data = {'username': 'test_user', 'email': 'test@example.com'} - response = self.client.post('/api/email/users', - data=json.dumps(user_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # DELETE user - response = self.client.delete('/api/email/users/test_user') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # GET status - response = self.client.get('/api/email/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_calendar_endpoints(self): - """Test calendar endpoints""" - # GET users - response = self.client.get('/api/calendar/users') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST create user - user_data = {'username': 'test_user', 'email': 'test@example.com'} - response = self.client.post('/api/calendar/users', - data=json.dumps(user_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # DELETE user - response = self.client.delete('/api/calendar/users/test_user') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # GET status - response = self.client.get('/api/calendar/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_files_endpoints(self): - """Test files endpoints""" - # GET users - response = self.client.get('/api/files/users') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST create user - user_data = {'username': 'test_user'} - response = self.client.post('/api/files/users', - data=json.dumps(user_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # DELETE user - response = self.client.delete('/api/files/users/test_user') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # GET status - response = self.client.get('/api/files/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - def test_api_routing_endpoints(self): - """Test routing endpoints""" - # GET status - response = self.client.get('/api/routing/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - # POST NAT rule - nat_data = {'type': 'masquerade', 'interface': 'eth0'} - response = self.client.post('/api/routing/nat', - data=json.dumps(nat_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('rule_id', data) - - # GET NAT rules - response = self.client.get('/api/routing/nat') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - def test_api_vault_endpoints(self): - """Test vault endpoints""" - # GET status - response = self.client.get('/api/vault/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('status', data) - - # GET certificates - response = self.client.get('/api/vault/certificates') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST generate certificate - cert_data = {'common_name': 'test.example.com'} - response = self.client.post('/api/vault/certificates', - data=json.dumps(cert_data), - content_type='application/json') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('certificate', data) - - # GET CA certificate - response = self.client.get('/api/vault/ca/certificate') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('certificate', data) - - def test_api_containers_endpoints(self): - """Test containers endpoints""" - # GET containers - response = self.client.get('/api/containers') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - # POST start container - response = self.client.post('/api/containers/test/start') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # POST stop container - response = self.client.post('/api/containers/test/stop') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('success', data) - - # GET container logs - response = self.client.get('/api/containers/test/logs') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - def test_api_services_status_endpoint(self): - """Test GET /api/services/status endpoint""" - response = self.client.get('/api/services/status') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('services', data) - - def test_api_services_connectivity_endpoint(self): - """Test GET /api/services/connectivity endpoint""" - response = self.client.get('/api/services/connectivity') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIn('results', data) - - def test_api_health_history_endpoint(self): - """Test GET /api/health/history endpoint""" - response = self.client.get('/api/health/history') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - - def test_api_logs_endpoint(self): - """Test GET /api/logs endpoint""" - response = self.client.get('/api/logs') - self.assertEqual(response.status_code, 200) - data = json.loads(response.data) - self.assertIsInstance(data, list) - -if __name__ == '__main__': - unittest.main() \ No newline at end of file diff --git a/tests/test_firewall_manager.py b/tests/test_firewall_manager.py index 99aff5d..5acf637 100644 --- a/tests/test_firewall_manager.py +++ b/tests/test_firewall_manager.py @@ -133,7 +133,8 @@ class TestGenerateCorefile(unittest.TestCase): self.assertIn('reload', content) def test_returns_false_on_write_error(self): - result = firewall_manager.generate_corefile([], '/nonexistent/path/Corefile') + with unittest.mock.patch('builtins.open', side_effect=OSError('Permission denied')): + result = firewall_manager.generate_corefile([], '/any/path/Corefile') self.assertFalse(result) diff --git a/tests/test_ip_utils.py b/tests/test_ip_utils.py index 5a5aa7f..cfcd907 100644 --- a/tests/test_ip_utils.py +++ b/tests/test_ip_utils.py @@ -5,6 +5,7 @@ import sys import os import tempfile import unittest +import unittest.mock from pathlib import Path api_dir = Path(__file__).parent.parent / 'api' @@ -98,7 +99,8 @@ class TestWriteEnvFile(unittest.TestCase): self.assertTrue(result) def test_returns_false_on_unwritable_path(self): - result = ip_utils.write_env_file('172.20.0.0/16', '/nonexistent/deep/path/.env') + with unittest.mock.patch('builtins.open', side_effect=OSError('Permission denied')): + result = ip_utils.write_env_file('172.20.0.0/16', '/any/path/.env') self.assertFalse(result) def test_contains_cell_network(self):