Chapter 39: Zero Trust Implementation — From Principle to Production¶

# Conceptual risk signal aggregation for Zero Trust policy engine
from dataclasses import dataclass
from enum import Enum
import math

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class AuthContext:
    user: str
    device_compliance: bool
    device_risk_score: float        # 0–1 from EDR
    location_known: bool
    ip_reputation: float            # 0–1 (1 = clean)
    sign_in_risk: float             # 0–1 from Identity Protection
    user_risk: float                # 0–1 long-term user risk
    time_of_day: int                # 0–23 hour
    mfa_method: str                 # "fido2" | "totp" | "sms" | "none"
    previous_session_normal: bool

class ZeroTrustPolicyEngine:
    MFA_STRENGTH = {"fido2": 1.0, "totp": 0.7, "push": 0.6, "sms": 0.3, "none": 0.0}

    def evaluate(self, ctx: AuthContext) -> dict:
        risk = self._calculate_risk(ctx)
        mfa_strength = self.MFA_STRENGTH.get(ctx.mfa_method, 0)
        decision = self._decide(risk, ctx, mfa_strength)
        return {
            "risk_score": round(risk, 2),
            "risk_level": self._level(risk).value,
            "mfa_strength": mfa_strength,
            "decision": decision,
            "required_controls": self._required_controls(risk, ctx)
        }

    def _calculate_risk(self, ctx: AuthContext) -> float:
        score = 0.0
        if not ctx.device_compliance:       score += 0.30
        if ctx.device_risk_score > 0.5:    score += ctx.device_risk_score * 0.20
        if not ctx.location_known:         score += 0.15
        if ctx.ip_reputation < 0.5:        score += (1 - ctx.ip_reputation) * 0.20
        score += ctx.sign_in_risk * 0.25
        score += ctx.user_risk * 0.20
        # After-hours penalty
        if ctx.time_of_day < 6 or ctx.time_of_day > 22:
            score += 0.05
        if not ctx.previous_session_normal: score += 0.10
        return min(score, 1.0)

    def _decide(self, risk: float, ctx: AuthContext, mfa: float) -> str:
        if risk > 0.80:                    return "BLOCK"
        if risk > 0.60 and mfa < 0.7:     return "REQUIRE_STRONG_MFA"
        if risk > 0.40 and not ctx.device_compliance: return "BLOCK_DOWNLOADS"
        if risk > 0.20:                    return "REQUIRE_MFA"
        return "ALLOW"

    def _level(self, score: float) -> RiskLevel:
        if score > 0.75: return RiskLevel.CRITICAL
        if score > 0.50: return RiskLevel.HIGH
        if score > 0.25: return RiskLevel.MEDIUM
        return RiskLevel.LOW

    def _required_controls(self, risk: float, ctx: AuthContext) -> list:
        controls = []
        if risk > 0.20:   controls.append("MFA")
        if risk > 0.40:   controls.append("Compliant device required")
        if risk > 0.60:   controls.append("Phishing-resistant MFA (FIDO2)")
        if risk > 0.75:   controls.append("Block all access; investigate")
        return controls

import requests

class DeviceRiskIntegration:
    """
    Pull MDE device risk scores for use in Zero Trust policy decisions.
    Integrates with Conditional Access Named Locations or MCAS policies.
    """

    def __init__(self, tenant_id: str, client_id: str, client_secret: str):
        self.base_url = "https://api.securitycenter.microsoft.com/api"
        self.token = self._get_token(tenant_id, client_id, client_secret)

    def _get_token(self, tenant_id, client_id, client_secret) -> str:
        resp = requests.post(
            f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
            data={"grant_type": "client_credentials",
                  "client_id": client_id,
                  "client_secret": client_secret,
                  "scope": "https://api.securitycenter.microsoft.com/.default"}
        )
        return resp.json()["access_token"]

    def get_high_risk_devices(self) -> list:
        """Fetch all devices with HIGH or CRITICAL risk scores."""
        resp = requests.get(
            f"{self.base_url}/machines?$filter=riskScore in ('High','Critical')"
            "&$select=id,computerDnsName,riskScore,healthStatus,lastSeen",
            headers={"Authorization": f"Bearer {self.token}"}
        )
        return resp.json().get("value", [])

    def isolate_device(self, machine_id: str, reason: str) -> dict:
        """Isolate a compromised device — removes from network."""
        resp = requests.post(
            f"{self.base_url}/machines/{machine_id}/isolate",
            headers={"Authorization": f"Bearer {self.token}",
                     "Content-Type": "application/json"},
            json={"Comment": reason, "IsolationType": "Selective"}
        )
        return resp.json()

ZT_METRICS = {
    # Identity
    "mfa_adoption_rate": {
        "formula": "users_with_mfa / total_users * 100",
        "target": 100,
        "current_baseline": 67,
        "unit": "%"
    },
    "phishing_resistant_mfa_rate": {
        "formula": "users_with_fido2_or_cert / total_users * 100",
        "target": 80,
        "unit": "%"
    },
    "conditional_access_coverage": {
        "formula": "apps_under_ca / total_apps * 100",
        "target": 100,
        "unit": "%"
    },

    # Devices
    "device_compliance_rate": {
        "formula": "compliant_devices / enrolled_devices * 100",
        "target": 95,
        "unit": "%"
    },
    "edr_coverage": {
        "formula": "edr_installed / total_endpoints * 100",
        "target": 99,
        "unit": "%"
    },

    # Networks
    "vpn_to_ztna_migration": {
        "formula": "ztna_users / total_remote_users * 100",
        "target": 100,
        "unit": "%"
    },
    "east_west_segmented_flows": {
        "formula": "microsegmented_flows / total_lateral_flows * 100",
        "target": 85,
        "unit": "%"
    },

    # Applications
    "sso_adoption": {
        "formula": "apps_in_sso / total_apps * 100",
        "target": 95,
        "unit": "%"
    },
    "privileged_standing_access": {
        "formula": "priv_accounts_with_standing_access / total_priv_accounts",
        "target": 0,
        "note": "Lower is better — target zero standing privilege"
    },

    # Data
    "classified_data_pct": {
        "formula": "labeled_documents / total_documents * 100",
        "target": 90,
        "unit": "%"
    },
    "dlp_policy_coverage": {
        "formula": "data_channels_with_dlp / total_channels * 100",
        "target": 100,
        "unit": "%"
    }
}

def generate_zt_scorecard(current_values: dict) -> None:
    print(f"\n{'Metric':<40} {'Current':>10} {'Target':>10} {'Status':>10}")
    print("-" * 75)
    for metric, config in ZT_METRICS.items():
        current = current_values.get(metric, 0)
        target = config.get("target", 100)
        # For "lower is better" metrics
        lower_better = config.get("note", "").startswith("Lower")
        if lower_better:
            ok = current <= target
        else:
            ok = current >= target * 0.9  # Within 10% of target = acceptable
        status = "✓ OK" if ok else "✗ GAP"
        print(f"{metric:<40} {str(current):>10} {str(target):>10} {status:>10}")