diff --git a/netengine/core/state.py b/netengine/core/state.py
index d103efa..2adf958 100644
--- a/netengine/core/state.py
+++ b/netengine/core/state.py
@@ -71,6 +71,13 @@ class RuntimeState:
     issued_certificates: Dict[str, Dict[str, Any]] = field(default_factory=dict)
     pki_rotation_state: Dict[str, Any] = field(default_factory=dict)
 
+    # Extended PKI state
+    intermediate_ca_cert: Optional[str] = None
+    dnssec_output: Optional[Dict[str, Any]] = None
+
+    # Gateway portal state
+    gateway_portal_output: Optional[Dict[str, Any]] = None
+
     @classmethod
     def load(cls) -> "RuntimeState":
         state_file = get_state_file()
diff --git a/netengine/handlers/gateway_handler.py b/netengine/handlers/gateway_handler.py
index 4dac90a..368ad89 100644
--- a/netengine/handlers/gateway_handler.py
+++ b/netengine/handlers/gateway_handler.py
@@ -1,10 +1,13 @@
 import os
 import tempfile
-from typing import Any
+from typing import TYPE_CHECKING, Any, Optional
 
 from netengine.errors import GatewayError
 from netengine.gateways.base import BaseGatewayHandler
 
+if TYPE_CHECKING:
+    from netengine.spec.models import RealInternetConfig
+
 
 class GatewayHandler(BaseGatewayHandler):
     def __init__(self, docker: Any) -> None:
@@ -116,3 +119,199 @@ async def reload(self) -> None:
         exit_code, output = await self.docker.exec_command(self.gateway_container, cmd)
         if exit_code != 0:
             raise GatewayError(f"Gateway nftables reload failed: {output}")
