Amazonaws Virus — Why Email Gateways Whitelist Malware URLs

import re
from dataclasses import dataclass
from typing import List, Optional
from urllib.parse import urlparse
from io.thecodeforge.security.models import ThreatIndicator

@dataclass
class AWSAbusePattern:
    name: str
    pattern: str
    threat_type: str
    confidence: str


class AmazonawsThreatDetector:
    """
    Detects potentially malicious content hosted on AWS infrastructure.
    Analyzes URL patterns, file extensions, and behavioral indicators.
    """
    
    AWS_PATTERNS = [
        AWSAbusePattern(
            name="s3_phishing_page",
            pattern=r"https?://[\w.-]*\.s3[\w.-]*\.amazonaws\.com/[\w/.-]*\.(html|htm|php|aspx)",
            threat_type="phishing",
            confidence="medium"
        ),
        AWSAbusePattern(
            name="s3_malware_download",
            pattern=r"https?://[\w.-]*\.s3[\w.-]*\.amazonaws\.com/[\w/.-]*\.(exe|dll|scr|bat|cmd|ps1|vbs|js)",
            threat_type="malware",
            confidence="high"
        ),
        AWSAbusePattern(
            name="s3_document_exploit",
            pattern=r"https?://[\w.-]*\.s3[\w.-]*\.amazonaws\.com/[\w/.-]*\.(docm|xlsm|pptm|hta)",
            threat_type="exploit",
            confidence="high"
        ),
        AWSAbusePattern(
            name="ec2_c2_server",
            pattern=r"https?://ec2-\d+-\d+-\d+-\d+[\w.-]*\.amazonaws\.com",
            threat_type="c2",
            confidence="low"
        )
    ]
    
    SUSPICIOUS_EXTENSIONS = {
        ".exe", ".dll", ".scr", ".bat", ".cmd",
        ".ps1", ".vbs", ".js", ".hta", ".docm",
        ".xlsm", ".pptm", ".iso", ".img"
    }
    
    def analyze_url(self, url: str) -> Optional[ThreatIndicator]:
        """
        Analyze a URL for AWS-hosted threat indicators.
        """
        parsed = urlparse(url)
        
        if "amazonaws.com" not in parsed.hostname:
            return None
        
        for pattern in self.AWS_PATTERNS:
            if re.match(pattern.pattern, url, re.IGNORECASE):
                return ThreatIndicator(
                    url=url,
                    threat_type=pattern.threat_type,
                    confidence=pattern.confidence,
                    pattern_name=pattern.name,
                    action="block" if pattern.confidence == "high" else "monitor"
                )
        
        path = parsed.path.lower()
        for ext in self.SUSPICIOUS_EXTENSIONS:
            if path.endswith(ext):
                return ThreatIndicator(
                    url=url,
                    threat_type="suspicious_download",
                    confidence="medium",
                    pattern_name="suspicious_extension",
                    action="quarantine"
                )
        
        return None
    
    def analyze_s3_bucket_policy(self, policy: dict) -> List[str]:
        """
        Analyze S3 bucket policy for public access risks.
        """
        warnings = []
        
        for statement in policy.get("Statement", []):
            principal = statement.get("Principal", {})
            effect = statement.get("Effect", "")
            
            if effect == "Allow" and principal == "*":
                warnings.append(
                    "Bucket allows public access via Principal: *"
                )
            
            actions = statement.get("Action", [])
            if isinstance(actions, str):
                actions = [actions]
            
            if effect == "Allow" and "s3:GetObject" in actions:
                if principal == "*" or principal.get("AWS") == "*":
                    warnings.append(
                        "Bucket allows public GetObject — anyone can download files"
                    )
        
        return warnings


