Daily Access Audit

Comprehensive access auditing script that generates detailed reports on user access patterns, privileged access, and compliance issues.

---

Overview

This Python script performs a comprehensive audit of all user access across your Cakewalk-managed applications. It identifies privileged access, excessive permissions, and generates detailed reports for compliance and security review.

Features

• Complete access inventory: Audits all users and their application access

• Privileged access identification: Identifies users with elevated permissions

• Excessive access detection: Finds users with access to too many applications

• CSV export capabilities: Export audit results in multiple formats

• Automated reporting: Generate summary reports with actionable insights

• Pagination support: Handles large datasets efficiently

Prerequisites

Python Dependencies

pip install requests

Environment Variables

Set these environment variables:

export CAKEWALK_API_KEY="your-api-key"
export CAKEWALK_API_SECRET="your-api-secret"

Required Information

• API Credentials: Your Cakewalk API key and secret

• Sensitive App Names: List of applications considered sensitive

• Access Thresholds: Define what constitutes "excessive" access

Setup Instructions

1. Create the Script File

Create cakewalk_audit.py:

#!/usr/bin/env python3
"""
Daily Cakewalk Access Audit Script

Generates a comprehensive report of:
- All users with access to each WorkApp
- Privileged access (isPrivileged permission levels)
- Access patterns and anomalies
- Users with access to sensitive applications

Usage:
    python cakewalk_audit.py --output audit-report.json
    python cakewalk_audit.py --format csv --output audit.csv
"""

import os
import sys
import json
import csv
import argparse
from datetime import datetime
from typing import List, Dict, Any
import requests


