#!/usr/bin/env python3
"""CyberArk PAM configuration audit agent.

Audits CyberArk Privileged Access Management via the REST API to
verify safe configurations, privileged account inventory, platform
assignments, and password rotation compliance.
"""
import argparse
import json
import os
import sys
from datetime import datetime, timezone

try:
    import requests
except ImportError:
    print("[!] 'requests' required: pip install requests", file=sys.stderr)
    sys.exit(1)


def get_cyberark_config():
    """Return CyberArk PVWA URL."""
    pvwa_url = os.environ.get("CYBERARK_PVWA_URL", "").rstrip("/")
    if not pvwa_url:
        print("[!] Set CYBERARK_PVWA_URL env var", file=sys.stderr)
        sys.exit(1)
    return pvwa_url


def authenticate(pvwa_url, username, password, auth_type="CyberArk"):
    """Authenticate and get session token."""
    url = f"{pvwa_url}/PasswordVault/API/Auth/{auth_type}/Logon"
    resp = requests.post(url, json={"username": username, "password": password},
                         verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=30)  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
    resp.raise_for_status()
    token = resp.json().strip('"')
    print(f"[+] Authenticated as {username}")
    return token


def api_call(pvwa_url, token, endpoint, method="GET", data=None, params=None):
    """Make authenticated API call."""
    url = f"{pvwa_url}/PasswordVault/API{endpoint}"
    headers = {"Authorization": token, "Content-Type": "application/json"}
    if method == "POST":
        resp = requests.post(url, headers=headers, json=data, params=params,
                             verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=30)  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
    else:
        resp = requests.get(url, headers=headers, params=params,
                            verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=30)  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
    resp.raise_for_status()
    return resp.json()


def audit_safes(pvwa_url, token):
    """Audit safe configurations."""
    findings = []
    print("[*] Auditing safes...")
    data = api_call(pvwa_url, token, "/Safes", params={"limit": 1000})
    safes = data.get("value", data.get("Safes", []))

    for safe in safes:
        name = safe.get("safeName", safe.get("SafeName", ""))
        member_count = safe.get("numberOfMembers", safe.get("NumberOfMembers", 0))
        days_retention = safe.get("numberOfDaysRetention", safe.get("NumberOfDaysRetention", 0))
        versions = safe.get("numberOfVersionsRetention", safe.get("NumberOfVersionsRetention", 0))

        if member_count == 0:
            findings.append({
                "safe": name, "check": "Empty safe (no members)",
                "severity": "MEDIUM", "detail": "Safe has no members assigned",
            })
        if days_retention == 0 and versions == 0:
            findings.append({
                "safe": name, "check": "No retention policy",
                "severity": "HIGH",
                "detail": "No password history retention configured",
            })

    print(f"[+] Audited {len(safes)} safes")
    return findings, safes


def audit_accounts(pvwa_url, token, safe_name=None):
    """Audit privileged accounts."""
    findings = []
    print("[*] Auditing privileged accounts...")
    params = {"limit": 1000}
    if safe_name:
        params["filter"] = f"safeName eq {safe_name}"
    data = api_call(pvwa_url, token, "/Accounts", params=params)
    accounts = data.get("value", [])

    now = datetime.now(timezone.utc)
    for acct in accounts:
        acct_name = acct.get("name", "")
        platform = acct.get("platformId", "")
        safe = acct.get("safeName", "")
        secret_mgmt = acct.get("secretManagement", {})
        last_modified = secret_mgmt.get("lastModifiedTime", 0)
        auto_mgmt = secret_mgmt.get("automaticManagementEnabled", False)
        status = secret_mgmt.get("status", "")

        if not auto_mgmt:
            findings.append({
                "account": acct_name, "safe": safe, "platform": platform,
                "check": "Automatic password management disabled",
                "severity": "HIGH",
                "detail": "Password not managed by CyberArk CPM",
            })

        if last_modified:
            last_mod_dt = datetime.fromtimestamp(last_modified, tz=timezone.utc)
            age_days = (now - last_mod_dt).days
            if age_days > 90:
                findings.append({
                    "account": acct_name, "safe": safe,
                    "check": "Password age exceeds 90 days",
                    "severity": "HIGH",
                    "detail": f"Last rotated {age_days} days ago",
                })

        if status and status != "success":
            findings.append({
                "account": acct_name, "safe": safe,
                "check": f"CPM status: {status}",
                "severity": "CRITICAL" if "fail" in status.lower() else "MEDIUM",
                "detail": f"Password management status: {status}",
            })

    print(f"[+] Audited {len(accounts)} accounts")
    return findings, accounts


