#!/usr/bin/env python3
"""Agent for auditing CyberArk Zero Standing Privilege (ZSP) configuration via REST API."""

import os
import requests
import json
import argparse
from datetime import datetime, timezone
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def authenticate(base_url, username, password, auth_method="CyberArk"):
    """Authenticate to CyberArk PVWA and obtain session token."""
    url = f"{base_url}/api/auth/{auth_method}/Logon"
    payload = {"username": username, "password": password}
    resp = requests.post(url, json=payload, 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 to CyberArk PVWA as {username}")
    return {"Authorization": token}


def list_safes(base_url, headers):
    """List all safes to audit access policies."""
    url = f"{base_url}/api/Safes"
    resp = requests.get(url, headers=headers, 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()
    safes = resp.json().get("value", [])
    print(f"[*] Found {len(safes)} safes")
    for s in safes[:20]:
        print(f"  {s['safeName']} (retention: {s.get('numberOfDaysRetention', 'N/A')} days)")
    return safes


def audit_safe_members(base_url, headers, safe_name):
    """Audit members and permissions of a specific safe."""
    url = f"{base_url}/api/Safes/{safe_name}/Members"
    resp = requests.get(url, headers=headers, 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()
    members = resp.json().get("value", [])
    findings = []
    for m in members:
        perms = m.get("permissions", {})
        if perms.get("useAccounts") and perms.get("retrieveAccounts"):
            if not m.get("memberType") == "Role":
                findings.append({
                    "safe": safe_name, "member": m.get("memberName"),
                    "issue": "Standing retrieve+use privileges (not JIT)",
                    "severity": "HIGH",
                })
                print(f"  [!] {m.get('memberName')} has standing access to {safe_name}")
    return findings


def list_platforms(base_url, headers):
    """List platforms to verify JIT/ZSP configuration."""
    url = f"{base_url}/api/Platforms"
    resp = requests.get(url, headers=headers, 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()
    platforms = resp.json().get("Platforms", [])
    print(f"[*] Found {len(platforms)} platforms")
    for p in platforms:
        print(f"  {p.get('general', {}).get('name', 'Unknown')} - "
              f"Active: {p.get('general', {}).get('active', False)}")
    return platforms


def check_jit_sessions(base_url, headers, days=7):
    """Check recent privileged sessions for JIT compliance."""
    url = f"{base_url}/api/LiveSessions"
    resp = requests.get(url, headers=headers, 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()
    sessions = resp.json().get("LiveSessions", [])
    print(f"[*] Active privileged sessions: {len(sessions)}")
    long_sessions = []
    for s in sessions:
        duration = s.get("Duration", 0)
        if duration > 3600:
            long_sessions.append({
                "user": s.get("User"), "target": s.get("AccountName"),
                "duration_sec": duration, "severity": "MEDIUM",
            })
            print(f"  [!] Long session: {s.get('User')} -> {s.get('AccountName')} "
                  f"({duration // 60} min)")
    return long_sessions


def audit_accounts_standing_access(base_url, headers):
    """Find privileged accounts with standing (non-JIT) access enabled."""
    url = f"{base_url}/api/Accounts"
    params = {"limit": 100, "offset": 0}
    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()
    accounts = resp.json().get("value", [])
    findings = []
    for a in accounts:
        props = a.get("platformAccountProperties", {})
        if not props.get("JITEnabled", False):
            findings.append({
                "account": a.get("name"), "safe": a.get("safeName"),
                "platform": a.get("platformId"), "issue": "JIT not enabled",
                "severity": "HIGH",
            })
    print(f"[*] Accounts without JIT: {len(findings)}/{len(accounts)}")
    return findings


def generate_report(safe_findings, session_findings, account_findings, output_path):
    """Generate ZSP compliance audit report."""
    report = {
        "audit_date": datetime.now(timezone.utc).isoformat(),
        "summary": {
            "standing_access_findings": len(safe_findings),
            "long_session_findings": len(session_findings),
            "non_jit_accounts": len(account_findings),
        },
        "safe_findings": safe_findings,
        "session_findings": session_findings,
        "account_findings": account_findings,
    }
    with open(output_path, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"\n[*] Report saved to {output_path}")


def main():
    parser = argparse.ArgumentParser(description="CyberArk Zero Standing Privilege Audit Agent")
    parser.add_argument("action", choices=["safes", "audit-safe", "platforms",
                                           "sessions", "accounts", "full-audit"])
    parser.add_argument("--url", required=True, help="CyberArk PVWA base URL")
    parser.add_argument("--username", required=True)
    parser.add_argument("--password", required=True)
    parser.add_argument("--safe", help="Specific safe name to audit")
    parser.add_argument("-o", "--output", default="zsp_audit.json")
    args = parser.parse_args()

    headers = authenticate(args.url, args.username, args.password)
    if args.action == "safes":
        list_safes(args.url, headers)
    elif args.action == "audit-safe" and args.safe:
        audit_safe_members(args.url, headers, args.safe)
    elif args.action == "platforms":
        list_platforms(args.url, headers)
    elif args.action == "sessions":
        check_jit_sessions(args.url, headers)
    elif args.action == "accounts":
        audit_accounts_standing_access(args.url, headers)
    elif args.action == "full-audit":
        safes = list_safes(args.url, headers)
        sf = []
        for s in safes:
            sf.extend(audit_safe_members(args.url, headers, s["safeName"]))
        sess = check_jit_sessions(args.url, headers)
        acct = audit_accounts_standing_access(args.url, headers)
        generate_report(sf, sess, acct, args.output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Zero Standing Privilege Audit Tool

Discovers standing privileged access across AWS, Azure, and GCP,
compares against CyberArk ZSP policies, and identifies accounts
that should be migrated to just-in-time access.

Requirements:
    pip install boto3 requests
"""

import json
import logging
import sys
from datetime import datetime, timedelta, timezone

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


class StandingPrivilegeDiscovery:
    """Discover standing privileged access across cloud environments."""

    def __init__(self):
        self.findings = []

    def discover_aws_standing_privileges(self, profile_name=None):
        """Find AWS IAM users/roles with standing admin access."""
        try:
            import boto3
        except ImportError:
            logger.error("boto3 required: pip install boto3")
            return []

        session = boto3.Session(profile_name=profile_name)
        iam = session.client("iam")

        admin_policy_arns = {
            "arn:aws:iam::aws:policy/AdministratorAccess",
            "arn:aws:iam::aws:policy/PowerUserAccess",
            "arn:aws:iam::aws:policy/IAMFullAccess",
        }

        standing = []

        # Check IAM users
        paginator = iam.get_paginator("list_users")
        for page in paginator.paginate():
            for user in page["Users"]:
                username = user["UserName"]
                attached = iam.list_attached_user_policies(UserName=username)
                user_policies = {p["PolicyArn"] for p in attached["AttachedPolicies"]}

                has_admin = bool(user_policies & admin_policy_arns)

                # Check groups
                groups = iam.list_groups_for_user(UserName=username)
                for group in groups["Groups"]:
                    grp_policies = iam.list_attached_group_policies(
                        GroupName=group["GroupName"]
                    )
                    for gp in grp_policies["AttachedPolicies"]:
                        if gp["PolicyArn"] in admin_policy_arns:
                            has_admin = True

                if has_admin:
                    access_keys = iam.list_access_keys(UserName=username)
                    active_keys = [
                        k for k in access_keys["AccessKeyMetadata"]
                        if k["Status"] == "Active"
                    ]

                    standing.append({
                        "cloud": "AWS",
                        "identity_type": "User",
                        "identity": username,
                        "privilege_level": "Admin",
                        "policies": list(user_policies & admin_policy_arns),
                        "active_access_keys": len(active_keys),
                        "created": user["CreateDate"].isoformat(),
                        "last_used": str(user.get("PasswordLastUsed", "Never")),
                        "recommendation": "Migrate to CyberArk ZSP",
                    })

        # Check IAM roles (non-service-linked)
        role_paginator = iam.get_paginator("list_roles")
        for page in role_paginator.paginate():
            for role in page["Roles"]:
                if role["Path"].startswith("/aws-service-role/"):
                    continue

                role_policies = iam.list_attached_role_policies(
                    RoleName=role["RoleName"]
                )
                role_admin_policies = {
                    p["PolicyArn"] for p in role_policies["AttachedPolicies"]
                } & admin_policy_arns

                if role_admin_policies:
                    standing.append({
                        "cloud": "AWS",
                        "identity_type": "Role",
                        "identity": role["RoleName"],
                        "privilege_level": "Admin",
                        "policies": list(role_admin_policies),
                        "created": role["CreateDate"].isoformat(),
                        "recommendation": "Convert to ephemeral CyberArk SCA role",
                    })

        self.findings.extend(standing)
        logger.info(f"Discovered {len(standing)} AWS standing privileged identities")
        return standing

    def generate_migration_plan(self):
        """Generate a migration plan to move from standing to ZSP."""
        if not self.findings:
            return {"status": "No standing privileges found"}

        plan = {
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "total_standing_privileges": len(self.findings),
            "by_cloud": {},
            "by_type": {},
            "migration_waves": [],
        }

        # Categorize
        for finding in self.findings:
            cloud = finding["cloud"]
            plan["by_cloud"][cloud] = plan["by_cloud"].get(cloud, 0) + 1
            id_type = finding["identity_type"]
            plan["by_type"][id_type] = plan["by_type"].get(id_type, 0) + 1

        # Create migration waves
        users = [f for f in self.findings if f["identity_type"] == "User"]
        roles = [f for f in self.findings if f["identity_type"] == "Role"]

        wave_size = 5
        wave_num = 1

        for i in range(0, len(users), wave_size):
            batch = users[i:i + wave_size]
            plan["migration_waves"].append({
                "wave": wave_num,
                "type": "Users",
                "identities": [u["identity"] for u in batch],
                "count": len(batch),
                "suggested_week": f"Week {wave_num}",
            })
            wave_num += 1

        for i in range(0, len(roles), wave_size):
            batch = roles[i:i + wave_size]
            plan["migration_waves"].append({
                "wave": wave_num,
                "type": "Roles",
                "identities": [r["identity"] for r in batch],
                "count": len(batch),
                "suggested_week": f"Week {wave_num}",
            })
            wave_num += 1

        return plan

    def export_report(self, output_path):
        """Export standing privilege findings and migration plan."""
        report = {
            "report_title": "Standing Privilege Discovery Report",
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "findings": self.findings,
            "migration_plan": self.generate_migration_plan(),
        }

        with open(output_path, "w") as f:
            json.dump(report, f, indent=2)

        logger.info(f"Report exported to {output_path}")
        return report


if __name__ == "__main__":
    print("=" * 60)
    print("Zero Standing Privilege Audit Tool")
    print("=" * 60)
    print()
    print("Usage:")
    print("  discovery = StandingPrivilegeDiscovery()")
    print("  discovery.discover_aws_standing_privileges(profile='prod')")
    print("  plan = discovery.generate_migration_plan()")
    print("  discovery.export_report('zsp_report.json')")