zero-trust

name	zero-trust
description	Design and implement zero-trust network architecture — never trust, always verify. Outputs identity-aware proxy config, mTLS policies, device trust enforcement, micro-segmentation rules, and continuous verification pipelines.
argument-hint	["current perimeter model","identity provider","cloud provider","compliance requirements"]
allowed-tools	Read, Write, Bash

Zero Trust Architecture

Zero trust replaces the castle-and-moat perimeter model — "trust everything inside the network" — with continuous verification: every request is authenticated, authorized, and encrypted regardless of where it originates. The network is assumed hostile.

Core Principles

Old Model	Zero Trust
Trust the network perimeter	Trust no network, including internal
Verify once at login	Verify every request, continuously
Broad network access	Least-privilege micro-segmentation
VPN for remote access	Identity-aware proxy for every resource
Static firewall rules	Dynamic policy based on identity + device posture

Process

1. Inventory assets — every service, database, device, and human identity that needs access to what.
2. Define protect surfaces — critical data, applications, assets (not the attack surface — the thing worth protecting).
3. Map transaction flows — how data flows between systems; who talks to what and why.
4. Implement identity foundation — SSO, MFA, short-lived credentials, device certificates.
5. Deploy identity-aware proxy — all internal resources accessed via proxy, never directly.
6. Enforce mTLS — mutual TLS for service-to-service; workload identity via SPIFFE/SPIRE.
7. Micro-segment networks — replace broad VLANs with per-service network policies.
8. Continuous monitoring — log every request, detect anomalies, revoke on signals.
9. Automate posture checks — device health, patch status, certificate validity checked on every access.

Output Format

Identity Foundation (SPIFFE/SPIRE for Workload Identity)

# spire-server-config.hcl
server {
  bind_address = "0.0.0.0"
  bind_port    = "8081"
  trust_domain = "prod.example.com"
  data_dir     = "/opt/spire/data/server"
  log_level    = "INFO"

  ca_subject {
    country      = ["US"]
    organization = ["Example Corp"]
    common_name  = ""
  }

  ca_ttl        = "24h"   # Short-lived CA
  default_svid_ttl = "1h" # Short-lived workload certs
}

plugins {
  DataStore "sql" {
    plugin_data {
      database_type = "postgres"
      connection_string = "postgresql://spire:${SPIRE_DB_PASS}@postgres:5432/spire?sslmode=require"
    }
  }

  NodeAttestor "k8s_psat" {
    plugin_data {
      clusters = {
        "prod-cluster" = {
          service_account_allow_list  []
        }
      }
    }
  }

    {
     {
        
    }
  }
}

# spire-agent-config.hcl
agent {
  data_dir     = "/opt/spire/data/agent"
  log_level    = "INFO"
  server_address = "spire-server"
  server_port  = "8081"
  trust_domain = "prod.example.com"
  socket_path  = "/run/spire/sockets/agent.sock"
}

plugins {
  NodeAttestor "k8s_psat" {
    plugin_data {
      cluster    = "prod-cluster"
      token_path = "/var/run/secrets/tokens/spire-agent"
    }
  }

  WorkloadAttestor "k8s" {
    plugin_data {
      skip_kubelet_verification = false
    }
  }

  KeyManager "memory" { plugin_data {} }
}

# SPIFFE registration entry — order-service gets an identity
apiVersion: spire.spiffe.io/v1alpha1
kind: ClusterSPIFFEID
metadata:
  name: order-service
spec:
  spiffeIDTemplate: "spiffe://prod.example.com/ns/{{ .PodMeta.Namespace }}/sa/{{ .PodSpec.ServiceAccountName }}"
  podSelector:
    matchLabels:
      app: order-service
  dnsNameTemplates:
    - "order-service.production.svc.cluster.local"

Identity-Aware Proxy (BeyondCorp / Cloudflare Access pattern)

# proxy/zero_trust_proxy.py
import jwt
import httpx
import time
import hashlib
from dataclasses import dataclass
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import Response
import logging

logger = logging.getLogger("zero-trust-proxy")

@dataclass
class AccessContext:
    user_id: str
    email: str
    groups: list[str]
    device_id: str
    device_trusted: bool
    device_patch_level: str
    auth_time: int       # When the user last authenticated
    risk_score: float    # 0.0 = low risk, 1.0 = high risk