class AWSAbuseReporter:
    """
    Handles reporting of AWS-hosted abuse to AWS Security.
    """
    
    ABUSE_EMAIL = "abuse@amazonaws.com"
    
    @staticmethod
    def format_abuse_report(
        malicious_url: str,
        threat_type: str,
        evidence: str
    ) -> str:
        return f"""
AWS Abuse Report

Malicious URL: {malicious_url}
Threat Type: {threat_type}
Evidence: {evidence}

Please investigate and take appropriate action.
"""
    
    @staticmethod
    def report_to_aws(url: str, threat_type: str, evidence: str) -> None:
        report = AWSAbuseReporter.format_abuse_report(
            url, threat_type, evidence
        )
        print(f"Send report to {AWSAbuseReporter.ABUSE_EMAIL}:")
        print(report)

import boto3
from datetime import datetime, timedelta
from typing import List, Dict
from io.thecodeforge.security.models import S3AuditFinding


class S3SecurityAuditor:
    """
    Audits S3 buckets for public access and security misconfigurations
    that could be exploited for malware hosting.
    """
    
    def __init__(self):
        self.s3_client = boto3.client('s3')
        self.s3control_client = boto3.client('s3control')
    
    def audit_all_buckets(self) -> List[S3AuditFinding]:
        """
        Audit all S3 buckets in the account for security issues.
        """
        findings = []
        
        response = self.s3_client.list_buckets()
        for bucket in response['Buckets']:
            bucket_name = bucket['Name']
            findings.extend(self._audit_bucket(bucket_name))
        
        return findings
    
    def _audit_bucket(self, bucket_name: str) -> List[S3AuditFinding]:
        """
        Audit a single bucket for common security issues.
        """
        findings = []
        
        # Check public access block
        try:
            pab = self.s3_client.get_public_access_block(Bucket=bucket_name)
            config = pab['PublicAccessBlockConfiguration']
            
            if not config.get('BlockPublicAcls', False):
                findings.append(S3AuditFinding(
                    bucket=bucket_name,
                    issue="BlockPublicAcls is disabled",
                    severity="high",
                    remediation="Enable BlockPublicAcls on the bucket"
                ))
            
            if not config.get('BlockPublicPolicy', False):
                findings.append(S3AuditFinding(
                    bucket=bucket_name,
                    issue="BlockPublicPolicy is disabled",
                    severity="high",
                    remediation="Enable BlockPublicPolicy on the bucket"
                ))
        except self.s3_client.exceptions.NoSuchPublicAccessBlockConfiguration:
            findings.append(S3AuditFinding(
                bucket=bucket_name,
                issue="No public access block configuration",
                severity="critical",
                remediation="Enable all public access block settings"
            ))
        
        # Check bucket policy for public access
        try:
            policy = self.s3_client.get_bucket_policy(Bucket=bucket_name)
            import json
            policy_doc = json.loads(policy['Policy'])
            
            for statement in policy_doc.get('Statement', []):
                if (statement.get('Effect') == 'Allow' and
                    statement.get('Principal') == '*'):
                    findings.append(S3AuditFinding(
                        bucket=bucket_name,
                        issue="Bucket policy allows public access",
                        severity="critical",
                        remediation="Remove Principal: * from bucket policy"
                    ))
        except self.s3_client.exceptions.NoSuchBucketPolicy:
            pass
        
        # Check bucket ACL for public access
        try:
            acl = self.s3_client.get_bucket_acl(Bucket=bucket_name)
            for grant in acl['Grants']:
                grantee = grant.get('Grantee', {})
                if grantee.get('URI') == 'http://acs.amazonaws.com/groups/global/AllUsers':
                    findings.append(S3AuditFinding(
                        bucket=bucket_name,
                        issue="Bucket ACL grants public access",
                        severity="critical",
                        remediation="Remove AllUsers from bucket ACL"
                    ))
        except Exception:
            pass
        
        return findings
    
    def enable_account_level_block(self) -> bool:
        """
        Enable S3 Block Public Access at the account level.
        This is the strongest protection against accidental public exposure.
        """
        try:
            account_id = boto3.client('sts').get_caller_identity()['Account']
            
            self.s3control_client.put_public_access_block(
                AccountId=account_id,
                PublicAccessBlockConfiguration={
                    'BlockPublicAcls': True,
                    'IgnorePublicAcls': True,
                    'BlockPublicPolicy': True,
                    'RestrictPublicBuckets': True
                }
            )
            return True
        except Exception as e:
            print(f"Failed to enable account-level block: {e}")
            return False

