Spaces:
Sleeping
Sleeping
SOC Simulator OpenEnv Space — deploy snapshot without binary eval plots (see GitHub for PNGs)
8c25076 | """ | |
| EASY TASK — Detect Phishing / Obvious Malicious Login | |
| ====================================================== | |
| Scenario: | |
| A corporate user receives a phishing email containing a credential-harvesting | |
| link. Shortly after, a login from an unusual foreign IP succeeds — a clear | |
| account-takeover indicator. The agent must detect either (or both) events | |
| and take appropriate action within 5 steps. | |
| Difficulty: EASY | |
| Max steps : 5 | |
| Deterministic seed: 42 | |
| """ | |
| from __future__ import annotations | |
| from datetime import datetime | |
| from typing import List | |
| from models import ( | |
| ActionType, Alert, EventType, SecurityEvent, SOCObservation, | |
| SystemState, ThreatLevel, | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Static synthetic data (deterministic — never randomised) | |
| # --------------------------------------------------------------------------- | |
| PHISHING_EVENT = SecurityEvent( | |
| event_id="EVT-001", | |
| timestamp=datetime(2024, 6, 15, 9, 3, 12), | |
| event_type=EventType.PHISHING_EMAIL, | |
| source_ip="185.220.101.47", # Known Tor exit node | |
| dest_ip=None, | |
| user_id="alice.chen", | |
| host_id="WORKSTATION-07", | |
| country="RU", | |
| details={ | |
| "email_subject": "URGENT: Reset your corporate VPN password NOW", | |
| "sender": "it-support@corp-help-desk.ru", | |
| "link_url": "http://corp-vpn-reset.malicious.ru/login", | |
| "attachment": None, | |
| "spf_pass": False, | |
| "dkim_pass": False, | |
| "header_mismatch": True, | |
| }, | |
| raw_log=( | |
| "2024-06-15T09:03:12Z [EMAIL-GATEWAY] INBOUND from=it-support@corp-help-desk.ru " | |
| "to=alice.chen@acmecorp.com subj='URGENT: Reset your corporate VPN password NOW' " | |
| "SPF=FAIL DKIM=FAIL link=http://corp-vpn-reset.malicious.ru/login" | |
| ), | |
| ) | |
| MALICIOUS_LOGIN_EVENT = SecurityEvent( | |
| event_id="EVT-002", | |
| timestamp=datetime(2024, 6, 15, 9, 17, 44), | |
| event_type=EventType.LOGIN_SUCCESS, | |
| source_ip="185.220.101.47", | |
| dest_ip="10.0.0.5", | |
| user_id="alice.chen", | |
| host_id="VPN-GATEWAY-01", | |
| country="RU", | |
| details={ | |
| "auth_method": "password", | |
| "usual_country": "US", | |
| "risk_signal": "impossible_travel", | |
| "time_since_phishing_email_min": 14, | |
| "mfa_bypassed": True, | |
| "session_id": "sess-8f2a1b", | |
| }, | |
| raw_log=( | |
| "2024-06-15T09:17:44Z [VPN-AUTH] SUCCESS user=alice.chen " | |
| "src=185.220.101.47 (RU) dst=10.0.0.5 method=password MFA=BYPASSED" | |
| ), | |
| ) | |
| BENIGN_LOGIN_EVENT = SecurityEvent( | |
| event_id="EVT-003", | |
| timestamp=datetime(2024, 6, 15, 9, 5, 0), | |
| event_type=EventType.LOGIN_SUCCESS, | |
| source_ip="203.0.113.55", | |
| dest_ip="10.0.0.5", | |
| user_id="bob.smith", | |
| host_id="VPN-GATEWAY-01", | |
| country="US", | |
| details={ | |
| "auth_method": "mfa", | |
| "usual_country": "US", | |
| "risk_signal": None, | |
| "mfa_bypassed": False, | |
| }, | |
| raw_log=( | |
| "2024-06-15T09:05:00Z [VPN-AUTH] SUCCESS user=bob.smith " | |
| "src=203.0.113.55 (US) dst=10.0.0.5 method=mfa MFA=OK" | |
| ), | |
| ) | |
| PHISHING_ALERT = Alert( | |
| alert_id="ALT-001", | |
| title="Suspected Phishing Email Received by alice.chen", | |
| description=( | |
| "An inbound email to alice.chen failed SPF and DKIM checks. " | |
| "The sender domain (corp-help-desk.ru) was registered 3 days ago. " | |
| "The embedded link resolves to a known credential-harvesting domain." | |
| ), | |
| threat_level=ThreatLevel.HIGH, | |
| related_events=["EVT-001"], | |
| anomaly_score=0.91, | |
| triggered_at=datetime(2024, 6, 15, 9, 3, 30), | |
| mitre_tactic="Initial Access", | |
| mitre_technique="T1566.001", # Spearphishing Link | |
| ) | |
| IMPOSSIBLE_TRAVEL_ALERT = Alert( | |
| alert_id="ALT-002", | |
| title="Impossible Travel: alice.chen logged in from Russia", | |
| description=( | |
| "alice.chen successfully authenticated from IP 185.220.101.47 (RU) — " | |
| "14 minutes after receiving a phishing email from the same IP block. " | |
| "MFA was bypassed, suggesting credential compromise." | |
| ), | |
| threat_level=ThreatLevel.CRITICAL, | |
| related_events=["EVT-001", "EVT-002"], | |
| anomaly_score=0.97, | |
| triggered_at=datetime(2024, 6, 15, 9, 18, 0), | |
| mitre_tactic="Initial Access", | |
| mitre_technique="T1078", # Valid Accounts | |
| ) | |
| # Ground truth: agent must block this IP and/or flag this user | |
| THREAT_IP = "185.220.101.47" | |
| THREAT_USER = "alice.chen" | |
| THREAT_EVENTS = {"EVT-001", "EVT-002"} | |
| class EasyTask: | |
| """ | |
| Task definition for the EASY scenario. | |
| The environment exposes phishing + malicious-login events from step 0. | |
| Correct responses: block_ip(target_ip) and/or flag_user(target_user). | |
| """ | |
| TASK_ID = "easy_phishing_login" | |
| MAX_STEPS = 5 | |
| DESCRIPTION = ( | |
| "A phishing email was sent to a user from a malicious IP. " | |
| "Shortly after, a successful VPN login from the same IP was detected " | |
| "with MFA bypassed. Identify and mitigate the threat." | |
| ) | |
| def __init__(self, **kwargs): | |
| # Allow RedAgent or custom_params to override the static defaults | |
| # Threat IP logic (Check if RedAgent mutated, OR if Live Intel fed real IPs) | |
| live_ips = kwargs.get("live_threat_ips", []) | |
| if kwargs.get("attacker_ip"): | |
| self.threat_ip = kwargs["attacker_ip"] | |
| elif live_ips: | |
| # Randomly pick one of the active live threat IPs if available | |
| import random | |
| self.threat_ip = random.choice(live_ips) | |
| else: | |
| self.threat_ip = THREAT_IP | |
| self.target_user = kwargs.get("target_user") or THREAT_USER | |
| def get_initial_observation(self, episode_id: str) -> SOCObservation: | |
| """Return the starting state of the easy task with dynamic substitutions.""" | |
| import copy | |
| # Deepcopy the static baseline structure so we don't pollute global models | |
| p_evt = PHISHING_EVENT.model_copy(deep=True) | |
| m_evt = MALICIOUS_LOGIN_EVENT.model_copy(deep=True) | |
| b_evt = BENIGN_LOGIN_EVENT.model_copy(deep=True) | |
| p_alert = PHISHING_ALERT.model_copy(deep=True) | |
| i_alert = IMPOSSIBLE_TRAVEL_ALERT.model_copy(deep=True) | |
| # Override values with the potentially mutated strings | |
| p_evt.source_ip = self.threat_ip | |
| p_evt.user_id = self.target_user | |
| p_evt.raw_log = p_evt.raw_log.replace(THREAT_IP, self.threat_ip).replace(THREAT_USER, self.target_user) | |
| m_evt.source_ip = self.threat_ip | |
| m_evt.user_id = self.target_user | |
| m_evt.raw_log = m_evt.raw_log.replace(THREAT_IP, self.threat_ip).replace(THREAT_USER, self.target_user) | |
| p_alert.title = p_alert.title.replace(THREAT_USER, self.target_user) | |
| p_alert.description = p_alert.description.replace(THREAT_USER, self.target_user) | |
| i_alert.title = i_alert.title.replace(THREAT_USER, self.target_user) | |
| i_alert.description = i_alert.description.replace(THREAT_IP, self.threat_ip).replace(THREAT_USER, self.target_user) | |
| return SOCObservation( | |
| done=False, | |
| reward=0.0, | |
| metadata={"episode_id": episode_id}, | |
| task_id=self.TASK_ID, | |
| step_number=0, | |
| timestamp=datetime(2024, 6, 15, 9, 18, 5), | |
| recent_events=[b_evt, p_evt, m_evt], | |
| active_alerts=[p_alert, i_alert], | |
| system_state=SystemState( | |
| active_connections=148, | |
| blocked_ips=[], | |
| flagged_users=[], | |
| isolated_hosts=[], | |
| open_incidents=2, | |
| risk_score=0.82, | |
| ), | |
| network_anomaly_score=0.84, | |
| hint=( | |
| "Pay close attention to the email gateway alert and the " | |
| "impossible-travel login from the same source IP." | |
| ), | |
| ) | |
| def get_threat_targets(self) -> dict: | |
| """Return dynamically-adjusted ground-truth targets for grading.""" | |
| return { | |
| "ip": self.threat_ip, | |
| "user": self.target_user, | |
| "event_ids": THREAT_EVENTS, | |
| "alert_ids": {"ALT-001", "ALT-002"}, | |
| } | |