zero_trust_validator.py

"""
zero_trust_validator.py — Zero Trust Architecture Posture Assessment
NIST SP 800-207 & CISA Zero Trust Maturity Model
Author: securekamal
"""

import json
import logging
import argparse
from dataclasses import dataclass, field
from typing import Optional
from pathlib import Path

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


class MaturityLevel:
    TRADITIONAL = ("Traditional", 0, 24)
    INITIAL = ("Initial", 25, 49)
    ADVANCED = ("Advanced", 50, 74)
    OPTIMAL = ("Optimal", 75, 100)

    @staticmethod
    def from_score(score: int) -> str:
        if score < 25:
            return "Traditional"
        elif score < 50:
            return "Initial"
        elif score < 75:
            return "Advanced"
        return "Optimal"


@dataclass
class PillarScore:
    name: str
    score: int
    max_score: int
    maturity: str
    findings: list[str] = field(default_factory=list)
    quick_wins: list[tuple[str, int]] = field(default_factory=list)  # (action, score_gain)


@dataclass
class ZTAReport:
    organization: str
    overall_score: int
    maturity_level: str
    pillars: list[PillarScore] = field(default_factory=list)
    nist_coverage: dict[str, bool] = field(default_factory=dict)
    roadmap: list[str] = field(default_factory=list)

    def to_text(self) -> str:
        bar = lambda s, m: "█" * (s * 10 // m) + "░" * (10 - s * 10 // m)
        lines = [
            f"\nZERO TRUST MATURITY ASSESSMENT — {self.organization}",
            "=" * 54,
            f"\nOverall Maturity: {self.maturity_level} ({self.overall_score}/100)\n",
            "Pillar Scores:",
        ]
        for p in self.pillars:
            pct = p.score * 100 // p.max_score if p.max_score else 0
            lines.append(f"  {p.name:<12} {bar(pct, 100)}  {pct}/100  [{p.maturity}]")

        lines += ["\nTop Quick Wins:"]
        all_wins = []
        for p in self.pillars:
            for action, gain in p.quick_wins:
                all_wins.append((p.name, action, gain))
        all_wins.sort(key=lambda x: -x[2])
        for i, (pillar, action, gain) in enumerate(all_wins[:5], 1):
            lines.append(f"  {i}. [{pillar}] {action}  +{gain}pts")

        lines += ["\nNIST 800-207 Tenet Coverage:"]
        for tenet, covered in self.nist_coverage.items():
            icon = "✅" if covered else "❌"
            lines.append(f"  {icon} {tenet}")

        if self.roadmap:
            lines += ["\nRemediation Roadmap:"]
            for item in self.roadmap:
                lines.append(f"  • {item}")

        return "\n".join(lines)


# ─────────────────────────────────────────────
#  PILLAR ANALYZERS
# ─────────────────────────────────────────────

class IdentityAnalyzer:
    def analyze(self, config: dict) -> PillarScore:
        score = 0
        findings = []
        wins = []

        mfa_pct = config.get("mfa_enforced_percent", 0)
        if mfa_pct >= 95:
            score += 25
        elif mfa_pct >= 80:
            score += 15
            wins.append(("Enforce MFA on remaining users", 10))
        else:
            findings.append(f"MFA only {mfa_pct}% — critical gap")
            wins.append(("Enforce MFA org-wide (target 100%)", 25))

        if config.get("phishing_resistant_mfa"):
            score += 20
        else:
            findings.append("Phishing-resistant MFA (FIDO2/passkeys) not enforced")
            wins.append(("Deploy FIDO2 passkeys for all privileged users", 12))

        if config.get("privileged_access_management"):
            score += 20
        else:
            findings.append("No PAM solution — privileged accounts unmanaged")
            wins.append(("Deploy CyberArk/BeyondTrust for PAM", 10))

        if config.get("just_in_time_access"):
            score += 20
        else:
            findings.append("Standing privileged access — JIT not implemented")
            wins.append(("Implement JIT access for admin roles", 8))

        sa_gov = config.get("service_account_governance", "none")
        if sa_gov == "full":
            score += 15
        elif sa_gov == "partial":
            score += 7
            findings.append("Service account governance partial — some unmanaged SAs")

        return PillarScore("Identity", score, 100, MaturityLevel.from_score(score), findings, wins)


class NetworkAnalyzer:
    def analyze(self, config: dict) -> PillarScore:
        score = 0
        findings = []
        wins = []

        seg = config.get("microsegmentation", "none")
        if seg == "full":
            score += 30
        elif seg == "partial":
            score += 15
            findings.append("Microsegmentation partial — east-west traffic gaps")
            wins.append(("Complete workload microsegmentation (Illumio/Guardicore)", 15))
        else:
            findings.append("No microsegmentation — flat network enables lateral movement")
            wins.append(("Deploy microsegmentation platform", 30))

        if config.get("east_west_tls"):
            score += 20
        else:
            findings.append("East-west traffic not encrypted with mTLS")
            wins.append(("Enable mTLS in service mesh (Istio/Linkerd)", 10))

        if config.get("dns_security"):
            score += 10
        else:
            findings.append("No DNS security filtering — C2 over DNS possible")

        if config.get("network_detection_response"):
            score += 25
        else:
            findings.append("No NDR deployed — east-west threats invisible")
            wins.append(("Deploy NDR (Darktrace/ExtraHop) for behavioral detection", 8))

        if config.get("beyondcorp_proxy"):
            score += 15
        else:
            findings.append("No application proxy — implicit trust in network location")

        return PillarScore("Network", score, 100, MaturityLevel.from_score(score), findings, wins)


class DeviceAnalyzer:
    def analyze(self, config: dict) -> PillarScore:
        score = 0
        findings = []
        wins = []

        mdm_pct = config.get("mdm_enrollment_percent", 0)
        if mdm_pct >= 95:
            score += 25
        elif mdm_pct >= 75:
            score += 15
            wins.append(("Enroll remaining devices in MDM", 10))
        else:
            findings.append(f"MDM only {mdm_pct}% — unmanaged devices accessing resources")

        edr_pct = config.get("edr_coverage_percent", 0)
        if edr_pct >= 95:
            score += 25
        elif edr_pct >= 75:
            score += 15
            wins.append(("Expand EDR to all endpoints", 10))
        else:
            findings.append(f"EDR only {edr_pct}% — detection gaps on endpoints")

        if config.get("certificate_auth"):
            score += 25
        else:
            findings.append("No certificate-based device auth — device identity unverified")

        if config.get("compliance_checks_at_auth"):
            score += 25
        else:
            findings.append("Device compliance not checked at auth time — stale posture data")
            wins.append(("Enforce real-time device compliance in Conditional Access", 7))

        return PillarScore("Devices", score, 100, MaturityLevel.from_score(score), findings, wins)


class ApplicationAnalyzer:
    def analyze(self, config: dict) -> PillarScore:
        score = 0
        findings = []
        wins = []

        if config.get("api_gateway"):
            score += 20
        else:
            findings.append("No API gateway — direct backend exposure")

        if config.get("waap_waf"):
            score += 20
        else:
            findings.append("No WAAP/WAF — application layer unprotected")

        if config.get("continuous_authz"):
            score += 25
        else:
            findings.append("Static authorization — no continuous authz re-evaluation")
            wins.append(("Implement continuous authorization in API gateway", 7))

        if config.get("step_up_auth"):
            score += 15
        else:
            findings.append("No step-up auth for sensitive actions")

        secrets = config.get("secrets_management", "none")
        if secrets in ("vault", "aws-secrets-manager", "azure-keyvault"):
            score += 20
        elif secrets == "partial":
            score += 10
        else:
            findings.append("No centralized secrets management — credentials in config/env")

        return PillarScore("Apps", score, 100, MaturityLevel.from_score(score), findings, wins)


class DataAnalyzer:
    def analyze(self, config: dict) -> PillarScore:
        score = 0
        findings = []
        wins = []

        classify = config.get("classification_coverage", "none")
        if classify == "full":
            score += 25
        elif classify == "partial":
            score += 12
            wins.append(("Complete data classification across all stores", 6))
        else:
            findings.append("No data classification — DLP and access controls ineffective")
            wins.append(("Deploy Purview/Macie for data classification", 25))

        if config.get("dlp_deployed"):
            score += 20
        else:
            findings.append("No DLP — data exfiltration undetected")

        if config.get("encryption_at_rest"):
            score += 20
        else:
            findings.append("Data not encrypted at rest — storage compromise = data breach")

        if config.get("encryption_in_transit"):
            score += 20
        else:
            findings.append("Data not encrypted in transit — interception risk")

        if config.get("rights_management"):
            score += 15
        else:
            findings.append("No IRM/DRM — data protection doesn't follow documents")

        return PillarScore("Data", score, 100, MaturityLevel.from_score(score), findings, wins)


# ─────────────────────────────────────────────
#  MAIN ASSESSOR
# ─────────────────────────────────────────────

NIST_TENETS = [
    "All data sources and computing services considered resources",
    "All communication secured regardless of network location",
    "Access to individual resources granted per-session",
    "Access determined by dynamic policy including observable state",
    "Enterprise monitors and measures integrity of all assets",
    "All resource auth is dynamic and strictly enforced",
    "Enterprise collects telemetry for improving security posture",
]


class ZeroTrustAssessor:

    def assess(self, config: dict) -> ZTAReport:
        org = config.get("organization", "Unknown")
        pillars = [
            IdentityAnalyzer().analyze(config.get("identity", {})),
            NetworkAnalyzer().analyze(config.get("network", {})),
            DeviceAnalyzer().analyze(config.get("devices", {})),
            ApplicationAnalyzer().analyze(config.get("applications", {})),
            DataAnalyzer().analyze(config.get("data", {})),
        ]

        overall = sum(p.score for p in pillars) // len(pillars)
        maturity = MaturityLevel.from_score(overall)

        # Simplified NIST tenet coverage heuristic
        nist_coverage = {
            NIST_TENETS[0]: True,
            NIST_TENETS[1]: config.get("network", {}).get("east_west_tls", False),
            NIST_TENETS[2]: config.get("applications", {}).get("continuous_authz", False),
            NIST_TENETS[3]: config.get("identity", {}).get("just_in_time_access", False),
            NIST_TENETS[4]: config.get("devices", {}).get("compliance_checks_at_auth", False),
            NIST_TENETS[5]: config.get("identity", {}).get("mfa_enforced_percent", 0) >= 95,
            NIST_TENETS[6]: config.get("network", {}).get("network_detection_response", False),
        }

        roadmap = self._generate_roadmap(pillars, maturity)
        return ZTAReport(org, overall, maturity, pillars, nist_coverage, roadmap)

    def _generate_roadmap(self, pillars: list[PillarScore], current: str) -> list[str]:
        next_level = {"Traditional": "Initial", "Initial": "Advanced", "Advanced": "Optimal"}.get(current)
        if not next_level:
            return ["🏆 Optimal maturity achieved — maintain and continuously improve"]
        roadmap = [f"Roadmap to reach {next_level}:"]
        all_wins = []
        for p in pillars:
            for action, gain in p.quick_wins:
                all_wins.append((p.name, action, gain))
        all_wins.sort(key=lambda x: -x[2])
        for pillar, action, gain in all_wins[:8]:
            roadmap.append(f"  [{pillar}] {action} (+{gain} pts)")
        return roadmap


def main():
    parser = argparse.ArgumentParser(description="Zero Trust Validator")
    parser.add_argument("command", choices=["assess", "quick-check"])
    parser.add_argument("--config", help="ZTA config JSON file")
    parser.add_argument("--format", choices=["text", "json"], default="text")
    parser.add_argument("--out", help="Output file")
    args = parser.parse_args()

    if args.command == "assess":
        config = json.loads(Path(args.config).read_text()) if args.config else {}
        assessor = ZeroTrustAssessor()
        report = assessor.assess(config)
        output = report.to_text() if args.format == "text" else json.dumps(report.__dict__, indent=2, default=str)
        if args.out:
            Path(args.out).write_text(output)
        else:
            print(output)

    elif args.command == "quick-check":
        print("Quick check: run with --config zt-config.json for full assessment")
        print("See README for config schema.")


if __name__ == "__main__":
    main()