class CakewalkAuditor:
    def __init__(self, api_key: str, api_secret: str):
        self.base_url = "https://open-api.getcakewalk.io"
        self.headers = {
            "X-API-KEY": api_key,
            "X-API-SECRET": api_secret
        }
        
    def _get_paginated(self, endpoint: str) -> List[Dict[str, Any]]:
        """Fetch all pages of results from a paginated endpoint."""
        results = []
        cursor = None
        
        while True:
            url = f"{self.base_url}{endpoint}"
            if cursor:
                separator = "&" if "?" in endpoint else "?"
                url += f"{separator}cursor={cursor}"
            
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            
            data = response.json()
            results.extend(data.get("data", []))
            
            # Check for next page
            next_link = data.get("links", {}).get("next")
            if not next_link or "cursor=" not in next_link:
                break
                
            cursor = next_link.split("cursor=")[-1]
                
        return results
    
    def get_all_users(self) -> List[Dict[str, Any]]:
        """Fetch all users."""
        return self._get_paginated("/api/v1/Users")
    
    def get_all_workapps(self) -> List[Dict[str, Any]]:
        """Fetch all WorkApps."""
        return self._get_paginated("/api/v1/WorkApps")
    
    def get_workapp_accesses(self, workapp_id: str) -> List[Dict[str, Any]]:
        """Fetch all accesses for a specific WorkApp."""
        return self._get_paginated(f"/api/v1/WorkApps/{workapp_id}/Accesses")
    
    def generate_audit_report(self) -> Dict[str, Any]:
        """Generate comprehensive audit report."""
        print("Fetching users...", file=sys.stderr)
        users = self.get_all_users()
        
        print("Fetching WorkApps...", file=sys.stderr)
        workapps = self.get_all_workapps()
        
        # Build lookup dictionaries
        user_lookup = {u["id"]: u for u in users}
        
        # Define sensitive applications (customize this)
        SENSITIVE_APP_NAMES = {"AWS", "Production Database", "Admin Panel", "Payroll", "HR System"}
        
        sensitive_workapps = [
            wa for wa in workapps 
            if wa.get("name") in SENSITIVE_APP_NAMES
        ]
        
        # Analyze access per WorkApp
        print("Analyzing access patterns...", file=sys.stderr)
        
        all_access_data = []
        privileged_access = []
        user_access_count = {}
        workapp_user_counts = {}
        
        for workapp in workapps:
            workapp_id = workapp["id"]
            workapp_name = workapp["name"]
            
            print(f"  Checking {workapp_name}...", file=sys.stderr)
            
            try:
                accesses = self.get_workapp_accesses(workapp_id)
            except Exception as e:
                print(f"    Warning: Could not fetch accesses for {workapp_name}: {e}", file=sys.stderr)
                continue
            
            workapp_user_counts[workapp_id] = len(accesses)
            
            for access in accesses:
                user = access.get("user", {})
                perm_level = access.get("permissionLevel", {})
                is_owner = access.get("isOwner", False)
                
                user_id = user.get("id")
                if user_id:
                    user_access_count[user_id] = user_access_count.get(user_id, 0) + 1
                
                # Track privileged access
                if perm_level.get("isPrivileged") or is_owner:
                    privileged_access.append({
                        "user_name": user.get("name"),
                        "user_email": user.get("email"),
                        "workapp": workapp_name,
                        "permission_level": perm_level.get("name"),
                        "is_owner": is_owner,
                        "is_privileged": perm_level.get("isPrivileged", False)
                    })
                
                # Track all access for sensitive apps
                if workapp_name in SENSITIVE_APP_NAMES:
                    all_access_data.append({
                        "user_name": user.get("name"),
                        "user_email": user.get("email"),
                        "workapp": workapp_name,
                        "permission_level": perm_level.get("name"),
                        "is_owner": is_owner,
                        "last_login": user.get("lastLogin"),
                        "user_status": user.get("statusName")
                    })
        
        # Find users with excessive access
        excessive_access = [
            {
                "user_name": user_lookup.get(user_id, {}).get("name", "Unknown"),
                "user_email": user_lookup.get(user_id, {}).get("email", "Unknown"),
                "access_count": count
            }
            for user_id, count in user_access_count.items()
            if count > 20  # Threshold for "excessive"
        ]
        excessive_access.sort(key=lambda x: x["access_count"], reverse=True)
        
        # Find WorkApps with most users
        top_workapps = sorted(
            [{"workapp": wa["name"], "user_count": workapp_user_counts.get(wa["id"], 0)}
             for wa in workapps],
            key=lambda x: x["user_count"],
            reverse=True
        )[:20]
        
        return {
            "generated_at": datetime.now().isoformat(),
            "summary": {
                "total_users": len(users),
                "total_workapps": len(workapps),
                "total_privileged_access": len(privileged_access),
                "total_sensitive_access": len(all_access_data),
                "users_with_excessive_access": len(excessive_access)
            },
            "privileged_access": privileged_access,
            "sensitive_app_access": all_access_data,
            "excessive_access": excessive_access,
            "top_workapps_by_users": top_workapps
        }
    
    def export_to_csv(self, report: Dict[str, Any], filename: str, section: str = "privileged_access"):
        """Export a section to CSV."""
        data = report.get(section, [])
        
        if not data:
            print(f"No data in section '{section}' to export", file=sys.stderr)
            return
        
        with open(filename, 'w', newline='') as csvfile:
            fieldnames = data[0].keys()
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(data)
        
        print(f"CSV report saved to {filename}", file=sys.stderr)


def main():
    parser = argparse.ArgumentParser(description="Cakewalk Access Audit Tool")
    parser.add_argument("--output", default="audit-report.json", help="Output filename")
    parser.add_argument("--format", choices=["json", "csv"], default="json", help="Output format")
    parser.add_argument("--section", default="privileged_access", 
                       help="CSV section to export (privileged_access, sensitive_app_access, excessive_access)")
    
    args = parser.parse_args()
    
    # Get credentials from environment
    api_key = os.getenv("CAKEWALK_API_KEY")
    api_secret = os.getenv("CAKEWALK_API_SECRET")
    
    if not api_key or not api_secret:
        print("Error: Set CAKEWALK_API_KEY and CAKEWALK_API_SECRET environment variables")
        sys.exit(1)
    
    # Run audit
    auditor = CakewalkAuditor(api_key, api_secret)
    report = auditor.generate_audit_report()
    
    # Save report
    if args.format == "json":
        with open(args.output, 'w') as f:
            json.dump(report, f, indent=2)
        print(f"JSON report saved to {args.output}", file=sys.stderr)
    else:
        auditor.export_to_csv(report, args.output, args.section)
    
    # Print summary to stdout
    print("\n=== AUDIT SUMMARY ===")
    print(f"Generated: {report['generated_at']}")
    print(f"Total Users: {report['summary']['total_users']}")
    print(f"Total WorkApps: {report['summary']['total_workapps']}")
    print(f"Privileged Access Grants: {report['summary']['total_privileged_access']}")
    print(f"Sensitive App Access: {report['summary']['total_sensitive_access']}")
    
    if report["excessive_access"]:
        print(f"\n⚠️  {len(report['excessive_access'])} users have excessive access (>20 apps)")
        print("\nTop 5 users by access count:")
        for user in report["excessive_access"][:5]:
            print(f"  - {user['user_name']} ({user['user_email']}): {user['access_count']} apps")


if __name__ == "__main__":
    main()

2. Make the Script Executable

chmod +x cakewalk_audit.py

3. Test the Script

# Test with JSON output
python cakewalk_audit.py --output test-audit.json

# Test with CSV output
python cakewalk_audit.py --format csv --section privileged_access --output privileged-access.csv

Usage Examples

Basic Audit

# Generate JSON report
python cakewalk_audit.py --output daily-audit.json

# Generate CSV report for privileged access
python cakewalk_audit.py --format csv --section privileged_access --output privileged-access.csv

Scheduled Audits

# Add to crontab for daily execution
# Run at 9 AM every day
0 9 * * * /path/to/cakewalk_audit.py --output /var/log/cakewalk-audit-$(date +\%Y\%m\%d).json

Custom Sensitive Apps

Edit the script to customize sensitive applications:

# Define sensitive applications (customize this)
SENSITIVE_APP_NAMES = {
    "AWS", 
    "Production Database", 
    "Admin Panel", 
    "Payroll", 
    "HR System",
    "Your Custom App"  # Add your sensitive apps here
}

Customization

Adjusting Access Thresholds

Modify the excessive access threshold:

# Find users with excessive access
excessive_access = [
    {
        "user_name": user_lookup.get(user_id, {}).get("name", "Unknown"),
        "user_email": user_lookup.get(user_id, {}).get("email", "Unknown"),
        "access_count": count
    }
    for user_id, count in user_access_count.items()
    if count > 10  # Change from 20 to 10
]

Adding Email Notifications

Add email notification functionality:

import smtplib
from email.mime.text import MIMEText

def send_report(report, recipients):
    msg = MIMEText(json.dumps(report, indent=2))
    msg['Subject'] = f'Cakewalk Audit Report - {datetime.now().date()}'
    msg['From'] = '[email protected]'
    msg['To'] = ', '.join(recipients)
    
    with smtplib.SMTP('smtp.company.com') as server:
        server.send_message(msg)

Adding Slack Notifications

Add Slack notification functionality:

import requests

def send_slack_notification(webhook_url, report):
    summary = report['summary']
    message = f"""
    🚨 Cakewalk Audit Report
    
    • Total Users: {summary['total_users']}
    • Total WorkApps: {summary['total_workapps']}
    • Privileged Access: {summary['total_privileged_access']}
    • Excessive Access: {summary['users_with_excessive_access']}
    """
    
    requests.post(webhook_url, json={"text": message})

Monitoring & Troubleshooting

Logging

The script includes detailed logging:

• Progress indicators for long-running operations

• Error messages for failed API calls

• Summary statistics

Common Issues

"User not found" Error:

• Verify the user exists in Cakewalk

• Check email spelling and case sensitivity

• Ensure the user has been invited to your organization

"401 Unauthorized" Error:

• Verify API key and secret are correct

• Check that credentials haven't expired

• Ensure proper header formatting

"Empty response" Error:

• Verify the API endpoint URL

• Check network connectivity

• Review API rate limits

Security Considerations

• API Credentials: Store as environment variables, never in code

• Access Review: Regularly review audit results

• Audit Logging: All audit operations are logged

• Data Handling: Ensure audit data is stored securely

Next Steps

• Set up monitoring: Configure alerts for audit failures

• Add notifications: Integrate with Slack, email, or other systems

• Automate further: Consider triggering from other systems

• Customize reporting: Add custom report formats and metrics