def audit_platforms(pvwa_url, token):
    """Audit platform configurations."""
    findings = []
    print("[*] Auditing platforms...")
    data = api_call(pvwa_url, token, "/Platforms")
    platforms = data.get("Platforms", data.get("value", []))

    for plat in platforms:
        name = plat.get("Name", plat.get("PlatformID", ""))
        active = plat.get("Active", True)
        if not active:
            continue
        priv_session = plat.get("PrivilegedSessionManagement", {})
        if not priv_session.get("PSMServerId"):
            findings.append({
                "platform": name,
                "check": "No PSM configured",
                "severity": "MEDIUM",
                "detail": "Platform missing privileged session management",
            })

    print(f"[+] Audited {len(platforms)} platforms")
    return findings, platforms


def logoff(pvwa_url, token):
    """End the CyberArk session."""
    try:
        requests.post(f"{pvwa_url}/PasswordVault/API/Auth/Logoff",
                      headers={"Authorization": token},
                      verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=10)  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
        print("[+] Session ended")
    except requests.RequestException:
        pass


def format_summary(safe_findings, acct_findings, plat_findings, safes, accounts):
    """Print audit summary."""
    all_findings = safe_findings + acct_findings + plat_findings
    print(f"\n{'='*60}")
    print(f"  CyberArk PAM Audit Report")
    print(f"{'='*60}")
    print(f"  Safes        : {len(safes)}")
    print(f"  Accounts     : {len(accounts)}")
    print(f"  Findings     : {len(all_findings)}")

    severity_counts = {}
    for f in all_findings:
        sev = f.get("severity", "INFO")
        severity_counts[sev] = severity_counts.get(sev, 0) + 1

    print(f"\n  By Severity:")
    for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"]:
        count = severity_counts.get(sev, 0)
        if count:
            print(f"    {sev:10s}: {count}")

    if all_findings:
        print(f"\n  Top Issues:")
        for f in all_findings[:15]:
            if f["severity"] in ("CRITICAL", "HIGH"):
                print(f"    [{f['severity']:8s}] {f['check']}: "
                      f"{f.get('account', f.get('safe', f.get('platform', '')))}")

    return severity_counts


def main():
    parser = argparse.ArgumentParser(description="CyberArk PAM configuration audit agent")
    parser.add_argument("--pvwa-url", help="PVWA URL (or CYBERARK_PVWA_URL env)")
    parser.add_argument("--username", required=True, help="CyberArk username")
    parser.add_argument("--password", required=True, help="CyberArk password")
    parser.add_argument("--auth-type", default="CyberArk",
                        choices=["CyberArk", "LDAP", "RADIUS", "Windows"])
    parser.add_argument("--safe", help="Audit specific safe only")
    parser.add_argument("--output", "-o", help="Output JSON report")
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    if args.pvwa_url:
        os.environ["CYBERARK_PVWA_URL"] = args.pvwa_url
    pvwa_url = get_cyberark_config()

    token = authenticate(pvwa_url, args.username, args.password, args.auth_type)
    try:
        safe_findings, safes = audit_safes(pvwa_url, token)
        acct_findings, accounts = audit_accounts(pvwa_url, token, args.safe)
        plat_findings, platforms = audit_platforms(pvwa_url, token)
    finally:
        logoff(pvwa_url, token)

    severity_counts = format_summary(safe_findings, acct_findings, plat_findings, safes, accounts)

    report = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "tool": "CyberArk PAM Audit",
        "safes_count": len(safes),
        "accounts_count": len(accounts),
        "findings": safe_findings + acct_findings + plat_findings,
        "severity_counts": severity_counts,
        "risk_level": (
            "CRITICAL" if severity_counts.get("CRITICAL", 0) > 0
            else "HIGH" if severity_counts.get("HIGH", 0) > 0
            else "MEDIUM" if severity_counts.get("MEDIUM", 0) > 0
            else "LOW"
        ),
    }

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"\n[+] Report saved to {args.output}")
    elif args.verbose:
        print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
CyberArk PAM Health Monitor and Audit Script

