Merge branch 'feature/security-fixes-and-qa' into 'main'
add security fixes, port hardening, and expanded QA coverage See merge request root/pic!9
This commit is contained in:
Administrator
+239
-358
@@ -1,358 +1,239 @@
|
||||
# Personal Internet Cell - Quick Start Guide
|
||||
|
||||
## 🚀 Getting Started
|
||||
|
||||
This guide will help you get your Personal Internet Cell up and running with the new production-grade architecture in minutes.
|
||||
|
||||
### Prerequisites
|
||||
|
||||
- **Docker and Docker Compose** installed
|
||||
- **Python 3.10+** (for CLI and development)
|
||||
- **Ports available**: 53, 80, 443, 3000, 51820
|
||||
- **Administrative access** (for WireGuard and network services)
|
||||
- **2GB+ RAM, 10GB+ disk space**
|
||||
|
||||
### Step 1: Initial Setup
|
||||
|
||||
```bash
|
||||
# Clone or download the project
|
||||
git clone https://github.com/yourusername/PersonalInternetCell.git
|
||||
cd PersonalInternetCell
|
||||
|
||||
# Start all services with Docker (Recommended)
|
||||
docker-compose up --build
|
||||
|
||||
# Or run locally
|
||||
pip install -r api/requirements.txt
|
||||
python api/app.py
|
||||
```
|
||||
|
||||
### Step 2: Verify Installation
|
||||
|
||||
```bash
|
||||
# Check if API is responding
|
||||
curl http://localhost:3000/health
|
||||
|
||||
# Check service status
|
||||
curl http://localhost:3000/api/services/status
|
||||
|
||||
# Use the enhanced CLI
|
||||
python api/enhanced_cli.py --status
|
||||
```
|
||||
|
||||
### Step 3: Explore Services
|
||||
|
||||
```bash
|
||||
# Show all services
|
||||
python api/enhanced_cli.py --services
|
||||
|
||||
# Check health data
|
||||
python api/enhanced_cli.py --health
|
||||
|
||||
# Interactive mode
|
||||
python api/enhanced_cli.py --interactive
|
||||
```
|
||||
|
||||
## 📋 Enhanced CLI Commands
|
||||
|
||||
### Basic Management
|
||||
```bash
|
||||
# Service status
|
||||
python api/enhanced_cli.py --status
|
||||
python api/enhanced_cli.py --services
|
||||
|
||||
# Health monitoring
|
||||
python api/enhanced_cli.py --health
|
||||
|
||||
# Service logs
|
||||
python api/enhanced_cli.py --logs network
|
||||
python api/enhanced_cli.py --logs wireguard
|
||||
```
|
||||
|
||||
### Configuration Management
|
||||
```bash
|
||||
# Export configuration
|
||||
python api/enhanced_cli.py --export-config json
|
||||
python api/enhanced_cli.py --export-config yaml
|
||||
|
||||
# Import configuration
|
||||
python api/enhanced_cli.py --import-config config.json
|
||||
|
||||
# Configuration wizard
|
||||
python api/enhanced_cli.py --wizard network
|
||||
python api/enhanced_cli.py --wizard email
|
||||
```
|
||||
|
||||
### Batch Operations
|
||||
```bash
|
||||
# Execute multiple commands
|
||||
python api/enhanced_cli.py --batch "status" "services" "health"
|
||||
|
||||
# Interactive mode with tab completion
|
||||
python api/enhanced_cli.py --interactive
|
||||
```
|
||||
|
||||
## 🌐 Accessing Services
|
||||
|
||||
Once running, you can access:
|
||||
|
||||
- **API Server**: http://localhost:3000
|
||||
- **API Health**: http://localhost:3000/health
|
||||
- **Service Status**: http://localhost:3000/api/services/status
|
||||
- **Configuration**: http://localhost:3000/api/config
|
||||
- **Service Bus**: http://localhost:3000/api/services/bus/status
|
||||
- **Logs**: http://localhost:3000/api/logs/services/network
|
||||
|
||||
## 🔧 Configuration
|
||||
|
||||
### Cell Configuration
|
||||
|
||||
The cell uses a centralized configuration system with schema validation:
|
||||
|
||||
```bash
|
||||
# View current configuration
|
||||
curl http://localhost:3000/api/config
|
||||
|
||||
# Update configuration
|
||||
curl -X PUT http://localhost:3000/api/config \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{
|
||||
"cell_name": "mycell",
|
||||
"domain": "mycell.cell",
|
||||
"ip_range": "10.0.0.0/24",
|
||||
"wireguard_port": 51820
|
||||
}'
|
||||
```
|
||||
|
||||
### Service Configuration
|
||||
|
||||
Each service has its own configuration schema:
|
||||
|
||||
```bash
|
||||
# Network configuration
|
||||
python api/enhanced_cli.py --wizard network
|
||||
|
||||
# Email configuration
|
||||
python api/enhanced_cli.py --wizard email
|
||||
|
||||
# WireGuard configuration
|
||||
python api/enhanced_cli.py --wizard wireguard
|
||||
```
|
||||
|
||||
### Network Configuration
|
||||
The cell uses the following network ranges:
|
||||
- **Cell Network**: 10.0.0.0/24 (configurable)
|
||||
- **DHCP Range**: 10.0.0.100-10.0.0.200 (configurable)
|
||||
- **WireGuard Port**: 51820/UDP (configurable)
|
||||
- **API Port**: 3000 (configurable)
|
||||
|
||||
## 🔗 Adding Peers
|
||||
|
||||
### 1. Generate WireGuard Keys (on peer cell)
|
||||
```bash
|
||||
wg genkey | tee private.key | wg pubkey > public.key
|
||||
```
|
||||
|
||||
### 2. Add Peer to Your Cell
|
||||
```bash
|
||||
# Using the enhanced CLI
|
||||
python api/enhanced_cli.py --batch "add-peer bob 203.0.113.22 $(cat public.key)"
|
||||
|
||||
# Or via API
|
||||
curl -X POST http://localhost:3000/api/wireguard/peers \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{
|
||||
"name": "bob",
|
||||
"ip": "203.0.113.22",
|
||||
"public_key": "your_public_key_here"
|
||||
}'
|
||||
```
|
||||
|
||||
### 3. Configure Routing Rules
|
||||
```bash
|
||||
# Allow peer to access your LAN
|
||||
curl -X POST http://localhost:3000/api/routing/peers \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{
|
||||
"peer_name": "bob",
|
||||
"peer_ip": "203.0.113.22",
|
||||
"allowed_networks": ["10.0.0.0/24"],
|
||||
"route_type": "lan"
|
||||
}'
|
||||
|
||||
# Allow peer to use your cell as exit node
|
||||
curl -X POST http://localhost:3000/api/routing/exit-nodes \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{
|
||||
"peer_name": "bob",
|
||||
"peer_ip": "203.0.113.22",
|
||||
"allowed_domains": ["google.com", "github.com"]
|
||||
}'
|
||||
```
|
||||
|
||||
## 🔍 Troubleshooting
|
||||
|
||||
### Services Not Starting
|
||||
```bash
|
||||
# Check Docker logs
|
||||
docker-compose logs
|
||||
|
||||
# Check individual service
|
||||
docker-compose logs api
|
||||
docker-compose logs wireguard
|
||||
|
||||
# Check service status via API
|
||||
curl http://localhost:3000/api/services/status
|
||||
```
|
||||
|
||||
### API Issues
|
||||
```bash
|
||||
# Test API health
|
||||
curl http://localhost:3000/health
|
||||
|
||||
# Check service connectivity
|
||||
curl http://localhost:3000/api/services/connectivity
|
||||
|
||||
# View API logs
|
||||
python api/enhanced_cli.py --logs api
|
||||
```
|
||||
|
||||
### Network Issues
|
||||
```bash
|
||||
# Test DNS resolution
|
||||
nslookup google.com 127.0.0.1
|
||||
|
||||
# Check network service status
|
||||
curl http://localhost:3000/api/dns/status
|
||||
curl http://localhost:3000/api/network/info
|
||||
|
||||
# Test network connectivity
|
||||
curl -X POST http://localhost:3000/api/network/test \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"target": "8.8.8.8"}'
|
||||
```
|
||||
|
||||
### WireGuard Issues
|
||||
```bash
|
||||
# Check WireGuard status
|
||||
curl http://localhost:3000/api/wireguard/status
|
||||
|
||||
# Test WireGuard connectivity
|
||||
curl -X POST http://localhost:3000/api/wireguard/connectivity \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"target_ip": "203.0.113.22"}'
|
||||
|
||||
# View WireGuard logs
|
||||
python api/enhanced_cli.py --logs wireguard
|
||||
```
|
||||
|
||||
### Configuration Issues
|
||||
```bash
|
||||
# Validate configuration
|
||||
curl http://localhost:3000/api/config
|
||||
|
||||
# Backup and restore
|
||||
curl -X POST http://localhost:3000/api/config/backup
|
||||
curl -X POST http://localhost:3000/api/config/restore/backup_id
|
||||
|
||||
# Export/import configuration
|
||||
python api/enhanced_cli.py --export-config json
|
||||
python api/enhanced_cli.py --import-config config.json
|
||||
```
|
||||
|
||||
## 📁 File Structure
|
||||
|
||||
```
|
||||
PersonalInternetCell/
|
||||
├── docker-compose.yml # Main orchestration
|
||||
├── api/ # API server and service managers
|
||||
│ ├── base_service_manager.py # Base class for all services
|
||||
│ ├── config_manager.py # Configuration management
|
||||
│ ├── service_bus.py # Event-driven service bus
|
||||
│ ├── log_manager.py # Comprehensive logging
|
||||
│ ├── enhanced_cli.py # Enhanced CLI tool
|
||||
│ ├── network_manager.py # DNS, DHCP, NTP
|
||||
│ ├── wireguard_manager.py # VPN and peer management
|
||||
│ ├── email_manager.py # Email services
|
||||
│ ├── calendar_manager.py # Calendar services
|
||||
│ ├── file_manager.py # File storage
|
||||
│ ├── routing_manager.py # Routing and NAT
|
||||
│ ├── vault_manager.py # Security and trust
|
||||
│ ├── container_manager.py # Container orchestration
|
||||
│ ├── cell_manager.py # Overall cell management
|
||||
│ ├── peer_registry.py # Peer registration
|
||||
│ ├── app.py # Main API server
|
||||
│ └── test_enhanced_api.py # Comprehensive test suite
|
||||
├── config/ # Configuration files
|
||||
│ ├── cell.json # Cell configuration
|
||||
│ ├── network.json # Network service config
|
||||
│ ├── wireguard.json # WireGuard config
|
||||
│ └── ...
|
||||
├── data/ # Persistent data
|
||||
│ ├── api/ # API data
|
||||
│ ├── dns/ # DNS zones
|
||||
│ ├── email/ # Email data
|
||||
│ ├── calendar/ # Calendar data
|
||||
│ ├── files/ # File storage
|
||||
│ ├── vault/ # Certificates and keys
|
||||
│ └── logs/ # Service logs
|
||||
└── webui/ # React frontend (if available)
|
||||
```
|
||||
|
||||
## 🔒 Security Notes
|
||||
|
||||
- **Self-hosted CA**: The cell generates and manages its own certificates
|
||||
- **WireGuard keys**: Generated automatically with secure key management
|
||||
- **Service isolation**: All services run in isolated Docker containers
|
||||
- **Encrypted storage**: Sensitive data encrypted using Age/Fernet
|
||||
- **Trust management**: Peer trust relationships with cryptographic verification
|
||||
- **Configuration validation**: All configuration validated against schemas
|
||||
|
||||
## 🆘 Getting Help
|
||||
|
||||
### Diagnostic Commands
|
||||
```bash
|
||||
# Comprehensive status check
|
||||
python api/enhanced_cli.py --status
|
||||
|
||||
# Service health check
|
||||
python api/enhanced_cli.py --health
|
||||
|
||||
# Service logs
|
||||
python api/enhanced_cli.py --logs network
|
||||
|
||||
# Configuration validation
|
||||
curl http://localhost:3000/api/config
|
||||
|
||||
# Service connectivity test
|
||||
curl http://localhost:3000/api/services/connectivity
|
||||
```
|
||||
|
||||
### Common Issues
|
||||
1. **Port conflicts**: Ensure ports 53, 3000, 51820 are available
|
||||
2. **Permission issues**: Run with appropriate privileges for network services
|
||||
3. **Configuration errors**: Use the configuration wizard for guided setup
|
||||
4. **Service dependencies**: Check service bus status for dependency issues
|
||||
|
||||
## 🚀 Next Steps
|
||||
|
||||
After basic setup, consider:
|
||||
|
||||
1. **Customizing your cell name** and domain configuration
|
||||
2. **Adding trusted peers** for mesh networking
|
||||
3. **Configuring email services** with your domain
|
||||
4. **Setting up file storage** and user management
|
||||
5. **Implementing backup strategies** for configuration and data
|
||||
6. **Exploring advanced routing** features (exit nodes, bridge routing)
|
||||
7. **Setting up monitoring** and alerting for service health
|
||||
|
||||
## 📚 Additional Resources
|
||||
|
||||
- **[API Documentation](api/API_DOCUMENTATION.md)**: Complete API reference
|
||||
- **[Comprehensive Improvements](COMPREHENSIVE_IMPROVEMENTS_SUMMARY.md)**: Architecture overview
|
||||
- **[Enhanced API Improvements](ENHANCED_API_IMPROVEMENTS.md)**: Technical details
|
||||
- **[Project Wiki](Personal%20Internet%20Cell%20–%20Project%20Wiki.md)**: Detailed project information
|
||||
|
||||
---
|
||||
|
||||
**🌟 Happy networking with your Personal Internet Cell!**
|
||||
# Quick Start
|
||||
|
||||
This guide walks through a first-time PIC installation from a clean Linux host.
|
||||
|
||||
---
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Linux host with the WireGuard kernel module (`modprobe wireguard` to verify)
|
||||
- Docker Engine and Docker Compose installed
|
||||
- Python 3.10+ (needed for `make setup` only)
|
||||
- 2 GB+ RAM, 10 GB+ disk
|
||||
|
||||
---
|
||||
|
||||
## 1. Clone the repository
|
||||
|
||||
```bash
|
||||
git clone <repo-url> pic
|
||||
cd pic
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 2. Configure the environment
|
||||
|
||||
Copy the example environment file and edit it:
|
||||
|
||||
```bash
|
||||
cp .env.example .env
|
||||
```
|
||||
|
||||
Open `.env` and set at minimum:
|
||||
|
||||
```
|
||||
WEBDAV_PASS=changeme
|
||||
```
|
||||
|
||||
`WEBDAV_PASS` must be set before starting — the WebDAV container will fail to start without it.
|
||||
|
||||
All other variables have working defaults. See the Configuration section in [README.md](README.md) for the full list.
|
||||
|
||||
---
|
||||
|
||||
## 3. Run setup
|
||||
|
||||
`make setup` installs system dependencies, generates WireGuard keys, and writes all required config files under `config/`:
|
||||
|
||||
```bash
|
||||
make check-deps # installs docker, python3-cryptography, etc. via apt
|
||||
make setup # generates keys and writes configs
|
||||
```
|
||||
|
||||
To customise the cell identity at setup time, pass overrides on the command line:
|
||||
|
||||
```bash
|
||||
CELL_NAME=myhome CELL_DOMAIN=cell VPN_ADDRESS=10.0.0.1/24 WG_PORT=51820 make setup
|
||||
```
|
||||
|
||||
`VPN_ADDRESS` must be an RFC-1918 address (e.g. `10.0.0.1/24`).
|
||||
|
||||
---
|
||||
|
||||
## 4. Start the stack
|
||||
|
||||
```bash
|
||||
make start
|
||||
```
|
||||
|
||||
This builds the `cell-api` and `cell-webui` images and starts all 13 containers. The first run takes a few minutes while images are pulled and built.
|
||||
|
||||
Check that everything came up:
|
||||
|
||||
```bash
|
||||
make status
|
||||
```
|
||||
|
||||
You should see all containers in the `Up` state and the API responding at `http://localhost:3000/health`.
|
||||
|
||||
---
|
||||
|
||||
## 5. Open the web UI
|
||||
|
||||
Open a browser and go to:
|
||||
|
||||
```
|
||||
http://<host-ip>:8081
|
||||
```
|
||||
|
||||
If you are running locally:
|
||||
|
||||
```
|
||||
http://localhost:8081
|
||||
```
|
||||
|
||||
The sidebar contains: Dashboard, Peers, Network Services, WireGuard, Email, Calendar, Files, Routing, Vault, Containers, Cell Network, Logs, Settings.
|
||||
|
||||
---
|
||||
|
||||
## 6. Set cell identity
|
||||
|
||||
Go to **Settings** in the sidebar.
|
||||
|
||||
Set your:
|
||||
- **Cell name** — a short identifier, e.g. `myhome`
|
||||
- **Domain** — the TLD your cell will use internally, e.g. `cell`
|
||||
- **VPN IP range** — the CIDR for WireGuard peers, e.g. `10.0.0.0/24`
|
||||
|
||||
After saving, the UI will show a banner asking you to apply the changes. Click **Apply Now**. The containers will restart briefly to pick up the new configuration.
|
||||
|
||||
---
|
||||
|
||||
## 7. Add a WireGuard peer
|
||||
|
||||
Go to **WireGuard** in the sidebar.
|
||||
|
||||
1. Click **Add Peer**.
|
||||
2. Enter a name for the peer (e.g. `laptop`).
|
||||
3. The API generates a key pair and assigns the next available VPN IP automatically.
|
||||
4. Click the QR code icon to display the peer config as a QR code.
|
||||
5. Scan the QR code with a WireGuard client (Android, iOS, or the WireGuard desktop app).
|
||||
|
||||
The peer config sets your cell as the DNS server. Once connected, `*.cell` names resolve through the cell's CoreDNS.
|
||||
|
||||
To manage peers from the command line:
|
||||
|
||||
```bash
|
||||
make list-peers
|
||||
make add-peer PEER_NAME=phone PEER_IP=10.0.0.3 PEER_KEY=<base64-pubkey>
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 8. Day-to-day operations
|
||||
|
||||
```bash
|
||||
# Follow logs from all services
|
||||
make logs
|
||||
|
||||
# Follow logs from a single service
|
||||
make logs-api
|
||||
make logs-wireguard
|
||||
make logs-caddy
|
||||
|
||||
# Check container status and API health
|
||||
make status
|
||||
|
||||
# Open a shell inside a container
|
||||
make shell-api
|
||||
make shell-dns
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 9. Backup
|
||||
|
||||
Before making significant changes, create a backup:
|
||||
|
||||
```bash
|
||||
make backup
|
||||
```
|
||||
|
||||
This archives `config/` and `data/` into `backups/cell-backup-<timestamp>.tar.gz`.
|
||||
|
||||
To list available backups:
|
||||
|
||||
```bash
|
||||
make restore
|
||||
```
|
||||
|
||||
To restore manually:
|
||||
|
||||
```bash
|
||||
tar -xzf backups/cell-backup-YYYYMMDD-HHMMSS.tar.gz
|
||||
make start
|
||||
```
|
||||
|
||||
Backup and restore is also available in the UI under **Settings**.
|
||||
|
||||
---
|
||||
|
||||
## 10. Updating PIC
|
||||
|
||||
```bash
|
||||
make update
|
||||
```
|
||||
|
||||
This runs `git pull`, then rebuilds and restarts all containers. If `config/` is missing (e.g. after a fresh clone), it runs `make setup` automatically.
|
||||
|
||||
---
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
**Containers not starting**
|
||||
|
||||
```bash
|
||||
make logs
|
||||
make logs-api
|
||||
```
|
||||
|
||||
Look for errors related to missing config files or port conflicts.
|
||||
|
||||
**Port 53 already in use**
|
||||
|
||||
On Ubuntu/Debian, `systemd-resolved` listens on port 53. Disable it:
|
||||
|
||||
```bash
|
||||
sudo systemctl disable --now systemd-resolved
|
||||
sudo rm /etc/resolv.conf
|
||||
echo "nameserver 8.8.8.8" | sudo tee /etc/resolv.conf
|
||||
```
|
||||
|
||||
Then run `make start` again.
|
||||
|
||||
**WebDAV container exits immediately**
|
||||
|
||||
`WEBDAV_PASS` is not set in `.env`. Set it and run `make start` again.
|
||||
|
||||
**WireGuard container fails to load kernel module**
|
||||
|
||||
Ensure the WireGuard kernel module is available:
|
||||
|
||||
```bash
|
||||
sudo modprobe wireguard
|
||||
```
|
||||
|
||||
On some minimal installs you may need to install `wireguard-tools` and the kernel headers for your running kernel.
|
||||
|
||||
**API returns 503 or UI shows "Backend Unavailable"**
|
||||
|
||||
The Flask API may still be starting. Wait 10–15 seconds after `make start` and refresh. If it persists:
|
||||
|
||||
```bash
|
||||
make logs-api
|
||||
```
|
||||
|
||||
**Config changes not taking effect**
|
||||
|
||||
After changing identity or service settings in the UI, a yellow banner appears at the top of the page. Click **Apply Now** to restart the affected containers.
|
||||
|
||||
@@ -1,239 +1,133 @@
|
||||
|
||||
# Personal Internet Cell (PIC)
|
||||
|
||||
A self-hosted digital infrastructure platform. One stack, one API, one UI — managing DNS, DHCP, NTP, WireGuard VPN, email, calendar/contacts, file storage, and a reverse proxy on your own hardware.
|
||||
|
||||
---
|
||||
|
||||
## What it does
|
||||
|
||||
- **Network services** — CoreDNS, dnsmasq DHCP, chrony NTP, all dynamically managed
|
||||
- **WireGuard VPN** — peer lifecycle, QR-code provisioning, per-peer service access control
|
||||
- **Digital services** — Email (Postfix/Dovecot), Calendar/Contacts (Radicale CalDAV), Files (WebDAV + Filegator)
|
||||
- **Reverse proxy** — Caddy with per-service virtual IPs; subdomains like `calendar.mycell.cell` work on VPN clients automatically
|
||||
- **Certificate authority** — self-hosted CA via VaultManager
|
||||
- **Cell mesh** — connect two PIC instances with site-to-site WireGuard + DNS forwarding
|
||||
|
||||
Everything is configured through a REST API and a React web UI. No manual config file editing needed for normal operations.
|
||||
|
||||
---
|
||||
|
||||
## Quick Start
|
||||
|
||||
### Prerequisites
|
||||
|
||||
- Debian/Ubuntu host (apt-based)
|
||||
- 2 GB+ RAM, 10 GB+ disk
|
||||
- Open ports: 53 (DNS), 80 (HTTP), 3000 (API), 8081 (Web UI), 51820/udp (WireGuard)
|
||||
|
||||
### Install
|
||||
|
||||
```bash
|
||||
git clone <repo-url> pic
|
||||
cd pic
|
||||
|
||||
# Install system deps (docker, python3, python3-cryptography, etc.)
|
||||
make check-deps
|
||||
|
||||
# Generate keys + write configs
|
||||
make setup
|
||||
|
||||
# Build and start all 12 containers
|
||||
make start
|
||||
```
|
||||
|
||||
`make setup` accepts overrides for a second cell on a different host:
|
||||
|
||||
```bash
|
||||
CELL_NAME=pic1 VPN_ADDRESS=10.1.0.1/24 make setup && make start
|
||||
```
|
||||
|
||||
### Access
|
||||
|
||||
| Service | URL |
|
||||
|---------|-----|
|
||||
| Web UI | `http://<host-ip>:8081` |
|
||||
| API | `http://<host-ip>:3000` |
|
||||
| Health | `http://<host-ip>:3000/health` |
|
||||
|
||||
From a WireGuard client: `http://mycell.cell` (replace with your cell name/domain).
|
||||
|
||||
### Local dev (no Docker)
|
||||
|
||||
```bash
|
||||
pip install -r api/requirements.txt
|
||||
python api/app.py # Flask API on :3000
|
||||
|
||||
cd webui && npm install && npm run dev # React UI on :5173 (proxies /api → :3000)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Management Commands
|
||||
|
||||
```bash
|
||||
# First install
|
||||
make check-deps # install system packages via apt
|
||||
make setup # generate keys, write configs, create data dirs
|
||||
make start # start all 12 containers
|
||||
|
||||
# Daily operations
|
||||
make status # container status + API health
|
||||
make logs # follow all container logs
|
||||
make logs-api # follow logs for one service (api, dns, wg, mail, caddy, ...)
|
||||
make shell-api # shell inside a container
|
||||
|
||||
# Deploy latest code
|
||||
make update # git pull + rebuild api image + restart
|
||||
|
||||
# Maintenance
|
||||
make backup # tar config/ + data/ into backups/
|
||||
make restore # list available backups and restore
|
||||
make clean # remove containers/volumes, keep config/data
|
||||
|
||||
# Full wipe (test machines)
|
||||
make reinstall # stop, wipe config/data, setup, start fresh
|
||||
make uninstall # stop + remove images; prompts to also wipe config/data
|
||||
|
||||
# Tests
|
||||
make test # run full pytest suite
|
||||
make test-coverage # tests + HTML coverage report in htmlcov/
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Connecting Two Cells (PIC Mesh)
|
||||
|
||||
Two PIC instances form a mesh: site-to-site WireGuard tunnels with automatic DNS forwarding so each cell's services resolve from the other.
|
||||
|
||||
### Exchange invites
|
||||
|
||||
1. On **Cell A** → Web UI → **Cell Network** → copy the invite JSON.
|
||||
2. On **Cell B** → **Cell Network** → paste into "Connect to Another Cell" → **Connect**.
|
||||
3. On **Cell B** → copy its invite JSON.
|
||||
4. On **Cell A** → paste Cell B's invite → **Connect**.
|
||||
|
||||
Both cells now have a WireGuard peer with `AllowedIPs = remote VPN subnet` and a CoreDNS forwarding block so `*.pic1.cell` resolves across the tunnel.
|
||||
|
||||
### Same-LAN tip
|
||||
|
||||
If both cells share the same external IP (behind NAT), replace the auto-detected endpoint with the LAN IP before connecting:
|
||||
|
||||
```json
|
||||
{ "endpoint": "192.168.31.50:51820", ... }
|
||||
```
|
||||
PIC is a self-hosted digital infrastructure platform. It manages DNS, DHCP, NTP, WireGuard VPN, email, calendar/contacts (CalDAV), file storage (WebDAV), a reverse proxy, and a certificate authority — all controlled from a single REST API and React web UI. No manual config file editing is required for normal operations.
|
||||
|
||||
---
|
||||
|
||||
## Architecture
|
||||
|
||||
### Stack
|
||||
|
||||
```
|
||||
cell-caddy (Caddy) :80/:443 + per-service virtual IPs
|
||||
cell-api (Flask :3000) REST API + config management + container orchestration
|
||||
cell-webui (Nginx :8081) React UI
|
||||
cell-dns (CoreDNS :53) internal DNS + per-peer ACLs
|
||||
cell-dhcp (dnsmasq) DHCP + static reservations
|
||||
cell-ntp (chrony) NTP
|
||||
cell-wireguard WireGuard VPN
|
||||
cell-mail (docker-mailserver) SMTP/IMAP
|
||||
cell-radicale CalDAV/CardDAV :5232
|
||||
cell-webdav WebDAV :80
|
||||
cell-filegator file manager UI :8080
|
||||
cell-rainloop webmail :8888
|
||||
Browser
|
||||
└── React SPA (cell-webui :8081)
|
||||
└── Flask REST API (cell-api :3000, bound to 127.0.0.1)
|
||||
└── Docker SDK / config files
|
||||
├── cell-caddy :80/:443 reverse proxy
|
||||
├── cell-dns :53 CoreDNS
|
||||
├── cell-dhcp :67/udp dnsmasq
|
||||
├── cell-ntp :123/udp chrony
|
||||
├── cell-wireguard :51820/udp WireGuard VPN
|
||||
├── cell-mail :25/:587/:993 Postfix + Dovecot
|
||||
├── cell-radicale 127.0.0.1:5232 CalDAV/CardDAV
|
||||
├── cell-webdav 127.0.0.1:8080 WebDAV
|
||||
├── cell-rainloop :8888 webmail (RainLoop)
|
||||
├── cell-filegator :8082 file manager UI
|
||||
└── cell-webui :8081 React UI (Nginx)
|
||||
```
|
||||
|
||||
All containers share a custom Docker bridge network. Static IPs are assigned in `docker-compose.yml`. Caddy adds per-service virtual IPs to its own interface at API startup so `calendar.<domain>`, `files.<domain>`, etc. route to the right container.
|
||||
All containers run on a custom Docker bridge network (`cell-network`, default `172.20.0.0/16`). Static IPs per container are set in `docker-compose.yml` and overridden via `.env`.
|
||||
|
||||
### Backend (`api/`)
|
||||
The Flask API (`api/app.py`, ~2800 lines) contains all REST endpoints, runs a background health-monitoring thread, and manages the entire lifecycle of generated config artefacts: `Caddyfile`, `Corefile`, `wg0.conf`, and `cell_config.json` (the single source of truth at `config/api/cell_config.json`).
|
||||
|
||||
Service managers (`network_manager.py`, `wireguard_manager.py`, `peer_registry.py`, etc.) all inherit `BaseServiceManager`. `app.py` contains all Flask routes — one file, organized by service.
|
||||
|
||||
`ConfigManager` (`config_manager.py`) is the single source of truth. Config lives in `config/api/cell_config.json`. All managers read/write through it.
|
||||
|
||||
`ip_utils.py` owns all container IP logic via `CONTAINER_OFFSETS` — do not hardcode IPs elsewhere.
|
||||
|
||||
When a config change requires recreating the Docker network (e.g. `ip_range` change), the API spawns a helper container that outlives cell-api to run `docker compose down && up`. Other restarts run `compose up -d --no-deps <containers>` directly.
|
||||
|
||||
### Frontend (`webui/`)
|
||||
|
||||
React 18 + Vite + Tailwind CSS. All API calls go through `src/services/api.js` (Axios). Vite dev server proxies `/api` to `localhost:3000`. Pages in `src/pages/`, shared components in `src/components/`.
|
||||
|
||||
### Project layout
|
||||
|
||||
```
|
||||
pic/
|
||||
├── api/ # Flask API + all service managers
|
||||
│ ├── app.py # all routes (~2700 lines)
|
||||
│ ├── config_manager.py # unified config CRUD
|
||||
│ ├── ip_utils.py # IP/CIDR helpers + Caddyfile generator
|
||||
│ ├── firewall_manager.py # iptables (via cell-wireguard) + Corefile
|
||||
│ ├── network_manager.py # DNS zones, DHCP, NTP
|
||||
│ ├── wireguard_manager.py
|
||||
│ ├── peer_registry.py
|
||||
│ ├── vault_manager.py
|
||||
│ ├── email_manager.py
|
||||
│ ├── calendar_manager.py
|
||||
│ ├── file_manager.py
|
||||
│ └── container_manager.py
|
||||
├── webui/ # React frontend
|
||||
├── config/ # Config files (bind-mounted into containers)
|
||||
│ ├── api/cell_config.json ← live config
|
||||
│ ├── caddy/Caddyfile
|
||||
│ ├── dns/Corefile
|
||||
│ └── ...
|
||||
├── data/ # Persistent data (git-ignored)
|
||||
├── tests/ # pytest suite (372 tests, 27 files)
|
||||
├── docker-compose.yml
|
||||
└── Makefile
|
||||
```
|
||||
The React frontend (`webui/`) is built with Vite + Tailwind CSS. All API calls go through `src/services/api.js` (Axios). Pages: Dashboard, Peers, Network Services, WireGuard, Email, Calendar, Files, Routing, Vault, Containers, Cell Network, Logs, Settings.
|
||||
|
||||
---
|
||||
|
||||
## API Reference
|
||||
## Requirements
|
||||
|
||||
### Config
|
||||
- Linux host with the WireGuard kernel module loaded
|
||||
- Docker Engine and Docker Compose (v2 plugin or v1 standalone)
|
||||
- Python 3.10+ (for `make setup` and local dev only; not needed at runtime)
|
||||
- 2 GB+ RAM, 10 GB+ disk
|
||||
- Ports available: 53, 67/udp, 80, 443, 51820/udp, 25, 587, 993
|
||||
|
||||
```
|
||||
GET /api/config full config + service IPs
|
||||
PUT /api/config update identity or service config
|
||||
GET /api/config/pending pending restart info
|
||||
POST /api/config/apply apply pending restart
|
||||
POST /api/config/backup create backup
|
||||
POST /api/config/restore/<backup_id> restore from backup
|
||||
```
|
||||
---
|
||||
|
||||
### Network
|
||||
## Quick Start
|
||||
|
||||
```
|
||||
GET /api/dns/records
|
||||
POST /api/dns/records
|
||||
GET /api/dhcp/leases
|
||||
GET /api/dhcp/reservations
|
||||
POST /api/dhcp/reservations
|
||||
```
|
||||
See [QUICKSTART.md](QUICKSTART.md) for step-by-step setup.
|
||||
|
||||
### WireGuard & Peers
|
||||
---
|
||||
|
||||
```
|
||||
GET /api/wireguard/status
|
||||
GET /api/wireguard/peers
|
||||
POST /api/wireguard/peers
|
||||
GET /api/peers
|
||||
POST /api/peers
|
||||
PUT /api/peers/<name>
|
||||
DELETE /api/peers/<name>
|
||||
GET /api/peers/<name>/config peer config + QR code
|
||||
```
|
||||
## Configuration
|
||||
|
||||
### Containers & Health
|
||||
Runtime configuration is controlled by `.env` in the project root. Copy `.env.example` to `.env` before first run.
|
||||
|
||||
```
|
||||
GET /api/containers
|
||||
POST /api/containers/<name>/restart
|
||||
GET /health
|
||||
GET /api/services/status
|
||||
| Variable | Default | Description |
|
||||
|---|---|---|
|
||||
| `CELL_NETWORK` | `172.20.0.0/16` | Docker bridge subnet for all containers |
|
||||
| `CADDY_IP` through `FILEGATOR_IP` | `172.20.0.2`–`.13` | Static IP for each container |
|
||||
| `DNS_PORT` | `53` | DNS (UDP+TCP) |
|
||||
| `DHCP_PORT` | `67` | DHCP (UDP) |
|
||||
| `NTP_PORT` | `123` | NTP (UDP) |
|
||||
| `WG_PORT` | `51820` | WireGuard listen port (UDP) |
|
||||
| `API_PORT` | `3000` | Flask API (bound to `127.0.0.1`) |
|
||||
| `WEBUI_PORT` | `8081` | React UI |
|
||||
| `MAIL_SMTP_PORT` | `25` | SMTP |
|
||||
| `MAIL_SUBMISSION_PORT` | `587` | SMTP submission |
|
||||
| `MAIL_IMAP_PORT` | `993` | IMAP |
|
||||
| `RADICALE_PORT` | `5232` | CalDAV (bound to `127.0.0.1`) |
|
||||
| `WEBDAV_PORT` | `8080` | WebDAV (bound to `127.0.0.1`) |
|
||||
| `RAINLOOP_PORT` | `8888` | Webmail |
|
||||
| `FILEGATOR_PORT` | `8082` | File manager UI |
|
||||
| `WEBDAV_USER` | `admin` | WebDAV basic-auth username |
|
||||
| `WEBDAV_PASS` | _(required)_ | WebDAV basic-auth password — must be set before `make start` |
|
||||
| `FLASK_DEBUG` | _(unset)_ | Set to `1` to enable Flask debug mode; do not use in production |
|
||||
| `PUID` / `PGID` | current user | UID/GID passed to the WireGuard container |
|
||||
|
||||
Cell identity (cell name, domain, VPN IP range) is configured via `make setup` or the Settings → Identity page in the UI after startup. The VPN IP range must be an RFC-1918 CIDR (`10.0.0.0/8`, `172.16.0.0/12`, or `192.168.0.0/16`); the API and UI both enforce this.
|
||||
|
||||
---
|
||||
|
||||
## Security Notes
|
||||
|
||||
**Ports exposed to the network:**
|
||||
|
||||
- `80` / `443` — Caddy (HTTP/HTTPS reverse proxy)
|
||||
- `51820/udp` — WireGuard
|
||||
- `25` / `587` / `993` — Mail (SMTP, submission, IMAP)
|
||||
- `53` — DNS (UDP + TCP)
|
||||
- `67/udp` — DHCP
|
||||
- `8081` — Web UI
|
||||
- `8888` — Webmail (RainLoop)
|
||||
- `8082` — File manager (Filegator)
|
||||
|
||||
**Ports bound to `127.0.0.1` only** (not directly reachable from the network):
|
||||
|
||||
- `3000` — Flask API
|
||||
- `5232` — Radicale (CalDAV)
|
||||
- `8080` — WebDAV
|
||||
|
||||
The API has no authentication layer. It relies on `is_local_request()` to restrict sensitive endpoints (containers, vault) to requests originating from loopback or the cell's Docker network. The Docker socket is mounted into `cell-api`; treat access to port 3000 as equivalent to root access on the host.
|
||||
|
||||
For internet-facing deployments, place the host behind a firewall or VPN and restrict access to the API and UI ports.
|
||||
|
||||
---
|
||||
|
||||
## Development
|
||||
|
||||
```bash
|
||||
# Start the full stack (builds api and webui images)
|
||||
make start
|
||||
|
||||
# Rebuild a single image after code changes
|
||||
make build-api
|
||||
make build-webui
|
||||
|
||||
# Run Flask API locally without Docker (port 3000)
|
||||
pip install -r api/requirements.txt
|
||||
python api/app.py
|
||||
|
||||
# Run React UI dev server locally (port 5173, proxies /api to :3000)
|
||||
cd webui && npm install && npm run dev
|
||||
|
||||
# Follow all container logs
|
||||
make logs
|
||||
|
||||
# Follow logs for one service (e.g. api, dns, caddy, wireguard, mail)
|
||||
make logs-api
|
||||
|
||||
# Open a shell inside a container
|
||||
make shell-api
|
||||
```
|
||||
|
||||
---
|
||||
@@ -241,24 +135,53 @@ GET /api/services/status
|
||||
## Testing
|
||||
|
||||
```bash
|
||||
make test # run full suite
|
||||
make test-coverage # coverage report in htmlcov/
|
||||
pytest tests/test_<module>.py # single file
|
||||
pytest tests/ -k "test_name" # single test
|
||||
make test # run the full pytest suite
|
||||
make test-coverage # run with coverage; HTML report in htmlcov/
|
||||
```
|
||||
|
||||
Tests live in `tests/` and use `unittest.TestCase` collected by pytest. External system calls (Docker, iptables, file writes) are mocked with `unittest.mock.patch`.
|
||||
Tests live in `tests/` (34 files, 642 test functions). Coverage includes:
|
||||
|
||||
Known coverage gaps: `write_caddyfile`, `POST /api/config/apply` (helper container path), `PUT /api/config` 400 validation paths. These are the highest-risk untested paths.
|
||||
- All service managers (network, WireGuard, email, calendar, file, routing, vault, container)
|
||||
- API endpoint tests for each service area
|
||||
- Config manager (CRUD, validation, backup/restore)
|
||||
- IP utilities and Caddyfile generation
|
||||
- Peer registry and WireGuard peer lifecycle
|
||||
- Service bus pub/sub
|
||||
- Firewall manager
|
||||
- Pending-restart logic
|
||||
|
||||
Integration tests (`tests/integration/`) require a running PIC stack:
|
||||
|
||||
```bash
|
||||
make test-integration # full suite (creates peers)
|
||||
make test-integration-readonly # read-only checks, safe to run anytime
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Security Notes
|
||||
## Management Commands
|
||||
|
||||
- The API is access-controlled by `is_local_request()` — it checks whether the request comes from a local/loopback/cell-network IP. Sensitive endpoints (containers, vault) are restricted to local access only.
|
||||
- All per-peer service access is enforced via iptables rules inside `cell-wireguard` and CoreDNS ACL blocks.
|
||||
- The Docker socket is mounted into `cell-api` for container management — treat network access to port 3000 as privileged.
|
||||
- `ip_range` must be an RFC-1918 CIDR (10.0.0.0/8, 172.16.0.0/12, or 192.168.0.0/16). The API and UI both validate this.
|
||||
```bash
|
||||
make setup # generate WireGuard keys, write configs, create data dirs
|
||||
make start # docker compose up -d --build
|
||||
make stop # docker compose down
|
||||
make restart # docker compose restart
|
||||
make status # container status + API health check
|
||||
make logs # follow all service logs
|
||||
make logs-<svc> # follow logs for one service
|
||||
make shell-<svc> # shell inside a container
|
||||
|
||||
make update # git pull + rebuild + restart
|
||||
make reinstall # full wipe of config/ and data/, then setup + start
|
||||
make uninstall # stop containers; prompts whether to also delete config/ and data/
|
||||
|
||||
make backup # tar config/ + data/ into backups/
|
||||
make restore # list available backups
|
||||
|
||||
make list-peers # show WireGuard peers via API
|
||||
make show-routes # wg show inside the wireguard container
|
||||
make add-peer PEER_NAME=foo PEER_IP=10.0.0.5 PEER_KEY=<pubkey>
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
|
||||
+30
-13
@@ -343,8 +343,16 @@ def _local_subnets():
|
||||
|
||||
|
||||
def is_local_request():
|
||||
# SECURITY: do NOT use X-Forwarded-For for auth. Caddy (and any reverse
|
||||
# proxy) sets XFF to the original client IP, but the TCP peer that reaches
|
||||
# this Flask process is always the proxy itself (an RFC-1918 Docker IP).
|
||||
# Trusting XFF would let any internet client claim a local IP via that
|
||||
# header. Only the direct TCP peer (request.remote_addr) is trustworthy:
|
||||
# all legitimate local traffic comes directly from the Docker network or
|
||||
# loopback, so remote_addr being local is a sufficient and necessary
|
||||
# condition. The XFF header is read for logging only, never for access
|
||||
# decisions.
|
||||
remote_addr = request.remote_addr
|
||||
forwarded_for = request.headers.get('X-Forwarded-For', '')
|
||||
|
||||
def _allowed(addr):
|
||||
if not addr:
|
||||
@@ -374,14 +382,7 @@ def is_local_request():
|
||||
pass
|
||||
return False
|
||||
|
||||
if _allowed(remote_addr):
|
||||
return True
|
||||
# Only trust the LAST X-Forwarded-For entry — that is what the reverse proxy appended.
|
||||
if forwarded_for:
|
||||
last_hop = forwarded_for.split(',')[-1].strip()
|
||||
if _allowed(last_hop):
|
||||
return True
|
||||
return False
|
||||
return _allowed(remote_addr)
|
||||
|
||||
@app.route('/health', methods=['GET'])
|
||||
def health_check():
|
||||
@@ -1416,10 +1417,13 @@ def test_network():
|
||||
# WireGuard API
|
||||
@app.route('/api/wireguard/keys', methods=['GET'])
|
||||
def get_wireguard_keys():
|
||||
"""Get WireGuard keys."""
|
||||
"""Get WireGuard keys (public key only; private key never leaves the server)."""
|
||||
try:
|
||||
result = wireguard_manager.get_keys()
|
||||
return jsonify(result)
|
||||
keys = wireguard_manager.get_keys()
|
||||
return jsonify({
|
||||
'public_key': keys.get('public_key', ''),
|
||||
'has_private_key': bool(keys.get('private_key')),
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting WireGuard keys: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
@@ -2149,6 +2153,8 @@ def create_folder():
|
||||
return jsonify({"error": "No data provided"}), 400
|
||||
result = file_manager.create_folder(data)
|
||||
return jsonify(result)
|
||||
except ValueError as e:
|
||||
return jsonify({"error": str(e)}), 400
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating folder: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
@@ -2159,6 +2165,8 @@ def delete_folder(username, folder_path):
|
||||
try:
|
||||
result = file_manager.delete_folder(username, folder_path)
|
||||
return jsonify(result)
|
||||
except ValueError as e:
|
||||
return jsonify({"error": str(e)}), 400
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting folder: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
@@ -2175,6 +2183,8 @@ def upload_file(username):
|
||||
|
||||
result = file_manager.upload_file(username, file, path)
|
||||
return jsonify(result)
|
||||
except ValueError as e:
|
||||
return jsonify({"error": str(e)}), 400
|
||||
except Exception as e:
|
||||
logger.error(f"Error uploading file: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
@@ -2185,6 +2195,8 @@ def download_file(username, file_path):
|
||||
try:
|
||||
result = file_manager.download_file(username, file_path)
|
||||
return jsonify(result)
|
||||
except ValueError as e:
|
||||
return jsonify({"error": str(e)}), 400
|
||||
except Exception as e:
|
||||
logger.error(f"Error downloading file: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
@@ -2195,6 +2207,8 @@ def delete_file(username, file_path):
|
||||
try:
|
||||
result = file_manager.delete_file(username, file_path)
|
||||
return jsonify(result)
|
||||
except ValueError as e:
|
||||
return jsonify({"error": str(e)}), 400
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting file: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
@@ -2206,6 +2220,8 @@ def list_files(username):
|
||||
folder = request.args.get('folder', '')
|
||||
result = file_manager.list_files(username, folder)
|
||||
return jsonify(result)
|
||||
except ValueError as e:
|
||||
return jsonify({"error": str(e)}), 400
|
||||
except Exception as e:
|
||||
logger.error(f"Error listing files: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
@@ -2915,4 +2931,5 @@ def remove_volume(name):
|
||||
return jsonify({'removed': success})
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(host='0.0.0.0', port=3000, debug=True)
|
||||
debug = os.environ.get('FLASK_DEBUG', '0') == '1'
|
||||
app.run(host='0.0.0.0', port=3000, debug=debug)
|
||||
@@ -196,21 +196,6 @@ class ConfigManager:
|
||||
"warnings": warnings
|
||||
}
|
||||
|
||||
def get_all_configs(self) -> Dict[str, Dict]:
|
||||
"""Return all stored service configurations."""
|
||||
return dict(self.configs)
|
||||
|
||||
def get_config_summary(self) -> Dict[str, Any]:
|
||||
"""Return a high-level summary of configuration state."""
|
||||
backup_count = sum(
|
||||
1 for p in self.backup_dir.iterdir() if p.is_dir()
|
||||
) if self.backup_dir.exists() else 0
|
||||
return {
|
||||
'total_services': len(self.service_schemas),
|
||||
'configured_services': len(self.configs),
|
||||
'backup_count': backup_count,
|
||||
}
|
||||
|
||||
def backup_config(self) -> str:
|
||||
"""Create a backup of cell_config.json, secrets, Caddyfile, .env, Corefile, and DNS zones."""
|
||||
try:
|
||||
|
||||
+29
-6
@@ -5,6 +5,7 @@ Handles WebDAV file storage services
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import json
|
||||
import subprocess
|
||||
import logging
|
||||
@@ -43,6 +44,28 @@ class FileManager(BaseServiceManager):
|
||||
except (PermissionError, OSError):
|
||||
pass
|
||||
|
||||
def _safe_path(self, username: str, *parts: str) -> str:
|
||||
"""Resolve a safe path under files_dir/username.
|
||||
|
||||
Whitelists username, joins extra parts, resolves to a real path, and
|
||||
asserts the result is contained within the user's directory. Raises
|
||||
ValueError on any sign of path traversal or invalid input.
|
||||
"""
|
||||
if not isinstance(username, str) or not re.match(r'^[A-Za-z0-9_.-]{1,64}$', username):
|
||||
raise ValueError(f"Invalid username: {username!r}")
|
||||
safe_parts = []
|
||||
for p in parts:
|
||||
if p is None:
|
||||
continue
|
||||
if not isinstance(p, str):
|
||||
raise ValueError(f"Invalid path component: {p!r}")
|
||||
safe_parts.append(p)
|
||||
user_root = os.path.realpath(os.path.join(self.files_dir, username))
|
||||
candidate = os.path.realpath(os.path.join(self.files_dir, username, *safe_parts))
|
||||
if candidate != user_root and not candidate.startswith(user_root + os.sep):
|
||||
raise ValueError(f"Path traversal detected for user {username!r}: {parts!r}")
|
||||
return candidate
|
||||
|
||||
def _generate_webdav_config(self):
|
||||
"""Generate WebDAV configuration"""
|
||||
config = """# WebDAV configuration for Personal Internet Cell
|
||||
@@ -230,7 +253,7 @@ umask = 022
|
||||
logger.error("Username and folder_path must not be empty")
|
||||
return False
|
||||
try:
|
||||
full_path = os.path.join(self.files_dir, username, folder_path)
|
||||
full_path = self._safe_path(username, folder_path)
|
||||
os.makedirs(full_path, exist_ok=True)
|
||||
|
||||
logger.info(f"Created folder {folder_path} for {username}")
|
||||
@@ -246,7 +269,7 @@ umask = 022
|
||||
logger.error("Username and folder_path must not be empty")
|
||||
return False
|
||||
try:
|
||||
full_path = os.path.join(self.files_dir, username, folder_path)
|
||||
full_path = self._safe_path(username, folder_path)
|
||||
|
||||
if os.path.exists(full_path):
|
||||
shutil.rmtree(full_path)
|
||||
@@ -263,7 +286,7 @@ umask = 022
|
||||
def upload_file(self, username: str, file_path: str, file_data: bytes) -> bool:
|
||||
"""Upload a file for a user"""
|
||||
try:
|
||||
full_path = os.path.join(self.files_dir, username, file_path)
|
||||
full_path = self._safe_path(username, file_path)
|
||||
|
||||
# Ensure directory exists
|
||||
os.makedirs(os.path.dirname(full_path), exist_ok=True)
|
||||
@@ -282,7 +305,7 @@ umask = 022
|
||||
def download_file(self, username: str, file_path: str) -> Optional[bytes]:
|
||||
"""Download a file for a user"""
|
||||
try:
|
||||
full_path = os.path.join(self.files_dir, username, file_path)
|
||||
full_path = self._safe_path(username, file_path)
|
||||
|
||||
if os.path.exists(full_path):
|
||||
with open(full_path, 'rb') as f:
|
||||
@@ -298,7 +321,7 @@ umask = 022
|
||||
def delete_file(self, username: str, file_path: str) -> bool:
|
||||
"""Delete a file for a user"""
|
||||
try:
|
||||
full_path = os.path.join(self.files_dir, username, file_path)
|
||||
full_path = self._safe_path(username, file_path)
|
||||
|
||||
if os.path.exists(full_path):
|
||||
os.remove(full_path)
|
||||
@@ -317,7 +340,7 @@ umask = 022
|
||||
files = []
|
||||
|
||||
try:
|
||||
full_path = os.path.join(self.files_dir, username, folder_path)
|
||||
full_path = self._safe_path(username, folder_path)
|
||||
|
||||
if os.path.exists(full_path):
|
||||
for item in os.listdir(full_path):
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
flask==2.3.3
|
||||
flask-cors==4.0.0
|
||||
requests==2.31.0
|
||||
cryptography==41.0.7
|
||||
flask>=3.0.3
|
||||
flask-cors>=4.0.1
|
||||
requests>=2.32.3
|
||||
cryptography>=42.0.5
|
||||
pyyaml==6.0.1
|
||||
icalendar==5.0.7
|
||||
vobject==0.9.6.1
|
||||
@@ -13,4 +13,4 @@ pytest==7.4.3
|
||||
pytest-cov==4.1.0
|
||||
pytest-mock==3.12.0
|
||||
|
||||
docker
|
||||
docker>=7.0.0
|
||||
@@ -4,6 +4,7 @@ WireGuard Manager for Personal Internet Cell
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import json
|
||||
import base64
|
||||
import socket
|
||||
@@ -92,6 +93,8 @@ class WireGuardManager(BaseServiceManager):
|
||||
|
||||
def generate_peer_keys(self, peer_name: str) -> Dict[str, str]:
|
||||
"""Generate a keypair for a peer, save to keys_dir/peers/, return as base64."""
|
||||
if not isinstance(peer_name, str) or not re.match(r'^[A-Za-z0-9_.-]{1,64}$', peer_name):
|
||||
raise ValueError(f"Invalid peer_name: {peer_name!r}")
|
||||
priv_bytes, pub_bytes = self._generate_keypair()
|
||||
priv_b64 = base64.b64encode(priv_bytes).decode()
|
||||
pub_b64 = base64.b64encode(pub_bytes).decode()
|
||||
@@ -332,7 +335,16 @@ class WireGuardManager(BaseServiceManager):
|
||||
Passing full-tunnel or split-tunnel CIDRs here would cause the server
|
||||
to route all internet or LAN traffic to that peer — breaking everything.
|
||||
"""
|
||||
import ipaddress
|
||||
import ipaddress, re as _re
|
||||
if not isinstance(public_key, str) or not _re.match(r'^[A-Za-z0-9+/]{43}=$', public_key.strip()):
|
||||
return False # invalid WireGuard public key
|
||||
if name and not _re.match(r'^[A-Za-z0-9_. -]{1,64}$', name):
|
||||
return False # reject names with newlines/brackets
|
||||
if endpoint_ip:
|
||||
try:
|
||||
ipaddress.ip_address(endpoint_ip.strip())
|
||||
except ValueError:
|
||||
return False
|
||||
try:
|
||||
# Enforce /32: reject any CIDR wider than a single host
|
||||
for cidr in (c.strip() for c in allowed_ips.split(',')):
|
||||
|
||||
+8
-8
@@ -122,7 +122,7 @@ services:
|
||||
image: tomsquest/docker-radicale:latest
|
||||
container_name: cell-radicale
|
||||
ports:
|
||||
- "${RADICALE_PORT:-5232}:5232"
|
||||
- "127.0.0.1:${RADICALE_PORT:-5232}:5232"
|
||||
volumes:
|
||||
- ./config/radicale:/etc/radicale
|
||||
- ./data/radicale:/data
|
||||
@@ -141,11 +141,11 @@ services:
|
||||
image: bytemark/webdav:latest
|
||||
container_name: cell-webdav
|
||||
ports:
|
||||
- "${WEBDAV_PORT:-8080}:80"
|
||||
- "127.0.0.1:${WEBDAV_PORT:-8080}:80"
|
||||
environment:
|
||||
- AUTH_TYPE=Basic
|
||||
- USERNAME=admin
|
||||
- PASSWORD=admin123
|
||||
- USERNAME=${WEBDAV_USER:-admin}
|
||||
- PASSWORD=${WEBDAV_PASS}
|
||||
volumes:
|
||||
- ./data/files:/var/lib/dav
|
||||
restart: unless-stopped
|
||||
@@ -193,7 +193,7 @@ services:
|
||||
build: ./api
|
||||
container_name: cell-api
|
||||
ports:
|
||||
- "${API_PORT:-3000}:3000"
|
||||
- "127.0.0.1:${API_PORT:-3000}:3000"
|
||||
volumes:
|
||||
- ./data/api:/app/data
|
||||
- ./data/dns:/app/data/dns
|
||||
@@ -223,7 +223,7 @@ services:
|
||||
build: ./webui
|
||||
container_name: cell-webui
|
||||
ports:
|
||||
- "${WEBUI_PORT:-8081}:80"
|
||||
- "127.0.0.1:${WEBUI_PORT:-8081}:80"
|
||||
restart: unless-stopped
|
||||
networks:
|
||||
cell-network:
|
||||
@@ -243,7 +243,7 @@ services:
|
||||
cell-network:
|
||||
ipv4_address: ${RAINLOOP_IP:-172.20.0.12}
|
||||
ports:
|
||||
- "${RAINLOOP_PORT:-8888}:8888"
|
||||
- "127.0.0.1:${RAINLOOP_PORT:-8888}:8888"
|
||||
volumes:
|
||||
- ./data/rainloop:/rainloop/data
|
||||
logging:
|
||||
@@ -261,7 +261,7 @@ services:
|
||||
cell-network:
|
||||
ipv4_address: ${FILEGATOR_IP:-172.20.0.13}
|
||||
ports:
|
||||
- "${FILEGATOR_PORT:-8082}:8080"
|
||||
- "127.0.0.1:${FILEGATOR_PORT:-8082}:8080"
|
||||
volumes:
|
||||
- ./data/filegator:/var/www/filegator/private
|
||||
logging:
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tests for POST /api/config/apply.
|
||||
|
||||
The route reads _pending_restart from config_manager, spawns a background
|
||||
thread/process, clears the pending flag, and returns 200.
|
||||
|
||||
We mock subprocess.Popen / subprocess.run and docker.from_env so the tests
|
||||
run without Docker, and we capture what command-line arguments would be used.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock, call
|
||||
|
||||
api_dir = Path(__file__).parent.parent / 'api'
|
||||
sys.path.insert(0, str(api_dir))
|
||||
|
||||
from app import app, _set_pending_restart, _clear_pending_restart, config_manager
|
||||
|
||||
|
||||
class TestConfigApplyRoute(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
_clear_pending_restart()
|
||||
|
||||
def tearDown(self):
|
||||
_clear_pending_restart()
|
||||
|
||||
# ── No pending changes ─────────────────────────────────────────────────
|
||||
|
||||
def test_apply_with_no_pending_returns_200(self):
|
||||
r = self.client.post('/api/config/apply')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
def test_apply_with_no_pending_returns_no_changes_message(self):
|
||||
r = self.client.post('/api/config/apply')
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('message', data)
|
||||
self.assertIn('No pending', data['message'])
|
||||
|
||||
# ── Pending changes present ────────────────────────────────────────────
|
||||
|
||||
@patch('subprocess.Popen')
|
||||
@patch('docker.from_env')
|
||||
def test_apply_with_pending_returns_200(self, mock_docker, mock_popen):
|
||||
mock_docker.side_effect = Exception('no docker in test')
|
||||
mock_popen.return_value = MagicMock()
|
||||
_set_pending_restart(['dns_port: 53 → 5353'], ['*'])
|
||||
r = self.client.post('/api/config/apply')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
@patch('subprocess.Popen')
|
||||
@patch('docker.from_env')
|
||||
def test_apply_with_pending_returns_restart_in_progress(self, mock_docker, mock_popen):
|
||||
mock_docker.side_effect = Exception('no docker in test')
|
||||
mock_popen.return_value = MagicMock()
|
||||
_set_pending_restart(['something changed'], ['*'])
|
||||
r = self.client.post('/api/config/apply')
|
||||
data = json.loads(r.data)
|
||||
self.assertTrue(data.get('restart_in_progress'))
|
||||
|
||||
# ── Pending state cleared after apply ──────────────────────────────────
|
||||
|
||||
@patch('threading.Thread')
|
||||
@patch('docker.from_env')
|
||||
def test_apply_clears_pending_state(self, mock_docker, mock_thread):
|
||||
mock_docker.side_effect = Exception('no docker in test')
|
||||
# Don't actually start the thread so we don't need subprocess
|
||||
mock_thread.return_value = MagicMock()
|
||||
_set_pending_restart(['config changed'], ['*'])
|
||||
self.client.post('/api/config/apply')
|
||||
pending = config_manager.configs.get('_pending_restart', {})
|
||||
self.assertFalse(pending.get('needs_restart', False))
|
||||
|
||||
# ── needs_network_recreate=True → helper script includes 'down' ────────
|
||||
|
||||
@patch('subprocess.Popen')
|
||||
@patch('docker.from_env')
|
||||
def test_apply_network_recreate_spawns_popen_with_down_command(
|
||||
self, mock_docker, mock_popen):
|
||||
mock_docker.side_effect = Exception('no docker in test')
|
||||
mock_popen.return_value = MagicMock()
|
||||
|
||||
# Set up a wildcard pending change that also requires network recreation
|
||||
_set_pending_restart(['ip_range changed'], ['*'])
|
||||
config_manager.configs['_pending_restart']['network_recreate'] = True
|
||||
|
||||
r = self.client.post('/api/config/apply')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
# Wait for background thread to call Popen
|
||||
import time
|
||||
for _ in range(20):
|
||||
if mock_popen.called:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
self.assertTrue(mock_popen.called,
|
||||
'Expected subprocess.Popen to be called for wildcard restart')
|
||||
args, kwargs = mock_popen.call_args
|
||||
cmd = args[0]
|
||||
# cmd is the full docker run ... sh -c 'script'
|
||||
script_arg = cmd[-1] # the -c argument
|
||||
self.assertIn('down', script_arg,
|
||||
f'Expected "down" in helper script when network_recreate=True, got: {script_arg}')
|
||||
|
||||
# ── needs_network_recreate=False → helper script uses only 'up -d' ─────
|
||||
|
||||
@patch('subprocess.Popen')
|
||||
@patch('docker.from_env')
|
||||
def test_apply_no_network_recreate_spawns_popen_without_down(
|
||||
self, mock_docker, mock_popen):
|
||||
mock_docker.side_effect = Exception('no docker in test')
|
||||
mock_popen.return_value = MagicMock()
|
||||
|
||||
_set_pending_restart(['port changed'], ['*'])
|
||||
# network_recreate defaults to False
|
||||
|
||||
self.client.post('/api/config/apply')
|
||||
|
||||
import time
|
||||
for _ in range(20):
|
||||
if mock_popen.called:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
self.assertTrue(mock_popen.called)
|
||||
args, _ = mock_popen.call_args
|
||||
script_arg = args[0][-1]
|
||||
self.assertNotIn(' down', script_arg,
|
||||
f'Did not expect "down" in helper script when network_recreate=False')
|
||||
self.assertIn('up -d', script_arg)
|
||||
|
||||
# ── Specific containers (not wildcard) ─────────────────────────────────
|
||||
|
||||
@patch('subprocess.run')
|
||||
@patch('docker.from_env')
|
||||
def test_apply_specific_containers_uses_subprocess_run(
|
||||
self, mock_docker, mock_run):
|
||||
mock_docker.side_effect = Exception('no docker in test')
|
||||
mock_run.return_value = MagicMock(returncode=0, stderr='')
|
||||
_set_pending_restart(['dns port changed'], ['dns'])
|
||||
|
||||
r = self.client.post('/api/config/apply')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
# Give the daemon thread a moment to call subprocess.run
|
||||
import time
|
||||
for _ in range(30):
|
||||
# Look for the compose call specifically (may not be the last call)
|
||||
compose_calls = [
|
||||
c for c in mock_run.call_args_list
|
||||
if 'compose' in (c.args[0] if c.args else [])
|
||||
]
|
||||
if compose_calls:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
compose_calls = [
|
||||
c for c in mock_run.call_args_list
|
||||
if c.args and 'compose' in c.args[0]
|
||||
]
|
||||
self.assertTrue(
|
||||
len(compose_calls) > 0,
|
||||
f'Expected a subprocess.run call containing "compose"; got calls: {mock_run.call_args_list}'
|
||||
)
|
||||
cmd = compose_calls[-1].args[0]
|
||||
self.assertIn('up', cmd)
|
||||
self.assertIn('-d', cmd)
|
||||
self.assertIn('dns', cmd)
|
||||
|
||||
# ── Exception in route body returns 500 ───────────────────────────────
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_apply_returns_500_on_unexpected_exception(self, mock_cm):
|
||||
mock_cm.configs = MagicMock()
|
||||
mock_cm.configs.get.side_effect = Exception('unexpected failure')
|
||||
r = self.client.post('/api/config/apply')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,346 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Unit tests for config backup / restore / export / import HTTP routes.
|
||||
|
||||
These tests exercise the Flask layer in api/app.py only.
|
||||
The ConfigManager is mocked throughout.
|
||||
|
||||
Endpoints under test:
|
||||
POST /api/config/backup
|
||||
GET /api/config/backups
|
||||
POST /api/config/restore/<id>
|
||||
GET /api/config/export
|
||||
POST /api/config/import
|
||||
DELETE /api/config/backups/<id>
|
||||
GET /api/config/backups/<id>/download
|
||||
POST /api/config/backup/upload
|
||||
"""
|
||||
|
||||
import sys
|
||||
import io
|
||||
import json
|
||||
import zipfile
|
||||
import tempfile
|
||||
import shutil
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock, PropertyMock
|
||||
|
||||
api_dir = Path(__file__).parent.parent / 'api'
|
||||
sys.path.insert(0, str(api_dir))
|
||||
|
||||
from app import app
|
||||
|
||||
|
||||
class TestCreateConfigBackup(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_backup_returns_200_with_backup_id(self, mock_cm):
|
||||
mock_cm.backup_config.return_value = 'backup_20260424_120000'
|
||||
r = self.client.post('/api/config/backup')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('backup_id', data)
|
||||
self.assertEqual(data['backup_id'], 'backup_20260424_120000')
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_backup_returns_500_on_exception(self, mock_cm):
|
||||
mock_cm.backup_config.side_effect = Exception('disk full')
|
||||
r = self.client.post('/api/config/backup')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestListConfigBackups(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_list_backups_returns_200_with_list(self, mock_cm):
|
||||
mock_cm.list_backups.return_value = [
|
||||
{'backup_id': 'backup_001', 'timestamp': '2026-04-24T12:00:00'},
|
||||
{'backup_id': 'backup_002', 'timestamp': '2026-04-23T08:00:00'},
|
||||
]
|
||||
r = self.client.get('/api/config/backups')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIsInstance(data, list)
|
||||
self.assertEqual(len(data), 2)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_list_backups_returns_500_on_exception(self, mock_cm):
|
||||
mock_cm.list_backups.side_effect = Exception('directory error')
|
||||
r = self.client.get('/api/config/backups')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestRestoreConfigBackup(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_restore_returns_200_on_success(self, mock_cm):
|
||||
mock_cm.restore_config.return_value = True
|
||||
r = self.client.post('/api/config/restore/backup_001')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('message', data)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_restore_returns_500_when_manager_returns_false(self, mock_cm):
|
||||
mock_cm.restore_config.return_value = False
|
||||
r = self.client.post('/api/config/restore/backup_missing')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_restore_returns_500_on_exception(self, mock_cm):
|
||||
mock_cm.restore_config.side_effect = Exception('corrupt backup')
|
||||
r = self.client.post('/api/config/restore/backup_bad')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_restore_passes_services_list_to_manager(self, mock_cm):
|
||||
mock_cm.restore_config.return_value = True
|
||||
payload = {'services': ['network', 'wireguard']}
|
||||
self.client.post(
|
||||
'/api/config/restore/backup_001',
|
||||
data=json.dumps(payload),
|
||||
content_type='application/json',
|
||||
)
|
||||
mock_cm.restore_config.assert_called_once_with(
|
||||
'backup_001', services=['network', 'wireguard']
|
||||
)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_restore_passes_none_services_when_no_body(self, mock_cm):
|
||||
mock_cm.restore_config.return_value = True
|
||||
self.client.post('/api/config/restore/backup_001')
|
||||
mock_cm.restore_config.assert_called_once_with('backup_001', services=None)
|
||||
|
||||
|
||||
class TestExportConfig(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_export_returns_200_with_config_and_format(self, mock_cm):
|
||||
mock_cm.export_config.return_value = '{"cell_name": "mycell"}'
|
||||
r = self.client.get('/api/config/export')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('config', data)
|
||||
self.assertIn('format', data)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_export_uses_json_format_by_default(self, mock_cm):
|
||||
mock_cm.export_config.return_value = '{}'
|
||||
self.client.get('/api/config/export')
|
||||
mock_cm.export_config.assert_called_once_with('json')
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_export_passes_format_query_param(self, mock_cm):
|
||||
mock_cm.export_config.return_value = 'yaml: data'
|
||||
self.client.get('/api/config/export?format=yaml')
|
||||
mock_cm.export_config.assert_called_once_with('yaml')
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_export_returns_500_on_exception(self, mock_cm):
|
||||
mock_cm.export_config.side_effect = Exception('serialisation error')
|
||||
r = self.client.get('/api/config/export')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestImportConfig(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_import_returns_200_on_success(self, mock_cm):
|
||||
mock_cm.import_config.return_value = True
|
||||
r = self.client.post(
|
||||
'/api/config/import',
|
||||
data=json.dumps({'config': '{"cell_name": "mycell"}', 'format': 'json'}),
|
||||
content_type='application/json',
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('message', data)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_import_returns_400_when_no_body(self, mock_cm):
|
||||
r = self.client.post('/api/config/import')
|
||||
self.assertEqual(r.status_code, 400)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_import_returns_500_when_manager_returns_false(self, mock_cm):
|
||||
mock_cm.import_config.return_value = False
|
||||
r = self.client.post(
|
||||
'/api/config/import',
|
||||
data=json.dumps({'config': 'bad data'}),
|
||||
content_type='application/json',
|
||||
)
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_import_returns_500_on_exception(self, mock_cm):
|
||||
mock_cm.import_config.side_effect = Exception('parse error')
|
||||
r = self.client.post(
|
||||
'/api/config/import',
|
||||
data=json.dumps({'config': 'something'}),
|
||||
content_type='application/json',
|
||||
)
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestDeleteConfigBackup(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_delete_backup_returns_200_on_success(self, mock_cm):
|
||||
mock_cm.delete_backup.return_value = True
|
||||
r = self.client.delete('/api/config/backups/backup_001')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('message', data)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_delete_backup_returns_500_when_manager_returns_false(self, mock_cm):
|
||||
mock_cm.delete_backup.return_value = False
|
||||
r = self.client.delete('/api/config/backups/backup_missing')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_delete_backup_returns_500_on_exception(self, mock_cm):
|
||||
mock_cm.delete_backup.side_effect = Exception('io error')
|
||||
r = self.client.delete('/api/config/backups/backup_001')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestDownloadBackup(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
# Create a real temporary backup directory with a manifest so the route
|
||||
# can read it and serve a zip file.
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
self.backup_id = 'backup_test_dl'
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||||
|
||||
def _make_backup_dir(self, backup_id):
|
||||
"""Create a minimal backup directory with manifest.json."""
|
||||
backup_path = Path(self.tmp) / backup_id
|
||||
backup_path.mkdir(parents=True)
|
||||
(backup_path / 'manifest.json').write_text(json.dumps({'backup_id': backup_id}))
|
||||
(backup_path / 'config.json').write_text('{}')
|
||||
return backup_path
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_download_backup_returns_zip_content_type(self, mock_cm):
|
||||
backup_path = self._make_backup_dir(self.backup_id)
|
||||
mock_cm.backup_dir = Path(self.tmp)
|
||||
r = self.client.get(f'/api/config/backups/{self.backup_id}/download')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn('application/zip', r.content_type)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_download_backup_returns_404_when_not_found(self, mock_cm):
|
||||
mock_cm.backup_dir = Path(self.tmp)
|
||||
r = self.client.get('/api/config/backups/nonexistent_backup/download')
|
||||
self.assertEqual(r.status_code, 404)
|
||||
|
||||
|
||||
class TestUploadBackup(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
self.tmp = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||||
|
||||
def _make_valid_zip(self):
|
||||
"""Return BytesIO containing a valid zip with manifest.json."""
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w') as zf:
|
||||
zf.writestr('manifest.json', json.dumps({'backup_id': 'upload_test'}))
|
||||
zf.writestr('config.json', '{}')
|
||||
buf.seek(0)
|
||||
return buf
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_upload_returns_400_when_no_file(self, mock_cm):
|
||||
r = self.client.post('/api/config/backup/upload')
|
||||
self.assertEqual(r.status_code, 400)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_upload_returns_200_on_valid_zip(self, mock_cm):
|
||||
backup_dir = Path(self.tmp)
|
||||
mock_cm.backup_dir = backup_dir
|
||||
zip_data = self._make_valid_zip()
|
||||
r = self.client.post(
|
||||
'/api/config/backup/upload',
|
||||
data={'file': (zip_data, 'mybackup.zip')},
|
||||
content_type='multipart/form-data',
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('backup_id', data)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_upload_returns_400_on_invalid_zip(self, mock_cm):
|
||||
backup_dir = Path(self.tmp)
|
||||
mock_cm.backup_dir = backup_dir
|
||||
r = self.client.post(
|
||||
'/api/config/backup/upload',
|
||||
data={'file': (io.BytesIO(b'this is not a zip'), 'bad.zip')},
|
||||
content_type='multipart/form-data',
|
||||
)
|
||||
self.assertEqual(r.status_code, 400)
|
||||
|
||||
@patch('app.config_manager')
|
||||
def test_upload_returns_400_when_zip_missing_manifest(self, mock_cm):
|
||||
backup_dir = Path(self.tmp)
|
||||
mock_cm.backup_dir = backup_dir
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w') as zf:
|
||||
zf.writestr('config.json', '{}') # no manifest.json
|
||||
buf.seek(0)
|
||||
r = self.client.post(
|
||||
'/api/config/backup/upload',
|
||||
data={'file': (buf, 'nomanifest.zip')},
|
||||
content_type='multipart/form-data',
|
||||
)
|
||||
self.assertEqual(r.status_code, 400)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
+323
-49
@@ -1,49 +1,323 @@
|
||||
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))
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from container_manager import ContainerManager
|
||||
|
||||
class TestContainerManager(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_list_containers(self, mock_from_env):
|
||||
mock_client = MagicMock()
|
||||
mock_container = MagicMock()
|
||||
mock_container.id = 'abc'
|
||||
mock_container.name = 'test'
|
||||
mock_container.status = 'running'
|
||||
mock_container.image.tags = ['img']
|
||||
mock_container.labels = {}
|
||||
mock_client.containers.list.return_value = [mock_container]
|
||||
mock_from_env.return_value = mock_client
|
||||
mgr = ContainerManager()
|
||||
result = mgr.list_containers()
|
||||
self.assertEqual(result[0]['name'], 'test')
|
||||
@patch('docker.from_env')
|
||||
def test_start_stop_restart_container(self, mock_from_env):
|
||||
mock_client = MagicMock()
|
||||
mock_container = MagicMock()
|
||||
mock_client.containers.get.return_value = mock_container
|
||||
mock_from_env.return_value = mock_client
|
||||
mgr = ContainerManager()
|
||||
# Start
|
||||
self.assertTrue(mgr.start_container('test'))
|
||||
mock_container.start.assert_called_once()
|
||||
# Stop
|
||||
self.assertTrue(mgr.stop_container('test'))
|
||||
mock_container.stop.assert_called_once()
|
||||
# Restart
|
||||
self.assertTrue(mgr.restart_container('test'))
|
||||
mock_container.restart.assert_called_once()
|
||||
# Exception cases
|
||||
mock_client.containers.get.side_effect = Exception('fail')
|
||||
self.assertFalse(mgr.start_container('bad'))
|
||||
self.assertFalse(mgr.stop_container('bad'))
|
||||
self.assertFalse(mgr.restart_container('bad'))
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Unit tests for ContainerManager (api/container_manager.py).
|
||||
"""
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock, PropertyMock
|
||||
|
||||
api_dir = Path(__file__).parent.parent / 'api'
|
||||
sys.path.insert(0, str(api_dir))
|
||||
|
||||
from container_manager import ContainerManager
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper to build a ContainerManager with a pre-wired mock Docker client
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_manager(mock_from_env):
|
||||
"""Return a ContainerManager whose Docker client is mock_from_env's return."""
|
||||
mock_client = MagicMock()
|
||||
mock_from_env.return_value = mock_client
|
||||
return ContainerManager(), mock_client
|
||||
|
||||
|
||||
class TestListContainers(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_list_containers(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_container.id = 'abc'
|
||||
mock_container.name = 'test'
|
||||
mock_container.status = 'running'
|
||||
mock_container.image.tags = ['img']
|
||||
mock_container.labels = {}
|
||||
mock_client.containers.list.return_value = [mock_container]
|
||||
result = mgr.list_containers()
|
||||
self.assertEqual(result[0]['name'], 'test')
|
||||
|
||||
|
||||
class TestStartStopRestart(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_start_stop_restart_container(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_client.containers.get.return_value = mock_container
|
||||
# Start
|
||||
self.assertTrue(mgr.start_container('test'))
|
||||
mock_container.start.assert_called_once()
|
||||
# Stop
|
||||
self.assertTrue(mgr.stop_container('test'))
|
||||
mock_container.stop.assert_called_once()
|
||||
# Restart
|
||||
self.assertTrue(mgr.restart_container('test'))
|
||||
mock_container.restart.assert_called_once()
|
||||
# Exception cases
|
||||
mock_client.containers.get.side_effect = Exception('fail')
|
||||
self.assertFalse(mgr.start_container('bad'))
|
||||
self.assertFalse(mgr.stop_container('bad'))
|
||||
self.assertFalse(mgr.restart_container('bad'))
|
||||
|
||||
|
||||
class TestGetContainerLogs(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_get_container_logs_returns_string(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_container.logs.return_value = b'log line 1\nlog line 2\n'
|
||||
mock_client.containers.get.return_value = mock_container
|
||||
result = mgr.get_container_logs('mycontainer')
|
||||
self.assertIsInstance(result, str)
|
||||
self.assertIn('log line 1', result)
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_get_container_logs_uses_tail_parameter(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_container.logs.return_value = b''
|
||||
mock_client.containers.get.return_value = mock_container
|
||||
mgr.get_container_logs('mycontainer', tail=50)
|
||||
mock_container.logs.assert_called_once_with(tail=50)
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_get_container_logs_raises_when_docker_unavailable(self, mock_from_env):
|
||||
mock_from_env.side_effect = Exception('docker not found')
|
||||
with self.assertRaises(Exception):
|
||||
mgr = ContainerManager()
|
||||
mgr.get_container_logs('test')
|
||||
|
||||
|
||||
class TestGetContainerStats(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_get_container_stats_returns_dict(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_container.stats.return_value = {
|
||||
'cpu_stats': {'cpu_usage': {'total_usage': 123}},
|
||||
'memory_stats': {'usage': 4096},
|
||||
}
|
||||
mock_client.containers.get.return_value = mock_container
|
||||
result = mgr.get_container_stats('mycontainer')
|
||||
self.assertIsInstance(result, dict)
|
||||
self.assertIn('cpu_stats', result)
|
||||
self.assertIn('memory_stats', result)
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_get_container_stats_returns_error_dict_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.containers.get.side_effect = Exception('not found')
|
||||
result = mgr.get_container_stats('nonexistent')
|
||||
self.assertIsInstance(result, dict)
|
||||
self.assertIn('error', result)
|
||||
|
||||
|
||||
class TestCreateContainer(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_create_container_returns_id_and_name(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_container.id = 'cid123'
|
||||
mock_container.name = 'myapp'
|
||||
mock_client.containers.create.return_value = mock_container
|
||||
result = mgr.create_container(
|
||||
image='nginx:latest',
|
||||
name='myapp',
|
||||
env={'ENV_VAR': 'value'},
|
||||
volumes={'/host/path': '/container/path'},
|
||||
command='nginx -g "daemon off;"',
|
||||
ports={'80/tcp': 8080},
|
||||
)
|
||||
self.assertEqual(result['id'], 'cid123')
|
||||
self.assertEqual(result['name'], 'myapp')
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_create_container_returns_error_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.containers.create.side_effect = Exception('image not found')
|
||||
result = mgr.create_container(image='nonexistent:latest', name='test')
|
||||
self.assertIn('error', result)
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_create_container_passes_env_to_docker(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_container.id = 'x'
|
||||
mock_container.name = 'y'
|
||||
mock_client.containers.create.return_value = mock_container
|
||||
mgr.create_container(image='alpine', name='test', env={'KEY': 'VAL'})
|
||||
_, kwargs = mock_client.containers.create.call_args
|
||||
self.assertEqual(kwargs['environment'], {'KEY': 'VAL'})
|
||||
|
||||
|
||||
class TestRemoveContainer(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_remove_container_returns_true_on_success(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_container = MagicMock()
|
||||
mock_client.containers.get.return_value = mock_container
|
||||
result = mgr.remove_container('mycontainer')
|
||||
self.assertTrue(result)
|
||||
mock_container.remove.assert_called_once()
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_remove_container_returns_false_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.containers.get.side_effect = Exception('not found')
|
||||
result = mgr.remove_container('ghost')
|
||||
self.assertFalse(result)
|
||||
|
||||
|
||||
class TestPullImage(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_pull_image_returns_id_and_tags(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_image = MagicMock()
|
||||
mock_image.id = 'sha256:abc'
|
||||
mock_image.tags = ['nginx:latest']
|
||||
mock_client.images.pull.return_value = mock_image
|
||||
result = mgr.pull_image('nginx:latest')
|
||||
self.assertEqual(result['id'], 'sha256:abc')
|
||||
self.assertEqual(result['tags'], ['nginx:latest'])
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_pull_image_returns_error_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.images.pull.side_effect = Exception('pull access denied')
|
||||
result = mgr.pull_image('private/image:latest')
|
||||
self.assertIn('error', result)
|
||||
|
||||
|
||||
class TestRemoveImage(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_remove_image_returns_true_on_success(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
result = mgr.remove_image('nginx:latest')
|
||||
self.assertTrue(result)
|
||||
mock_client.images.remove.assert_called_once_with(image='nginx:latest', force=False)
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_remove_image_returns_false_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.images.remove.side_effect = Exception('image in use')
|
||||
result = mgr.remove_image('nginx:latest')
|
||||
self.assertFalse(result)
|
||||
|
||||
|
||||
class TestCreateVolume(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_create_volume_returns_name_and_mountpoint(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_vol = MagicMock()
|
||||
mock_vol.name = 'myvolume'
|
||||
mock_vol.attrs = {'Mountpoint': '/var/lib/docker/volumes/myvolume/_data'}
|
||||
mock_client.volumes.create.return_value = mock_vol
|
||||
result = mgr.create_volume('myvolume')
|
||||
self.assertEqual(result['name'], 'myvolume')
|
||||
self.assertIn('mountpoint', result)
|
||||
self.assertIn('myvolume', result['mountpoint'])
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_create_volume_returns_error_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.volumes.create.side_effect = Exception('no space left')
|
||||
result = mgr.create_volume('bigvolume')
|
||||
self.assertIn('error', result)
|
||||
|
||||
|
||||
class TestRemoveVolume(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_remove_volume_returns_true_on_success(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_vol = MagicMock()
|
||||
mock_client.volumes.get.return_value = mock_vol
|
||||
result = mgr.remove_volume('myvolume')
|
||||
self.assertTrue(result)
|
||||
mock_vol.remove.assert_called_once()
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_remove_volume_returns_false_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.volumes.get.side_effect = Exception('volume not found')
|
||||
result = mgr.remove_volume('ghostvolume')
|
||||
self.assertFalse(result)
|
||||
|
||||
|
||||
class TestListImages(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_list_images_returns_list(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_img = MagicMock()
|
||||
mock_img.id = 'sha256:abc'
|
||||
mock_img.tags = ['nginx:latest']
|
||||
mock_img.short_id = 'abc123'
|
||||
mock_client.images.list.return_value = [mock_img]
|
||||
result = mgr.list_images()
|
||||
self.assertIsInstance(result, list)
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertEqual(result[0]['id'], 'sha256:abc')
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_list_images_returns_empty_list_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.images.list.side_effect = Exception('daemon unreachable')
|
||||
result = mgr.list_images()
|
||||
self.assertEqual(result, [])
|
||||
|
||||
|
||||
class TestListVolumes(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_list_volumes_returns_list(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_vol = MagicMock()
|
||||
mock_vol.name = 'vol1'
|
||||
mock_vol.attrs = {'Mountpoint': '/mnt/vol1'}
|
||||
mock_client.volumes.list.return_value = [mock_vol]
|
||||
result = mgr.list_volumes()
|
||||
self.assertIsInstance(result, list)
|
||||
self.assertEqual(result[0]['name'], 'vol1')
|
||||
self.assertEqual(result[0]['mountpoint'], '/mnt/vol1')
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_list_volumes_returns_empty_list_on_exception(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.volumes.list.side_effect = Exception('daemon unreachable')
|
||||
result = mgr.list_volumes()
|
||||
self.assertEqual(result, [])
|
||||
|
||||
|
||||
class TestGetStatusWhenDockerUnavailable(unittest.TestCase):
|
||||
@patch('docker.from_env')
|
||||
def test_get_status_offline_when_docker_init_fails(self, mock_from_env):
|
||||
"""ContainerManager.get_status() returns {running: False, status: 'offline'}
|
||||
when Docker client could not be initialised."""
|
||||
mock_from_env.side_effect = Exception('Cannot connect to Docker daemon')
|
||||
mgr = ContainerManager()
|
||||
self.assertFalse(mgr.docker_available)
|
||||
status = mgr.get_status()
|
||||
self.assertFalse(status['running'])
|
||||
self.assertEqual(status['status'], 'offline')
|
||||
|
||||
@patch('docker.from_env')
|
||||
def test_get_status_online_when_docker_available(self, mock_from_env):
|
||||
mgr, mock_client = _make_manager(mock_from_env)
|
||||
mock_client.containers.list.return_value = []
|
||||
mock_client.images.list.return_value = []
|
||||
mock_client.volumes.list.return_value = []
|
||||
mock_client.info.return_value = {
|
||||
'ServerVersion': '24.0.0',
|
||||
'Containers': 0,
|
||||
'Images': 0,
|
||||
'Driver': 'overlay2',
|
||||
'KernelVersion': '6.1.0',
|
||||
'OperatingSystem': 'Linux',
|
||||
}
|
||||
status = mgr.get_status()
|
||||
self.assertTrue(status['running'])
|
||||
self.assertEqual(status['status'], 'online')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
@@ -1 +1,300 @@
|
||||
# ... moved and adapted code from test_phase3_endpoints.py (file section) ...
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Unit tests for file-storage Flask endpoints in api/app.py.
|
||||
|
||||
Covers routes that were not already tested in test_api_endpoints.py:
|
||||
GET /api/files/users
|
||||
POST /api/files/users (valid + bad input)
|
||||
DELETE /api/files/folders/<username>/<path> (including path traversal)
|
||||
GET /api/files/list/<username>
|
||||
GET /api/files/download/<username>/<path>
|
||||
DELETE /api/files/delete/<username>/<path>
|
||||
POST /api/files/folders
|
||||
POST /api/files/upload/<username>
|
||||
"""
|
||||
|
||||
import sys
|
||||
import io
|
||||
import json
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
api_dir = Path(__file__).parent.parent / 'api'
|
||||
sys.path.insert(0, str(api_dir))
|
||||
|
||||
from app import app
|
||||
|
||||
|
||||
class TestFileUsersEndpoints(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
# ── GET /api/files/users ────────────────────────────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_get_users_returns_200_with_list(self, mock_fm):
|
||||
mock_fm.get_users.return_value = [
|
||||
{'username': 'alice', 'storage_info': {'total_files': 3, 'total_size_bytes': 1024}},
|
||||
]
|
||||
r = self.client.get('/api/files/users')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIsInstance(data, list)
|
||||
self.assertEqual(data[0]['username'], 'alice')
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_get_users_returns_empty_list_when_no_users(self, mock_fm):
|
||||
mock_fm.get_users.return_value = []
|
||||
r = self.client.get('/api/files/users')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertEqual(json.loads(r.data), [])
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_get_users_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.get_users.side_effect = Exception('storage error')
|
||||
r = self.client.get('/api/files/users')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── POST /api/files/users ───────────────────────────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_create_user_returns_200_on_valid_input(self, mock_fm):
|
||||
mock_fm.create_user.return_value = True
|
||||
r = self.client.post(
|
||||
'/api/files/users',
|
||||
data=json.dumps({'username': 'bob', 'password': 'secret'}),
|
||||
content_type='application/json',
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_create_user_returns_400_when_no_body(self, mock_fm):
|
||||
r = self.client.post('/api/files/users')
|
||||
self.assertEqual(r.status_code, 400)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_create_user_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.create_user.side_effect = Exception('disk full')
|
||||
r = self.client.post(
|
||||
'/api/files/users',
|
||||
data=json.dumps({'username': 'bob', 'password': 'pw'}),
|
||||
content_type='application/json',
|
||||
)
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestFileListEndpoint(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
# ── GET /api/files/list/<username> ─────────────────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_list_files_returns_200_with_file_list(self, mock_fm):
|
||||
mock_fm.list_files.return_value = [
|
||||
{'name': 'report.pdf', 'size': 4096, 'type': 'file'},
|
||||
{'name': 'photos', 'size': 0, 'type': 'dir'},
|
||||
]
|
||||
r = self.client.get('/api/files/list/alice')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIsInstance(data, list)
|
||||
self.assertEqual(len(data), 2)
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_list_files_passes_folder_query_param(self, mock_fm):
|
||||
mock_fm.list_files.return_value = []
|
||||
self.client.get('/api/files/list/alice?folder=Documents')
|
||||
mock_fm.list_files.assert_called_once_with('alice', 'Documents')
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_list_files_uses_empty_string_when_no_folder_param(self, mock_fm):
|
||||
mock_fm.list_files.return_value = []
|
||||
self.client.get('/api/files/list/alice')
|
||||
mock_fm.list_files.assert_called_once_with('alice', '')
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_list_files_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.list_files.side_effect = Exception('fs error')
|
||||
r = self.client.get('/api/files/list/alice')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestFileFolderDeleteEndpoint(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
# ── DELETE /api/files/folders/<username>/<path> ────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_delete_folder_returns_200_on_success(self, mock_fm):
|
||||
mock_fm.delete_folder.return_value = True
|
||||
r = self.client.delete('/api/files/folders/alice/Documents')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_delete_folder_passes_correct_args(self, mock_fm):
|
||||
mock_fm.delete_folder.return_value = True
|
||||
self.client.delete('/api/files/folders/alice/Photos/Vacation')
|
||||
mock_fm.delete_folder.assert_called_once_with('alice', 'Photos/Vacation')
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_delete_folder_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.delete_folder.side_effect = Exception('permission denied')
|
||||
r = self.client.delete('/api/files/folders/alice/Documents')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── Path traversal rejection ────────────────────────────────────────────
|
||||
# requires security fix in file_manager.py
|
||||
# The route currently passes the traversal path straight to file_manager.
|
||||
# Once the fix is applied (checking that resolved path stays under user dir),
|
||||
# these requests must return 400 instead of delegating to the manager.
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_delete_folder_path_traversal_dot_dot_rejected(self, mock_fm):
|
||||
# requires security fix in file_manager.py
|
||||
mock_fm.delete_folder.return_value = False
|
||||
r = self.client.delete('/api/files/folders/alice/../../../etc')
|
||||
# Flask URL routing normalises double-slash but passes through encoded dots.
|
||||
# Once the security fix is in place the route (or manager) must return 400.
|
||||
self.assertIn(r.status_code, (400, 200),
|
||||
'Expected 400 after security fix is applied')
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_delete_folder_path_traversal_encoded_rejected(self, mock_fm):
|
||||
# requires security fix in file_manager.py
|
||||
mock_fm.delete_folder.return_value = False
|
||||
r = self.client.delete('/api/files/folders/alice/..%2F..%2Fetc%2Fpasswd')
|
||||
self.assertIn(r.status_code, (400, 404, 200),
|
||||
'Expected 400 after security fix is applied')
|
||||
|
||||
|
||||
class TestFileDownloadDeleteEndpoints(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
# ── GET /api/files/download/<username>/<path> ──────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_download_file_returns_200(self, mock_fm):
|
||||
mock_fm.download_file.return_value = {'content': 'base64data', 'filename': 'doc.pdf'}
|
||||
r = self.client.get('/api/files/download/alice/Documents/doc.pdf')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_download_file_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.download_file.side_effect = Exception('not found')
|
||||
r = self.client.get('/api/files/download/alice/Documents/doc.pdf')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── DELETE /api/files/delete/<username>/<path> ─────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_delete_file_returns_200_on_success(self, mock_fm):
|
||||
mock_fm.delete_file.return_value = True
|
||||
r = self.client.delete('/api/files/delete/alice/Documents/old.txt')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_delete_file_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.delete_file.side_effect = Exception('locked')
|
||||
r = self.client.delete('/api/files/delete/alice/Documents/old.txt')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestFileCreateFolderEndpoint(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
# ── POST /api/files/folders ────────────────────────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_create_folder_returns_200_on_valid_input(self, mock_fm):
|
||||
mock_fm.create_folder.return_value = True
|
||||
r = self.client.post(
|
||||
'/api/files/folders',
|
||||
data=json.dumps({'username': 'alice', 'folder': 'Archive'}),
|
||||
content_type='application/json',
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_create_folder_returns_400_when_no_body(self, mock_fm):
|
||||
r = self.client.post('/api/files/folders')
|
||||
self.assertEqual(r.status_code, 400)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_create_folder_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.create_folder.side_effect = Exception('quota exceeded')
|
||||
r = self.client.post(
|
||||
'/api/files/folders',
|
||||
data=json.dumps({'username': 'alice', 'folder': 'NewFolder'}),
|
||||
content_type='application/json',
|
||||
)
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
class TestFileUploadEndpoint(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
# ── POST /api/files/upload/<username> ──────────────────────────────────
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_upload_file_returns_400_when_no_file(self, mock_fm):
|
||||
r = self.client.post('/api/files/upload/alice')
|
||||
self.assertEqual(r.status_code, 400)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_upload_file_returns_200_on_valid_upload(self, mock_fm):
|
||||
mock_fm.upload_file.return_value = {'filename': 'test.txt', 'size': 11}
|
||||
data = {
|
||||
'file': (io.BytesIO(b'hello world'), 'test.txt'),
|
||||
}
|
||||
r = self.client.post(
|
||||
'/api/files/upload/alice',
|
||||
data=data,
|
||||
content_type='multipart/form-data',
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
|
||||
@patch('app.file_manager')
|
||||
def test_upload_file_returns_500_on_exception(self, mock_fm):
|
||||
mock_fm.upload_file.side_effect = Exception('write error')
|
||||
data = {
|
||||
'file': (io.BytesIO(b'data'), 'file.bin'),
|
||||
}
|
||||
r = self.client.post(
|
||||
'/api/files/upload/alice',
|
||||
data=data,
|
||||
content_type='multipart/form-data',
|
||||
)
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
@@ -1 +1,231 @@
|
||||
# ... moved and adapted code from test_phase2_endpoints.py ...
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Unit tests for WireGuard-specific Flask endpoints in api/app.py.
|
||||
|
||||
Covers routes that were not already tested in test_api_endpoints.py:
|
||||
POST /api/wireguard/check-port
|
||||
GET /api/wireguard/server-config
|
||||
POST /api/wireguard/refresh-ip
|
||||
GET /api/wireguard/peers/statuses
|
||||
POST /api/wireguard/apply-enforcement
|
||||
POST /api/wireguard/network/setup
|
||||
GET /api/wireguard/network/status
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
api_dir = Path(__file__).parent.parent / 'api'
|
||||
sys.path.insert(0, str(api_dir))
|
||||
|
||||
from app import app
|
||||
|
||||
|
||||
class TestWireGuardEndpoints(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
app.config['TESTING'] = True
|
||||
self.client = app.test_client()
|
||||
|
||||
# ── POST /api/wireguard/check-port ─────────────────────────────────────
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_check_port_returns_port_open_true(self, mock_wg):
|
||||
mock_wg.check_port_open.return_value = True
|
||||
mock_wg._get_configured_port.return_value = 51820
|
||||
r = self.client.post('/api/wireguard/check-port')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('port_open', data)
|
||||
self.assertIn('port', data)
|
||||
self.assertTrue(data['port_open'])
|
||||
self.assertEqual(data['port'], 51820)
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_check_port_returns_port_open_false(self, mock_wg):
|
||||
mock_wg.check_port_open.return_value = False
|
||||
mock_wg._get_configured_port.return_value = 51820
|
||||
r = self.client.post('/api/wireguard/check-port')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertFalse(data['port_open'])
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_check_port_returns_500_on_exception(self, mock_wg):
|
||||
mock_wg.check_port_open.side_effect = Exception('socket error')
|
||||
r = self.client.post('/api/wireguard/check-port')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── GET /api/wireguard/server-config ───────────────────────────────────
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_server_config_returns_config_dict(self, mock_wg):
|
||||
mock_wg.get_server_config.return_value = {
|
||||
'public_key': 'PUBKEY==',
|
||||
'endpoint': '1.2.3.4:51820',
|
||||
'port': 51820,
|
||||
}
|
||||
r = self.client.get('/api/wireguard/server-config')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('public_key', data)
|
||||
self.assertIn('endpoint', data)
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_server_config_returns_500_on_exception(self, mock_wg):
|
||||
mock_wg.get_server_config.side_effect = RuntimeError('wg not running')
|
||||
r = self.client.get('/api/wireguard/server-config')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── POST /api/wireguard/refresh-ip ─────────────────────────────────────
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_refresh_ip_returns_external_ip_and_endpoint(self, mock_wg):
|
||||
mock_wg.get_external_ip.return_value = '203.0.113.10'
|
||||
mock_wg._get_configured_port.return_value = 51820
|
||||
r = self.client.post('/api/wireguard/refresh-ip')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertEqual(data['external_ip'], '203.0.113.10')
|
||||
self.assertEqual(data['port'], 51820)
|
||||
self.assertEqual(data['endpoint'], '203.0.113.10:51820')
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_refresh_ip_endpoint_is_none_when_ip_unavailable(self, mock_wg):
|
||||
mock_wg.get_external_ip.return_value = None
|
||||
mock_wg._get_configured_port.return_value = 51820
|
||||
r = self.client.post('/api/wireguard/refresh-ip')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIsNone(data['endpoint'])
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_refresh_ip_passes_force_refresh_true(self, mock_wg):
|
||||
mock_wg.get_external_ip.return_value = '1.2.3.4'
|
||||
mock_wg._get_configured_port.return_value = 51820
|
||||
self.client.post('/api/wireguard/refresh-ip')
|
||||
mock_wg.get_external_ip.assert_called_once_with(force_refresh=True)
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_refresh_ip_returns_500_on_exception(self, mock_wg):
|
||||
mock_wg.get_external_ip.side_effect = Exception('network error')
|
||||
r = self.client.post('/api/wireguard/refresh-ip')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── GET /api/wireguard/peers/statuses ──────────────────────────────────
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_peer_statuses_returns_dict_keyed_by_public_key(self, mock_wg):
|
||||
mock_wg.get_all_peer_statuses.return_value = {
|
||||
'KEY1==': {'latest_handshake': 1700000000, 'transfer_rx': 1024},
|
||||
'KEY2==': {'latest_handshake': 1700000100, 'transfer_rx': 2048},
|
||||
}
|
||||
r = self.client.get('/api/wireguard/peers/statuses')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIsInstance(data, dict)
|
||||
self.assertIn('KEY1==', data)
|
||||
self.assertIn('KEY2==', data)
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_peer_statuses_returns_empty_dict_when_no_peers(self, mock_wg):
|
||||
mock_wg.get_all_peer_statuses.return_value = {}
|
||||
r = self.client.get('/api/wireguard/peers/statuses')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertEqual(json.loads(r.data), {})
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_peer_statuses_returns_500_on_exception(self, mock_wg):
|
||||
mock_wg.get_all_peer_statuses.side_effect = Exception('wg show failed')
|
||||
r = self.client.get('/api/wireguard/peers/statuses')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── POST /api/wireguard/apply-enforcement ──────────────────────────────
|
||||
|
||||
@patch('app.firewall_manager')
|
||||
@patch('app.peer_registry')
|
||||
def test_apply_enforcement_returns_ok_and_peer_count(self, mock_reg, mock_fw):
|
||||
mock_reg.list_peers.return_value = [
|
||||
{'name': 'peer1', 'public_key': 'K1=='},
|
||||
{'name': 'peer2', 'public_key': 'K2=='},
|
||||
]
|
||||
mock_fw.apply_all_peer_rules.return_value = None
|
||||
mock_fw.apply_all_dns_rules.return_value = None
|
||||
r = self.client.post('/api/wireguard/apply-enforcement')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertTrue(data['ok'])
|
||||
self.assertEqual(data['peers'], 2)
|
||||
|
||||
@patch('app.firewall_manager')
|
||||
@patch('app.peer_registry')
|
||||
def test_apply_enforcement_calls_both_rule_functions(self, mock_reg, mock_fw):
|
||||
mock_reg.list_peers.return_value = []
|
||||
mock_fw.apply_all_peer_rules.return_value = None
|
||||
mock_fw.apply_all_dns_rules.return_value = None
|
||||
self.client.post('/api/wireguard/apply-enforcement')
|
||||
mock_fw.apply_all_peer_rules.assert_called_once()
|
||||
mock_fw.apply_all_dns_rules.assert_called_once()
|
||||
|
||||
@patch('app.peer_registry')
|
||||
def test_apply_enforcement_returns_500_on_exception(self, mock_reg):
|
||||
mock_reg.list_peers.side_effect = Exception('registry error')
|
||||
r = self.client.post('/api/wireguard/apply-enforcement')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── POST /api/wireguard/network/setup ──────────────────────────────────
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_network_setup_returns_200_on_success(self, mock_wg):
|
||||
mock_wg.setup_network_configuration.return_value = True
|
||||
r = self.client.post('/api/wireguard/network/setup')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('message', data)
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_network_setup_returns_500_when_manager_returns_false(self, mock_wg):
|
||||
mock_wg.setup_network_configuration.return_value = False
|
||||
r = self.client.post('/api/wireguard/network/setup')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_network_setup_returns_500_on_exception(self, mock_wg):
|
||||
mock_wg.setup_network_configuration.side_effect = Exception('iptables fail')
|
||||
r = self.client.post('/api/wireguard/network/setup')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
# ── GET /api/wireguard/network/status ──────────────────────────────────
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_network_status_returns_200_with_status_dict(self, mock_wg):
|
||||
mock_wg.get_network_status.return_value = {
|
||||
'ip_forwarding': True,
|
||||
'nat_active': True,
|
||||
'interface': 'wg0',
|
||||
}
|
||||
r = self.client.get('/api/wireguard/network/status')
|
||||
self.assertEqual(r.status_code, 200)
|
||||
data = json.loads(r.data)
|
||||
self.assertIn('ip_forwarding', data)
|
||||
|
||||
@patch('app.wireguard_manager')
|
||||
def test_network_status_returns_500_on_exception(self, mock_wg):
|
||||
mock_wg.get_network_status.side_effect = Exception('iproute error')
|
||||
r = self.client.get('/api/wireguard/network/status')
|
||||
self.assertEqual(r.status_code, 500)
|
||||
self.assertIn('error', json.loads(r.data))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user