How to Detect New AWS Service Usage In Your Organization

As AWS organizations grow, keeping track of which services are being used across different teams and accounts becomes increasingly difficult. New services might be adopted without central IT awareness, leading to:

• Security blind spots - Unknown services may not follow security policies
• Cost surprises - Unexpected charges from services you didn't know were running
• Compliance gaps - New services might not meet regulatory requirements
• Shadow IT - Teams using services outside approved workflows

Solution Overview

AWS IAM provides Service Access Reports that deliver comprehensive service usage data across your entire organization. These reports answer the critical question: "For each service, view which account last accessed the service and when."

Key Features:

• Provides last access timestamps and entity details for each service
• Cross-account visibility with organizational hierarchy (OUs, accounts)
• Single API call generates organization-wide service usage data

This native AWS feature provides the most complete and authoritative view of actual service usage across your entire organization.

How Service Access Reports Work

AWS Service Access Reports analyze service usage by monitoring API authentication events across your entire organizational structure. This provides comprehensive visibility into which services are being accessed, when, by whom, and from which accounts.

Generating Service Access Reports

def generate_service_access_report():
    """Generate organization-wide service access report"""
    iam_client = boto3.client('iam')
    org_client = boto3.client('organizations')
    
    try:
        # Get organization details
        org_info = org_client.describe_organization()
        org_id = org_info['Organization']['Id']
        
        # Get organization root
        roots = org_client.list_roots()
        root_id = roots['Roots'][0]['Id']
        
        # Construct entity path for organization root
        entity_path = f"{org_id}/{root_id}"
        
        # Generate comprehensive organization access report
        response = iam_client.generate_organizations_access_report(
            EntityPath=entity_path
            # OrganizationsPolicyId is optional - omitting for complete coverage
        )
        
        job_id = response['JobId']
        print(f"Service access report generation started: {job_id}")
        
        # Poll for completion
        while True:
            details = iam_client.get_organizations_access_report(JobId=job_id)
            
            if details['JobStatus'] == 'COMPLETED':
                print(f"Report completed - {details['NumberOfServicesAccessible']} services analyzed")
                
                services = {}
                for access_detail in details.get('AccessDetails', []):
                    service_namespace = access_detail['ServiceNamespace']
                    
                    if service_namespace not in services:
                        services[service_namespace] = {
                            'service_name': access_detail['ServiceName'],
                            'accounts': [],
                            'last_access': None,
                            'total_entities': 0
                        }
                    
                    # Add account-level access information
                    if access_detail.get('LastAuthenticatedTime'):
                        account_info = {
                            'entity_path': access_detail.get('EntityPath'),
                            'last_authenticated': access_detail['LastAuthenticatedTime'],
                            'region': access_detail.get('Region'),
                            'entities': access_detail.get('TotalAuthenticatedEntities', 0)
                        }
                        
                        services[service_namespace]['accounts'].append(account_info)
                        
                        # Track most recent access across all accounts
                        if not services[service_namespace]['last_access'] or \
                           access_detail['LastAuthenticatedTime'] > services[service_namespace]['last_access']:
                            services[service_namespace]['last_access'] = access_detail['LastAuthenticatedTime']
                        
                        services[service_namespace]['total_entities'] += account_info['entities']
                
                return services
                
            elif details['JobStatus'] == 'FAILED':
                error_details = details.get('ErrorDetails', {})
                raise Exception(f"Service access report generation failed: {error_details}")
            
            time.sleep(10)  # Wait before polling again
            
    except ClientError as e:
        print(f"Error generating service access report: {e}")
        return {}

Processing Organization-Wide Service Data

The Organizations Access Report provides detailed information for each service:

def process_service_access_data(services_data):
    """Process and analyze organization-wide service access data"""
    
    # Categorize services by usage patterns
    active_services = {}
    unused_services = {}
    
    for service_namespace, service_info in services_data.items():
        if service_info['accounts']:
            # Service is actively used
            active_services[service_namespace] = {
                'service_name': service_info['service_name'],
                'accounts_using': len(service_info['accounts']),
                'last_access': service_info['last_access'],
                'total_entities': service_info['total_entities'],
                'regions': list(set(acc.get('region') for acc in service_info['accounts'] if acc.get('region')))
            }
        else:
            # Service is available but not accessed
            unused_services[service_namespace] = service_info['service_name']
    
    return {
        'active_services': active_services,
        'unused_services': unused_services,
        'total_services_available': len(services_data),
        'services_in_use': len(active_services)
    }

New Service Detection Logic

Establish baselines and detect changes using Organizations Access Report data:

def detect_new_services():
    """Detect new services using AWS Service Access Reports"""
    
    # Generate comprehensive organization-wide service access report
    current_services = generate_service_access_report()
    
    # Process the data for analysis
    processed_data = process_service_access_data(current_services)
    
    # Load baseline from previous scan
    baseline = load_baseline()
    
    # Find new services
    baseline_services = set(baseline.get('active_services', {}).keys())
    current_active_services = set(processed_data['active_services'].keys())
    new_services = current_active_services - baseline_services
    removed_services = baseline_services - current_active_services
    
    # Analyze new service usage with organizational context
    if new_services:
        print("NEW SERVICES DETECTED ACROSS ORGANIZATION:")
        for service in sorted(new_services):
            service_data = processed_data['active_services'][service]
            
            print(f"  + {service} ({service_data['service_name']})")
            print(f"    Accounts using: {service_data['accounts_using']}")
            print(f"    Last accessed: {service_data['last_access']}")
            print(f"    Total entities: {service_data['total_entities']}")
            print(f"    Regions: {', '.join(service_data['regions'])}")
            
            # Show which specific accounts are using this service
            service_details = current_services[service]
            for account_info in service_details['accounts']:
                entity_path = account_info['entity_path']
                account_id = entity_path.split('/')[-1] if entity_path else 'Unknown'
                print(f"      - Account {account_id}: {account_info['last_authenticated']}")
        
        send_comprehensive_alert(new_services, processed_data, current_services)
    
    # Track services no longer being accessed
    if removed_services:
        print("SERVICES NO LONGER ACCESSED:")
        for service in sorted(removed_services):
            print(f"  - {service}")
    
    # Update baseline with complete organizational data
    save_baseline({
        'active_services': processed_data['active_services'],
        'unused_services': processed_data['unused_services'],
        'last_updated': datetime.now().isoformat(),
        'total_services_available': processed_data['total_services_available'],
        'services_in_use': processed_data['services_in_use'],
        'detection_method': 'organizations_access_reports'
    })
    
    return {
        'new_services': list(new_services),
        'removed_services': list(removed_services),
        'total_services_available': processed_data['total_services_available'],
        'services_in_use': processed_data['services_in_use'],
        'service_details': {service: current_services[service] for service in new_services}
    }

Production Architecture

Deploy as a serverless solution that leverages AWS Service Access Reports:

def lambda_handler(event, context):
    """Lambda handler for Service Access Report-based detection"""
    
    detector = ServiceAccessDetector()
    
    # Generate comprehensive organization-wide service usage report
    results = detector.detect_new_services()
    
    if results['new_services']:
        # Send detailed alert with organizational context
        message = {
            'new_services': results['new_services'],
            'service_details': results['service_details'],
            'total_services_available': results['total_services_available'],
            'services_in_use': results['services_in_use'],
            'timestamp': datetime.now().isoformat(),
            'alert_type': 'new_organizational_service_usage',
            'detection_method': 'service_access_reports'
        }
        
        # Send to SNS for immediate alerting
        sns.publish(
            TopicArn=os.environ['SNS_TOPIC_ARN'],
            Message=json.dumps(message, indent=2, default=str),
            Subject='New AWS Services Detected Across Organization'
        )
        
        # Store comprehensive report in S3 for analysis
        s3.put_object(
            Bucket=os.environ['REPORTS_BUCKET'],
            Key=f"org-service-reports/{datetime.now().strftime('%Y/%m/%d')}/access-report.json",
            Body=json.dumps(message, indent=2, default=str)
        )
    
    return {
        'statusCode': 200,
        'body': json.dumps(results, default=str)
    }

Infrastructure Components:

• Lambda Function - Processes Service Access Reports for the entire organization
• EventBridge Rule - Schedules weekly scans (reports can take several minutes to generate)
• SNS Topic - Sends alerts with detailed organizational service usage information
• S3 Bucket - Stores baselines and comprehensive organizational service reports
• IAM Role - Includes permissions for Service Access Reports

Implementation Considerations

Required IAM Permissions

The Lambda execution role needs specific IAM and Organizations permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iam:GenerateOrganizationsAccessReport",
        "iam:GetOrganizationsAccessReport",
        "organizations:DescribeOrganization",
        "organizations:ListRoots"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sns:Publish",
        "s3:PutObject",
        "s3:GetObject"
      ],
      "Resource": [
        "arn:aws:sns:*:*:new-service-alerts",
        "arn:aws:s3:::org-service-reports-bucket/*"
      ]
    }
  ]
}

Baseline Management

The solution maintains a baseline of known services to detect changes over time:

def load_baseline():
    """Load baseline service data from storage"""
    try:
        # Load from S3, DynamoDB, or local file
        response = s3_client.get_object(
            Bucket='service-reports-bucket',
            Key='baselines/service-baseline.json'
        )
        baseline_data = json.loads(response['Body'].read())
        print(f"Loaded baseline with {len(baseline_data.get('active_services', {}))} services")
        return baseline_data
    except ClientError as e:
        if e.response['Error']['Code'] == 'NoSuchKey':
            print("No baseline found, creating new baseline")
            return {}
        else:
            print(f"Error loading baseline: {e}")
            return {}

def save_baseline(baseline_data):
    """Save current service data as new baseline"""
    try:
        # Add metadata
        baseline_data.update({
            'last_updated': datetime.now().isoformat(),
            'detection_method': 'organizations_access_reports',
            'baseline_version': '1.0'
        })
        
        # Save to S3, DynamoDB, or local file
        s3_client.put_object(
            Bucket='service-reports-bucket',
            Key='baselines/service-baseline.json',
            Body=json.dumps(baseline_data, indent=2, default=str),
            ContentType='application/json'
        )
        
        print(f"Baseline saved with {baseline_data['services_in_use']} active services")
        
        # Optional: Keep historical baselines
        timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
        s3_client.put_object(
            Bucket='service-reports-bucket',
            Key=f'baselines/history/baseline-{timestamp}.json',
            Body=json.dumps(baseline_data, indent=2, default=str),
            ContentType='application/json'
        )
        
    except Exception as e:
        print(f"Error saving baseline: {e}")
        raise

Store comprehensive organizational baselines sample:

{
  "active_services": {
    "s3": {
      "service_name": "Amazon S3",
      "accounts_using": 3,
      "last_access": "2024-01-15T10:30:00Z",
      "total_entities": 12,
      "regions": ["us-east-1", "us-west-2", "eu-west-1"]
    },
    "ec2": {
      "service_name": "Amazon EC2",
      "accounts_using": 2,
      "last_access": "2024-01-14T15:20:00Z", 
      "total_entities": 8,
      "regions": ["us-east-1", "us-west-2"]
    }
  },
  "unused_services": {
    "braket": "Amazon Braket",
    "honeycode": "Amazon Honeycode"
  },
  "last_updated": "2024-01-15T12:00:00Z",
  "total_services_available": 441,
  "services_in_use": 150,
  "detection_method": "service_access_reports"
}

Cost and Performance Considerations

API Costs:

• Service Access Report generation: No additional charge for IAM reports
• Organizations API calls: Included in AWS Organizations pricing
• Lambda execution: Low cost for weekly organizational scans
• S3 storage: Minimal cost for comprehensive organizational reports

Conclusion

AWS Service Access Reports provide the most comprehensive and efficient method for detecting new service usage across your entire organization. By leveraging this native AWS feature, you get:

• Cross-account visibility showing which accounts and entities access each service
• Detailed usage information with timestamps, regions, and entity details
• Efficient single API call that covers your entire organizational structure

This approach provides the definitive answer to "For each service, view which account last accessed the service and when" across your entire AWS organization. Unlike per-account or per-role approaches, Service Access Reports deliver complete organizational visibility in a single comprehensive report.

Start with Service Access Reports to establish comprehensive organizational service usage baselines, then implement automated monitoring to detect when new services are accessed across your AWS organization.

---

Service Access Reports provide the foundation for comprehensive organizational service governance and security oversight across AWS Organizations.