Monitors CyberArk vault health, checks credential rotation status,
audits safe permissions, and generates compliance reports for
privileged access management.
"""

import json
import datetime
import hashlib
from typing import Dict, List, Optional
from dataclasses import dataclass, field


@dataclass
class PrivilegedAccount:
    """Represents a privileged account in the vault."""
    account_id: str
    username: str
    address: str  # target system
    platform_id: str
    safe_name: str
    account_type: str  # domain_admin, local_admin, service_account, dba, network
    last_rotation: Optional[str] = None
    last_verification: Optional[str] = None
    last_access: Optional[str] = None
    rotation_interval_days: int = 30
    status: str = "active"  # active, locked, failed_rotation, disabled


@dataclass
class SafeConfig:
    """Represents a CyberArk safe configuration."""
    safe_name: str
    description: str
    members: List[Dict] = field(default_factory=list)  # {"user": "...", "role": "..."}
    retention_days: int = 365
    dual_control: bool = False
    require_reason: bool = True
    account_count: int = 0


@dataclass
class PAMAuditFinding:
    """PAM audit finding."""
    finding_id: str
    severity: str
    category: str
    title: str
    details: str
    affected_accounts: List[str] = field(default_factory=list)
    remediation: str = ""
    nist_control: str = ""


class CyberArkPAMAuditor:
    """Audits CyberArk PAM configuration and compliance."""

    ROTATION_POLICIES = {
        "domain_admin": 1,       # days
        "local_admin": 3,
        "service_account": 30,
        "dba": 1,
        "network": 7,
        "cloud_iam": 90,
        "root": 3,
    }

    def __init__(self):
        self.accounts: List[PrivilegedAccount] = []
        self.safes: List[SafeConfig] = []
        self.findings: List[PAMAuditFinding] = []
        self.finding_counter = 0

    def load_accounts(self, accounts: List[Dict]):
        """Load privileged accounts for audit."""
        for acct in accounts:
            self.accounts.append(PrivilegedAccount(**acct))

    def load_safes(self, safes: List[Dict]):
        """Load safe configurations for audit."""
        for safe in safes:
            self.safes.append(SafeConfig(**safe))

    def _next_finding_id(self) -> str:
        self.finding_counter += 1
        return f"PAM-{self.finding_counter:04d}"

    def audit_all(self) -> List[PAMAuditFinding]:
        """Run all PAM audit checks."""
        self.findings = []
        self.finding_counter = 0
        self._audit_rotation_compliance()
        self._audit_verification_status()
        self._audit_stale_accounts()
        self._audit_safe_permissions()
        self._audit_dual_control()
        self._audit_service_account_dependencies()
        self._audit_account_coverage()
        return self.findings

    def _audit_rotation_compliance(self):
        """Check if credentials are being rotated per policy."""
        now = datetime.datetime.now()
        overdue_accounts = []

        for acct in self.accounts:
            if acct.status == "disabled":
                continue
            max_days = self.ROTATION_POLICIES.get(acct.account_type, 30)
            if acct.last_rotation:
                try:
                    last_rot = datetime.datetime.fromisoformat(acct.last_rotation)
                    days_since = (now - last_rot).days
                    if days_since > max_days:
                        overdue_accounts.append(
                            f"{acct.username}@{acct.address} ({acct.account_type}): "
                            f"{days_since} days since rotation, policy: {max_days} days"
                        )
                except ValueError:
                    pass
            else:
                overdue_accounts.append(
                    f"{acct.username}@{acct.address}: Never rotated"
                )

        if overdue_accounts:
            self.findings.append(PAMAuditFinding(
                finding_id=self._next_finding_id(),
                severity="critical",
                category="Credential Rotation",
                title=f"{len(overdue_accounts)} accounts overdue for rotation",
                details="The following accounts exceed their rotation policy:\n" +
                        "\n".join(f"  - {a}" for a in overdue_accounts),
                affected_accounts=[a.split("@")[0] for a in overdue_accounts],
                remediation="Investigate CPM rotation failures. Check reconciliation accounts. Run manual rotation.",
                nist_control="IA-5(1)"
            ))
        else:
            self.findings.append(PAMAuditFinding(
                finding_id=self._next_finding_id(),
                severity="info",
                category="Credential Rotation",
                title="All accounts within rotation policy",
                details="All active accounts have been rotated within their policy window."
            ))

    def _audit_verification_status(self):
        """Check credential verification status."""
        unverified = []
        now = datetime.datetime.now()

        for acct in self.accounts:
            if acct.status == "disabled":
                continue
            if not acct.last_verification:
                unverified.append(f"{acct.username}@{acct.address}: Never verified")
            else:
                try:
                    last_ver = datetime.datetime.fromisoformat(acct.last_verification)
                    days_since = (now - last_ver).days
                    if days_since > 7:
                        unverified.append(
                            f"{acct.username}@{acct.address}: {days_since} days since verification"
                        )
                except ValueError:
                    pass

        if unverified:
            self.findings.append(PAMAuditFinding(
                finding_id=self._next_finding_id(),
                severity="high",
                category="Credential Verification",
                title=f"{len(unverified)} accounts have stale verification",
                details="Vault credentials may not match target systems:\n" +
                        "\n".join(f"  - {u}" for u in unverified),
                remediation="Run CPM verification task. Check target system connectivity.",
                nist_control="IA-5(2)"
            ))

    def _audit_stale_accounts(self):
        """Identify accounts that haven't been accessed."""
        stale = []
        now = datetime.datetime.now()

        for acct in self.accounts:
            if acct.status == "disabled":
                continue
            if not acct.last_access:
                stale.append(f"{acct.username}@{acct.address}: Never accessed from vault")
            else:
                try:
                    last_acc = datetime.datetime.fromisoformat(acct.last_access)
                    days_since = (now - last_acc).days
                    if days_since > 90:
                        stale.append(
                            f"{acct.username}@{acct.address}: {days_since} days since last access"
                        )
                except ValueError:
                    pass

        if stale:
            self.findings.append(PAMAuditFinding(
                finding_id=self._next_finding_id(),
                severity="medium",
                category="Stale Accounts",
                title=f"{len(stale)} accounts unused for 90+ days",
                details="These accounts may be candidates for decommissioning:\n" +
                        "\n".join(f"  - {s}" for s in stale),
                remediation="Review with account owners. Disable or remove if no longer needed.",
                nist_control="AC-2(3)"
            ))

    def _audit_safe_permissions(self):
        """Check safe member permissions for least privilege."""
        for safe in self.safes:
            admin_count = sum(1 for m in safe.members if m.get("role") == "admin")
            if admin_count > 3:
                self.findings.append(PAMAuditFinding(
                    finding_id=self._next_finding_id(),
                    severity="high",
                    category="Safe Permissions",
                    title=f"Safe '{safe.safe_name}' has {admin_count} admins",
                    details=f"Excessive admin access to safe '{safe.safe_name}'. "
                            "Admin role allows full control including member management.",
                    remediation="Review safe admins. Reduce to minimum required (typically 2).",
                    nist_control="AC-6"
                ))

            for member in safe.members:
                if member.get("role") == "full_access":
                    self.findings.append(PAMAuditFinding(
                        finding_id=self._next_finding_id(),
                        severity="medium",
                        category="Safe Permissions",
                        title=f"Full access granted in safe '{safe.safe_name}'",
                        details=f"User '{member.get('user')}' has full access to safe '{safe.safe_name}'.",
                        remediation="Assign minimum required permissions (retrieve, list) instead of full access.",
                        nist_control="AC-6(1)"
                    ))

    def _audit_dual_control(self):
        """Check dual control enforcement for sensitive safes."""
        for safe in self.safes:
            if not safe.dual_control:
                high_priv_types = any(
                    acct.account_type in ("domain_admin", "root", "dba")
                    for acct in self.accounts if acct.safe_name == safe.safe_name
                )
                if high_priv_types:
                    self.findings.append(PAMAuditFinding(
                        finding_id=self._next_finding_id(),
                        severity="high",
                        category="Dual Control",
                        title=f"Dual control not enabled for high-privilege safe '{safe.safe_name}'",
                        details=f"Safe '{safe.safe_name}' contains high-privilege accounts but "
                                "does not require dual control for credential checkout.",
                        remediation="Enable dual control requiring approval before credential release.",
                        nist_control="AC-5"
                    ))

    def _audit_service_account_dependencies(self):
        """Check for service accounts that may have rotation dependencies."""
        service_accounts = [a for a in self.accounts if a.account_type == "service_account"]
        if service_accounts:
            rapid_rotation = [
                a for a in service_accounts
                if a.rotation_interval_days < 7
            ]
            if rapid_rotation:
                self.findings.append(PAMAuditFinding(
                    finding_id=self._next_finding_id(),
                    severity="medium",
                    category="Service Account Rotation",
                    title=f"{len(rapid_rotation)} service accounts with aggressive rotation",
                    details="Service accounts with < 7-day rotation may cause application disruptions "
                            "if dependencies are not properly managed.",
                    remediation="Verify application dependency management before aggressive rotation. "
                                "Use account groups for coordinated rotation.",
                    nist_control="IA-5(1)"
                ))

    def _audit_account_coverage(self):
        """Check for potential gaps in privileged account coverage."""
        account_types = set(a.account_type for a in self.accounts)
        expected_types = {"domain_admin", "local_admin", "service_account", "dba", "network", "root"}
        missing = expected_types - account_types

        if missing:
            self.findings.append(PAMAuditFinding(
                finding_id=self._next_finding_id(),
                severity="medium",
                category="Coverage Gap",
                title=f"No {', '.join(missing)} accounts in vault",
                details=f"Account types not found in vault: {', '.join(missing)}. "
                        "These may be unmanaged privileged accounts.",
                remediation="Run privileged account discovery scan. Onboard missing account types.",
                nist_control="AC-2"
            ))

    def generate_report(self) -> str:
        """Generate comprehensive PAM audit report."""
        if not self.findings:
            self.audit_all()

        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
        sorted_findings = sorted(self.findings, key=lambda f: severity_order.get(f.severity, 5))

        lines = [
            "=" * 70,
            "CYBERARK PAM AUDIT REPORT",
            "=" * 70,
            f"Report Date: {datetime.datetime.now().isoformat()}",
            f"Total Accounts Audited: {len(self.accounts)}",
            f"Total Safes Audited: {len(self.safes)}",
            f"Total Findings: {len(self.findings)}",
            "-" * 70,
            ""
        ]

        by_severity = {}
        for f in sorted_findings:
            by_severity.setdefault(f.severity, []).append(f)

        for sev in ["critical", "high", "medium", "low", "info"]:
            count = len(by_severity.get(sev, []))
            lines.append(f"  {sev.upper()}: {count}")
        lines.append("")

        for f in sorted_findings:
            icon = {"critical": "[!!!]", "high": "[!!]", "medium": "[!]", "low": "[~]", "info": "[i]"}.get(f.severity, "")
            lines.append(f"{icon} {f.finding_id} [{f.severity.upper()}] {f.title}")
            lines.append(f"    Category: {f.category}")
            lines.append(f"    {f.details}")
            if f.remediation:
                lines.append(f"    Remediation: {f.remediation}")
            if f.nist_control:
                lines.append(f"    NIST Control: {f.nist_control}")
            lines.append("")

        lines.append("=" * 70)
        critical_count = len(by_severity.get("critical", []))
        overall = "FAIL" if critical_count > 0 else "PASS WITH FINDINGS" if len(self.findings) > len(by_severity.get("info", [])) else "PASS"
        lines.append(f"OVERALL: {overall}")
        lines.append("=" * 70)

        return "\n".join(lines)


