IBM Cloud Security Engineer (C1000-178) - Compliance & Security Operations¶

# automated-compliance.py
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
from ibm_scc import SecurityAndComplianceCenterApiV3
import json
from datetime import datetime

class ComplianceAutomation:
    def __init__(self, api_key, instance_id):
        authenticator = IAMAuthenticator(api_key)
        self.scc_client = SecurityAndComplianceCenterApiV3(
            authenticator=authenticator
        )
        self.scc_client.set_service_url(
            'https://us-south.compliance.cloud.ibm.com/instances/{}'.format(instance_id)
        )

    def run_compliance_scan(self, profile_id, attachment_id):
        """Trigger compliance scan"""
        try:
            response = self.scc_client.create_scan(
                attachment_id=attachment_id
            )
            scan_id = response.get_result()['id']
            print(f"Scan initiated: {scan_id}")
            return scan_id
        except Exception as e:
            print(f"Scan failed: {str(e)}")
            return None

    def get_compliance_score(self, attachment_id, report_id):
        """Calculate compliance score"""
        try:
            results = self.scc_client.list_report_evaluations(
                report_id=report_id
            ).get_result()

            total_controls = len(results['evaluations'])
            passed_controls = sum(1 for e in results['evaluations']
                                 if e['status'] == 'pass')

            score = (passed_controls / total_controls) * 100

            return {
                'score': round(score, 2),
                'total_controls': total_controls,
                'passed': passed_controls,
                'failed': total_controls - passed_controls,
                'timestamp': datetime.now().isoformat()
            }
        except Exception as e:
            print(f"Error calculating score: {str(e)}")
            return None

    def export_compliance_report(self, report_id, output_file):
        """Export compliance report to JSON"""
        try:
            report = self.scc_client.get_report(
                report_id=report_id
            ).get_result()

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

            print(f"Report exported to {output_file}")
            return True
        except Exception as e:
            print(f"Export failed: {str(e)}")
            return False

    def check_critical_failures(self, report_id):
        """Identify critical control failures"""
        try:
            results = self.scc_client.list_report_evaluations(
                report_id=report_id
            ).get_result()

            critical_failures = []
            for evaluation in results['evaluations']:
                if (evaluation['status'] == 'fail' and
                    evaluation.get('severity') == 'critical'):
                    critical_failures.append({
                        'control_id': evaluation['control_id'],
                        'control_name': evaluation['control_name'],
                        'reason': evaluation.get('reason', 'N/A'),
                        'resource': evaluation.get('target', {})
                    })

            return critical_failures
        except Exception as e:
            print(f"Error checking failures: {str(e)}")
            return []

# Usage
if __name__ == "__main__":
    automation = ComplianceAutomation(
        api_key='your-api-key',
        instance_id='your-instance-id'
    )

    # Run scan
    scan_id = automation.run_compliance_scan(
        profile_id='profile-123',
        attachment_id='attachment-456'
    )

    # Get compliance score
    score = automation.get_compliance_score(
        attachment_id='attachment-456',
        report_id='report-789'
    )
    print(f"Compliance Score: {score['score']}%")

    # Check critical failures
    failures = automation.check_critical_failures('report-789')
    if failures:
        print(f"Critical Failures: {len(failures)}")
        for failure in failures:
            print(f"  - {failure['control_name']}: {failure['reason']}")

# security-event-handler.py
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
from ibm_platform_services import IamAccessGroupsV2
import requests
import json