import boto3
from datetime import datetime, timedelta
from typing import List, Dict
from io.thecodeforge.security.models import EC2AuditFinding


class EC2SecurityMonitor:
    """
    Monitors EC2 instances for unauthorized usage and
    potential malware hosting infrastructure.
    """
    
    def __init__(self):
        self.ec2_client = boto3.client('ec2')
        self.cloudtrail_client = boto3.client('cloudtrail')
    
    def find_unauthorized_instances(
        self,
        approved_instance_ids: List[str]
    ) -> List[Dict]:
        """
        Identify EC2 instances not in the approved inventory.
        """
        unauthorized = []
        
        response = self.ec2_client.describe_instances(
            Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
        )
        
        for reservation in response['Reservations']:
            for instance in reservation['Instances']:
                instance_id = instance['InstanceId']
                
                if instance_id not in approved_instance_ids:
                    unauthorized.append({
                        'instance_id': instance_id,
                        'launch_time': instance['LaunchTime'].isoformat(),
                        'instance_type': instance['InstanceType'],
                        'public_ip': instance.get('PublicIpAddress', 'none'),
                        'vpc_id': instance.get('VpcId', 'none'),
                        'security_groups': [
                            sg['GroupId'] for sg in instance.get('SecurityGroups', [])
                        ]
                    })
        
        return unauthorized
    
    def check_security_group_exposure(self) -> List[EC2AuditFinding]:
        """
        Find security groups with overly permissive ingress rules.
        """
        findings = []
        
        response = self.ec2_client.describe_security_groups()
        
        for sg in response['SecurityGroups']:
            for rule in sg.get('IpPermissions', []):
                for ip_range in rule.get('IpRanges', []):
                    cidr = ip_range.get('CidrIp', '')
                    
                    if cidr == '0.0.0.0/0':
                        from_port = rule.get('FromPort', 'all')
                        to_port = rule.get('ToPort', 'all')
                        protocol = rule.get('IpProtocol', 'all')
                        
                        severity = 'critical' if from_port == 22 else 'high'
                        
                        findings.append(EC2AuditFinding(
                            security_group_id=sg['GroupId'],
                            group_name=sg['GroupName'],
                            issue=f"Open to internet: {protocol}/{from_port}-{to_port}",
                            severity=severity,
                            remediation=f"Restrict {cidr} to specific IP ranges"
                        ))
        
        return findings
    
    def get_recent_instance_launches(
        self,
        hours: int = 24
    ) -> List[Dict]:
        """
        List EC2 instances launched in the last N hours.
        Useful for detecting unauthorized provisioning.
        """
        since = datetime.utcnow() - timedelta(hours=hours)
        
        response = self.ec2_client.describe_instances(
            Filters=[
                {'Name': 'instance-state-name', 'Values': ['running', 'pending']}
            ]
        )
        
        recent = []
        for reservation in response['Reservations']:
            for instance in reservation['Instances']:
                if instance['LaunchTime'].replace(tzinfo=None) > since:
                    recent.append({
                        'instance_id': instance['InstanceId'],
                        'launch_time': instance['LaunchTime'].isoformat(),
                        'instance_type': instance['InstanceType'],
                        'public_ip': instance.get('PublicIpAddress', 'pending'),
                        'image_id': instance['ImageId']
                    })
        
        return recent
    
    def alert_on_unauthorized_launch(self, instance_data: Dict) -> None:
        """
        Trigger alert for potentially unauthorized EC2 instance.
        """
        print(f"ALERT: Unauthorized EC2 instance detected")
        print(f"  Instance ID: {instance_data['instance_id']}")
        print(f"  Launch Time: {instance_data['launch_time']}")
        print(f"  Public IP: {instance_data['public_ip']}")
        print(f"  Action: Investigate and terminate if unauthorized")

