bothari01/secops-env

1"""2Public Access Task - Medium Security Task.3 4Identify and fix overly permissive cloud storage (S3 bucket) access.5"""6 7import random8from typing import Any, Dict, List, Optional, Tuple9from secops_env.models import SecOpsAction, TaskDifficulty, ActionType10 11 12class PublicAccessTask:13    """14    Public Access Task - Medium Difficulty.15 16    Objective: Identify S3 buckets with public access and fix their permissions.17 18    Success Criteria:19    - All public buckets correctly identified20    - All public buckets set to private/block public access21    - No false positives (private buckets correctly left alone)22 23    Reward Structure:24    - +0.15 per correctly identified public resource25    - +0.4 bonus for fixing all public resources26    - -0.1 per missed public resource27    - -0.2 per false positive (marking private as public)28    """29 30    def __init__(self, difficulty: Optional[str] = None):31        """Initialize the public access task."""32        self.max_steps = 733        self.difficulty = (34            TaskDifficulty.MEDIUM if difficulty is None else TaskDifficulty(difficulty)35        )36        self.objective = "Identify S3 buckets with public access enabled and fix their permissions to block public access."37 38        self._resources = []39        self._expected_public = []40        self._identified_public = []41        self._fixed_buckets = []42        self._total_issues = 043 44    def generate_scenario(self) -> Dict[str, Any]:45        """Generate a public access scenario."""46        all_buckets = [47            {"name": "logs-prod-2024", "public": False, "type": "s3"},48            {"name": "customer-data-backup", "public": True, "type": "s3"},49            {"name": "website-static-assets", "public": True, "type": "s3"},50            {"name": "internal-reports-q4", "public": False, "type": "s3"},51            {"name": "marketing-assets", "public": False, "type": "s3"},52            {"name": "user-uploads-prod", "public": True, "type": "s3"},53            {"name": "config-backups", "public": False, "type": "s3"},54            {"name": "analytics-data", "public": True, "type": "s3"},55            {"name": "ml-models-prod", "public": False, "type": "s3"},56            {"name": "public-documentation", "public": True, "type": "s3"},57            {"name": "employee-records", "public": False, "type": "s3"},58            {"name": "temp-storage-share", "public": True, "type": "s3"},59            {"name": "application-logs", "public": False, "type": "s3"},60            {"name": "public-media-bucket", "public": True, "type": "s3"},61            {"name": "database-exports", "public": False, "type": "s3"},62            {"name": "api-keys-storage", "public": False, "type": "s3"},63            {"name": "cdn-assets-prod", "public": True, "type": "s3"},64            {"name": "user-avatars", "public": True, "type": "s3"},65            {"name": "backup-archive-2023", "public": False, "type": "s3"},66            {"name": "audit-logs-secure", "public": False, "type": "s3"},67            {"name": "mobile-app-assets", "public": True, "type": "s3"},68            {"name": "billing-invoices", "public": False, "type": "s3"},69            {"name": "shared-team-files", "public": True, "type": "s3"},70            {"name": "product-images", "public": True, "type": "s3"},71        ]72 73        scenario_buckets = random.sample(all_buckets, min(8, len(all_buckets)))74 75        self._resources = scenario_buckets76        self._expected_public = [b["name"] for b in scenario_buckets if b["public"]]77        self._identified_public = []78        self._fixed_buckets = []79        self._total_issues = len(self._expected_public)80 81        return {82            "resources": scenario_buckets,83            "instructions": "Identify buckets with public access and apply fixes to block public access.",84        }85 86    def execute_action(87        self, action: SecOpsAction, grader, task_data: Dict[str, Any]88    ) -> Tuple[float, str, bool, bool]:89        """90        Execute a public access action.91 92        Returns:93            Tuple of (reward, feedback, done, success)94        """95        reward = 0.0196        feedback = ""97        done = False98        success = False99 100        if action.action_type == ActionType.ANALYZE:101            feedback = (102                f"Analyzing {len(self._resources)} resources for public access..."103            )104 105        elif action.action_type == ActionType.IDENTIFY:106            if action.public_resources:107                self._identified_public = action.public_resources108                score = grader.grade_identification(109                    identified=self._identified_public, expected=self._expected_public110                )111                reward = score * 0.5112                feedback = f"Identified {len(self._identified_public)} public resources. Score: {score:.2f}"113            else:114                feedback = "No public resources identified."115 116        elif action.action_type == ActionType.APPLY_FIX:117            if action.fixed_resources:118                self._fixed_buckets = action.fixed_resources119                score = grader.grade_fix(120                    fixed=self._fixed_buckets,121                    expected_public=self._expected_public,122                    identified=self._identified_public,123                )124                reward = score * 0.3125                feedback = f"Applied fixes to {len(self._fixed_buckets)} resources. Score: {score:.2f}"126            else:127                feedback = "No fixes applied."128 129        elif action.action_type == ActionType.FINALIZE:130            if action.fixed_resources:131                self._fixed_buckets = action.fixed_resources132 133            if not self._fixed_buckets and action.public_resources:134                self._fixed_buckets = action.public_resources135 136            score = grader.grade_fix(137                fixed=self._fixed_buckets,138                expected_public=self._expected_public,139                identified=self._identified_public,140            )141            reward = score142 143            if score >= 0.9:144                feedback = f"Perfect! All public access fixed. Score: {score:.2f}"145                success = True146                done = True147            elif score >= 0.5:148                feedback = f"Good progress. Some buckets may still be public. Score: {score:.2f}"149            else:150                feedback = f"Action required. Public buckets remain unfixed. Score: {score:.2f}"151 152        else:153            feedback = f"Unknown action type: {action.action_type}"154 155        return reward, feedback, done, success156 157    def get_info(self) -> Dict[str, Any]:158        """Get current task information."""159        return {160            "difficulty": self.difficulty,161            "objective": self.objective,162            "detected_issues": self._identified_public,163            "fixed_issues": self._fixed_buckets,164            "total_issues": self._total_issues,165        }166 167    def get_state(self) -> Dict[str, Any]:168        """Get current task state."""169        return {170            "total_resources": len(self._resources),171            "expected_public": len(self._expected_public),172            "identified_public": len(self._identified_public),173            "fixed_buckets": len(self._fixed_buckets),174            "remaining_public": len(self._expected_public) - len(self._fixed_buckets),175        }176