feat: replace hardcoded service names with ServiceRegistry-driven Caddy and CoreDNS config
Unit Tests / test (push) Failing after 11s

Previously, CaddyManager and NetworkManager contained hardcoded lists of
service names (calendar, files, mail, webdav, etc.), meaning every new
service required a code change to appear in Caddy routes and DNS records.
Now both managers accept a service_registry parameter and derive their
service lists dynamically from the registry at runtime.

- CaddyManager: new _build_registry_service_routes() and
  _http01_service_pairs() methods pull routes from the registry
- NetworkManager: new _get_service_subdomains() method returns registry
  subdomains with a hardcoded fallback when no registry is wired in;
  _build_dns_records, stale-record detection, and service name sets all
  use the registry
- managers.py: service_registry constructed before network_manager so it
  can be injected into both CaddyManager and NetworkManager
- service_registry.py: validation chokepoint in get_caddy_routes() rejects
  invalid subdomain/backend values and reserved service names
- service_store_manager.py: _validate_manifest now validates top-level
  subdomain, backend, extra_subdomains, and extra_backends fields
- tests: 24 new tests covering registry-driven routing and DNS subdomain
  generation (test_caddy_registry_integration.py)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-28 18:27:52 -04:00
parent 63c0dfb9d9
commit 16fb362df7
12 changed files with 1312 additions and 46 deletions
+45
View File
@@ -125,6 +125,17 @@ def add_peer():
return jsonify({"error": f"Peer {peer_name} already exists"}), 400
peer_added_to_registry = True
# Store credentials only after the peer is committed — avoids orphaned
# credential entries if peer_registry.add_peer rejects a duplicate name.
try:
from app import account_manager
_svc_names = {'email', 'calendar', 'files'}
for svc in provisioned:
if svc in _svc_names:
account_manager.store_credentials(svc, peer_name, {'password': password})
except Exception as _am_err:
logger.warning(f"Peer {peer_name}: credential storage failed (non-fatal): {_am_err}")
firewall_manager.apply_peer_rules(peer_info['ip'], peer_info,
wg_subnet=_wg_subnet, cell_subnets=_cell_subnets)
firewall_applied = True
@@ -320,12 +331,46 @@ def remove_peer(peer_name):
_cleanup()
except Exception:
pass
try:
from app import account_manager
account_manager.deprovision_peer(peer_name)
except Exception as _am_err:
logger.warning(f"Peer {peer_name}: account_manager cleanup failed (non-fatal): {_am_err}")
return jsonify({"message": f"Peer {peer_name} removed successfully"})
except Exception as e:
logger.error(f"Error removing peer: {e}")
return jsonify({"error": str(e)}), 500
@bp.route('/api/peers/<peer_name>/service-credentials', methods=['GET'])
def get_peer_service_credentials(peer_name: str):
"""Return service credentials for a peer across all provisioned services (admin only).
Returns filled peer_config_template values for each service the peer is provisioned on.
Intended for an admin to view or copy credentials to share with the peer during
device setup. The global enforce_auth gate already restricts this to admin sessions.
Phase 2 note: a peer-self-service variant should live at /api/peer/service-credentials
(no path arg) and restrict to session['username'] to prevent cross-peer enumeration.
"""
try:
from app import peer_registry, account_manager, service_registry, config_manager
peer = peer_registry.get_peer(peer_name)
if not peer:
return jsonify({'error': f'Peer {peer_name!r} not found'}), 404
raw_creds = account_manager.get_all_credentials(peer_name)
identity = config_manager.get_identity()
domain = config_manager.get_effective_domain() or identity.get('domain', '')
result = {}
for service_id, cred in raw_creds.items():
svc_info = service_registry.get_peer_service_info(service_id, peer_name, domain, cred)
result[service_id] = svc_info if svc_info is not None else cred
return jsonify({'peer': peer_name, 'services': result})
except Exception as e:
logger.error('get_peer_service_credentials(%s): %s', peer_name, e)
return jsonify({'error': str(e)}), 500
@bp.route('/api/peers/register', methods=['POST'])
def register_peer():
try: