Modern Identity and Access Management: Strategies, Challenges, and the Road Ahead

# Modern identity fabric design
Identity Fabric:
  Identity Providers (IDP):
    - Primary IDP: Azure AD (workforce identity)
    - CIAM IDP: Auth0 (customer identity)
    - Cloud IDP: AWS SSO, Google Cloud Identity

  Identity Governance:
    - SailPoint IdentityIQ
    - Automated provisioning/deprovisioning
    - Access certification campaigns (quarterly)
    - Privileged access management (CyberArk)

  Authentication Layer:
    - SSO: Okta (enterprise apps)
    - Federation: SAML 2.0, OIDC
    - MFA: Duo Security (adaptive MFA)
    - Passwordless: FIDO2 (WebAuthn)

  Authorization Layer:
    - Policy engine: Open Policy Agent (OPA)
    - Attribute-based access control (ABAC)
    - Just-in-time (JIT) access
    - Step-up authentication for sensitive resources

  Monitoring & Analytics:
    - SIEM: Splunk (identity events)
    - UEBA: Abnormal user behavior detection
    - Identity threat detection (Varonis)

# Phase 1: Discover all identity systems (Month 1-2)
import ldap
import requests
from azure.identity import DefaultAzureCredential
from azure.mgmt.resource import SubscriptionClient

class IdentityDiscovery:
    def __init__(self):
        self.identity_systems = []

    def discover_active_directory(self):
        """Scan for all AD forests and domains"""
        # Query DNS for _ldap._tcp.dc._msdcs records
        forests = self.query_ad_forests()

        for forest in forests:
            conn = ldap.initialize(f'ldap://{forest}')
            conn.simple_bind_s()

            # Get all users
            result = conn.search_s(
                'DC=example,DC=com',
                ldap.SCOPE_SUBTREE,
                '(objectClass=user)',
                ['sAMAccountName', 'mail', 'memberOf', 'lastLogon']
            )

            self.identity_systems.append({
                'type': 'Active Directory',
                'forest': forest,
                'user_count': len(result),
                'users': result
            })

    def discover_saas_applications(self):
        """Discover SaaS apps via SSO audit logs"""
        # Query Okta for all integrated apps
        okta_apps = requests.get(
            'https://your-domain.okta.com/api/v1/apps',
            headers={'Authorization': f'SSWS {api_token}'}
        ).json()

        for app in okta_apps:
            self.identity_systems.append({
                'type': 'SaaS Application',
                'name': app['label'],
                'users': app['users'],
                'last_accessed': app['lastUpdated']
            })

    def discover_cloud_iam(self):
        """Discover AWS, Azure, GCP identities"""
        # Azure AD
        credential = DefaultAzureCredential()
        client = SubscriptionClient(credential)

        # Get all Azure AD users
        azure_users = self.get_azure_ad_users()

        # AWS IAM users across all accounts
        aws_users = self.get_aws_iam_users()

        return {
            'azure': azure_users,
            'aws': aws_users
        }

    def generate_report(self):
        """Generate identity sprawl report"""
        report = {
            'total_systems': len(self.identity_systems),
            'total_identities': sum(s['user_count'] for s in self.identity_systems),
            'orphaned_accounts': self.find_orphaned_accounts(),
            'duplicate_accounts': self.find_duplicates(),
            'privileged_accounts': self.find_privileged_accounts()
        }

        return report

# Run discovery
discovery = IdentityDiscovery()
discovery.discover_active_directory()
discovery.discover_saas_applications()
discovery.discover_cloud_iam()

report = discovery.generate_report()
print(f"Found {report['total_systems']} identity systems")
print(f"Total identities: {report['total_identities']}")
print(f"Orphaned accounts: {report['orphaned_accounts']} (security risk!)")

// Production risk-based auth implementation
const riskEngine = require('./risk-engine');
const mfaService = require('./mfa-service');

async function authenticateUser(req, res) {
  const { username, password } = req.body;

  // 1. Verify credentials
  const user = await verifyCredentials(username, password);
  if (!user) {
    await logFailedAttempt(req.ip, username);
    return res.status(401).json({ error: 'Invalid credentials' });
  }

  // 2. Calculate risk score (0-100)
  const riskScore = await riskEngine.calculateRisk({
    user: user,
    ip: req.ip,
    userAgent: req.headers['user-agent'],
    location: await geolocate(req.ip),
    deviceFingerprint: req.body.deviceFingerprint,
    behavioralBiometrics: req.body.typingPattern,
    timestamp: Date.now()
  });

  console.log(`Risk score for ${username}: ${riskScore}`);

  // 3. Risk-based decision tree
  if (riskScore < 20) {
    // Low risk: Normal login (trusted device, usual location)
    await createSession(user);
    return res.json({
      success: true,
      sessionToken: generateToken(user)
    });

  } else if (riskScore >= 20 && riskScore < 60) {
    // Medium risk: Require MFA
    const mfaChallenge = await mfaService.sendChallenge(user.phone);
    return res.json({
      requiresMFA: true,
      challengeId: mfaChallenge.id,
      message: 'MFA required due to unusual login pattern'
    });

  } else if (riskScore >= 60 && riskScore < 80) {
    // High risk: Step-up authentication + fraud check
    await securityTeam.alertSuspiciousLogin(user, riskScore);

    const stepUpChallenge = await mfaService.sendPushNotification(user, {
      message: 'Suspicious login attempt detected',
      location: req.location,
      device: req.userAgent,
      requireApproval: true
    });

    return res.json({
      requiresStepUp: true,
      challengeId: stepUpChallenge.id
    });

  } else {
    // Critical risk: Block + investigate
    await blockUser(user.id, 'High-risk login attempt');
    await securityTeam.createIncident({
      type: 'potential_account_takeover',
      user: user.id,
      riskScore: riskScore,
      indicators: riskEngine.getIndicators()
    });

    return res.status(403).json({
      error: 'Account temporarily locked for security',
      contactSupport: true
    });
  }
}

// Risk scoring engine
class RiskEngine {
  async calculateRisk(context) {
    let riskScore = 0;

    // Check breached password databases
    if (await this.isPasswordBreached(context.password)) {
      riskScore += 40;  // High risk
    }

    // Velocity check: Multiple failed attempts
    const failedAttempts = await this.getRecentFailures(context.user.id);
    if (failedAttempts > 3) {
      riskScore += 25;
    }

    // Impossible travel: User in different country within 1 hour
    const lastLocation = await this.getLastLoginLocation(context.user.id);
    if (this.isImpossibleTravel(lastLocation, context.location)) {
      riskScore += 35;
    }

    // New device detection
    const knownDevices = await this.getKnownDevices(context.user.id);
    if (!knownDevices.includes(context.deviceFingerprint)) {
      riskScore += 15;
    }

    // Behavioral biometrics: Typing pattern doesn't match
    const typingMatch = await this.matchTypingPattern(
      context.user.id,
      context.behavioralBiometrics
    );
    if (typingMatch < 0.7) {  // <70% confidence
      riskScore += 20;
    }

    // Anonymous proxy / VPN / Tor
    if (await this.isAnonymousIP(context.ip)) {
      riskScore += 30;
    }

    // Time-based: Login at unusual hours
    const usualHours = await this.getTypicalLoginHours(context.user.id);
    const currentHour = new Date(context.timestamp).getHours();
    if (!usualHours.includes(currentHour)) {
      riskScore += 10;
    }

    return Math.min(riskScore, 100);  // Cap at 100
  }

  async isPasswordBreached(password) {
    // Check against HaveIBeenPwned API (k-anonymity)
    const sha1 = crypto.createHash('sha1').update(password).digest('hex');
    const prefix = sha1.substring(0, 5);
    const suffix = sha1.substring(5).toUpperCase();

    const response = await fetch(`https://api.pwnedpasswords.com/range/${prefix}`);
    const hashes = await response.text();

    return hashes.includes(suffix);
  }

  isImpossibleTravel(lastLocation, currentLocation) {
    const distance = this.calculateDistance(lastLocation, currentLocation);
    const timeDiff = Date.now() - lastLocation.timestamp;
    const hours = timeDiff / (1000 * 60 * 60);

    // Average speed > 500 mph (impossible without flight)
    const avgSpeed = distance / hours;
    return avgSpeed > 500;
  }
}

# JIT access system with approval workflow
from datetime import datetime, timedelta
import boto3

class JITAccessManager:
    def __init__(self):
        self.iam_client = boto3.client('iam')
        self.approval_required_roles = [
            'Administrator',
            'DatabaseAdmin',
            'SecurityAuditor'
        ]

    def request_access(self, user, resource, role, duration_hours, justification):
        """Request temporary privileged access"""

        # 1. Validate request
        if role in self.approval_required_roles and not justification:
            raise ValueError("Justification required for privileged access")

        # 2. Create approval ticket
        ticket = self.create_approval_ticket({
            'requestor': user,
            'resource': resource,
            'role': role,
            'duration': duration_hours,
            'justification': justification,
            'timestamp': datetime.now()
        })

        # 3. Notify approvers
        approvers = self.get_approvers(role)
        self.notify_approvers(approvers, ticket)

        return ticket

    def approve_access(self, ticket_id, approver):
        """Approve JIT access request"""
        ticket = self.get_ticket(ticket_id)

        # 1. Grant temporary IAM role
        expiration = datetime.now() + timedelta(hours=ticket['duration'])

        self.iam_client.attach_user_policy(
            UserName=ticket['requestor'],
            PolicyArn=self.get_policy_arn(ticket['role'])
        )

        # 2. Schedule automatic revocation
        self.schedule_revocation(
            user=ticket['requestor'],
            role=ticket['role'],
            expiration=expiration
        )

        # 3. Audit log
        self.audit_log({
            'event': 'jit_access_granted',
            'user': ticket['requestor'],
            'role': ticket['role'],
            'approver': approver,
            'expiration': expiration,
            'justification': ticket['justification']
        })

        return {
            'granted': True,
            'expires': expiration,
            'role': ticket['role']
        }

    def monitor_privileged_activity(self, user, role):
        """Real-time monitoring of privileged access usage"""

        # Watch for suspicious patterns
        activities = self.get_user_activities(user, since=datetime.now() - timedelta(hours=1))

        alerts = []

        # Alert 1: Bulk data download
        if self.detect_bulk_download(activities):
            alerts.append({
                'severity': 'HIGH',
                'type': 'bulk_data_exfiltration',
                'user': user,
                'action': 'Immediately revoke access and investigate'
            })

        # Alert 2: Access during off-hours
        if datetime.now().hour < 6 or datetime.now().hour > 20:
            alerts.append({
                'severity': 'MEDIUM',
                'type': 'off_hours_access',
                'user': user
            })

        # Alert 3: Privilege escalation attempt
        if self.detect_privilege_escalation(activities):
            alerts.append({
                'severity': 'CRITICAL',
                'type': 'privilege_escalation',
                'user': user,
                'action': 'Block user and create security incident'
            })

        if alerts:
            self.security_team.alert(alerts)

        return alerts