class ZeroTrustProxy:
    """
    Every request to internal services must pass through this proxy.
    No direct network access to services — even from inside the VPC.
    """
    
    def __init__(self):
        self.policies = PolicyEngine()
        self.device_trust = DeviceTrustService()
        self.risk_engine = RiskEngine()
    
    async def handle_request(self, request: Request) -> Response:
        # Step 1: Verify identity token (OIDC/JWT from IdP)
        context = await self._authenticate(request)
        
        
         ._verify_device(request, context)
        
        
         ._check_session_freshness(context)
        
        
        resource = ._extract_resource(request)
         ._authorize(context, resource, request.method)
        
        
        risk =  .risk_engine.score(context, request)
         risk.score > :
            logger.warning()
             HTTPException(, detail=)
        
        
        response =  ._proxy_upstream(request, context)
        
        
        ._audit(context, resource, request.method, response.status_code, risk.score)
        
         response
    
      () -> AccessContext:
        token = request.headers.get(, ).removeprefix()
          token:
             HTTPException(, detail=)
        
        :
            
            payload = jwt.decode(
                token,
                key=._get_jwks(),
                algorithms=[],
                audience=,
            )
         jwt.ExpiredSignatureError:
             HTTPException(, detail=)
         jwt.InvalidTokenError  e:
             HTTPException(, detail=)
        
         AccessContext(
            user_id=payload[],
            email=payload[],
            groups=payload.get(, []),
            device_id=request.headers.get(, ),
            device_trusted=,  
            device_patch_level=,
            auth_time=payload.get(, ),
            risk_score=,
        )
    
      ():
        device_cert = request.headers.get()
        
          device_cert:
            
            context.device_trusted = 
            
        
        device_info =  .device_trust.verify_certificate(device_cert)
        
          device_info.is_managed:
             HTTPException(, detail=)
        
          device_info.is_patched:
             HTTPException(, detail=)
        
          device_info.disk_encrypted:
             HTTPException(, detail=)
        
        context.device_trusted = 
        context.device_patch_level = device_info.patch_level
    
      ():
        
        age_minutes = (time.time() - context.auth_time) / 
        
        
         age_minutes > :
             HTTPException(, detail=, headers={: })
    
      ():
        decision =  .policies.evaluate(
            subject={
                : context.user_id,
                : context.groups,
                : context.device_trusted,
            },
            action=method,
            resource=resource,
        )
        
          decision.allowed:
            logger.warning()
             HTTPException(, detail=)
    
     ():
        logger.info(, extra={
            : context.user_id,
            : context.email,
            : context.device_id,
            : context.device_trusted,
            : resource,
            : method,
            : status,
            : (risk, ),
            : context.groups,
        })

Policy Definition (Open Policy Agent)

# policies/zero_trust.rego
package zero_trust

import future.keywords.if
import future.keywords.in

default allow := false

# Admin group has full access
allow if {
    "platform-admins" in input.subject.groups
    input.subject.device_trusted == true
}

# Engineers can read all internal services, write to non-production
allow if {
    "engineers" in input.subject.groups
    input.action in {"GET", "HEAD", "OPTIONS"}
    input.subject.device_trusted == true
}

allow if {
    "engineers" in input.subject.groups
    input.action in {"POST", "PUT", "PATCH", "DELETE"}
    not startswith(input.resource, "/production/")
    input.subject.device_trusted == true
}

# Production writes require both group membership AND MFA
allow if {
    "senior-engineers" in input.subject.groups
    startswith(input.resource, "/production/")
    input.subject.mfa_verified == true
    input.subject.device_trusted == true
    # Require fresh auth for production writes
    time.now_ns() - input.subject.auth_time_ns < 3600000000000  # 1 hour
}

# Service accounts (SPIFFE identities) get narrow access
allow if {
    startswith(input.subject.user_id, "spiffe://")
    service_allowed_resources[input.subject.user_id][input.resource]
    input.action in {"GET", "POST"}
}

service_allowed_resources := {
    "spiffe://prod.example.com/ns/production/sa/order-service": {
        "/inventory/check": true,
        "/payments/charge": true,
    },
    "spiffe://prod.example.com/ns/production/sa/user-service": {
        "/notifications/send": true,
    },
}

# Deny reason for audit
deny_reason := "untrusted device" if { input.subject.device_trusted == false }
deny_reason := "insufficient group membership" if {
    not "engineers" in input.subject.groups
    not "platform-admins" in input.subject.groups
}
deny_reason := "stale authentication" if {
    startswith(input.resource, "/production/")
    time.now_ns() - input.subject.auth_time_ns >= 3600000000000
}

Network Micro-Segmentation (Kubernetes NetworkPolicy)

# Default deny-all for production namespace
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: deny-all-default
  namespace: production
spec:
  podSelector: {}
  policyTypes: [Ingress, Egress]
