roof
09138fbc18
Extract 9 route groups out of app.py into routes/ blueprints:
- routes/network.py — DNS, DHCP, NTP, network info/test (10 routes)
- routes/wireguard.py — WireGuard keys, peers, config, enforcement (18 routes)
- routes/cells.py — cell-to-cell connections (5 routes)
- routes/peers.py — peer CRUD + IP update + _next_peer_ip helper (10 routes)
- routes/routing.py — NAT, peer routes, firewall, iptables (17 routes)
- routes/vault.py — certs, trust, secrets (19 routes)
- routes/containers.py — containers, images, volumes (14 routes)
- routes/services.py — service bus, logs, services status/connectivity (18 routes)
- routes/peer_dashboard.py — peer-scoped dashboard/services (2 routes)
All blueprints use lazy `from app import X` inside route bodies to preserve
test patch compatibility (patch('app.email_manager', mock) still works).
Also included in this commit:
- A1 fix: backup/restore now includes email/calendar user files
- A2 fix: apply_config sets applying=True flag via helper container
- A3 fix: add_peer rolls back firewall on DNS failure
app.py reduced: 3011 → 1294 lines. 1021 tests passing.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
237 lines
9.2 KiB
Python
237 lines
9.2 KiB
Python
import logging
|
|
from flask import Blueprint, request, jsonify
|
|
logger = logging.getLogger('picell')
|
|
bp = Blueprint('wireguard', __name__)
|
|
|
|
@bp.route('/api/wireguard/keys', methods=['GET'])
|
|
def get_wireguard_keys():
|
|
try:
|
|
from app import wireguard_manager
|
|
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
|
|
|
|
@bp.route('/api/wireguard/keys/peer', methods=['POST'])
|
|
def generate_peer_keys():
|
|
try:
|
|
from app import wireguard_manager
|
|
data = request.get_json(silent=True) or {}
|
|
name = data.get('name') or data.get('peer_name')
|
|
if not name:
|
|
return jsonify({"error": "Missing peer name"}), 400
|
|
return jsonify(wireguard_manager.generate_peer_keys(name))
|
|
except Exception as e:
|
|
logger.error(f"Error generating peer keys: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/config', methods=['GET'])
|
|
def get_wireguard_config():
|
|
try:
|
|
from app import wireguard_manager
|
|
return jsonify(wireguard_manager.get_config())
|
|
except Exception as e:
|
|
logger.error(f"Error getting WireGuard config: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/peers', methods=['GET'])
|
|
def get_wireguard_peers():
|
|
try:
|
|
from app import wireguard_manager
|
|
return jsonify(wireguard_manager.get_peers())
|
|
except Exception as e:
|
|
logger.error(f"Error getting WireGuard peers: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/peers', methods=['POST'])
|
|
def add_wireguard_peer():
|
|
try:
|
|
from app import wireguard_manager
|
|
data = request.get_json(silent=True) or {}
|
|
result = wireguard_manager.add_peer(
|
|
name=data.get('name', ''),
|
|
public_key=data.get('public_key', ''),
|
|
endpoint_ip=data.get('endpoint', data.get('endpoint_ip', '')),
|
|
allowed_ips=data.get('allowed_ips', ''),
|
|
persistent_keepalive=data.get('persistent_keepalive', 25)
|
|
)
|
|
return jsonify({"success": result})
|
|
except Exception as e:
|
|
logger.error(f"Error adding WireGuard peer: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/peers', methods=['DELETE'])
|
|
def remove_wireguard_peer():
|
|
try:
|
|
from app import wireguard_manager
|
|
data = request.get_json(silent=True) or {}
|
|
public_key = data.get('public_key') or data.get('name', '')
|
|
return jsonify({"success": wireguard_manager.remove_peer(public_key)})
|
|
except Exception as e:
|
|
logger.error(f"Error removing WireGuard peer: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/status', methods=['GET'])
|
|
def get_wireguard_status():
|
|
try:
|
|
from app import wireguard_manager
|
|
return jsonify(wireguard_manager.get_status())
|
|
except Exception as e:
|
|
logger.error(f"Error getting WireGuard status: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/connectivity', methods=['POST'])
|
|
def test_wireguard_connectivity():
|
|
try:
|
|
from app import wireguard_manager
|
|
data = request.get_json(silent=True)
|
|
if data is None:
|
|
return jsonify({"error": "No data provided"}), 400
|
|
return jsonify(wireguard_manager.test_connectivity(data))
|
|
except Exception as e:
|
|
logger.error(f"Error testing WireGuard connectivity: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/peers/ip', methods=['PUT'])
|
|
def update_peer_ip():
|
|
try:
|
|
from app import wireguard_manager
|
|
data = request.get_json(silent=True) or {}
|
|
result = wireguard_manager.update_peer_ip(
|
|
data.get('public_key', data.get('peer', '')),
|
|
data.get('ip', '')
|
|
)
|
|
return jsonify({"success": result})
|
|
except Exception as e:
|
|
logger.error(f"Error updating peer IP: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/peers/status', methods=['POST'])
|
|
def get_peer_status():
|
|
try:
|
|
from app import wireguard_manager
|
|
data = request.get_json(silent=True) or {}
|
|
public_key = data.get('public_key', '')
|
|
if not public_key:
|
|
return jsonify({"error": "Missing public_key"}), 400
|
|
return jsonify(wireguard_manager.get_peer_status(public_key))
|
|
except Exception as e:
|
|
logger.error(f"Error getting peer status: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/peers/statuses', methods=['GET'])
|
|
def get_all_peer_statuses():
|
|
try:
|
|
from app import wireguard_manager
|
|
return jsonify(wireguard_manager.get_all_peer_statuses())
|
|
except Exception as e:
|
|
logger.error(f"Error getting peer statuses: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/network/setup', methods=['POST'])
|
|
def setup_network():
|
|
try:
|
|
from app import wireguard_manager
|
|
success = wireguard_manager.setup_network_configuration()
|
|
if success:
|
|
return jsonify({"message": "Network configuration setup completed successfully"})
|
|
return jsonify({"error": "Failed to setup network configuration"}), 500
|
|
except Exception as e:
|
|
logger.error(f"Error setting up network configuration: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/network/status', methods=['GET'])
|
|
def get_network_status():
|
|
try:
|
|
from app import wireguard_manager
|
|
return jsonify(wireguard_manager.get_network_status())
|
|
except Exception as e:
|
|
logger.error(f"Error getting network status: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/peers/config', methods=['POST'])
|
|
def get_peer_config():
|
|
try:
|
|
from app import wireguard_manager, peer_registry
|
|
data = request.get_json(silent=True) or {}
|
|
peer_name = data.get('name', data.get('peer', ''))
|
|
|
|
peer_ip = data.get('ip', '')
|
|
peer_private_key = data.get('private_key', '')
|
|
registered = peer_registry.get_peer(peer_name) if peer_name else {}
|
|
if peer_name and (not peer_ip or not peer_private_key):
|
|
if registered:
|
|
peer_ip = peer_ip or registered.get('ip', '')
|
|
peer_private_key = peer_private_key or registered.get('private_key', '')
|
|
|
|
server_endpoint = data.get('server_endpoint', '')
|
|
if not server_endpoint:
|
|
srv = wireguard_manager.get_server_config()
|
|
server_endpoint = srv.get('endpoint') or '<SERVER_IP>'
|
|
|
|
allowed_ips = data.get('allowed_ips') or None
|
|
if not allowed_ips and registered:
|
|
internet_access = registered.get('internet_access', True)
|
|
allowed_ips = wireguard_manager.FULL_TUNNEL_IPS if internet_access else wireguard_manager.get_split_tunnel_ips()
|
|
|
|
result = wireguard_manager.get_peer_config(
|
|
peer_name=peer_name,
|
|
peer_ip=peer_ip,
|
|
peer_private_key=peer_private_key,
|
|
server_endpoint=server_endpoint,
|
|
allowed_ips=allowed_ips,
|
|
)
|
|
return jsonify({"config": result})
|
|
except Exception as e:
|
|
logger.error(f"Error getting peer config: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/server-config', methods=['GET'])
|
|
def get_server_config():
|
|
try:
|
|
from app import wireguard_manager
|
|
return jsonify(wireguard_manager.get_server_config())
|
|
except Exception as e:
|
|
logger.error(f"Error getting server config: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/refresh-ip', methods=['GET', 'POST'])
|
|
def refresh_external_ip():
|
|
try:
|
|
from app import wireguard_manager
|
|
ip = wireguard_manager.get_external_ip(force_refresh=True)
|
|
port = wireguard_manager._get_configured_port()
|
|
return jsonify({
|
|
'external_ip': ip,
|
|
'port': port,
|
|
'endpoint': f'{ip}:{port}' if ip else None,
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Error refreshing external IP: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/apply-enforcement', methods=['POST'])
|
|
def apply_wireguard_enforcement():
|
|
try:
|
|
from app import peer_registry, firewall_manager, cell_link_manager, _configured_domain, COREFILE_PATH
|
|
peers = peer_registry.list_peers()
|
|
firewall_manager.apply_all_peer_rules(peers)
|
|
firewall_manager.apply_all_dns_rules(peers, COREFILE_PATH, _configured_domain(),
|
|
cell_links=cell_link_manager.list_connections())
|
|
return jsonify({'ok': True, 'peers': len(peers)})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@bp.route('/api/wireguard/check-port', methods=['GET', 'POST'])
|
|
def check_wireguard_port():
|
|
try:
|
|
from app import wireguard_manager
|
|
port_open = wireguard_manager.check_port_open()
|
|
return jsonify({'port_open': port_open, 'port': wireguard_manager._get_configured_port()})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|