From f62965844a676d06f79f21562c8375c69720b8e6 Mon Sep 17 00:00:00 2001 From: aidankhogg Date: Tue, 30 Jun 2026 00:33:56 +0100 Subject: [PATCH] Complete services org admission consumer --- netengine/phases/phase_services.py | 265 +++++++++++++++++++++++++- tests/test_phase_services_consumer.py | 218 +++++++++++++++++++++ 2 files changed, 474 insertions(+), 9 deletions(-) create mode 100644 tests/test_phase_services_consumer.py diff --git a/netengine/phases/phase_services.py b/netengine/phases/phase_services.py index 74eab37..9257fef 100644 --- a/netengine/phases/phase_services.py +++ b/netengine/phases/phase_services.py @@ -9,11 +9,14 @@ """ import asyncio +import json +import secrets from datetime import UTC, datetime from typing import Any from netengine.events.emitter import emit_event from netengine.events.queues import Queue +from netengine.events.schema import EventEnvelope from netengine.handlers._base import BasePhaseHandler from netengine.handlers.context import PhaseContext from netengine.handlers.dns import DNSHandler @@ -88,7 +91,9 @@ async def execute(self, context: PhaseContext) -> None: # Deploy Storage service (MinIO) if spec.world_services.storage.enabled: logger.info("Deploying Storage service (MinIO)") - storage_handler = StorageHandler(context, docker, dns, pki, runtime_state) + storage_handler = StorageHandler( + context, docker, dns, pki, runtime_state + ) storage_output = await storage_handler.deploy_minio() services_output["storage"] = storage_output logger.info("Storage deployment complete") @@ -170,7 +175,9 @@ async def healthcheck(self, context: PhaseContext) -> bool: try: container = docker.client.containers.get(container_id) if container.status != "running": - logger.warning(f"Mail container not running: {container.status}") + logger.warning( + f"Mail container not running: {container.status}" + ) return False except Exception as e: logger.warning(f"Failed to check mail container: {e}") @@ -190,7 +197,9 @@ async def healthcheck(self, context: PhaseContext) -> bool: try: container = docker.client.containers.get(container_id) if container.status != "running": - logger.warning(f"Storage container not running: {container.status}") + logger.warning( + f"Storage container not running: {container.status}" + ) return False except Exception as e: logger.warning(f"Failed to check storage container: {e}") @@ -311,17 +320,252 @@ async def _consume_org_admission_events( await asyncio.sleep(1) continue - # Process org.admitted event - # (Implementation: create org-specific mail/storage if configured) - logger.debug("Processing org.admitted event for services provisioning") + try: + envelope = EventEnvelope(**json.loads(msg["message"])) + + if envelope.event_type != "org.admitted": + await context.pgmq_client.delete( + Queue.SERVICES_ADMISSIONS, msg["msg_id"] + ) + continue + + payload = envelope.payload + org_name = ( + payload.get("org_name") + or payload.get("org") + or payload.get("org_id") + ) + if not org_name: + logger.warning("org.admitted event missing org identifier") + await context.pgmq_client.delete( + Queue.SERVICES_ADMISSIONS, msg["msg_id"] + ) + continue + + service_config = ( + payload.get("service_config") + or payload.get("world_services") + or payload.get("services") + or {} + ) + if not isinstance(service_config, dict): + raise ValueError( + "org.admitted service configuration must be an object" + ) + + logger.info(f"Auto-provisioning world services for org: {org_name}") + result = await self._provision_org_services( + context, docker, dns, str(org_name), service_config + ) + await self._persist_org_service_provisioning( + context, str(org_name), result + ) + logger.info(f"Auto-provisioned world services for org {org_name}") + + await context.pgmq_client.delete( + Queue.SERVICES_ADMISSIONS, msg["msg_id"] + ) - # Mark message as processed - await context.pgmq_client.delete(Queue.SERVICES_ADMISSIONS, msg["msg_id"]) + except Exception as e: + logger.error( + f"Failed to process org admission event for services: {e}" + ) + try: + await context.pgmq_client.archive_to_dlq( + Queue.SERVICES_ADMISSIONS, msg["msg_id"], str(e) + ) + except Exception as dlq_err: + logger.error(f"Failed to archive to DLQ: {dlq_err}") except Exception as e: logger.error(f"Org admission consumer error: {e}") await asyncio.sleep(5) + async def _provision_org_services( + self, + context: PhaseContext, + docker: DockerHandler, + dns: DNSHandler, + org_name: str, + service_config: dict[str, Any], + ) -> dict[str, Any]: + """Provision mail and storage resources for a newly admitted org.""" + result: dict[str, Any] = { + "org_name": org_name, + "provisioned_at": datetime.now(UTC).isoformat(), + } + spec = context.spec + + mail_enabled = self._service_enabled( + service_config, "mail", spec.world_services.mail.enabled + ) + if mail_enabled: + mail_handler = MailHandler(context, docker, dns) + _private_key, public_key = await mail_handler._generate_dkim_keys() + mail_domain = await self._provision_org_mail_domain( + context, dns, mail_handler, org_name, public_key + ) + result["mail"] = mail_domain + + storage_enabled = self._service_enabled( + service_config, "storage", spec.world_services.storage.enabled + ) + if storage_enabled: + pki = PKIHandler(docker, context.runtime_state, spec) + storage_handler = StorageHandler( + context, docker, dns, pki, context.runtime_state + ) + storage_result = await self._provision_org_storage( + context, dns, storage_handler, org_name + ) + result["storage"] = storage_result + + if "mail" not in result and "storage" not in result: + result["skipped"] = "mail and storage disabled" + + return result + + @staticmethod + def _service_enabled( + service_config: dict[str, Any], service_name: str, default: bool + ) -> bool: + """Read per-service enabled overrides from org admission payloads.""" + value = service_config.get(service_name) + if isinstance(value, dict): + return bool(value.get("enabled", default)) + if value is None: + return default + return bool(value) + + async def _provision_org_mail_domain( + self, + context: PhaseContext, + dns: DNSHandler, + mail_handler: MailHandler, + org_name: str, + dkim_public_pem: str, + ) -> dict[str, Any]: + """Create DNS records for an org-specific mail domain.""" + org_domain = f"{org_name}.internal" + mail_config = context.spec.world_services.mail + records: list[dict[str, Any]] = [] + + async def add(record_type: str, name: str, value: str) -> None: + await dns.add_zone_record( + context, "internal", record_type, name, value, 300 + ) + records.append({"type": record_type, "name": name, "value": value}) + + await add("TXT", org_domain, mail_config.mailbox_policy.spf_default) + if mail_config.dkim.enabled: + await add( + "TXT", + f"_dkim._domainkey.{org_domain}", + mail_handler._dkim_txt_value(dkim_public_pem), + ) + if mail_config.dmarc.enabled: + await add( + "TXT", f"_dmarc.{org_domain}", mail_config.mailbox_policy.dmarc_default + ) + await add("MX", org_domain, f"10 {mail_config.canonical_name}.") + + return {"domain": org_domain, "records": records} + + async def _provision_org_storage( + self, + context: PhaseContext, + dns: DNSHandler, + storage_handler: StorageHandler, + org_name: str, + ) -> dict[str, Any]: + """Create an org storage bucket, credentials, DNS record, and certificate.""" + bucket = f"{org_name}-data" + access_key = secrets.token_urlsafe(16) + secret_key = secrets.token_urlsafe(32) + service_name = f"storage-{org_name}" + fqdn = f"{service_name}.platform.internal" + + cert, _key = await storage_handler.pki.issue_cert(fqdn, []) + expiry = storage_handler.pki.extract_cert_expiry(cert) + context.runtime_state.issued_certificates[fqdn] = { + "cert_type": "storage", + "issued_at": datetime.now(UTC).isoformat(), + "expires_at": expiry.isoformat(), + "sans": [], + "rotated_at": None, + "version": 1, + } + + root_access = ( + getattr(context.runtime_state, "minio_access_key", None) or access_key + ) + root_secret = ( + getattr(context.runtime_state, "minio_secret_key", None) or secret_key + ) + await storage_handler._create_bucket(bucket, root_access, root_secret) + await dns.add_zone_record( + context, + "platform.internal", + "CNAME", + service_name, + context.spec.world_services.storage.canonical_name, + 300, + ) + + return { + "bucket": bucket, + "access_key": access_key, + "secret_key": secret_key, + "dns": fqdn, + "certificate": {"common_name": fqdn, "expires_at": expiry.isoformat()}, + } + + async def _persist_org_service_provisioning( + self, context: PhaseContext, org_name: str, result: dict[str, Any] + ) -> None: + """Persist org service provisioning results in RuntimeState and Supabase tables.""" + state = context.runtime_state + output = dict(state.world_services_output or {}) + orgs = dict(output.get("orgs") or {}) + orgs[org_name] = result + output["orgs"] = orgs + state.world_services_output = output + state.save() + + try: + from netengine.core.supabase_client import get_db + + db = await get_db() + if "mail" in result: + await ( + db.table("mail_domains") + .upsert( + { + "org_name": org_name, + **result["mail"], + "created_at": result["provisioned_at"], + } + ) + .execute() + ) + if "storage" in result: + await ( + db.table("storage_buckets") + .upsert( + { + "org_name": org_name, + **result["storage"], + "created_at": result["provisioned_at"], + } + ) + .execute() + ) + await db.table("service_provisioning").upsert(result).execute() + except Exception as e: + context.logger.warning( + f"Failed to persist org service provisioning to DB: {e}" + ) + async def _emit_event( self, context: PhaseContext, @@ -336,5 +580,8 @@ async def _emit_event( payload: Event payload dict """ await emit_event( - context, event_type=event_type, emitted_by="services_handler", payload=payload + context, + event_type=event_type, + emitted_by="services_handler", + payload=payload, ) diff --git a/tests/test_phase_services_consumer.py b/tests/test_phase_services_consumer.py new file mode 100644 index 0000000..b3d7638 --- /dev/null +++ b/tests/test_phase_services_consumer.py @@ -0,0 +1,218 @@ +"""Tests for Phase 8 org-admission service provisioning consumer.""" + +import asyncio +import json +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from netengine.events.factory import org_admitted +from netengine.events.queues import Queue +from netengine.events.schema import EventEnvelope +from netengine.phases.phase_services import ServicesPhaseHandler + + +async def _stop_consumer(*_args, **_kwargs): + raise asyncio.CancelledError + + +def _message(envelope: EventEnvelope, msg_id: int = 42) -> dict: + return {"msg_id": msg_id, "message": json.dumps(envelope.to_dict())} + + +async def test_services_consumer_provisions_and_deletes_on_success( + phase_context_with_mocks, monkeypatch +): + context = phase_context_with_mocks + handler = ServicesPhaseHandler() + event = org_admitted(org_name="acme", capabilities=[], and_profile="business") + event.payload["service_config"] = { + "mail": {"enabled": True}, + "storage": {"enabled": True}, + } + context.pgmq_client.receive = AsyncMock(side_effect=[_message(event), None]) + + provision = AsyncMock(return_value={"org_name": "acme", "mail": {}, "storage": {}}) + persist = AsyncMock() + monkeypatch.setattr(handler, "_provision_org_services", provision) + monkeypatch.setattr(handler, "_persist_org_service_provisioning", persist) + monkeypatch.setattr(asyncio, "sleep", _stop_consumer) + + with pytest.raises(asyncio.CancelledError): + await handler._consume_org_admission_events(context, AsyncMock(), AsyncMock()) + + provision.assert_awaited_once() + persist.assert_awaited_once() + context.pgmq_client.delete.assert_awaited_once_with(Queue.SERVICES_ADMISSIONS, 42) + context.pgmq_client.archive_to_dlq.assert_not_awaited() + + +async def test_services_consumer_ignores_non_admission_events( + phase_context_with_mocks, monkeypatch +): + context = phase_context_with_mocks + handler = ServicesPhaseHandler() + event = EventEnvelope.create("org.updated", "test", {"org_name": "acme"}) + context.pgmq_client.receive = AsyncMock(side_effect=[_message(event, 7), None]) + provision = AsyncMock() + monkeypatch.setattr(handler, "_provision_org_services", provision) + monkeypatch.setattr(asyncio, "sleep", _stop_consumer) + + with pytest.raises(asyncio.CancelledError): + await handler._consume_org_admission_events(context, AsyncMock(), AsyncMock()) + + provision.assert_not_awaited() + context.pgmq_client.delete.assert_awaited_once_with(Queue.SERVICES_ADMISSIONS, 7) + context.pgmq_client.archive_to_dlq.assert_not_awaited() + + +async def test_services_consumer_archives_malformed_messages( + phase_context_with_mocks, monkeypatch +): + context = phase_context_with_mocks + handler = ServicesPhaseHandler() + context.pgmq_client.receive = AsyncMock( + side_effect=[{"msg_id": 99, "message": "not-json"}, None] + ) + monkeypatch.setattr(asyncio, "sleep", _stop_consumer) + + with pytest.raises(asyncio.CancelledError): + await handler._consume_org_admission_events(context, AsyncMock(), AsyncMock()) + + context.pgmq_client.delete.assert_not_awaited() + context.pgmq_client.archive_to_dlq.assert_awaited_once() + assert ( + context.pgmq_client.archive_to_dlq.await_args.args[0] + == Queue.SERVICES_ADMISSIONS + ) + assert context.pgmq_client.archive_to_dlq.await_args.args[1] == 99 + + +async def test_services_consumer_archives_provisioning_failure( + phase_context_with_mocks, monkeypatch +): + context = phase_context_with_mocks + handler = ServicesPhaseHandler() + event = org_admitted(org_name="acme", capabilities=[], and_profile="business") + context.pgmq_client.receive = AsyncMock(side_effect=[_message(event, 12), None]) + monkeypatch.setattr( + handler, "_provision_org_services", AsyncMock(side_effect=RuntimeError("boom")) + ) + monkeypatch.setattr(asyncio, "sleep", _stop_consumer) + + with pytest.raises(asyncio.CancelledError): + await handler._consume_org_admission_events(context, AsyncMock(), AsyncMock()) + + context.pgmq_client.delete.assert_not_awaited() + context.pgmq_client.archive_to_dlq.assert_awaited_once_with( + Queue.SERVICES_ADMISSIONS, 12, "boom" + ) + + +async def test_services_consumer_continues_when_dlq_archiving_fails( + phase_context_with_mocks, monkeypatch +): + context = phase_context_with_mocks + handler = ServicesPhaseHandler() + event = org_admitted(org_name="acme", capabilities=[], and_profile="business") + context.pgmq_client.receive = AsyncMock(side_effect=[_message(event, 13), None]) + context.pgmq_client.archive_to_dlq = AsyncMock(side_effect=RuntimeError("dlq down")) + monkeypatch.setattr( + handler, "_provision_org_services", AsyncMock(side_effect=RuntimeError("boom")) + ) + monkeypatch.setattr(asyncio, "sleep", _stop_consumer) + + with pytest.raises(asyncio.CancelledError): + await handler._consume_org_admission_events(context, AsyncMock(), AsyncMock()) + + context.pgmq_client.archive_to_dlq.assert_awaited_once_with( + Queue.SERVICES_ADMISSIONS, 13, "boom" + ) + + +async def test_provision_org_services_calls_mail_storage_handlers( + phase_context_with_mocks, monkeypatch +): + context = phase_context_with_mocks + handler = ServicesPhaseHandler() + dns = AsyncMock() + + monkeypatch.setattr( + "netengine.phases.phase_services.MailHandler._generate_dkim_keys", + AsyncMock(return_value=("private", "public")), + ) + monkeypatch.setattr( + "netengine.phases.phase_services.PKIHandler.issue_cert", + AsyncMock(return_value=("cert", "key")), + ) + monkeypatch.setattr( + "netengine.phases.phase_services.PKIHandler.extract_cert_expiry", + lambda _self, _cert: datetime.now(UTC) + timedelta(days=30), + ) + monkeypatch.setattr( + "netengine.phases.phase_services.StorageHandler._create_bucket", + AsyncMock(), + ) + + result = await handler._provision_org_services( + context, + AsyncMock(), + dns, + "acme", + {"mail": {"enabled": True}, "storage": {"enabled": True}}, + ) + + assert result["mail"]["domain"] == "acme.internal" + assert result["storage"]["bucket"] == "acme-data" + assert "storage-acme.platform.internal" in context.runtime_state.issued_certificates + assert dns.add_zone_record.await_count >= 5 + + +async def test_persist_org_service_provisioning_updates_runtime_and_db( + phase_context_with_mocks, monkeypatch +): + context = phase_context_with_mocks + handler = ServicesPhaseHandler() + calls = [] + + class Query: + def __init__(self, table): + self.table = table + + def upsert(self, data): + calls.append((self.table, data)) + return self + + async def execute(self): + return SimpleNamespace(data=[]) + + class DB: + def table(self, name): + return Query(name) + + monkeypatch.setattr( + "netengine.core.supabase_client.get_db", AsyncMock(return_value=DB()) + ) + + await handler._persist_org_service_provisioning( + context, + "acme", + { + "org_name": "acme", + "provisioned_at": "2026-01-01T00:00:00+00:00", + "mail": {"domain": "acme.internal"}, + "storage": {"bucket": "acme-data"}, + }, + ) + + assert ( + context.runtime_state.world_services_output["orgs"]["acme"]["storage"]["bucket"] + == "acme-data" + ) + assert [name for name, _ in calls] == [ + "mail_domains", + "storage_buckets", + "service_provisioning", + ]