class SecurityEventHandler:
    def __init__(self, api_key):
        self.api_key = api_key
        authenticator = IAMAuthenticator(api_key)
        self.iam_client = IamAccessGroupsV2(authenticator=authenticator)

    def handle_unauthorized_access(self, event):
        """Respond to unauthorized access attempts"""
        user_id = event.get('initiator', {}).get('id')
        resource = event.get('target', {}).get('name')

        print(f"Unauthorized access detected: {user_id} -> {resource}")

        # Revoke user's API keys
        self.revoke_user_api_keys(user_id)

        # Send alert
        self.send_security_alert({
            'severity': 'HIGH',
            'event_type': 'unauthorized_access',
            'user': user_id,
            'resource': resource,
            'action': 'API keys revoked'
        })

        # Create incident ticket
        self.create_incident_ticket(event)

    def handle_policy_violation(self, event):
        """Respond to policy violations"""
        violation_type = event.get('outcome', {}).get('reason')
        resource_crn = event.get('target', {}).get('id')

        print(f"Policy violation: {violation_type}")

        # Apply remediation
        if violation_type == 'encryption_disabled':
            self.enable_encryption(resource_crn)
        elif violation_type == 'public_access_enabled':
            self.disable_public_access(resource_crn)
        elif violation_type == 'weak_password':
            self.enforce_password_policy()

    def handle_compliance_drift(self, event):
        """Respond to compliance drift"""
        control_id = event.get('control_id')
        resource_id = event.get('resource_id')

        print(f"Compliance drift detected: {control_id}")

        # Run targeted scan
        self.run_compliance_check(control_id, resource_id)

        # Notify compliance team
        self.send_compliance_notification({
            'control': control_id,
            'resource': resource_id,
            'status': 'drift_detected'
        })

    def revoke_user_api_keys(self, user_id):
        """Revoke all API keys for a user"""
        # Implementation would use IBM Cloud IAM API
        print(f"Revoking API keys for user: {user_id}")

    def send_security_alert(self, alert_data):
        """Send security alert to monitoring system"""
        # Integration with PagerDuty, Slack, etc.
        print(f"Security alert sent: {alert_data}")

    def create_incident_ticket(self, event):
        """Create incident management ticket"""
        # Integration with ServiceNow, Jira, etc.
        print(f"Incident ticket created for event: {event.get('id')}")

# Event processing function for IBM Cloud Functions
def main(args):
    """Process Activity Tracker events"""
    api_key = args.get('api_key')
    event = args.get('event')

    handler = SecurityEventHandler(api_key)

    action = event.get('action')
    outcome = event.get('outcome')

    # Route to appropriate handler
    if outcome == 'failure' and 'unauthorized' in action.lower():
        handler.handle_unauthorized_access(event)
    elif 'policy' in action.lower():
        handler.handle_policy_violation(event)
    elif 'compliance' in action.lower():
        handler.handle_compliance_drift(event)

    return {'status': 'processed', 'event_id': event.get('id')}

# log-analysis.py
from datetime import datetime, timedelta
import json
import re

