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>
166 lines
6.1 KiB
Python
166 lines
6.1 KiB
Python
import logging
|
|
import os
|
|
from flask import Blueprint, request, jsonify, current_app
|
|
logger = logging.getLogger('picell')
|
|
bp = Blueprint('vault', __name__)
|
|
|
|
@bp.route('/api/vault/status', methods=['GET'])
|
|
def get_vault_status():
|
|
try:
|
|
return jsonify(current_app.vault_manager.get_status())
|
|
except Exception as e:
|
|
logger.error(f"Error getting vault status: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/certificates', methods=['GET'])
|
|
def get_certificates():
|
|
try:
|
|
return jsonify(current_app.vault_manager.list_certificates())
|
|
except Exception as e:
|
|
logger.error(f"Error getting certificates: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/certificates', methods=['POST'])
|
|
def generate_certificate():
|
|
try:
|
|
data = request.get_json(silent=True)
|
|
if data is None:
|
|
return jsonify({"error": "No data provided"}), 400
|
|
result = current_app.vault_manager.generate_certificate(
|
|
common_name=data['common_name'],
|
|
domains=data.get('domains', []),
|
|
key_size=data.get('key_size', 2048),
|
|
days=data.get('days', 365)
|
|
)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
logger.error(f"Error generating certificate: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/certificates/<common_name>', methods=['DELETE'])
|
|
def revoke_certificate(common_name):
|
|
try:
|
|
return jsonify({"revoked": current_app.vault_manager.revoke_certificate(common_name)})
|
|
except Exception as e:
|
|
logger.error(f"Error revoking certificate: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/ca/certificate', methods=['GET'])
|
|
def get_ca_certificate():
|
|
try:
|
|
return jsonify({"certificate": current_app.vault_manager.get_ca_certificate()})
|
|
except Exception as e:
|
|
logger.error(f"Error getting CA certificate: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/age/public-key', methods=['GET'])
|
|
def get_age_public_key():
|
|
try:
|
|
return jsonify({"public_key": current_app.vault_manager.get_age_public_key()})
|
|
except Exception as e:
|
|
logger.error(f"Error getting Age public key: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/trust/keys', methods=['GET'])
|
|
def get_trusted_keys():
|
|
try:
|
|
return jsonify(current_app.vault_manager.get_trusted_keys())
|
|
except Exception as e:
|
|
logger.error(f"Error getting trusted keys: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/trust/keys', methods=['POST'])
|
|
def add_trusted_key():
|
|
try:
|
|
data = request.get_json(silent=True)
|
|
if data is None:
|
|
return jsonify({"error": "No data provided"}), 400
|
|
result = current_app.vault_manager.add_trusted_key(
|
|
name=data['name'],
|
|
public_key=data['public_key'],
|
|
trust_level=data.get('trust_level', 'direct')
|
|
)
|
|
return jsonify({"added": result})
|
|
except Exception as e:
|
|
logger.error(f"Error adding trusted key: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/trust/keys/<name>', methods=['DELETE'])
|
|
def remove_trusted_key(name):
|
|
try:
|
|
return jsonify({"removed": current_app.vault_manager.remove_trusted_key(name)})
|
|
except Exception as e:
|
|
logger.error(f"Error removing trusted key: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/trust/verify', methods=['POST'])
|
|
def verify_trust_chain():
|
|
try:
|
|
data = request.get_json(silent=True)
|
|
if data is None:
|
|
return jsonify({"error": "No data provided"}), 400
|
|
result = current_app.vault_manager.verify_trust_chain(
|
|
peer_name=data['peer_name'],
|
|
signature=data['signature'],
|
|
data=data['data']
|
|
)
|
|
return jsonify({"verified": result})
|
|
except Exception as e:
|
|
logger.error(f"Error verifying trust chain: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/trust/chains', methods=['GET'])
|
|
def get_trust_chains():
|
|
try:
|
|
return jsonify(current_app.vault_manager.get_trust_chains())
|
|
except Exception as e:
|
|
logger.error(f"Error getting trust chains: {e}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
@bp.route('/api/vault/secrets', methods=['GET'])
|
|
def list_secrets():
|
|
try:
|
|
from app import is_local_request
|
|
if not is_local_request():
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
return jsonify({'secrets': current_app.vault_manager.list_secrets()})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@bp.route('/api/vault/secrets', methods=['POST'])
|
|
def store_secret():
|
|
try:
|
|
from app import is_local_request
|
|
if not is_local_request():
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
data = request.get_json(silent=True)
|
|
if not data or 'name' not in data or 'value' not in data:
|
|
return jsonify({'error': 'Missing name or value'}), 400
|
|
current_app.vault_manager.store_secret(data['name'], data['value'])
|
|
return jsonify({'stored': True})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@bp.route('/api/vault/secrets/<name>', methods=['GET'])
|
|
def get_secret(name):
|
|
try:
|
|
from app import is_local_request
|
|
if not is_local_request():
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
value = current_app.vault_manager.get_secret(name)
|
|
if value is None:
|
|
return jsonify({'error': 'Not found'}), 404
|
|
return jsonify({'name': name, 'value': value})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@bp.route('/api/vault/secrets/<name>', methods=['DELETE'])
|
|
def delete_secret(name):
|
|
try:
|
|
from app import is_local_request
|
|
if not is_local_request():
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
return jsonify({'deleted': current_app.vault_manager.delete_secret(name)})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|