+
+    # ─────────────────────────────────────────────
+    # Real Internet Gateway Policy
+    # ─────────────────────────────────────────────
+
+    async def apply_internet_policy(self, config: "RealInternetConfig") -> None:
+        """Apply real internet access policy rules to the gateway container.
+
+        Generates and loads an nftables ruleset that enforces the mode declared
+        in *config*.  CUSTOM mode is a no-op (operator manages rules directly).
+        """
+        from netengine.spec.types import GatewayRealInternetMode
+
+        if config.mode == GatewayRealInternetMode.CUSTOM:
+            return
+
+        rules = self._internet_rules(config)
+        dest_path = "/etc/nftables/rules/internet.nft"
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".nft", delete=False) as f:
+            f.write(rules)
+            tmp_path = f.name
+        try:
+            await self.docker.copy_to_container(self.gateway_container, tmp_path, dest_path)
+        finally:
+            os.unlink(tmp_path)
+
+        exit_code, output = await self.docker.exec_command(
+            self.gateway_container, ["nft", "-f", dest_path]
+        )
+        if exit_code != 0:
+            raise GatewayError(f"Failed to apply internet policy ({config.mode.value}): {output}")
+
+    async def remove_internet_policy(self) -> None:
+        """Remove internet policy rules (reverts to isolated state)."""
+        exit_code, output = await self.docker.exec_command(
+            self.gateway_container,
+            ["nft", "delete", "table", "inet", "netengine_internet"],
+        )
+        if exit_code != 0 and "No such table" not in output:
+            raise GatewayError(f"Failed to remove internet policy rules: {output}")
+        await self.docker.exec_command(
+            self.gateway_container,
+            ["rm", "-f", "/etc/nftables/rules/internet.nft"],
+        )
+
+    def _internet_rules(self, config: "RealInternetConfig") -> str:
+        """Dispatch to the correct rule generator for *config.mode*."""
+        from netengine.spec.types import GatewayRealInternetMode
+
+        dispatch = {
+            GatewayRealInternetMode.ISOLATED: self._isolated_internet_rules,
+            GatewayRealInternetMode.SHADOWED: self._shadowed_internet_rules,
+            GatewayRealInternetMode.MIRRORED: lambda: self._mirrored_internet_rules(config),
+            GatewayRealInternetMode.EXPOSED: self._exposed_internet_rules,
+        }
+        return dispatch[config.mode]()
+
+    def _isolated_internet_rules(self) -> str:
+        """Block all WAN ingress/egress; pass only internal traffic."""
+        return """\
+table inet netengine_internet {
+    chain input {
+        type filter hook input priority 0; policy drop;
+        ct state established,related accept
+        iifname "lo" accept
+        iifname != "eth_wan" accept
+    }
+    chain output {
+        type filter hook output priority 0; policy drop;
+        ct state established,related accept
+        oifname "lo" accept
+        oifname != "eth_wan" accept
+    }
+    chain forward {
+        type filter hook forward priority 0; policy drop;
+        iifname "eth_wan" drop
+        oifname "eth_wan" drop
+    }
+}
+"""
+
+    def _shadowed_internet_rules(self) -> str:
+        """Outbound HTTPS only (read-only shadow); all inbound WAN blocked."""
+        return """\
+table inet netengine_internet {
+    chain input {
+        type filter hook input priority 0; policy drop;
+        ct state established,related accept
+        iifname "lo" accept
+        iifname != "eth_wan" accept
+    }
+    chain forward {
+        type filter hook forward priority 0; policy drop;
+        ct state established,related accept
+        iifname != "eth_wan" oifname "eth_wan" tcp dport { 80, 443 } ct state new accept
+        iifname "eth_wan" drop
+    }
+    chain postrouting {
+        type nat hook postrouting priority 100; policy accept;
+        oifname "eth_wan" masquerade
+    }
+}
+"""
+
+    def _mirrored_internet_rules(self, config: "RealInternetConfig") -> str:
+        """Allow outbound to configured service mirrors; block all other WAN."""
+        mirror_accepts = ""
+        for mirror in config.service_mirrors:
+            # Allow traffic destined for the in-world service counterpart
+            mirror_accepts += (
+                f"\n        ip daddr {mirror.in_world_service} tcp dport {{ 80, 443 }}"
+                " ct state new accept"
+            )
+
+        return f"""\
+table inet netengine_internet {{
+    chain forward {{
+        type filter hook forward priority 0; policy drop;
+        ct state established,related accept{mirror_accepts}
+        iifname "eth_wan" drop
+    }}
+    chain postrouting {{
+        type nat hook postrouting priority 100; policy accept;
+        oifname "eth_wan" masquerade
+    }}
+}}
+"""
+
+    def _exposed_internet_rules(self) -> str:
+        """Full internet access with stateful inbound filtering."""
+        return """\
+table inet netengine_internet {
+    chain input {
+        type filter hook input priority 0; policy drop;
+        ct state established,related accept
+        iifname "lo" accept
+        tcp dport { 80, 443 } ct state new accept
+    }
+    chain forward {
+        type filter hook forward priority 0; policy accept;
+        ct state established,related accept
+    }
+    chain postrouting {
+        type nat hook postrouting priority 100; policy accept;
+        oifname "eth_wan" masquerade
+    }
+}
+"""
+
+    # ─────────────────────────────────────────────
+    # Cross-world peer routing
+    # ─────────────────────────────────────────────
+
+    async def apply_peer_routing(self, peer_name: str, peer_endpoint_ip: str) -> None:
+        """Allow forwarded traffic to/from a cross-world peer endpoint.
+
+        Adds a dedicated nftables table for the peer so rules can be removed
+        cleanly when the peer is removed.
+        """
+        rules = f"""\
+table inet netengine_peer_{peer_name} {{
+    chain forward {{
+        type filter hook forward priority 0; policy accept;
+        ip daddr {peer_endpoint_ip} ct state new accept
+        ip saddr {peer_endpoint_ip} ct state new accept
+    }}
+}}
+"""
+        dest_path = f"/etc/nftables/rules/peer_{peer_name}.nft"
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".nft", delete=False) as f:
+            f.write(rules)
+            tmp_path = f.name
+        try:
+            await self.docker.copy_to_container(self.gateway_container, tmp_path, dest_path)
+        finally:
+            os.unlink(tmp_path)
+
+        exit_code, output = await self.docker.exec_command(
+            self.gateway_container, ["nft", "-f", dest_path]
+        )
+        if exit_code != 0:
+            raise GatewayError(f"Failed to apply peer routing for {peer_name}: {output}")
+
+    async def remove_peer_routing(self, peer_name: str) -> None:
+        """Remove routing rules for a cross-world peer."""
+        exit_code, output = await self.docker.exec_command(
+            self.gateway_container,
+            ["nft", "delete", "table", "inet", f"netengine_peer_{peer_name}"],
+        )
+        if exit_code != 0 and "No such table" not in output:
+            raise GatewayError(f"Failed to remove peer routing for {peer_name}: {output}")
+        await self.docker.exec_command(
+            self.gateway_container,
+            ["rm", "-f", f"/etc/nftables/rules/peer_{peer_name}.nft"],
+        )
diff --git a/netengine/handlers/gateway_portal_handler.py b/netengine/handlers/gateway_portal_handler.py
new file mode 100644
index 0000000..f6f41cf
--- /dev/null
+++ b/netengine/handlers/gateway_portal_handler.py
@@ -0,0 +1,319 @@
+"""Gateway Portal handler — Real Internet access and Cross-World Federation.
+
+This is a boundary handler, not a numbered phase. It is invoked after Phase 7
+(ANDs) and applies two independent policies declared in the spec:
+
+  * ``gateway_portal.real_internet`` — controls how the world connects to
+    the public internet (ISOLATED / SHADOWED / MIRRORED / EXPOSED / CUSTOM).
+
+  * ``gateway_portal.cross_world`` — controls peering and federation with
+    other NetEngine worlds (NONE / PEERED / FEDERATED).
+"""
+
+from datetime import datetime
+from typing import Any
+
+from netengine.errors import GatewayError, PKIError
+from netengine.events.schema import EventEnvelope
+from netengine.handlers._base import BasePhaseHandler
+from netengine.handlers.context import PhaseContext
+from netengine.handlers.docker_handler import DockerHandler
+from netengine.handlers.gateway_handler import GatewayHandler
+from netengine.logging import get_logger
+from netengine.spec.models import CrossWorldPeer, GatewayPortal
+from netengine.spec.types import GatewayCrossWorldMode, GatewayRealInternetMode
+
+logger = get_logger(__name__)
+
+
+class GatewayPortalHandler(BasePhaseHandler):
+    """Apply real-internet and cross-world policies after ANDs are provisioned."""
+
+    async def execute(self, context: PhaseContext) -> None:
+        spec = context.spec
+        portal: GatewayPortal = spec.gateway_portal
+
+        if not portal.enabled:
+            context.logger.info("Gateway portal disabled — skipping")
+            context.runtime_state.gateway_portal_output = {
+                "enabled": False,
+                "deployed_at": datetime.utcnow().isoformat(),
+            }
+            context.runtime_state.save()
+            return
+
+        context.logger.info("Applying gateway portal policies")
+
+        if context.mock_mode:
+            context.runtime_state.gateway_portal_output = {
+                "enabled": True,
+                "internet_mode": portal.real_internet.mode.value,
+                "cross_world_mode": portal.cross_world.mode.value,
+                "peer_count": len(portal.cross_world.peers),
+                "mock": True,
+                "deployed_at": datetime.utcnow().isoformat(),
+            }
+            context.runtime_state.save()
+            await self._emit_event(
+                context, "gateway_portal.ready", context.runtime_state.gateway_portal_output
+            )
+            return
+
+        docker = context.docker_client or DockerHandler()  # type: ignore[no-untyped-call]
+        gateway = GatewayHandler(docker)
+
+        # ── Real Internet ───────────────────────────────────────────────────
+        internet_output = await self._apply_internet_policy(context, gateway, portal)
+
+        # ── Cross-World Federation ──────────────────────────────────────────
+        federation_output = await self._apply_cross_world(context, gateway, docker, portal)
+
+        # ── Persist ────────────────────────────────────────────────────────
+        context.runtime_state.gateway_portal_output = {
+            "enabled": True,
+            "internet_mode": portal.real_internet.mode.value,
+            "cross_world_mode": portal.cross_world.mode.value,
+            "peer_count": len(portal.cross_world.peers),
+            "internet": internet_output,
+            "federation": federation_output,
+            "deployed_at": datetime.utcnow().isoformat(),
+        }
+        context.runtime_state.save()
+
+        await self._emit_event(
+            context,
+            "gateway_portal.ready",
+            context.runtime_state.gateway_portal_output,
+        )
+        context.logger.info("Gateway portal policies applied")
+
+    async def healthcheck(self, context: PhaseContext) -> bool:
+        return context.runtime_state.gateway_portal_output is not None
+
+    async def should_skip(self, context: PhaseContext) -> bool:
+        return context.runtime_state.gateway_portal_output is not None
+
+    # ─────────────────────────────────────────────
+    # Internet policy
+    # ─────────────────────────────────────────────
+
+    async def _apply_internet_policy(
+        self,
+        context: PhaseContext,
+        gateway: GatewayHandler,
+        portal: GatewayPortal,
+    ) -> dict[str, Any]:
+        config = portal.real_internet
+        context.logger.info(f"Real-internet mode: {config.mode.value}")
+
+        await gateway.apply_internet_policy(config)
+
+        output: dict[str, Any] = {"mode": config.mode.value}
+
+        if config.upstream_resolver_enabled and config.upstream_resolver_ip:
+            await self._configure_upstream_resolver(context, config.upstream_resolver_ip)
+            output["upstream_resolver"] = config.upstream_resolver_ip
+
+        if config.mode == GatewayRealInternetMode.MIRRORED and config.service_mirrors:
+            output["mirrors"] = [
+                {"real": m.real_hostname, "in_world": m.in_world_service}
+                for m in config.service_mirrors
+            ]
+
+        return output
+
+    async def _configure_upstream_resolver(self, context: PhaseContext, resolver_ip: str) -> None:
+        """Inject an upstream forwarder into the CoreDNS root Corefile.
+
+        Adds a ``forward . <resolver_ip>`` directive so that names not
+        resolved within the world are forwarded to the real internet resolver.
+        This method is best-effort: a failure is logged but does not abort
+        the portal setup.
+        """
+        if context.docker_client is None:
+            return
+
+        try:
+            corefile_patch = (
+                f"\n# Upstream internet resolver (gateway portal)\n" f"forward . {resolver_ip}\n"
+            )
+            corefile_append_cmd = [
+                "sh",
+                "-c",
+                f"echo '{corefile_patch}' >> /etc/coredns/Corefile",
+            ]
+            exit_code, output = await context.docker_client.exec_command(
+                "netengine_coredns", corefile_append_cmd
+            )
+            if exit_code != 0:
+                context.logger.warning(f"Could not append upstream resolver to CoreDNS: {output}")
+            else:
+                # Reload CoreDNS to pick up the change
+                await context.docker_client.exec_command("netengine_coredns", ["kill", "-HUP", "1"])
+                context.logger.info(f"Upstream resolver configured: {resolver_ip}")
+        except Exception as exc:
+            context.logger.warning(f"Upstream resolver setup skipped: {exc}")
+
+    # ─────────────────────────────────────────────
+    # Cross-world federation
+    # ─────────────────────────────────────────────
+
+    async def _apply_cross_world(
+        self,
+        context: PhaseContext,
+        gateway: GatewayHandler,
+        docker: DockerHandler,
+        portal: GatewayPortal,
+    ) -> dict[str, Any]:
+        cross_world = portal.cross_world
+
+        if cross_world.mode == GatewayCrossWorldMode.NONE:
+            context.logger.info("Cross-world mode: NONE — no peering configured")
+            return {"mode": GatewayCrossWorldMode.NONE.value, "peers": []}
+
+        context.logger.info(
+            f"Cross-world mode: {cross_world.mode.value} " f"({len(cross_world.peers)} peer(s))"
+        )
+
+        peers_output = []
+        for peer in cross_world.peers:
+            peer_result = await self._setup_peer(context, gateway, docker, peer)
+            peers_output.append(peer_result)
+
+        return {
+            "mode": cross_world.mode.value,
+            "peers": peers_output,
+        }
+
+    async def _setup_peer(
+        self,
+        context: PhaseContext,
+        gateway: GatewayHandler,
+        docker: DockerHandler,
+        peer: CrossWorldPeer,
+    ) -> dict[str, Any]:
+        """Wire up a single cross-world peer: trust anchor + routing + DNS."""
+        context.logger.info(f"Setting up cross-world peer: {peer.name} ({peer.endpoint})")
+        result: dict[str, Any] = {
+            "name": peer.name,
+            "endpoint": peer.endpoint,
+            "mode": peer.mode.value,
+        }
+
+        # 1. Install trust anchor certificate
+        if peer.trust_anchor_cert:
+            try:
+                await self._install_trust_anchor(context, docker, peer.name, peer.trust_anchor_cert)
+                result["trust_anchor_installed"] = True
+            except Exception as exc:
+                context.logger.warning(f"Trust anchor install failed for peer {peer.name}: {exc}")
+                result["trust_anchor_installed"] = False
+                result["trust_anchor_error"] = str(exc)
+        else:
+            result["trust_anchor_installed"] = False
+
+        # 2. Configure nftables routing to peer endpoint
+        try:
+            # Extract host from endpoint (strip port if present)
+            peer_ip = peer.endpoint.split(":")[0]
+            await gateway.apply_peer_routing(peer.name, peer_ip)
+            result["routing_configured"] = True
+        except GatewayError as exc:
+            context.logger.warning(f"Peer routing failed for {peer.name}: {exc}")
+            result["routing_configured"] = False
+            result["routing_error"] = str(exc)
+
+        # 3. Configure DNS forwarding for peer domains
+        try:
+            await self._configure_peer_dns(context, peer)
+            result["dns_forwarding_configured"] = True
+        except Exception as exc:
+            context.logger.warning(f"Peer DNS forwarding failed for {peer.name}: {exc}")
+            result["dns_forwarding_configured"] = False
+            result["dns_error"] = str(exc)
+
+        return result
+
+    async def _install_trust_anchor(
+        self,
+        context: PhaseContext,
+        docker: DockerHandler,
+        peer_name: str,
+        cert_pem: str,
+    ) -> None:
+        """Install a peer's CA certificate into the gateway container trust store.
+
+        Writes the PEM cert to ``/usr/local/share/ca-certificates/<peer>.crt``
+        then runs ``update-ca-certificates`` so that TLS connections to the peer
+        are automatically trusted by any process running in the gateway.
+        """
+        import os
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".crt", delete=False) as f:
+            f.write(cert_pem)
+            tmp_path = f.name
+        try:
+            dest_path = f"/usr/local/share/ca-certificates/peer_{peer_name}.crt"
+            await docker.copy_to_container("netengine_gateway", tmp_path, dest_path)
+        finally:
+            os.unlink(tmp_path)
+
+        exit_code, output = await docker.exec_command(
+            "netengine_gateway", ["update-ca-certificates"]
+        )
+        if exit_code != 0:
+            raise PKIError(f"update-ca-certificates failed for peer {peer_name}: {output}")
+
+        context.logger.info(f"Trust anchor installed for peer: {peer_name}")
+
+    async def _configure_peer_dns(self, context: PhaseContext, peer: CrossWorldPeer) -> None:
+        """Add a CoreDNS forwarding zone for the peer world's TLD.
+
+        Derives the peer TLD from ``<peer.name>.internal`` and adds a
+        ``forward <tld> <peer_resolver>`` stub to the CoreDNS root Corefile.
+        The peer's DNS resolver is assumed to live at port 53 of the peer endpoint.
+        """
+        if context.docker_client is None:
+            return
+
+        peer_tld = f"{peer.name}.internal"
+        peer_ip = peer.endpoint.split(":")[0]
+
+        corefile_stub = (
+            f"\n# Cross-world peer: {peer.name}\n"
+            f"{peer_tld} {{\n"
+            f"    forward . {peer_ip}:53\n"
+            f"}}\n"
+        )
+        exit_code, output = await context.docker_client.exec_command(
+            "netengine_coredns",
+            ["sh", "-c", f"echo '{corefile_stub}' >> /etc/coredns/Corefile"],
+        )
+        if exit_code != 0:
+            raise GatewayError(f"Could not configure DNS forwarding for peer {peer.name}: {output}")
+
+        # Signal CoreDNS to reload config
+        await context.docker_client.exec_command("netengine_coredns", ["kill", "-HUP", "1"])
+        context.logger.info(f"DNS forwarding configured for peer TLD: {peer_tld}")
+
+    # ─────────────────────────────────────────────
+    # Event emission
+    # ─────────────────────────────────────────────
+
+    async def _emit_event(
+        self, context: PhaseContext, event_type: str, payload: dict[str, Any]
+    ) -> None:
+        event = EventEnvelope.create(
+            event_type=event_type,
+            emitted_by="gateway_portal_handler",
+            payload=payload,
+            correlation_id=getattr(context.runtime_state, "correlation_id", None),
+            parent_event_id=getattr(context.runtime_state, "parent_event_id", None),
+        )
+        context.logger.info(f"Event emitted: {event_type}")
+        if context.pgmq_client is not None:
+            try:
+                await context.pgmq_client.send("gateway_portal_events", event)
+            except Exception as exc:
+                context.logger.warning(f"Failed to queue gateway portal event: {exc}")
diff --git a/netengine/handlers/phase_pki.py b/netengine/handlers/phase_pki.py
index afcde4b..4b7ebec 100644
--- a/netengine/handlers/phase_pki.py
+++ b/netengine/handlers/phase_pki.py
@@ -63,14 +63,46 @@ async def execute(self, context: PhaseContext) -> None:
             ttl=300,
         )
 