---
# Allow order-service to talk ONLY to inventory and payment
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: order-service-egress
  namespace: production
spec:
  podSelector:
    matchLabels:
      app: order-service
  policyTypes: [Egress]
  egress:
    - to:
        - podSelector:
            matchLabels:
              app: inventory-service
      ports:
        - port: 8080
    - to:
        - podSelector:
            matchLabels:
              app: payment-service
      ports:
          
     
         
            
               
      
               


 
 

   
   

  
    
       
   []
  
     
         
            
               
      
          
     
         
            
                    

Device Trust Enforcement (MDM Integration)

# device_trust/mdm_checker.py
import httpx
from dataclasses import dataclass
from functools import lru_cache
import time

@dataclass
class DevicePosture:
    device_id: str
    is_managed: bool           # Enrolled in MDM (Jamf, Intune, etc.)
    is_patched: bool           # No critical patches outstanding
    disk_encrypted: bool       # FileVault / BitLocker enabled
    screen_lock_enabled: bool
    os_version: str
    patch_level: str
    compliance_status: str     # "compliant" | "warning" | "non-compliant"
    last_checked: float

class DeviceTrustService:
    def __init__(self, mdm_url: str, api_key: str):
        self.mdm_url = mdm_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self._cache: dict[str, tuple[DevicePosture, float]] = {}
        self.cache_ttl = 300  # 5 min — don't hammer MDM on every request
    
    async def verify_certificate() -> DevicePosture:
        
        device_id = ._extract_device_id(device_cert)
        
        
         device_id  ._cache:
            posture, cached_at = ._cache[device_id]
             time.time() - cached_at < .cache_ttl:
                 posture
        
        
          httpx.AsyncClient()  client:
            resp =  client.get(
                ,
                headers=.headers,
                timeout=
            )
        
         resp.status_code == :
             DevicePosture(
                device_id=device_id,
                is_managed=,
                is_patched=,
                disk_encrypted=,
                screen_lock_enabled=,
                os_version=,
                patch_level=,
                compliance_status=,
                last_checked=time.time(),
            )
        
        data = resp.json()
        posture = DevicePosture(
            device_id=device_id,
            is_managed=,
            is_patched=data.get(, ) == ,
            disk_encrypted=data.get(, ),
            screen_lock_enabled=data.get(, ),
            os_version=data.get(, ),
            patch_level=data.get(, ),
            compliance_status=data.get(, ),
            last_checked=time.time(),
        )
        
        ._cache[device_id] = (posture, time.time())
         posture

Continuous Monitoring & Anomaly Detection

# monitoring/anomaly_detector.py
from collections import defaultdict
import statistics

class RiskEngine:
    def __init__(self):
        self._user_baselines: dict[str, dict] = {}  # Historical access patterns
    
    async def score(self, context, request) -> "RiskScore":
        signals = []
        
        # Impossible travel: access from two distant locations in short time
        if await self._is_impossible_travel(context.user_id, request):
            signals.append(("impossible_travel", 0.9))
        
        # New device: user has never accessed from this device
        if context.device_id not in self._known_devices(context.user_id):
            signals.append(("new_device", 0.4))
        
        # Unusual time: access at 3am for a user who normally works 9-5
        if self._is_unusual_time(context.user_id):
            signals.append(("unusual_time", 0.2))
        
        # High-value target with unmanaged device
        if not context.device_trusted and "/admin" in str(request.url):
            signals.append((, ))
        
        
          ._is_bulk_access(context.user_id, request):
            signals.append((, ))
        
        
        risk_score = ((score  _, score  signals), default=)
        
         RiskScore(
            score=risk_score,
            signals=[name  name, _  signals],
            recommendation=  risk_score >     risk_score >   
        )

Rules

• Never trust the network — internal traffic requires the same authentication as external traffic.
• Short-lived credentials everywhere — tokens expire in hours, service certs expire in days; rotate automatically.
• Device posture is part of the policy — unmanaged or unpatched devices get no access to sensitive resources.
• Least-privilege network segmentation — services may only connect to the specific other services they need.
• Log every access decision — include user, device, resource, method, risk score, and grant/deny.
• Step-up authentication for sensitive actions — production writes require re-authentication within the last hour.
• Workload identity, not service passwords — use SPIFFE/SPIRE for service-to-service; never shared secrets.
• Fail closed, not open — if the policy engine is unavailable, deny access rather than defaulting to allow.
• Test with red team exercises — verify that lateral movement from a compromised host is actually blocked.
• Automate certificate rotation — expiring certs cause outages; automate renewal 72h before expiry.