#!/usr/bin/env python3
"""Azure AD Privileged Identity Management agent using Microsoft Graph API via requests."""

import argparse
import json
import logging
import os
import sys
from datetime import datetime
from typing import List

try:
    import requests
except ImportError:
    sys.exit("requests required: pip install requests")

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

GRAPH_BASE = "https://graph.microsoft.com/v1.0"


class PIMClient:
    """Client for Microsoft Entra PIM via Graph API."""

    def __init__(self, tenant_id: str, client_id: str, client_secret: str):
        self.tenant_id = tenant_id
        self.token = self._acquire_token(client_id, client_secret)
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {self.token}"})

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

    def list_role_definitions(self) -> List[dict]:
        """List available directory role definitions."""
        resp = self.session.get(f"{GRAPH_BASE}/roleManagement/directory/roleDefinitions", timeout=30)
        resp.raise_for_status()
        return resp.json().get("value", [])

    def list_eligible_assignments(self) -> List[dict]:
        """List PIM eligible role assignments."""
        resp = self.session.get(
            f"{GRAPH_BASE}/roleManagement/directory/roleEligibilityScheduleInstances", timeout=30)
        resp.raise_for_status()
        return resp.json().get("value", [])

    def list_active_assignments(self) -> List[dict]:
        """List currently active (activated) role assignments."""
        resp = self.session.get(
            f"{GRAPH_BASE}/roleManagement/directory/roleAssignmentScheduleInstances", timeout=30)
        resp.raise_for_status()
        return resp.json().get("value", [])

    def list_role_settings(self) -> List[dict]:
        """List PIM role management policy assignments."""
        resp = self.session.get(
            f"{GRAPH_BASE}/policies/roleManagementPolicyAssignments?"
            "$filter=scopeId eq '/' and scopeType eq 'DirectoryRole'", timeout=30)
        resp.raise_for_status()
        return resp.json().get("value", [])

    def get_activation_requests(self, top: int = 50) -> List[dict]:
        """List recent role activation requests."""
        resp = self.session.get(
            f"{GRAPH_BASE}/roleManagement/directory/roleEligibilityScheduleRequests?"
            f"$top={top}&$orderby=createdDateTime desc", timeout=30)
        resp.raise_for_status()
        return resp.json().get("value", [])


def audit_permanent_assignments(active: List[dict], eligible: List[dict]) -> List[dict]:
    """Identify permanent (non-PIM) role assignments that should be converted to eligible."""
    eligible_ids = {a.get("principalId") for a in eligible}
    permanent = []
    for a in active:
        if a.get("assignmentType") == "Assigned" and a.get("principalId") not in eligible_ids:
            permanent.append({
                "principal_id": a.get("principalId", ""),
                "role": a.get("roleDefinition", {}).get("displayName", ""),
                "start": a.get("startDateTime", ""),
                "recommendation": "Convert to eligible assignment with JIT activation",
            })
    return permanent


def compute_pim_coverage(active: List[dict], eligible: List[dict]) -> dict:
    """Calculate PIM coverage metrics."""
    total = len(active)
    eligible_count = len(eligible)
    pim_pct = (eligible_count / (total + eligible_count) * 100) if (total + eligible_count) else 0
    return {
        "active_assignments": total,
        "eligible_assignments": eligible_count,
        "pim_coverage_pct": round(pim_pct, 1),
    }


def generate_report(client: PIMClient) -> dict:
    """Generate PIM compliance report."""
    roles = client.list_role_definitions()
    eligible = client.list_eligible_assignments()
    active = client.list_active_assignments()
    permanent = audit_permanent_assignments(active, eligible)
    coverage = compute_pim_coverage(active, eligible)

    report = {
        "analysis_date": datetime.utcnow().isoformat(),
        "role_definitions": len(roles),
        "coverage": coverage,
        "permanent_assignments": permanent,
        "permanent_count": len(permanent),
        "recommendations": [],
    }
    if permanent:
        report["recommendations"].append(
            f"Convert {len(permanent)} permanent assignments to PIM-eligible")
    if coverage["pim_coverage_pct"] < 80:
        report["recommendations"].append("Increase PIM coverage above 80%")
    return report