class AuditLogAnalyzer:
    def __init__(self, activity_tracker_logs):
        self.logs = activity_tracker_logs

    def detect_suspicious_patterns(self):
        """Detect suspicious activity patterns"""
        patterns = {
            'brute_force': self.detect_brute_force(),
            'privilege_escalation': self.detect_privilege_escalation(),
            'data_exfiltration': self.detect_data_exfiltration(),
            'unusual_access': self.detect_unusual_access_patterns()
        }
        return patterns

    def detect_brute_force(self):
        """Detect brute force login attempts"""
        failed_logins = {}
        threshold = 5
        time_window = timedelta(minutes=10)

        for log in self.logs:
            if log.get('action') == 'iam.policy.authenticate':
                if log.get('outcome') == 'failure':
                    user = log.get('initiator', {}).get('id')
                    timestamp = datetime.fromisoformat(log.get('eventTime'))

                    if user not in failed_logins:
                        failed_logins[user] = []
                    failed_logins[user].append(timestamp)

        # Check for threshold breaches
        alerts = []
        for user, timestamps in failed_logins.items():
            timestamps.sort()
            for i in range(len(timestamps) - threshold + 1):
                window = timestamps[i:i+threshold]
                if window[-1] - window[0] <= time_window:
                    alerts.append({
                        'user': user,
                        'failed_attempts': threshold,
                        'time_range': f"{window[0]} to {window[-1]}",
                        'severity': 'HIGH'
                    })
                    break

        return alerts

    def detect_privilege_escalation(self):
        """Detect privilege escalation attempts"""
        escalation_events = []

        for log in self.logs:
            action = log.get('action', '')
            if any(keyword in action for keyword in [
                'iam.policy.create',
                'iam.role.assign',
                'iam.serviceId.create'
            ]):
                initiator = log.get('initiator', {}).get('id')
                target = log.get('target', {}).get('id')

                if initiator != target:  # Someone modifying another user
                    escalation_events.append({
                        'initiator': initiator,
                        'target': target,
                        'action': action,
                        'timestamp': log.get('eventTime'),
                        'outcome': log.get('outcome')
                    })

        return escalation_events

    def detect_data_exfiltration(self):
        """Detect potential data exfiltration"""
        download_events = []
        threshold_size = 1024 * 1024 * 100  # 100 MB

        for log in self.logs:
            if 'download' in log.get('action', '').lower():
                size = log.get('requestData', {}).get('size', 0)
                if size > threshold_size:
                    download_events.append({
                        'user': log.get('initiator', {}).get('id'),
                        'resource': log.get('target', {}).get('name'),
                        'size_mb': size / (1024 * 1024),
                        'timestamp': log.get('eventTime'),
                        'ip_address': log.get('initiator', {}).get('host', {}).get('address')
                    })

        return download_events

    def detect_unusual_access_patterns(self):
        """Detect unusual access patterns"""
        user_access = {}

        for log in self.logs:
            user = log.get('initiator', {}).get('id')
            timestamp = datetime.fromisoformat(log.get('eventTime'))
            hour = timestamp.hour

            if user not in user_access:
                user_access[user] = {'hours': [], 'locations': []}

            user_access[user]['hours'].append(hour)
            user_access[user]['locations'].append(
                log.get('initiator', {}).get('host', {}).get('address')
            )

        # Detect unusual access times (outside business hours)
        alerts = []
        for user, data in user_access.items():
            unusual_hours = [h for h in data['hours'] if h < 6 or h > 22]
            if len(unusual_hours) > 5:
                alerts.append({
                    'user': user,
                    'unusual_access_count': len(unusual_hours),
                    'pattern': 'after_hours_access',
                    'severity': 'MEDIUM'
                })

            # Detect access from multiple locations
            unique_locations = set(data['locations'])
            if len(unique_locations) > 3:
                alerts.append({
                    'user': user,
                    'location_count': len(unique_locations),
                    'pattern': 'multi_location_access',
                    'severity': 'MEDIUM'
                })

        return alerts

    def generate_forensic_report(self, incident_time, time_range_hours=24):
        """Generate forensic report for incident investigation"""
        incident_dt = datetime.fromisoformat(incident_time)
        start_time = incident_dt - timedelta(hours=time_range_hours)
        end_time = incident_dt + timedelta(hours=time_range_hours)

        relevant_logs = [
            log for log in self.logs
            if start_time <= datetime.fromisoformat(log.get('eventTime')) <= end_time
        ]

        report = {
            'incident_time': incident_time,
            'analysis_window': f"{start_time} to {end_time}",
            'total_events': len(relevant_logs),
            'event_breakdown': self._analyze_events(relevant_logs),
            'affected_resources': self._get_affected_resources(relevant_logs),
            'involved_users': self._get_involved_users(relevant_logs),
            'timeline': self._build_timeline(relevant_logs),
            'suspicious_patterns': self.detect_suspicious_patterns()
        }

        return report

    def _analyze_events(self, logs):
        """Analyze event types and outcomes"""
        event_types = {}
        outcomes = {'success': 0, 'failure': 0}

        for log in logs:
            action = log.get('action')
            outcome = log.get('outcome')

            event_types[action] = event_types.get(action, 0) + 1
            outcomes[outcome] = outcomes.get(outcome, 0) + 1

        return {
            'by_action': event_types,
            'by_outcome': outcomes
        }

    def _get_affected_resources(self, logs):
        """Extract affected resources"""
        resources = set()
        for log in logs:
            target = log.get('target', {}).get('name')
            if target:
                resources.add(target)
        return list(resources)

    def _get_involved_users(self, logs):
        """Extract involved users"""
        users = set()
        for log in logs:
            initiator = log.get('initiator', {}).get('id')
            if initiator:
                users.add(initiator)
        return list(users)

    def _build_timeline(self, logs):
        """Build event timeline"""
        timeline = sorted(logs, key=lambda x: x.get('eventTime'))
        return [
            {
                'time': log.get('eventTime'),
                'user': log.get('initiator', {}).get('id'),
                'action': log.get('action'),
                'resource': log.get('target', {}).get('name'),
                'outcome': log.get('outcome')
            }
            for log in timeline
        ]

# Usage example
if __name__ == "__main__":
    # Load logs from Activity Tracker
    with open('activity_tracker_logs.json', 'r') as f:
        logs = json.load(f)

    analyzer = AuditLogAnalyzer(logs)

    # Detect suspicious patterns
    patterns = analyzer.detect_suspicious_patterns()

    print("=== Brute Force Attempts ===")
    for alert in patterns['brute_force']:
        print(f"User: {alert['user']}, Attempts: {alert['failed_attempts']}")

    print("\n=== Privilege Escalation ===")
    for event in patterns['privilege_escalation']:
        print(f"{event['initiator']} -> {event['target']}: {event['action']}")

    print("\n=== Data Exfiltration ===")
    for event in patterns['data_exfiltration']:
        print(f"User: {event['user']}, Size: {event['size_mb']:.2f} MB")

    # Generate forensic report
    report = analyzer.generate_forensic_report('2024-01-15T14:30:00Z')
    with open('forensic_report.json', 'w') as f:
        json.dump(report, f, indent=2)