from typing import List, Dict, Optional
from dataclasses import dataclass
from io.thecodeforge.security.models import ThreatResponse


@dataclass
class DetectionRule:
    name: str
    description: str
    action: str
    enabled: bool


class AmazonawsDefenseSystem:
    """
    Comprehensive defense system for AWS-hosted threats.
    Combines URL analysis, content inspection, and account monitoring.
    """
    
    def __init__(self):
        self.detection_rules: List[DetectionRule] = []
        self.blocked_urls: List[str] = []
        self.whitelisted_buckets: List[str] = []
    
    def configure_detection_rules(self) -> None:
        """
        Set up detection rules for AWS-hosted threats.
        """
        self.detection_rules = [
            DetectionRule(
                name="s3_executable_download",
                description="Block executable file downloads from S3",
                action="block",
                enabled=True
            ),
            DetectionRule(
                name="s3_phishing_page",
                description="Inspect HTML pages on S3 for credential harvesting",
                action="quarantine",
                enabled=True
            ),
            DetectionRule(
                name="ec2_direct_ip_access",
                description="Monitor direct IP access to EC2 instances",
                action="log",
                enabled=True
            ),
            DetectionRule(
                name="s3_bucket_enumeration",
                description="Detect bucket name enumeration attempts",
                action="alert",
                enabled=True
            )
        ]
    
    def should_allow_url(self, url: str) -> bool:
        """
        Determine if a URL should be allowed based on whitelist
        and detection rules.
        """
        for bucket in self.whitelisted_buckets:
            if bucket in url:
                return True
        
        for blocked in self.blocked_urls:
            if blocked in url:
                return False
        
        return True
    
    def add_whitelisted_bucket(self, bucket_name: str) -> None:
        """
        Whitelist a specific S3 bucket used by your applications.
        Only whitelisted buckets should be allowed through filters.
        """
        if bucket_name not in self.whitelisted_buckets:
            self.whitelisted_buckets.append(bucket_name)
    
    def get_defense_recommendations(self) -> List[str]:
        """
        Return prioritized defense recommendations.
        """
        return [
            "Enable S3 Block Public Access at the account level",
            "Enable CloudTrail in all regions with log file validation",
            "Implement IAM least privilege — no wildcard permissions",
            "Rotate IAM access keys every 90 days maximum",
            "Enable GuardDuty for automated threat detection",
            "Whitelist specific S3 bucket names, not the entire amazonaws.com domain",
            "Deploy browser isolation for all email-borne links",
            "Inspect page content for credential harvesting regardless of domain",
            "Monitor VPC flow logs for unexpected S3 and EC2 traffic",
            "Report abuse to abuse@amazonaws.com for takedown"
        ]


class AWSCloudTrailAnalyzer:
    """
    Analyzes CloudTrail logs for indicators of compromise.
    """
    
    SUSPICIOUS_EVENTS = [
        "RunInstances",
        "CreateBucket",
        "PutBucketPolicy",
        "PutObject",
        "CreateAccessKey",
        "CreateUser",
        "AttachUserPolicy",
        "StopLogging"
    ]
    
    def __init__(self):
        self.cloudtrail_client = boto3.client('cloudtrail')
    
    def find_suspicious_events(
        self,
        hours: int = 24
    ) -> List[Dict]:
        """
        Search CloudTrail for suspicious API calls.
        """
        suspicious = []
        
        for event_name in self.SUSPICIOUS_EVENTS:
            response = self.cloudtrail_client.lookup_events(
                LookupAttributes=[
                    {'AttributeKey': 'EventName', 'AttributeValue': event_name}
                ],
                MaxResults=50
            )
            
            for event in response.get('Events', []):
                suspicious.append({
                    'event_name': event_name,
                    'event_time': event['EventTime'].isoformat(),
                    'user_identity': event.get('Username', 'unknown'),
                    'source_ip': event.get('SourceIPAddress', 'unknown'),
                    'event_source': event.get('EventSource', 'unknown')
                })
        
        return suspicious

