Implementing Zero-Trust Boundaries in DR Sandboxes

# dr-sandbox-network-policy.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: dr-drill-isolation
  namespace: dr-sandbox-
spec:
  podSelector:
    matchLabels:
      app: restored-db
  policyTypes:
    - Ingress
    - Egress
  ingress:
    - from:
        - podSelector:
            matchLabels:
              role: validation-agent
      ports:
        - protocol: TCP
          port: 5432
  egress: []

import logging
import time
from typing import Optional
from pydantic import BaseModel, Field, ValidationError
from kubernetes import client, config
from kubernetes.client.rest import ApiException
import hvac  # HashiCorp Vault client for credential generation

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

class DrillManifest(BaseModel):
    drill_id: str
    backup_snapshot: str
    db_engine: str
    target_port: int = 5432
    orchestrator_cidr: str
    vault_path: str
    ttl_seconds: int = 900

class ZeroTrustDrillOrchestrator:
    def __init__(self, manifest: DrillManifest):
        self.manifest = manifest
        self.namespace = f"dr-sandbox-{manifest.drill_id}"
        self.vault_client = hvac.Client(url="https://vault.internal", token="DRILL_SERVICE_TOKEN")
        config.load_incluster_config()
        self.k8s_net = client.NetworkingV1Api()
        self.k8s_core = client.CoreV1Api()

    def apply_network_policy(self) -> None:
        policy = client.V1NetworkPolicy(
            api_version="networking.k8s.io/v1",
            kind="NetworkPolicy",
            metadata=client.V1ObjectMeta(name="dr-drill-isolation", namespace=self.namespace),
            spec=client.V1NetworkPolicySpec(
                pod_selector=client.V1LabelSelector(match_labels={"app": "restored-db"}),
                policy_types=["Ingress", "Egress"],
                ingress=[client.V1NetworkPolicyIngressRule(
                    from_=[client.V1NetworkPolicyPeer(
                        pod_selector=client.V1LabelSelector(match_labels={"role": "validation-agent"})
                    )],
                    ports=[client.V1NetworkPolicyPort(protocol="TCP", port=self.manifest.target_port)]
                )],
                egress=[]  # default-deny egress, matching the YAML manifest
            )
        )
        try:
            self.k8s_net.create_namespaced_network_policy(namespace=self.namespace, body=policy)
            logging.info("Ephemeral NetworkPolicy applied to %s", self.namespace)
        except ApiException as e:
            logging.error("NetworkPolicy creation failed: %s", e.reason)
            raise

    def inject_credentials(self) -> dict:
        # Request short-lived DB credentials bound to drill scope
        # TTL is enforced on the Vault role (default_ttl/max_ttl), not per request;
        # generate_credentials() takes only the role name (and optional mount_point).
        response = self.vault_client.secrets.database.generate_credentials(
            name=self.manifest.vault_path
        )
        creds = response["data"]
        logging.info("Short-lived credentials issued. TTL: %ds", self.manifest.ttl_seconds)
        return creds

    def run_validation(self, creds: dict) -> bool:
        # Placeholder for actual DB validation logic (schema check, checksum, row count)
        # In production, use psycopg2/mysql-connector with strict connection timeouts
        logging.info("Executing automated backup validation with scoped credentials...")
        # Simulate validation execution
        time.sleep(2)
        return True

    def cleanup(self) -> None:
        try:
            self.k8s_net.delete_namespaced_network_policy(
                name="dr-drill-isolation", namespace=self.namespace
            )
            logging.info("NetworkPolicy removed. Sandbox teardown initiated.")
        except ApiException as e:
            logging.warning("Policy cleanup skipped: %s", e.reason)

def execute_drill(manifest_path: str) -> None:
    try:
        with open(manifest_path, "r") as f:
            import json
            data = json.load(f)
        manifest = DrillManifest(**data)
    except (ValidationError, FileNotFoundError) as e:
        logging.error("Invalid drill manifest: %s", e)
        return

    orchestrator = ZeroTrustDrillOrchestrator(manifest)
    orchestrator.apply_network_policy()
    creds = orchestrator.inject_credentials()
    success = orchestrator.run_validation(creds)
    orchestrator.cleanup()
    
    if success:
        logging.info("DR drill %s passed zero-trust validation.", manifest.drill_id)
    else:
        logging.error("DR drill %s failed validation checks.", manifest.drill_id)

if __name__ == "__main__":
    execute_drill("drill_manifest.json")