# Usage example
jit_manager = JITAccessManager()

# Employee requests database admin access for 2 hours
ticket = jit_manager.request_access(
    user='[email protected]',
    resource='production-database',
    role='DatabaseAdmin',
    duration_hours=2,
    justification='Emergency fix for customer-reported data issue (Ticket #12345)'
)

# Manager approves
jit_manager.approve_access(ticket.id, approver='[email protected]')

# System monitors all activities during the 2-hour window
jit_manager.monitor_privileged_activity('[email protected]', 'DatabaseAdmin')

# Access automatically revoked after 2 hours

# Zero Trust IAM architecture (Kubernetes + Istio + OPA)
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: zero-trust-access-policy
  namespace: production
spec:
  # Default deny all
  action: DENY
  rules: []
---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: allow-authenticated-users
  namespace: production
spec:
  action: ALLOW
  rules:
  - from:
    - source:
        # Only allow JWT-authenticated requests
        requestPrincipals: ["*"]
    to:
    - operation:
        methods: ["GET", "POST"]
    when:
    # Additional context-based checks via OPA
    - key: request.auth.claims[email_verified]
      values: ["true"]
    - key: request.auth.claims[mfa_verified]
      values: ["true"]
    # Check device compliance
    - key: request.headers[x-device-compliant]
      values: ["true"]
---
# OPA policy for attribute-based access control
apiVersion: v1
kind: ConfigMap
metadata:
  name: opa-policy
data:
  policy.rego: |
    package authz

    # Default deny
    default allow = false

    # Allow if all conditions met
    allow {
      input.user.email_verified == true
      input.user.mfa_verified == true
      input.device.compliant == true
      input.device.managed == true
      not is_risky_location
      not is_impossible_travel
      has_required_role
    }

    is_risky_location {
      input.location.country in ["XX", "YY"]  # High-risk countries
    }

    is_impossible_travel {
      distance_km := geo.distance(input.location.coords, input.user.last_location)
      time_hours := (input.timestamp - input.user.last_login) / 3600
      speed_kmh := distance_km / time_hours
      speed_kmh > 800  # Impossible without air travel
    }

    has_required_role {
      required_roles := data.resources[input.resource].required_roles
      user_roles := input.user.roles
      intersection := required_roles & user_roles
      count(intersection) > 0
    }

// WebAuthn (FIDO2) registration
const credential = await navigator.credentials.create({
  publicKey: {
    challenge: Uint8Array.from(challenge, c => c.charCodeAt(0)),
    rp: {
      name: "Your Company",
      id: "yourcompany.com"
    },
    user: {
      id: Uint8Array.from(userId, c => c.charCodeAt(0)),
      name: userEmail,
      displayName: userName
    },
    pubKeyCredParams: [
      { alg: -7, type: "public-key" },   // ES256
      { alg: -257, type: "public-key" }  // RS256
    ],
    authenticatorSelection: {
      authenticatorAttachment: "platform",  // Built-in (TouchID, FaceID, Windows Hello)
      userVerification: "required"
    },
    attestation: "direct"
  }
});

{
  "@context": "https://www.w3.org/2018/credentials/v1",
  "type": ["VerifiableCredential", "EmployeeCredential"],
  "issuer": "did:example:company123",
  "issuanceDate": "2025-01-01T00:00:00Z",
  "credentialSubject": {
    "id": "did:example:employee456",
    "employeeId": "E12345",
    "department": "Engineering",
    "clearanceLevel": "Secret"
  },
  "proof": {
    "type": "Ed25519Signature2020",
    "created": "2025-01-01T00:00:00Z",
    "verificationMethod": "did:example:company123#key-1",
    "proofPurpose": "assertionMethod",
    "proofValue": "z58DAdFfa9SkqZMVPxAQpic7ndSayn1PzZs6ZjWp1CktyGesjuTSwRdo..."
  }
}