import boto3
from typing import List, Dict
from io.thecodeforge.security.models import HardeningCheck


class AWSAccountHardening:
    """
    Implements AWS security best practices to prevent account
    abuse for malware hosting and distribution.
    """
    
    def __init__(self):
        self.findings: List[HardeningCheck] = []
    
    def check_all_controls(self) -> List[HardeningCheck]:
        """
        Run all hardening checks against the AWS account.
        """
        self.findings = []
        
        self._check_root_account_mfa()
        self._check_iam_password_policy()
        self._check_cloudtrail_enabled()
        self._check_guardduty_enabled()
        self._check_s3_block_public_access()
        self._check_unused_access_keys()
        self._check_security_hub_enabled()
        
        return self.findings
    
    def _check_root_account_mfa(self) -> None:
        iam = boto3.client('iam')
        summary = iam.get_account_summary()['SummaryMap']
        
        if summary.get('AccountMFAEnabled', 0) == 0:
            self.findings.append(HardeningCheck(
                control="Root Account MFA",
                status="FAIL",
                severity="critical",
                remediation="Enable MFA on the root account immediately"
            ))
        else:
            self.findings.append(HardeningCheck(
                control="Root Account MFA",
                status="PASS",
                severity="info",
                remediation=""
            ))
    
    def _check_cloudtrail_enabled(self) -> None:
        ct = boto3.client('cloudtrail')
        trails = ct.describe_trails()['trailList']
        
        if not trails:
            self.findings.append(HardeningCheck(
                control="CloudTrail Enabled",
                status="FAIL",
                severity="critical",
                remediation="Enable CloudTrail in all regions with log file validation"
            ))
        else:
            multi_region = any(t.get('IsMultiRegionTrail') for t in trails)
            if not multi_region:
                self.findings.append(HardeningCheck(
                    control="CloudTrail Multi-Region",
                    status="FAIL",
                    severity="high",
                    remediation="Enable multi-region CloudTrail to capture all API activity"
                ))
    
    def _check_guardduty_enabled(self) -> None:
        gd = boto3.client('guardduty')
        detectors = gd.list_detectors()['DetectorIds']
        
        if not detectors:
            self.findings.append(HardeningCheck(
                control="GuardDuty Enabled",
                status="FAIL",
                severity="high",
                remediation="Enable GuardDuty for automated threat detection"
            ))
    
    def _check_s3_block_public_access(self) -> None:
        s3control = boto3.client('s3control')
        sts = boto3.client('sts')
        account_id = sts.get_caller_identity()['Account']
        
        try:
            pab = s3control.get_public_access_block(AccountId=account_id)
            config = pab['PublicAccessBlockConfiguration']
            
            all_blocked = all([
                config.get('BlockPublicAcls', False),
                config.get('IgnorePublicAcls', False),
                config.get('BlockPublicPolicy', False),
                config.get('RestrictPublicBuckets', False)
            ])
            
            if not all_blocked:
                self.findings.append(HardeningCheck(
                    control="S3 Block Public Access",
                    status="FAIL",
                    severity="critical",
                    remediation="Enable all four Block Public Access settings at account level"
                ))
        except Exception:
            self.findings.append(HardeningCheck(
                control="S3 Block Public Access",
                status="FAIL",
                severity="critical",
                remediation="Configure S3 Block Public Access at account level"
            ))
    
    def _check_unused_access_keys(self) -> None:
        iam = boto3.client('iam')
        users = iam.list_users()['Users']
        
        for user in users:
            keys = iam.list_access_keys(UserName=user['UserName'])['AccessKeyMetadata']
            for key in keys:
                if key['Status'] == 'Active':
                    last_used = iam.get_access_key_last_used(
                        AccessKeyId=key['AccessKeyId']
                    )
                    last_used_date = last_used.get('AccessKeyLastUsed', {}).get('LastUsedDate')
                    
                    if last_used_date is None:
                        self.findings.append(HardeningCheck(
                            control=f"Unused Access Key: {key['AccessKeyId'][:8]}...",
                            status="FAIL",
                            severity="medium",
                            remediation=f"Deactivate unused key for user {user['UserName']}"
                        ))
    
    def _check_iam_password_policy(self) -> None:
        iam = boto3.client('iam')
        try:
            policy = iam.get_account_password_policy()['PasswordPolicy']
            
            if not policy.get('RequireUppercaseCharacters') or \
               not policy.get('RequireLowercaseCharacters') or \
               not policy.get('RequireNumbers') or \
               not policy.get('RequireSymbols'):
                self.findings.append(HardeningCheck(
                    control="IAM Password Policy",
                    status="FAIL",
                    severity="medium",
                    remediation="Enforce strong password policy with all character types"
                ))
        except iam.exceptions.NoSuchEntityException:
            self.findings.append(HardeningCheck(
                control="IAM Password Policy",
                status="FAIL",
                severity="medium",
                remediation="Configure IAM password policy"
            ))
    
    def _check_security_hub_enabled(self) -> None:
        sh = boto3.client('securityhub')
        try:
            sh.describe_hub()
        except sh.exceptions.InvalidAccessException:
            self.findings.append(HardeningCheck(
                control="Security Hub Enabled",
                status="FAIL",
                severity="medium",
                remediation="Enable Security Hub for centralized security findings"
            ))

