bothari01/secops-env

1"""2Ghost User Task - Hard Security Task.3 4Identify and disable orphaned/inactive user accounts.5"""6 7import random8from datetime import datetime, timedelta9from typing import Any, Dict, List, Optional, Tuple10from secops_env.models import SecOpsAction, TaskDifficulty, ActionType11 12 13class GhostUserTask:14    """15    Ghost User Task - Hard Difficulty.16 17    Objective: Identify orphaned/inactive user accounts and disable them.18 19    Ghost User Criteria:20    - No login in 90+ days21    - No active resources associated22    - No role in recent deployments23    - Account created > 1 year ago and never used24 25    Success Criteria:26    - All ghost users correctly identified27    - No false positives (active users disabled)28    - All ghost users disabled29 30    Reward Structure:31    - +0.1 per correctly identified ghost user32    - +0.3 bonus for correctly disabling all identified users33    - -0.2 per incorrect disable (active user marked as ghost)34    """35 36    def __init__(self, difficulty: Optional[str] = None):37        """Initialize the ghost user task."""38        self.max_steps = 1039        self.difficulty = (40            TaskDifficulty.HARD if difficulty is None else TaskDifficulty(difficulty)41        )42        self.objective = "Analyze user accounts and identify orphaned/inactive 'ghost' users. Disable only confirmed ghost accounts without affecting active users."43 44        self._users = []45        self._expected_ghosts = []46        self._identified_ghosts = []47        self._disabled_users = []48        self._total_issues = 049 50    def generate_scenario(self) -> Dict[str, Any]:51        """Generate a ghost user scenario."""52        now = datetime.now()53 54        all_users = [55            {56                "username": "john.doe@company.com",57                "last_login": (now - timedelta(days=5)).isoformat(),58                "created": (now - timedelta(days=365)).isoformat(),59                "active_resources": ["ec2-prod-1", "rds-primary"],60                "recent_deployments": ["v2.1.0", "v2.0.0"],61                "is_ghost": False,62            },63            {64                "username": "jane.smith@company.com",65                "last_login": (now - timedelta(days=120)).isoformat(),66                "created": (now - timedelta(days=500)).isoformat(),67                "active_resources": [],68                "recent_deployments": [],69                "is_ghost": True,70            },71            {72                "username": "bob.wilson@company.com",73                "last_login": (now - timedelta(days=2)).isoformat(),74                "created": (now - timedelta(days=200)).isoformat(),75                "active_resources": ["lambda-processor"],76                "recent_deployments": ["v2.2.0"],77                "is_ghost": False,78            },79            {80                "username": "alice.chen@company.com",81                "last_login": (now - timedelta(days=200)).isoformat(),82                "created": (now - timedelta(days=700)).isoformat(),83                "active_resources": [],84                "recent_deployments": [],85                "is_ghost": True,86            },87            {88                "username": "temp.contractor@company.com",89                "last_login": (now - timedelta(days=95)).isoformat(),90                "created": (now - timedelta(days=100)).isoformat(),91                "active_resources": [],92                "recent_deployments": [],93                "is_ghost": True,94            },95            {96                "username": "service.pipeline@company.com",97                "last_login": (now - timedelta(days=1)).isoformat(),98                "created": (now - timedelta(days=800)).isoformat(),99                "active_resources": ["s3-pipeline-bucket"],100                "recent_deployments": ["v2.2.1"],101                "is_ghost": False,102            },103            {104                "username": "former.employee@company.com",105                "last_login": (now - timedelta(days=450)).isoformat(),106                "created": (now - timedelta(days=900)).isoformat(),107                "active_resources": [],108                "recent_deployments": [],109                "is_ghost": True,110            },111            {112                "username": "mike.johnson@company.com",113                "last_login": (now - timedelta(days=15)).isoformat(),114                "created": (now - timedelta(days=180)).isoformat(),115                "active_resources": ["eks-prod"],116                "recent_deployments": ["v2.1.5"],117                "is_ghost": False,118            },119            {120                "username": "intern.summer2023@company.com",121                "last_login": (now - timedelta(days=300)).isoformat(),122                "created": (now - timedelta(days=400)).isoformat(),123                "active_resources": [],124                "recent_deployments": [],125                "is_ghost": True,126            },127            {128                "username": "devops.automation@company.com",129                "last_login": (now - timedelta(days=3)).isoformat(),130                "created": (now - timedelta(days=1000)).isoformat(),131                "active_resources": ["ecs-cluster", "codedeploy"],132                "recent_deployments": ["v2.3.0", "v2.2.9"],133                "is_ghost": False,134            },135            {136                "username": "legacy.integration@company.com",137                "last_login": (now - timedelta(days=150)).isoformat(),138                "created": (now - timedelta(days=800)).isoformat(),139                "active_resources": [],140                "recent_deployments": [],141                "is_ghost": True,142            },143            {144                "username": "sarah.connor@company.com",145                "last_login": (now - timedelta(days=10)).isoformat(),146                "created": (now - timedelta(days=500)).isoformat(),147                "active_resources": ["s3-bucket"],148                "recent_deployments": ["v2.2.5"],149                "is_ghost": False,150            },151            {152                "username": "project.terminated@company.com",153                "last_login": (now - timedelta(days=500)).isoformat(),154                "created": (now - timedelta(days=1100)).isoformat(),155                "active_resources": [],156                "recent_deployments": [],157                "is_ghost": True,158            },159            {160                "username": "data.science@company.com",161                "last_login": (now - timedelta(days=7)).isoformat(),162                "created": (now - timedelta(days=400)).isoformat(),163                "active_resources": ["sagemaker-endpoint"],164                "recent_deployments": ["ml-pipeline-v3"],165                "is_ghost": False,166            },167            {168                "username": "vacation.replacement@company.com",169                "last_login": (now - timedelta(days=100)).isoformat(),170                "created": (now - timedelta(days=200)).isoformat(),171                "active_resources": [],172                "recent_deployments": [],173                "is_ghost": True,174            },175        ]176 177        scenario_users = random.sample(all_users, min(8, len(all_users)))178 179        self._users = scenario_users180        self._expected_ghosts = [u["username"] for u in scenario_users if u["is_ghost"]]181        self._identified_ghosts = []182        self._disabled_users = []183        self._total_issues = len(self._expected_ghosts)184 185        return {186            "users": scenario_users,187            "ghost_criteria": {188                "no_login_days": 90,189                "no_resources": True,190                "no_recent_deployments": True,191                "created_over_year_ago_unused": True,192            },193            "instructions": "Identify ghost users (inactive >90 days, no resources, no deployments) and disable them.",194        }195 196    def execute_action(197        self, action: SecOpsAction, grader, task_data: Dict[str, Any]198    ) -> Tuple[float, str, bool, bool]:199        """200        Execute a ghost user action.201 202        Returns:203            Tuple of (reward, feedback, done, success)204        """205        reward = 0.01206        feedback = ""207        done = False208        success = False209 210        if action.action_type == ActionType.ANALYZE:211            feedback = f"Analyzing {len(self._users)} user accounts..."212 213        elif action.action_type == ActionType.IDENTIFY:214            if action.ghost_users:215                self._identified_ghosts = action.ghost_users216                score = grader.grade_identification(217                    identified=self._identified_ghosts, expected=self._expected_ghosts218                )219                reward = score * 0.4220                feedback = f"Identified {len(self._identified_ghosts)} ghost users. Accuracy: {score:.2f}"221            else:222                feedback = "No ghost users identified."223 224        elif action.action_type == ActionType.APPLY_FIX:225            if action.disabled_users:226                self._disabled_users = action.disabled_users227                score = grader.grade_disabling(228                    disabled=self._disabled_users,229                    expected_ghosts=self._expected_ghosts,230                    identified_ghosts=self._identified_ghosts,231                )232                reward = score * 0.4233                feedback = (234                    f"Disabled {len(self._disabled_users)} users. Score: {score:.2f}"235                )236            else:237                feedback = "No users disabled."238 239        elif action.action_type == ActionType.FINALIZE:240            if action.disabled_users:241                self._disabled_users = action.disabled_users242 243            if not self._disabled_users and action.ghost_users:244                self._disabled_users = action.ghost_users245 246            score = grader.grade_disabling(247                disabled=self._disabled_users,248                expected_ghosts=self._expected_ghosts,249                identified_ghosts=self._identified_ghosts,250            )251            reward = score252 253            if score >= 0.9:254                feedback = (255                    f"Excellent! All ghost users properly handled. Score: {score:.2f}"256                )257                success = True258                done = True259            elif score >= 0.5:260                feedback = f"Good work. Some ghost users may remain. Score: {score:.2f}"261            else:262                feedback = f"Ghost users remain active. Score: {score:.2f}"263 264        else:265            feedback = f"Unknown action type: {action.action_type}"266 267        return reward, feedback, done, success268 269    def get_info(self) -> Dict[str, Any]:270        """Get current task information."""271        return {272            "difficulty": self.difficulty,273            "objective": self.objective,274            "detected_issues": self._identified_ghosts,275            "fixed_issues": self._disabled_users,276            "total_issues": self._total_issues,277        }278 279    def get_state(self) -> Dict[str, Any]:280        """Get current task state."""281        return {282            "total_users": len(self._users),283            "expected_ghosts": len(self._expected_ghosts),284            "identified_ghosts": len(self._identified_ghosts),285            "disabled_users": len(self._disabled_users),286            "remaining_ghosts": len(self._expected_ghosts) - len(self._disabled_users),287        }288