-        # 3. Persist outputs
-        context.runtime_state.pki_output = {
+        # 3. Setup optional PKI features
+        pki_output: dict = {
             "ca_ip": pki.ca_ip,
             "ca_dns": pki.ca_dns,
             "container_id": context.runtime_state.step_ca_container_id,
             "bootstrapped": True,
             "deployed_at": datetime.utcnow().isoformat(),
         }
+
+        if spec.pki.crl_enabled:
+            pki_output["crl_url"] = f"https://{pki.ca_ip}/1.0/crl"
+            pki_output["crl_enabled"] = True
+            logger.info(f"CRL endpoint: {pki_output['crl_url']}")
+
+        if spec.pki.ocsp_enabled:
+            pki_output["ocsp_url"] = f"https://{pki.ca_ip}/ocsp"
+            pki_output["ocsp_enabled"] = True
+            logger.info(f"OCSP endpoint: {pki_output['ocsp_url']}")
+
+        if spec.pki.intermediate_ca_enabled:
+            pki_output["intermediate_ca_enabled"] = True
+            if context.runtime_state.intermediate_ca_cert:
+                pki_output["intermediate_ca_cert_available"] = True
+            logger.info("Intermediate CA enabled and tracked in state")
+
+        if spec.pki.dnssec_enabled:
+            dnssec_info = await pki.setup_dnssec(
+                zone="internal",
+                ksk_lifetime_days=spec.pki.dnssec_ksk_lifetime_days,
+                zsk_lifetime_days=spec.pki.dnssec_zsk_lifetime_days,
+            )
+            context.runtime_state.dnssec_output = dnssec_info
+            pki_output["dnssec_enabled"] = True
+            pki_output["dnssec_zone"] = dnssec_info["zone"]
+            pki_output["dnssec_ksk"] = dnssec_info["ksk_name"]
+            pki_output["dnssec_zsk"] = dnssec_info["zsk_name"]
+            logger.info(f"DNSSEC keys generated for zone: {dnssec_info['zone']}")
+
+        # 4. Persist outputs
+        context.runtime_state.pki_output = pki_output
         context.runtime_state.pki_bootstrapped = True
         context.runtime_state.phase_completed["3"] = True
         context.runtime_state.completed_at = datetime.utcnow()
@@ -102,7 +134,12 @@ async def should_skip(self, context: PhaseContext) -> bool:
         return context.runtime_state.phase_completed.get("3", False)
 
     def _register_rotation_worker(self, context: PhaseContext, pki: PKIHandler, spec: Any) -> None:
-        """Register certificate rotation worker with ConsumerSupervisor."""
+        """Register certificate rotation worker with ConsumerSupervisor.
+
+        Builds rotation configs for all cert types: the four well-known built-in
+        types (platform_identity, inworld_identity, app, storage) plus any
+        additional types declared in policy.cert_type_overrides.
+        """
         if not context.consumer_supervisor:
             return
 
@@ -111,106 +148,41 @@ def _register_rotation_worker(self, context: PhaseContext, pki: PKIHandler, spec
             context.logger.info("PKI certificate rotation disabled")
             return
 
-        # Build rotation configs from spec (with defaults)
-        rotation_configs = []
-
-        # Platform identity cert
-        platform_cfg_dict = policy.cert_type_overrides.get("platform_identity")
-        if platform_cfg_dict and isinstance(platform_cfg_dict, dict):
-            platform_interval = platform_cfg_dict.get(
-                "rotation_interval_hours", policy.default_interval_hours
-            )
-            platform_warning = platform_cfg_dict.get(
-                "expiry_warning_days", policy.default_warning_days
-            )
-            platform_cfg = CertTypeRotationConfig(
-                cert_type="platform_identity",
-                rotation_interval_hours=platform_interval,
-                expiry_warning_days=platform_warning,
-                rotation_callback=self._prepare_app_cert_rotation,
-            )
-        else:
-            platform_cfg = CertTypeRotationConfig(
-                cert_type="platform_identity",
-                rotation_interval_hours=policy.default_interval_hours,
-                expiry_warning_days=policy.default_warning_days,
-                rotation_callback=self._prepare_app_cert_rotation,
-            )
-        rotation_configs.append(platform_cfg)
-
-        # Inworld identity cert
-        inworld_cfg_dict = policy.cert_type_overrides.get("inworld_identity")
-        if inworld_cfg_dict and isinstance(inworld_cfg_dict, dict):
-            inworld_interval = inworld_cfg_dict.get(
-                "rotation_interval_hours", policy.default_interval_hours
-            )
-            inworld_warning = inworld_cfg_dict.get(
-                "expiry_warning_days", policy.default_warning_days
-            )
-            inworld_cfg = CertTypeRotationConfig(
-                cert_type="inworld_identity",
-                rotation_interval_hours=inworld_interval,
-                expiry_warning_days=inworld_warning,
-                rotation_callback=self._prepare_app_cert_rotation,
-            )
-        else:
-            inworld_cfg = CertTypeRotationConfig(
-                cert_type="inworld_identity",
-                rotation_interval_hours=policy.default_interval_hours,
-                expiry_warning_days=policy.default_warning_days,
-                rotation_callback=self._prepare_app_cert_rotation,
-            )
-        rotation_configs.append(inworld_cfg)
+        # Cert types that receive the graceful-transition rotation callback
+        _callback_types = {"platform_identity", "inworld_identity", "app"}
 
-        # App certs (with rotation callback for graceful transition)
-        app_cfg_dict = policy.cert_type_overrides.get("app")
-        if app_cfg_dict and isinstance(app_cfg_dict, dict):
-            app_interval = app_cfg_dict.get(
-                "rotation_interval_hours", policy.default_interval_hours
-            )
-            app_warning = app_cfg_dict.get("expiry_warning_days", policy.default_warning_days)
-            app_cfg = CertTypeRotationConfig(
-                cert_type="app",
-                rotation_interval_hours=app_interval,
-                expiry_warning_days=app_warning,
-                rotation_callback=self._prepare_app_cert_rotation,
-            )
-        else:
-            app_cfg = CertTypeRotationConfig(
-                cert_type="app",
-                rotation_interval_hours=policy.default_interval_hours,
-                expiry_warning_days=policy.default_warning_days,
-                rotation_callback=self._prepare_app_cert_rotation,
-            )
-        rotation_configs.append(app_cfg)
+        # Merge built-in cert types with any extras declared in spec overrides
+        _builtin = ["platform_identity", "inworld_identity", "app", "storage"]
+        _extra = [t for t in policy.cert_type_overrides if t not in _builtin]
+        all_cert_types = _builtin + _extra
 
-        # Storage/MinIO cert
-        storage_cfg_dict = policy.cert_type_overrides.get("storage")
-        if storage_cfg_dict and isinstance(storage_cfg_dict, dict):
-            storage_interval = storage_cfg_dict.get(
-                "rotation_interval_hours", policy.default_interval_hours
-            )
-            storage_warning = storage_cfg_dict.get(
-                "expiry_warning_days", policy.default_warning_days
-            )
-            storage_cfg = CertTypeRotationConfig(
-                cert_type="storage",
-                rotation_interval_hours=storage_interval,
-                expiry_warning_days=storage_warning,
-            )
-        else:
-            storage_cfg = CertTypeRotationConfig(
-                cert_type="storage",
-                rotation_interval_hours=policy.default_interval_hours,
-                expiry_warning_days=policy.default_warning_days,
+        rotation_configs = []
+        for cert_type in all_cert_types:
+            override = policy.cert_type_overrides.get(cert_type)
+            if isinstance(override, dict):
+                interval = override.get("rotation_interval_hours", policy.default_interval_hours)
+                warning = override.get("expiry_warning_days", policy.default_warning_days)
+            else:
+                interval = policy.default_interval_hours
+                warning = policy.default_warning_days
+
+            rotation_configs.append(
+                CertTypeRotationConfig(
+                    cert_type=cert_type,
+                    rotation_interval_hours=interval,
+                    expiry_warning_days=warning,
+                    rotation_callback=(
+                        self._prepare_app_cert_rotation if cert_type in _callback_types else None
+                    ),
+                )
             )
-        rotation_configs.append(storage_cfg)
 
-        # Register rotation worker
         if context.pgmq_client:
             rotation_worker = PKICertRotationWorker(pki, context.pgmq_client, rotation_configs)
             context.consumer_supervisor.register("pki_cert_rotation", rotation_worker.run)
-            context.logger.info("PKI certificate rotation worker registered")
+            context.logger.info(
+                f"PKI certificate rotation worker registered ({len(rotation_configs)} cert types)"
+            )
 
     async def _prepare_app_cert_rotation(self, cn: str, cert_metadata: dict) -> None:
         """Called before rotating app cert - prepare for transition.
diff --git a/netengine/handlers/pki_handler.py b/netengine/handlers/pki_handler.py
index 90aee98..251081a 100644
--- a/netengine/handlers/pki_handler.py
+++ b/netengine/handlers/pki_handler.py
@@ -39,6 +39,14 @@ def __init__(self, docker: DockerHandler, state: RuntimeState, spec):
     # ─────────────────────────────────────────────
     # Main bootstrap (idempotent)
     # ─────────────────────────────────────────────
+    def _pki_flag(self, flag: str) -> bool:
+        """Return a boolean PKI spec flag, handling both model and dict specs."""
+        if hasattr(self.spec, "pki"):
+            return bool(getattr(self.spec.pki, flag, False))
+        if isinstance(self.spec, dict):
+            return bool(self.spec.get("pki", {}).get(flag, False))
+        return False
+
     async def bootstrap(self) -> None:
         """Ensure CA is generated and step‑ca is running."""
         # 1. Create volume if missing
@@ -47,6 +55,11 @@ async def bootstrap(self) -> None:
         # 2. Generate CA if not present in state
         if not self.state.ca_cert_pem:
             await self._generate_ca()
+            # Inject optional features into ca.json before server starts
+            if self._pki_flag("crl_enabled"):
+                await self._inject_crl_config()
+            if self._pki_flag("ocsp_enabled"):
+                await self._inject_ocsp_config()
 
         # 3. Start container if not running
         if not self.state.step_ca_container_id:
@@ -56,6 +69,11 @@ async def bootstrap(self) -> None:
         if not await self.healthcheck():
             raise PKIError("step‑ca is not responding after bootstrap")
 
+        # 5. Read intermediate CA cert when enabled
+        if self._pki_flag("intermediate_ca_enabled") and not self.state.intermediate_ca_cert:
+            self.state.intermediate_ca_cert = await self.read_intermediate_cert()
+            self.state.save()
+
     # ─────────────────────────────────────────────
     # CA generation (one‑off)
     # ─────────────────────────────────────────────
@@ -228,6 +246,158 @@ async def issue_cert(self, common_name: str, sans: list[str] = None) -> tuple[st
         key_content = await self._read_file_from_volume(f"/home/step/{common_name}.key")
         return cert_content, key_content
 
+    # ─────────────────────────────────────────────
+    # Intermediate CA
+    # ─────────────────────────────────────────────
+
+    async def read_intermediate_cert(self) -> str:
+        """Read the intermediate CA certificate from the step-ca volume.
+
+        step-ca generates a root + intermediate CA pair by default; the
+        intermediate is what signs leaf certs.  When intermediate_ca_enabled
+        is True the caller should store and distribute this cert separately.
+        """
+        try:
+            return await self._read_file_from_volume("/home/step/certs/intermediate_ca.crt")
+        except Exception as exc:
+            raise PKIError(f"Failed to read intermediate CA certificate: {exc}") from exc
+
+    # ─────────────────────────────────────────────
+    # CRL and OCSP
+    # ─────────────────────────────────────────────
+
+    async def _read_ca_config(self) -> dict:
+        """Read and parse the step-ca ca.json from the PKI volume."""
+        import json as _json
+
+        volumes = {self.volume_name: {"bind": "/home/step", "mode": "ro"}}
+        result = await self.docker.run_container_one_off(
+            image=self.image,
+            command=["cat", "/home/step/config/ca.json"],
+            volumes=volumes,
+            environment={},
+        )
+        if result["exit_code"] != 0:
+            raise PKIError(f"Failed to read CA config: {result['logs']}")
+        return _json.loads(result["logs"])
+
+    async def _write_ca_config(self, config: dict) -> None:
+        """Write an updated ca.json back to the step-ca volume."""
+        import json as _json
+
+        updated = _json.dumps(config, indent=2)
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            f.write(updated)
+            tmp_path = f.name
+        try:
+            updated_volumes = {
+                self.volume_name: {"bind": "/home/step", "mode": "rw"},
+                tmp_path: {"bind": "/tmp/ca_updated.json", "mode": "ro"},
+            }
+            result = await self.docker.run_container_one_off(
+                image=self.image,
+                command=["cp", "/tmp/ca_updated.json", "/home/step/config/ca.json"],
+                volumes=updated_volumes,
+                environment={},
+            )
+            if result["exit_code"] != 0:
+                raise PKIError(f"Failed to write CA config: {result['logs']}")
+        finally:
+            os.unlink(tmp_path)
+
+    async def _inject_crl_config(self) -> None:
+        """Add CRL generation config to ca.json (called before server start)."""
+        config = await self._read_ca_config()
+        config["crl"] = {
+            "enabled": True,
+            "cacheDuration": "3h",
+            "renewalDisabled": False,
+            "cacheDisabled": False,
+        }
+        await self._write_ca_config(config)
+        logger.info("CRL enabled in step-ca configuration")
+
+    async def _inject_ocsp_config(self) -> None:
+        """Add OCSP responder config to ca.json (called before server start)."""
+        config = await self._read_ca_config()
+        authority = config.setdefault("authority", {})
+        authority.setdefault("claims", {})["enableOCSP"] = True
+        await self._write_ca_config(config)
+        logger.info("OCSP responder enabled in step-ca configuration")
+
+    # ─────────────────────────────────────────────
+    # DNSSEC
+    # ─────────────────────────────────────────────
+
+    async def setup_dnssec(
+        self,
+        zone: str,
+        ksk_lifetime_days: int = 365,
+        zsk_lifetime_days: int = 30,
+    ) -> dict:
+        """Generate DNSSEC KSK and ZSK keys for *zone* using BIND's dnssec-keygen.
+
+        Keys are stored in the ``netengines_dnssec_keys`` Docker volume so that
+        CoreDNS can mount them and activate the ``dnssec`` plugin.  Returns a
+        dict of key metadata that the caller should persist in state for later
+        rotation.
+        """
+        dnssec_volume = "netengines_dnssec_keys"
+        await self.docker.ensure_volume(dnssec_volume)
+
+        bind_image = "internetsystemsconsortium/bind9:9.18"
+        volumes = {dnssec_volume: {"bind": "/keys", "mode": "rw"}}
+
+        # KSK — signs only the DNSKEY RRset
+        ksk_result = await self.docker.run_container_one_off(
+            image=bind_image,
+            command=[
+                "dnssec-keygen",
+                "-f",
+                "KSK",
+                "-a",
+                "ECDSAP256SHA256",
+                "-n",
+                "ZONE",
+                zone,
+            ],
+            volumes=volumes,
+            environment={},
+            working_dir="/keys",
+        )
+        if ksk_result["exit_code"] != 0:
+            raise PKIError(f"DNSSEC KSK generation failed for zone '{zone}': {ksk_result['logs']}")
+        ksk_name = ksk_result["logs"].strip()
+
+        # ZSK — signs all other RRsets
+        zsk_result = await self.docker.run_container_one_off(
+            image=bind_image,
+            command=[
+                "dnssec-keygen",
+                "-a",
+                "ECDSAP256SHA256",
+                "-n",
+                "ZONE",
+                zone,
+            ],
+            volumes=volumes,
+            environment={},
+            working_dir="/keys",
+        )
+        if zsk_result["exit_code"] != 0:
+            raise PKIError(f"DNSSEC ZSK generation failed for zone '{zone}': {zsk_result['logs']}")
+        zsk_name = zsk_result["logs"].strip()
+
+        return {
+            "zone": zone,
+            "ksk_name": ksk_name,
+            "zsk_name": zsk_name,
+            "volume": dnssec_volume,
+            "algorithm": "ECDSAP256SHA256",
+            "ksk_lifetime_days": ksk_lifetime_days,
+            "zsk_lifetime_days": zsk_lifetime_days,
+        }
+
     def extract_cert_expiry(self, cert_pem: str) -> datetime:
         """Extract the notAfter date from a PEM certificate."""
         try:
diff --git a/tests/integration/test_dns_add_zone_record_callers.py b/tests/integration/test_dns_add_zone_record_callers.py
index 986c22b..074d087 100644
--- a/tests/integration/test_dns_add_zone_record_callers.py
+++ b/tests/integration/test_dns_add_zone_record_callers.py
@@ -29,6 +29,17 @@ async def test_phase_3_pki_inserts_ca_dns_record(context_with_zone_files):
         ca_ip="10.0.0.6",
         ca_dns="ca.platform.internal",
         bootstrap=AsyncMock(),
+        setup_dnssec=AsyncMock(
+            return_value={
+                "zone": "internal",
+                "ksk_name": "Kinternal.+013+00001",
+                "zsk_name": "Kinternal.+013+00002",
+                "volume": "netengines_dnssec_keys",
+                "algorithm": "ECDSAP256SHA256",
+                "ksk_lifetime_days": 365,
+                "zsk_lifetime_days": 30,
+            }
+        ),
     )
 
     with (
diff --git a/tests/test_gateway_portal.py b/tests/test_gateway_portal.py
new file mode 100644
index 0000000..0c56857
--- /dev/null
+++ b/tests/test_gateway_portal.py
@@ -0,0 +1,364 @@
+"""Tests for Gateway Portal — Real Internet and Cross-World Federation.
+
+Covers:
+- Real internet policy for all five modes (ISOLATED, SHADOWED, MIRRORED, EXPOSED, CUSTOM)
+- Peer routing (apply + remove)
+- GatewayPortalHandler execute in mock mode
+- Trust anchor installation
+- DNS forwarding for peers
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from netengine.core.state import RuntimeState
+from netengine.errors import GatewayError
+from netengine.handlers.gateway_handler import GatewayHandler
+from netengine.handlers.gateway_portal_handler import GatewayPortalHandler
+from netengine.spec.models import (
+    CrossWorldConfig,
+    CrossWorldPeer,
+    GatewayPortal,
+    RealInternetConfig,
+    ServiceMirror,
+)
+from netengine.spec.types import GatewayCrossWorldMode, GatewayRealInternetMode
+
+# ─────────────────────────────────────────────
+# Fixtures
+# ─────────────────────────────────────────────
+
+
+@pytest.fixture
+def mock_docker():
+    d = MagicMock()
+    d.copy_to_container = AsyncMock()
+    d.exec_command = AsyncMock(return_value=(0, ""))
+    return d
+
+
+@pytest.fixture
+def gateway(mock_docker):
+    return GatewayHandler(mock_docker)
+
+
+# ─────────────────────────────────────────────
+# Real Internet Policy — rule generation
+# ─────────────────────────────────────────────
+
+
+class TestInternetRuleGeneration:
+    def test_isolated_blocks_wan_forward(self, gateway):
+        rules = gateway._isolated_internet_rules()
+        assert "oifname" in rules and "eth_wan" in rules
+        assert "drop" in rules
+
+    def test_isolated_uses_inet_table(self, gateway):
+        rules = gateway._isolated_internet_rules()
+        assert "table inet netengine_internet" in rules
+
+    def test_shadowed_allows_https_outbound(self, gateway):
+        rules = gateway._shadowed_internet_rules()
+        assert "tcp dport { 80, 443 }" in rules
+        assert "ct state new accept" in rules
+
+    def test_shadowed_blocks_wan_inbound(self, gateway):
+        rules = gateway._shadowed_internet_rules()
+        assert 'iifname "eth_wan" drop' in rules
+
+    def test_shadowed_has_masquerade(self, gateway):
+        rules = gateway._shadowed_internet_rules()
+        assert "masquerade" in rules
+
+    def test_mirrored_includes_mirror_addresses(self, gateway):
+        config = RealInternetConfig(
+            mode=GatewayRealInternetMode.MIRRORED,
+            service_mirrors=[
+                ServiceMirror(real_hostname="example.com", in_world_service="10.0.1.50"),
+            ],
+        )
+        rules = gateway._mirrored_internet_rules(config)
+        assert "10.0.1.50" in rules
+
+    def test_mirrored_multiple_mirrors(self, gateway):
+        config = RealInternetConfig(
+            mode=GatewayRealInternetMode.MIRRORED,
+            service_mirrors=[
+                ServiceMirror(real_hostname="a.com", in_world_service="10.0.1.1"),
+                ServiceMirror(real_hostname="b.com", in_world_service="10.0.1.2"),
+            ],
+        )
+        rules = gateway._mirrored_internet_rules(config)
+        assert "10.0.1.1" in rules
+        assert "10.0.1.2" in rules
+
+    def test_exposed_has_policy_accept_on_forward(self, gateway):
+        rules = gateway._exposed_internet_rules()
+        assert "policy accept" in rules
+
+    def test_exposed_allows_http_inbound(self, gateway):
+        rules = gateway._exposed_internet_rules()
+        assert "tcp dport { 80, 443 }" in rules
+
+    def test_exposed_has_masquerade(self, gateway):
+        rules = gateway._exposed_internet_rules()
+        assert "masquerade" in rules
+
+
+class TestApplyInternetPolicy:
+    async def test_custom_mode_is_noop(self, gateway, mock_docker):
+        config = RealInternetConfig(mode=GatewayRealInternetMode.CUSTOM)
+        await gateway.apply_internet_policy(config)
+        mock_docker.copy_to_container.assert_not_called()
+        mock_docker.exec_command.assert_not_called()
+
+    async def test_isolated_copies_rules_file(self, gateway, mock_docker):
+        config = RealInternetConfig(mode=GatewayRealInternetMode.ISOLATED)
+        await gateway.apply_internet_policy(config)
+        mock_docker.copy_to_container.assert_called_once()
+        args = mock_docker.copy_to_container.call_args[0]
+        assert args[2] == "/etc/nftables/rules/internet.nft"
+
+    async def test_shadowed_loads_rules_with_nft(self, gateway, mock_docker):
+        config = RealInternetConfig(mode=GatewayRealInternetMode.SHADOWED)
+        await gateway.apply_internet_policy(config)
+        mock_docker.exec_command.assert_called_once()
+        cmd = mock_docker.exec_command.call_args[0][1]
+        assert cmd == ["nft", "-f", "/etc/nftables/rules/internet.nft"]
+
+    async def test_exposed_raises_on_nft_failure(self, gateway, mock_docker):
+        mock_docker.exec_command.return_value = (1, "nft error")
+        config = RealInternetConfig(mode=GatewayRealInternetMode.EXPOSED)
+        with pytest.raises(GatewayError, match="internet policy"):
+            await gateway.apply_internet_policy(config)
+
+    async def test_mirrored_passes_config_to_rule_generator(self, gateway, mock_docker):
+        config = RealInternetConfig(
+            mode=GatewayRealInternetMode.MIRRORED,
+            service_mirrors=[ServiceMirror(real_hostname="a.com", in_world_service="10.1.2.3")],
+        )
+        written_content = []
+
+        import builtins
+        import os
+
+        real_open = builtins.open
+
+        def capture_open(path, mode="r", **kw):
+            if isinstance(path, str) and path.endswith(".nft") and "w" in mode:
+                import io
+
+                buf = io.StringIO()
+                buf.name = path
+                written_content.append(buf)
+                return buf
+            return real_open(path, mode, **kw)
+
+        with patch("tempfile.NamedTemporaryFile") as mock_tmp, patch("os.unlink"):
+            import io
+
+            buf = io.StringIO()
+            buf.name = "/tmp/fake.nft"
+            mock_tmp.return_value.__enter__ = MagicMock(
+                return_value=MagicMock(
+                    write=lambda s: written_content.append(s),
+                    name="/tmp/fake.nft",
+                )
+            )
+            mock_tmp.return_value.__exit__ = MagicMock(return_value=False)
+            await gateway.apply_internet_policy(config)
+
+        # Verify copy was called with the internet.nft path
+        mock_docker.copy_to_container.assert_called_once()
+
+
+class TestRemoveInternetPolicy:
+    async def test_removes_table_and_file(self, gateway, mock_docker):
+        await gateway.remove_internet_policy()
+        assert mock_docker.exec_command.call_count == 2
+        first_cmd = mock_docker.exec_command.call_args_list[0][0][1]
+        assert "delete" in first_cmd and "netengine_internet" in first_cmd
+
+    async def test_table_not_found_is_tolerated(self, gateway, mock_docker):
+        mock_docker.exec_command.side_effect = [
+            (1, "No such table"),
+            (0, ""),
+        ]
+        await gateway.remove_internet_policy()
+
+    async def test_other_failure_raises(self, gateway, mock_docker):
+        mock_docker.exec_command.return_value = (1, "permission denied")
+        with pytest.raises(GatewayError, match="internet policy"):
+            await gateway.remove_internet_policy()
+
+
+# ─────────────────────────────────────────────
+# Cross-World Peer Routing
+# ─────────────────────────────────────────────
+
+
+class TestPeerRouting:
+    async def test_apply_peer_routing_uses_peer_name_in_table(self, gateway, mock_docker):
+        await gateway.apply_peer_routing("worldb", "192.168.100.1")
+        copy_args = mock_docker.copy_to_container.call_args[0]
+        assert "peer_worldb" in copy_args[2]
+
+    async def test_apply_peer_routing_loads_rules(self, gateway, mock_docker):
+        await gateway.apply_peer_routing("worldb", "192.168.100.1")
+        cmd = mock_docker.exec_command.call_args[0][1]
+        assert cmd[0] == "nft" and "-f" in cmd
+
+    async def test_apply_peer_routing_raises_on_failure(self, gateway, mock_docker):
+        mock_docker.exec_command.return_value = (1, "error")
+        with pytest.raises(GatewayError, match="worldb"):
+            await gateway.apply_peer_routing("worldb", "192.168.100.1")
+
+    async def test_remove_peer_routing_deletes_table(self, gateway, mock_docker):
+        await gateway.remove_peer_routing("worldb")
+        first_cmd = mock_docker.exec_command.call_args_list[0][0][1]
+        assert "netengine_peer_worldb" in first_cmd
+
+    async def test_remove_peer_routing_tolerates_not_found(self, gateway, mock_docker):
+        mock_docker.exec_command.side_effect = [(1, "No such table"), (0, "")]
+        await gateway.remove_peer_routing("worldb")
+
+    async def test_remove_peer_routing_raises_on_other_failure(self, gateway, mock_docker):
+        mock_docker.exec_command.return_value = (1, "permission denied")
+        with pytest.raises(GatewayError, match="worldb"):
+            await gateway.remove_peer_routing("worldb")
+
+
+# ─────────────────────────────────────────────
+# GatewayPortalHandler — execute in mock mode
+# ─────────────────────────────────────────────
+
+
+class TestGatewayPortalHandlerMockMode:
+    def _make_context(self, portal: GatewayPortal, mock_mode: bool = True):
+        state = RuntimeState()
+        spec = MagicMock()
+        spec.gateway_portal = portal
+        ctx = MagicMock()
+        ctx.spec = spec
+        ctx.runtime_state = state
+        ctx.mock_mode = mock_mode
+        ctx.docker_client = None
+        ctx.pgmq_client = None
+        ctx.logger = MagicMock()
+        return ctx
+
+    async def test_disabled_portal_sets_output_and_returns(self):
+        portal = GatewayPortal(
+            enabled=False,
+            real_internet=RealInternetConfig(mode=GatewayRealInternetMode.ISOLATED),
+            cross_world=CrossWorldConfig(mode=GatewayCrossWorldMode.NONE),
+        )
+        ctx = self._make_context(portal)
+        handler = GatewayPortalHandler()
+        await handler.execute(ctx)
+        assert ctx.runtime_state.gateway_portal_output is not None
+        assert ctx.runtime_state.gateway_portal_output["enabled"] is False
+
+    async def test_mock_mode_populates_output(self):
+        portal = GatewayPortal(
+            enabled=True,
+            real_internet=RealInternetConfig(mode=GatewayRealInternetMode.SHADOWED),
+            cross_world=CrossWorldConfig(
+                mode=GatewayCrossWorldMode.PEERED,
+                peers=[
+                    CrossWorldPeer(
+                        name="worldb",
+                        endpoint="192.168.200.1:9000",
+                        mode=GatewayCrossWorldMode.PEERED,
+                    )
+                ],
+            ),
+        )
+        ctx = self._make_context(portal, mock_mode=True)
+        handler = GatewayPortalHandler()
+        await handler.execute(ctx)
+        output = ctx.runtime_state.gateway_portal_output
+        assert output["enabled"] is True
+        assert output["internet_mode"] == "shadowed"
+        assert output["cross_world_mode"] == "peered"
+        assert output["peer_count"] == 1
+        assert output["mock"] is True
+
+    async def test_healthcheck_returns_false_before_execute(self):
+        portal = GatewayPortal(
+            enabled=True,
+            real_internet=RealInternetConfig(),
+            cross_world=CrossWorldConfig(),
+        )
+        ctx = self._make_context(portal)
+        handler = GatewayPortalHandler()
+        assert await handler.healthcheck(ctx) is False
+
+    async def test_healthcheck_returns_true_after_execute(self):
+        portal = GatewayPortal(
+            enabled=True,
+            real_internet=RealInternetConfig(),
+            cross_world=CrossWorldConfig(),
+        )
+        ctx = self._make_context(portal, mock_mode=True)
+        handler = GatewayPortalHandler()
+        await handler.execute(ctx)
+        assert await handler.healthcheck(ctx) is True
+
+    async def test_should_skip_after_output_exists(self):
+        portal = GatewayPortal(
+            enabled=True,
+            real_internet=RealInternetConfig(),
+            cross_world=CrossWorldConfig(),
+        )
+        ctx = self._make_context(portal, mock_mode=True)
+        handler = GatewayPortalHandler()
+        await handler.execute(ctx)
+        assert await handler.should_skip(ctx) is True
+
+    async def test_should_not_skip_before_execute(self):
+        portal = GatewayPortal(
+            enabled=True,
+            real_internet=RealInternetConfig(),
+            cross_world=CrossWorldConfig(),
+        )
+        ctx = self._make_context(portal)
+        handler = GatewayPortalHandler()
+        assert await handler.should_skip(ctx) is False
+
+
+# ─────────────────────────────────────────────
+# RuntimeState — new fields
+# ─────────────────────────────────────────────
+
+
+class TestRuntimeStateNewFields:
+    def test_dnssec_output_defaults_to_none(self):
+        state = RuntimeState()
+        assert state.dnssec_output is None
+
+    def test_gateway_portal_output_defaults_to_none(self):
+        state = RuntimeState()
+        assert state.gateway_portal_output is None
+
+    def test_intermediate_ca_cert_defaults_to_none(self):
+        state = RuntimeState()
+        assert state.intermediate_ca_cert is None
+
+    def test_new_fields_survive_save_load_cycle(self, tmp_path, monkeypatch):
+        import os
+
+        state_path = tmp_path / "state.json"
+        monkeypatch.setenv("NETENGINE_STATE_FILE", str(state_path))
+
+        state = RuntimeState()
+        state.dnssec_output = {"zone": "internal", "ksk_name": "Kinternal.+013+00001"}
+        state.gateway_portal_output = {"enabled": True, "internet_mode": "exposed"}
+        state.intermediate_ca_cert = "-----BEGIN CERTIFICATE-----\nABC\n-----END CERTIFICATE-----"
+        state.save()
+
+        loaded = RuntimeState.load()
+        assert loaded.dnssec_output == {"zone": "internal", "ksk_name": "Kinternal.+013+00001"}
+        assert loaded.gateway_portal_output["internet_mode"] == "exposed"
+        assert "ABC" in loaded.intermediate_ca_cert
diff --git a/tests/test_pki_features.py b/tests/test_pki_features.py
new file mode 100644
index 0000000..99d0c3f
--- /dev/null
+++ b/tests/test_pki_features.py
@@ -0,0 +1,358 @@
+"""Tests for declared-but-not-implemented PKI features.
+
+Covers:
+- Intermediate CA cert exposure
+- CRL config injection
+- OCSP config injection
+- DNSSEC key generation
+- Dynamic PKI rotation policy wiring
+"""
+
+import json
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from netengine.core.state import RuntimeState
+from netengine.errors import PKIError
+from netengine.handlers.phase_pki import PKIPhaseHandler
+from netengine.handlers.pki_handler import PKIHandler
+from netengine.workers.pki_cert_rotation_worker import CertTypeRotationConfig
+
+# ─────────────────────────────────────────────
+# Fixtures
+# ─────────────────────────────────────────────
+
+
+@pytest.fixture
+def mock_docker():
+    d = MagicMock()
+    d.ensure_volume = AsyncMock()
+    d.run_container_one_off = AsyncMock(return_value={"exit_code": 0, "logs": ""})
+    d.start_container = AsyncMock(return_value="container-abc")
+    d.exec_command = AsyncMock(return_value=(0, ""))
+    d.copy_to_container = AsyncMock()
+    return d
+
+
+@pytest.fixture
+def minimal_spec_dict():
+    return {
+        "pki": {
+            "acme": {"listen_ip": "10.0.0.6", "canonical_name": "ca.platform.internal"},
+            "crl_enabled": False,
+            "ocsp_enabled": False,
+            "intermediate_ca_enabled": False,
+            "dnssec_enabled": False,
+        }
+    }
+
+
+@pytest.fixture
+def state():
+    return RuntimeState()
+
+
+@pytest.fixture
+def pki_handler(mock_docker, state, minimal_spec_dict):
+    return PKIHandler(mock_docker, state, minimal_spec_dict)
+
+
+# ─────────────────────────────────────────────
+# Intermediate CA
+# ─────────────────────────────────────────────
+
+
+class TestIntermediateCACert:
+    async def test_read_intermediate_cert_returns_content(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off.return_value = {
+            "exit_code": 0,
+            "logs": "-----BEGIN CERTIFICATE-----\nMIIB...\n-----END CERTIFICATE-----\n",
+        }
+        cert = await pki_handler.read_intermediate_cert()
+        assert "BEGIN CERTIFICATE" in cert
+
+    async def test_read_intermediate_cert_raises_on_failure(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off.return_value = {"exit_code": 1, "logs": "not found"}
+        with pytest.raises(PKIError, match="intermediate CA"):
+            await pki_handler.read_intermediate_cert()
+
+    async def test_read_intermediate_uses_correct_path(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off.return_value = {"exit_code": 0, "logs": "CERT"}
+        await pki_handler.read_intermediate_cert()
+        call_args = mock_docker.run_container_one_off.call_args
+        cmd = call_args[1].get("command") or call_args[0][1]
+        # _read_file_from_volume remaps /home/step → /data
+        assert "intermediate_ca.crt" in " ".join(cmd)
+
+    async def test_bootstrap_stores_intermediate_cert_when_enabled(self, mock_docker, state):
+        spec = {
+            "pki": {
+                "acme": {"listen_ip": "10.0.0.6", "canonical_name": "ca.platform.internal"},
+                "crl_enabled": False,
+                "ocsp_enabled": False,
+                "intermediate_ca_enabled": True,
+                "dnssec_enabled": False,
+            }
+        }
+
+        def side_effect(**kwargs):
+            cmd = kwargs.get("command", [])
+            if "intermediate_ca.crt" in " ".join(cmd):
+                return {
+                    "exit_code": 0,
+                    "logs": "-----BEGIN CERTIFICATE-----\nINTERMEDIATE\n-----END CERTIFICATE-----",
+                }
+            if "ca.crt" in " ".join(cmd):
+                return {
+                    "exit_code": 0,
+                    "logs": "-----BEGIN CERTIFICATE-----\nROOT\n-----END CERTIFICATE-----",
+                }
+            if "password.txt" in " ".join(cmd) and cmd[0] == "cat":
+                return {"exit_code": 0, "logs": "secret-password"}
+            return {"exit_code": 0, "logs": ""}
+
+        mock_docker.run_container_one_off = AsyncMock(side_effect=lambda **kw: side_effect(**kw))
+
+        # Pre-populate state so _generate_ca is skipped (already generated)
+        state.ca_cert_pem = "EXISTING_CA"
+        state.step_ca_container_id = "existing-container"
+
+        handler = PKIHandler(mock_docker, state, spec)
+        with patch.object(handler, "healthcheck", AsyncMock(return_value=True)):
+            await handler.bootstrap()
+
+        assert state.intermediate_ca_cert is not None
+        assert "INTERMEDIATE" in state.intermediate_ca_cert
+
+
+# ─────────────────────────────────────────────
+# CRL
+# ─────────────────────────────────────────────
+
+
+class TestCRLConfig:
+    async def test_inject_crl_config_adds_crl_section(self, pki_handler, mock_docker):
+        ca_config = {
+            "root": "/home/step/certs/root_ca.crt",
+            "authority": {"provisioners": []},
+        }
+        # First call reads config, subsequent calls are writes
+        mock_docker.run_container_one_off = AsyncMock(
+            side_effect=[
+                {"exit_code": 0, "logs": json.dumps(ca_config)},  # _read_ca_config
+                {"exit_code": 0, "logs": ""},  # _write_ca_config (cp)
+            ]
+        )
+
+        with patch("tempfile.NamedTemporaryFile") as mock_tmp, patch("os.unlink"):
+            mock_tmp.return_value.__enter__ = MagicMock(
+                return_value=MagicMock(name="f", write=MagicMock())
+            )
+            mock_tmp.return_value.__exit__ = MagicMock(return_value=False)
+            mock_tmp.return_value.__enter__.return_value.name = "/tmp/fake.json"
+            await pki_handler._inject_crl_config()
+
+        # Verify the write call contained a config with crl.enabled = true
+        write_call = mock_docker.run_container_one_off.call_args_list[1]
+        # The copy command will have been called; the config was written to a temp file
+        # which we can't easily inspect here — just verify the call was made
+        assert mock_docker.run_container_one_off.call_count == 2
+
+    async def test_inject_crl_raises_on_read_failure(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off.return_value = {"exit_code": 1, "logs": "error"}
+        with pytest.raises(PKIError, match="CA config"):
+            await pki_handler._inject_crl_config()
+
+
+# ─────────────────────────────────────────────
+# OCSP
+# ─────────────────────────────────────────────
+
+
+class TestOCSPConfig:
+    async def test_inject_ocsp_config_enables_ocsp_in_authority(self, pki_handler, mock_docker):
+        ca_config = {"authority": {"provisioners": []}}
+
+        def capture_write(**kwargs):
+            cmd = kwargs.get("command", [])
+            if cmd[0] == "cat":
+                return {"exit_code": 0, "logs": json.dumps(ca_config)}
+            return {"exit_code": 0, "logs": ""}
+
+        mock_docker.run_container_one_off = AsyncMock(side_effect=lambda **kw: capture_write(**kw))
+
+        with patch("tempfile.NamedTemporaryFile") as mock_tmp, patch("os.unlink"):
+            mock_file = MagicMock()
+            mock_file.name = "/tmp/fake.json"
+            mock_tmp.return_value.__enter__ = MagicMock(return_value=mock_file)
+            mock_tmp.return_value.__exit__ = MagicMock(return_value=False)
+            await pki_handler._inject_ocsp_config()
+
+        assert mock_docker.run_container_one_off.call_count == 2
+
+    async def test_inject_ocsp_raises_on_read_failure(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off.return_value = {"exit_code": 1, "logs": "io error"}
+        with pytest.raises(PKIError, match="CA config"):
+            await pki_handler._inject_ocsp_config()
+
+
+# ─────────────────────────────────────────────
+# DNSSEC
+# ─────────────────────────────────────────────
+
+
+class TestDNSSEC:
+    async def test_setup_dnssec_creates_volume(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off = AsyncMock(
+            side_effect=[
+                {"exit_code": 0, "logs": "Kinternal.+013+01234"},  # KSK
+                {"exit_code": 0, "logs": "Kinternal.+013+05678"},  # ZSK
+            ]
+        )
+        await pki_handler.setup_dnssec("internal")
+        mock_docker.ensure_volume.assert_called_once_with("netengines_dnssec_keys")
+
+    async def test_setup_dnssec_returns_key_names(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off = AsyncMock(
+            side_effect=[
+                {"exit_code": 0, "logs": "Kinternal.+013+01234"},
+                {"exit_code": 0, "logs": "Kinternal.+013+05678"},
+            ]
+        )
+        result = await pki_handler.setup_dnssec(
+            "internal", ksk_lifetime_days=365, zsk_lifetime_days=30
+        )
+        assert result["zone"] == "internal"
+        assert result["ksk_name"] == "Kinternal.+013+01234"
+        assert result["zsk_name"] == "Kinternal.+013+05678"
+        assert result["algorithm"] == "ECDSAP256SHA256"
+        assert result["ksk_lifetime_days"] == 365
+        assert result["zsk_lifetime_days"] == 30
+
+    async def test_setup_dnssec_uses_ksk_flag_for_ksk(self, pki_handler, mock_docker):
+        calls = []
+
+        def capture(**kwargs):
+            calls.append(kwargs.get("command", []))
+            return {"exit_code": 0, "logs": "Kinternal.+013+00001"}
+
+        mock_docker.run_container_one_off = AsyncMock(side_effect=lambda **kw: capture(**kw))
+        await pki_handler.setup_dnssec("internal")
+
+        ksk_cmd = calls[0]
+        zsk_cmd = calls[1]
+        assert "-f" in ksk_cmd and "KSK" in ksk_cmd
+        assert "-f" not in zsk_cmd or "KSK" not in zsk_cmd
+
+    async def test_setup_dnssec_raises_on_ksk_failure(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off = AsyncMock(
+            return_value={"exit_code": 1, "logs": "permission denied"}
+        )
+        with pytest.raises(PKIError, match="KSK generation failed"):
+            await pki_handler.setup_dnssec("internal")
+
+    async def test_setup_dnssec_raises_on_zsk_failure(self, pki_handler, mock_docker):
+        mock_docker.run_container_one_off = AsyncMock(
+            side_effect=[
+                {"exit_code": 0, "logs": "Kinternal.+013+01234"},
+                {"exit_code": 1, "logs": "keygen error"},
+            ]
+        )
+        with pytest.raises(PKIError, match="ZSK generation failed"):
+            await pki_handler.setup_dnssec("internal")
+
+    async def test_pki_flag_reads_from_pydantic_model(self):
+        from unittest.mock import MagicMock
+
+        spec = MagicMock()
+        spec.pki.dnssec_enabled = True
+        handler = PKIHandler(MagicMock(), RuntimeState(), spec)
+        assert handler._pki_flag("dnssec_enabled") is True
+
+    async def test_pki_flag_reads_from_dict_spec(self, mock_docker, state):
+        spec = {"pki": {"acme": {}, "dnssec_enabled": True}}
+        handler = PKIHandler(mock_docker, state, spec)
+        assert handler._pki_flag("dnssec_enabled") is True
+
+    async def test_pki_flag_defaults_to_false(self, pki_handler):
+        assert pki_handler._pki_flag("dnssec_enabled") is False
+
+
+# ─────────────────────────────────────────────
+# PKI Rotation Policy — dynamic cert types
+# ─────────────────────────────────────────────
+
+
+class TestPKIRotationPolicyWiring:
+    """_register_rotation_worker wires all cert types from the spec dynamically."""
+
+    def _make_context(self, overrides: dict):
+        spec = MagicMock()
+        spec.pki.rotation_policy.enabled = True
+        spec.pki.rotation_policy.default_interval_hours = 24
+        spec.pki.rotation_policy.default_warning_days = 30
+        spec.pki.rotation_policy.cert_type_overrides = overrides
+
+        ctx = MagicMock()
+        ctx.consumer_supervisor = MagicMock()
+        ctx.consumer_supervisor.register = MagicMock()
+        ctx.pgmq_client = AsyncMock()
+        ctx.runtime_state = RuntimeState()
+        ctx.logger = MagicMock()
+
+        return ctx, spec
+
+    def _run_and_capture(self, ctx, spec, handler) -> list:
+        """Patch PKICertRotationWorker and return the configs it was constructed with."""
+        captured = []
+
+        def fake_worker(pki, pgmq, configs):
+            captured.extend(configs)
+            m = MagicMock()
+            m.run = MagicMock()
+            return m
+
+        with patch("netengine.handlers.phase_pki.PKICertRotationWorker", side_effect=fake_worker):
+            handler._register_rotation_worker(ctx, MagicMock(), spec)
+
+        return captured
+
+    def test_builtin_four_cert_types_always_registered(self):
+        ctx, spec = self._make_context({})
+        configs = self._run_and_capture(ctx, spec, PKIPhaseHandler())
+        types = [c.cert_type for c in configs]
+        assert "platform_identity" in types
+        assert "inworld_identity" in types
+        assert "app" in types
+        assert "storage" in types
+
+    def test_extra_cert_types_from_overrides_are_included(self):
+        ctx, spec = self._make_context({"mail": {"rotation_interval_hours": 12}})
+        configs = self._run_and_capture(ctx, spec, PKIPhaseHandler())
+        types = [c.cert_type for c in configs]
+        assert "mail" in types
+
+    def test_override_values_are_applied_to_cert_type(self):
+        ctx, spec = self._make_context(
+            {"app": {"rotation_interval_hours": 6, "expiry_warning_days": 7}}
+        )
+        configs = self._run_and_capture(ctx, spec, PKIPhaseHandler())
+        app_cfg = next(c for c in configs if c.cert_type == "app")
+        assert app_cfg.rotation_interval_hours == 6
+        assert app_cfg.expiry_warning_days == 7
+
+    def test_disabled_policy_skips_registration(self):
+        ctx, spec = self._make_context({})
+        spec.pki.rotation_policy.enabled = False
+        handler = PKIPhaseHandler()
+        handler._register_rotation_worker(ctx, MagicMock(), spec)
+        ctx.consumer_supervisor.register.assert_not_called()
+
+    def test_no_supervisor_skips_registration(self):
+        ctx, spec = self._make_context({})
+        ctx.consumer_supervisor = None
+        handler = PKIPhaseHandler()
+        # Should not raise
+        handler._register_rotation_worker(ctx, MagicMock(), spec)