// io.thecodeforge — cs-fundamentals tutorial

import boto3
from datetime import datetime, timezone, timedelta

# Non-negotiable: catch regions where you have zero legitimate activity
session = boto3.Session()
ec2_client = session.client('ec2', region_name='us-east-1')

# Get all regions available to your account
regions = [r['RegionName'] for r in ec2_client.describe_regions()['Regions']]

active_regions = []
suspicious_regions = []

for region in regions:
    try:
        # 10-day lookback — long enough to catch stealthy miners
        ec2_regional = session.client('ec2', region_name=region)
        instances = ec2_regional.describe_instances(
            Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
        )
        # Count running instances
        count = sum(len(r['Instances']) for r in instances['Reservations'])
        if count > 0:
            active_regions.append((region, count))
        else:
            suspicious_regions.append(region)
    except Exception as e:
        print(f"Can't reach {region}: {e}")

# If you have instances in a region you never approved, that's your breach fingerprint
print("Active regions (known good):", active_regions)
print("Zero-instance regions (potential abuse zone):", suspicious_regions)

// io.thecodeforge — cs-fundamentals tutorial

import boto3
from datetime import datetime, timezone, timedelta

session = boto3.Session()
client = session.client('cloudtrail')

# Look back 48 hours — attackers often strike in a narrow window
start_time = datetime.now(timezone.utc) - timedelta(hours=48)

# Get all GetObject events on the specific bucket
response = client.lookup_events(
    LookupAttributes=[
        {'AttributeKey': 'EventName', 'AttributeValue': 'GetObject'},
        {'AttributeKey': 'ResourceName', 'AttributeValue': 'arn:aws:s3:::infected-bucket-prod/*'}
    ],
    StartTime=start_time
)

for event in response['Events']:
    # Extract the actual user or role that made the call
    user = event.get('Username', 'unknown')
    # The IP address of the caller — critical for attribution
    source_ip = event['CloudTrailEvent']['sourceIPAddress']
    # Object key being accessed
    object_key = event['Resources'][0]['ResourceName']
    # Timestamp to the second
    timestamp = event['EventTime']

    print(f"{timestamp} | {user} | {source_ip} -> {object_key}")