def main():
    parser = argparse.ArgumentParser(description="Azure AD PIM Audit Agent")
    parser.add_argument("--tenant-id", required=True, help="Azure AD tenant ID")
    parser.add_argument("--client-id", required=True, help="App registration client ID")
    parser.add_argument("--client-secret", required=True, help="App registration secret")
    parser.add_argument("--output-dir", default=".")
    parser.add_argument("--output", default="pim_report.json")
    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)
    client = PIMClient(args.tenant_id, args.client_id, args.client_secret)
    report = generate_report(client)
    out_path = os.path.join(args.output_dir, args.output)
    with open(out_path, "w") as f:
        json.dump(report, f, indent=2, default=str)
    logger.info("Report saved to %s", out_path)
    print(json.dumps(report, indent=2, default=str))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Azure AD PIM Audit and Configuration Tool

Audits Entra ID role assignments, identifies permanent privileged assignments
that should be converted to PIM eligible, and monitors PIM activation events.

Requirements:
    pip install msal requests pandas
"""

import json
import sys
from datetime import datetime, timezone

try:
    import requests
    import msal
except ImportError:
    print("[ERROR] Required: pip install msal requests")
    sys.exit(1)


class EntraPIMAuditor:
    """Audit and manage Microsoft Entra PIM configuration."""

    def __init__(self, tenant_id, client_id, client_secret):
        self.tenant_id = tenant_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.token = None
        self._authenticate()

    def _authenticate(self):
        """Acquire Microsoft Graph access token using client credentials."""
        app = msal.ConfidentialClientApplication(
            self.client_id,
            authority=f"https://login.microsoftonline.com/{self.tenant_id}",
            client_credential=self.client_secret,
        )
        result = app.acquire_token_for_client(
            scopes=["https://graph.microsoft.com/.default"]
        )
        if "access_token" in result:
            self.token = result["access_token"]
            print("[OK] Authenticated to Microsoft Graph")
        else:
            raise Exception(f"Auth failed: {result.get('error_description')}")

    def _graph_get(self, endpoint):
        """Make authenticated GET request to Microsoft Graph."""
        headers = {"Authorization": f"Bearer {self.token}"}
        url = f"https://graph.microsoft.com/v1.0{endpoint}"
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        return response.json()

    def get_directory_roles(self):
        """List all Entra directory role definitions."""
        result = self._graph_get("/roleManagement/directory/roleDefinitions")
        roles = {}
        for role in result.get("value", []):
            roles[role["id"]] = {
                "displayName": role["displayName"],
                "description": role.get("description", ""),
                "isBuiltIn": role.get("isBuiltIn", False),
            }
        return roles

    def get_active_assignments(self):
        """List all permanently active role assignments (non-PIM)."""
        result = self._graph_get(
            "/roleManagement/directory/roleAssignments?$expand=principal"
        )
        assignments = []
        for item in result.get("value", []):
            assignments.append({
                "roleDefinitionId": item["roleDefinitionId"],
                "principalId": item["principalId"],
                "directoryScopeId": item.get("directoryScopeId", "/"),
                "principalDisplayName": item.get("principal", {}).get("displayName", ""),
                "principalType": item.get("principal", {}).get("@odata.type", ""),
            })
        return assignments

    def get_eligible_assignments(self):
        """List all PIM eligible role assignments."""
        result = self._graph_get(
            "/roleManagement/directory/roleEligibilityScheduleInstances"
        )
        eligible = []
        for item in result.get("value", []):
            eligible.append({
                "roleDefinitionId": item["roleDefinitionId"],
                "principalId": item["principalId"],
                "directoryScopeId": item.get("directoryScopeId", "/"),
                "startDateTime": item.get("startDateTime"),
                "endDateTime": item.get("endDateTime"),
            })
        return eligible

    def get_pim_activation_history(self, days=30):
        """Retrieve PIM role activation audit events."""
        result = self._graph_get(
            f"/auditLogs/directoryAudits?"
            f"$filter=category eq 'RoleManagement' and "
            f"activityDisplayName eq 'Add member to role completed (PIM activation)'"
            f"&$top=100"
        )
        activations = []
        for event in result.get("value", []):
            activations.append({
                "activityDateTime": event.get("activityDateTime"),
                "activityDisplayName": event.get("activityDisplayName"),
                "initiatedBy": event.get("initiatedBy", {}).get("user", {}).get("displayName", ""),
                "targetResources": [
                    t.get("displayName", "") for t in event.get("targetResources", [])
                ],
                "result": event.get("result"),
            })
        return activations

    def identify_permanent_admins(self):
        """Find permanently active admin assignments that should be PIM eligible."""
        roles = self.get_directory_roles()
        active = self.get_active_assignments()
        eligible = self.get_eligible_assignments()

        critical_roles = {
            rid: info for rid, info in roles.items()
            if info["displayName"] in [
                "Global Administrator",
                "Privileged Role Administrator",
                "Security Administrator",
                "Exchange Administrator",
                "SharePoint Administrator",
                "User Administrator",
                "Application Administrator",
                "Cloud Application Administrator",
                "Intune Administrator",
                "Compliance Administrator",
            ]
        }

        findings = []
        eligible_principals = {
            (e["roleDefinitionId"], e["principalId"]) for e in eligible
        }

        for assignment in active:
            role_id = assignment["roleDefinitionId"]
            if role_id in critical_roles:
                is_also_eligible = (role_id, assignment["principalId"]) in eligible_principals
                findings.append({
                    "role": critical_roles[role_id]["displayName"],
                    "principal": assignment["principalDisplayName"],
                    "principalType": assignment["principalType"],
                    "hasEligibleAssignment": is_also_eligible,
                    "recommendation": "Convert to PIM eligible" if not is_also_eligible else "Review - has both active and eligible",
                })

        return findings

    def generate_audit_report(self):
        """Generate comprehensive PIM audit report."""
        roles = self.get_directory_roles()
        active = self.get_active_assignments()
        eligible = self.get_eligible_assignments()
        permanent_findings = self.identify_permanent_admins()

        report = {
            "report_title": "Microsoft Entra PIM Audit Report",
            "tenant_id": self.tenant_id,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "summary": {
                "total_directory_roles": len(roles),
                "total_active_assignments": len(active),
                "total_eligible_assignments": len(eligible),
                "permanent_admin_findings": len(permanent_findings),
            },
            "findings": permanent_findings,
            "recommendations": [],
        }

        if len(permanent_findings) > 0:
            report["recommendations"].append({
                "priority": "Critical",
                "finding": f"{len(permanent_findings)} permanent privileged role assignments found",
                "action": "Convert to PIM eligible assignments with MFA and approval requirements"
            })

        active_global_admins = sum(
            1 for f in permanent_findings
            if f["role"] == "Global Administrator"
        )
        if active_global_admins > 2:
            report["recommendations"].append({
                "priority": "High",
                "finding": f"{active_global_admins} permanent Global Administrators (should be max 2 break-glass)",
                "action": "Reduce to 2 break-glass accounts, convert rest to PIM eligible"
            })

        return report


if __name__ == "__main__":
    print("=" * 60)
    print("Microsoft Entra PIM Audit Tool")
    print("=" * 60)
    print()
    print("Usage:")
    print("  auditor = EntraPIMAuditor(tenant_id, client_id, client_secret)")
    print("  report = auditor.generate_audit_report()")
    print("  print(json.dumps(report, indent=2))")
    print()
    print("Required Microsoft Graph permissions:")
    print("  - RoleManagement.Read.All")
    print("  - AuditLog.Read.All")
    print("  - Directory.Read.All")