def main():
    """Run PAM audit with sample data."""
    auditor = CyberArkPAMAuditor()

    sample_accounts = [
        {"account_id": "1", "username": "domain_admin01", "address": "dc01.corp.local",
         "platform_id": "WinDomain", "safe_name": "DomainAdmins", "account_type": "domain_admin",
         "last_rotation": "2026-02-20", "last_verification": "2026-02-22", "last_access": "2026-02-21"},
        {"account_id": "2", "username": "root", "address": "webserver01.corp.local",
         "platform_id": "UnixSSH", "safe_name": "LinuxRoot", "account_type": "root",
         "last_rotation": "2026-01-15", "last_verification": "2026-01-16"},
        {"account_id": "3", "username": "sa", "address": "sqlserver01.corp.local",
         "platform_id": "MSSQL", "safe_name": "DatabaseAdmins", "account_type": "dba",
         "last_rotation": "2026-02-22", "last_verification": "2026-02-22", "last_access": "2026-02-23"},
        {"account_id": "4", "username": "svc_backup", "address": "backup01.corp.local",
         "platform_id": "WinService", "safe_name": "ServiceAccounts", "account_type": "service_account",
         "last_rotation": "2026-02-01", "rotation_interval_days": 5},
        {"account_id": "5", "username": "admin", "address": "switch01.corp.local",
         "platform_id": "CiscoIOS", "safe_name": "NetworkDevices", "account_type": "network",
         "last_rotation": "2026-02-10", "last_verification": "2026-02-10"},
    ]

    sample_safes = [
        {"safe_name": "DomainAdmins", "description": "Domain admin accounts", "dual_control": True,
         "members": [{"user": "admin1", "role": "admin"}, {"user": "admin2", "role": "admin"}], "account_count": 5},
        {"safe_name": "LinuxRoot", "description": "Linux root accounts", "dual_control": False,
         "members": [{"user": "admin1", "role": "admin"}, {"user": "admin2", "role": "admin"},
                     {"user": "admin3", "role": "admin"}, {"user": "admin4", "role": "admin"}], "account_count": 20},
        {"safe_name": "ServiceAccounts", "description": "Service accounts", "dual_control": False,
         "members": [{"user": "svc_team", "role": "full_access"}], "account_count": 50},
    ]

    auditor.load_accounts(sample_accounts)
    auditor.load_safes(sample_safes)
    report = auditor.generate_report()
    print(report)


if __name__ == "__main__":
    main()