# Sample output from a real incident — pattern is obvious
print("\n=== Suspicious pattern detected ===")

// io.thecodeforge — cs-fundamentals tutorial

import boto3

def find_public_buckets():
    s3 = boto3.client('s3')
    buckets = s3.list_buckets()['Buckets']
    
    for bucket in buckets:
        name = bucket['Name']
        try:
            acl = s3.get_bucket_acl(Bucket=name)
            for grant in acl['Grants']:
                uri = grant['Grantee'].get('URI', '')
                if 'AllUsers' in uri or 'AuthenticatedUsers' in uri:
                    print(f"[!] PUBLIC: {name}")
        except Exception as e:
            print(f"[!] ERROR: {name} - {e}")

if __name__ == "__main__":
    find_public_buckets()

// io.thecodeforge — cs-fundamentals tutorial

import boto3

ec2 = boto3.client('ec2')
instances = ec2.describe_instances(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}])

for reservation in instances['Reservations']:
    for inst in reservation['Instances']:
        inst_id = inst['InstanceId']
        metadata = inst.get('MetadataOptions', {})
        version = metadata.get('HttpTokens', 'not-set')
        if version != 'required':
            print(f"[!] {inst_id} uses IMDSv{version} — upgrade to IMDSv2 now")
        else:
            print(f"[OK] {inst_id} IMDSv2 enforced")

// io.thecodeforge — cs-fundamentals tutorial

import re
import sys

def scan_file_for_keys(filepath):
    patterns = [
        r'AKIA[0-9A-Z]{16}',  # Access Key ID
        r'(?i)aws_secret_access_key\s*[=:]\s*\S+',
        r'(?i)aws_access_key_id\s*[=:]\s*\S+',
    ]
    try:
        with open(filepath, 'r') as f:
            content = f.read()
        for i, pattern in enumerate(patterns):
            matches = re.findall(pattern, content)
            if matches:
                print(f"[!] LEAK in {filepath}: pattern {i+1}")
                return True
    except Exception as e:
        print(f"[-] Could not read {filepath}: {e}")
    return False

if __name__ == "__main__":
    for arg in sys.argv[1:]:
        scan_file_for_keys(arg)

// io.thecodeforge — cs-fundamentals tutorial

import boto3
import hashlib

def check_trojan(bucket, key):
    s3 = boto3.client('s3')
    obj = s3.get_object(Bucket=bucket, Key=key)
    content = obj['Body'].read()
    # Common Trojan hash: simulate malicious signature
    if hashlib.sha256(content).hexdigest() == 'abc123':
        print(f"ALERT: Trojan detected in s3://{bucket}/{key}")
        return True
    print("Clean object")
    return False

check_trojan('my-bucket', 'invoice.pdf')

// io.thecodeforge — cs-fundamentals tutorial

import boto3

ec2 = boto3.client('ec2')

def block_public_buckets():
    s3 = boto3.client('s3control')
    s3.put_public_access_block(
        AccountId='123456789012',
        PublicAccessBlockConfiguration={
            'BlockPublicAcls': True,
            'IgnorePublicAcls': True,
            'BlockPublicPolicy': True,
            'RestrictPublicBuckets': True
        }
    )
    print("Public access blocked across account")

block_public_buckets()

// io.thecodeforge — cs-fundamentals tutorial

import boto3

def enable_cloudtrail():
    ct = boto3.client('cloudtrail')
    ct.create_trail(
        Name='audit-trail',
        S3BucketName='my-cloudtrail-bucket',
        IsMultiRegionTrail=True,
        EnableLogFileValidation=True
    )
    ct.start_logging(Name='audit-trail')
    print("CloudTrail enabled across all regions")

enable_cloudtrail()