Spaces:
Sleeping
Sleeping
Update server/sentinel_env_environment.py
Browse files
server/sentinel_env_environment.py
CHANGED
|
@@ -51,7 +51,7 @@ class SentinelEnv:
|
|
| 51 |
|
| 52 |
def step(self, action: SentinelAction) -> Tuple[SentinelObservation, float, bool, Dict[str, Any]]:
|
| 53 |
self._state.step_count += 1
|
| 54 |
-
reward = 0.
|
| 55 |
done = False
|
| 56 |
msg = "No change detected."
|
| 57 |
task = self._state.current_task_id
|
|
@@ -60,30 +60,28 @@ class SentinelEnv:
|
|
| 60 |
if task == "easy-lockdown":
|
| 61 |
if action.command == "lockdown" and action.target_id == "s3-vault":
|
| 62 |
self.resources[0].status = "private"
|
| 63 |
-
reward =
|
| 64 |
done = True
|
| 65 |
msg = "SUCCESS: S3 Bucket is now private."
|
| 66 |
elif action.command == "scan" and action.target_id == "s3-vault":
|
| 67 |
-
# incremental reward for scanning before acting
|
| 68 |
reward = 0.1
|
| 69 |
msg = "Scan complete: S3 bucket is public. Apply lockdown."
|
| 70 |
else:
|
| 71 |
-
reward =
|
| 72 |
msg = "Wrong action. Try: command='lockdown', target_id='s3-vault'."
|
| 73 |
|
| 74 |
# ββ TASK 2: MEDIUM β Least Privilege ββ
|
| 75 |
elif task == "medium-access":
|
| 76 |
if action.command == "revoke_admin" and action.target_id == "user-dev-01":
|
| 77 |
self.resources[0].status = "read_only"
|
| 78 |
-
reward =
|
| 79 |
done = True
|
| 80 |
msg = "SUCCESS: Admin rights revoked. User set to Read-Only."
|
| 81 |
elif action.command == "audit" and action.target_id == "user-dev-01":
|
| 82 |
-
# incremental reward for auditing before revoking
|
| 83 |
reward = 0.1
|
| 84 |
msg = "Audit complete: user-dev-01 has admin_access. Revoke it."
|
| 85 |
else:
|
| 86 |
-
reward =
|
| 87 |
msg = "Wrong action. Try: command='revoke_admin', target_id='user-dev-01'."
|
| 88 |
|
| 89 |
# ββ TASK 3: HARD β Incident Response ββ
|
|
@@ -93,29 +91,28 @@ class SentinelEnv:
|
|
| 93 |
for r in self.resources
|
| 94 |
)
|
| 95 |
if action.command == "investigate" and action.target_id == "attacker-ip":
|
| 96 |
-
# incremental reward for investigating before blocking
|
| 97 |
reward = 0.1
|
| 98 |
-
msg = "Investigation complete: brute force
|
| 99 |
elif action.command == "block_ip" and action.target_id == "attacker-ip":
|
| 100 |
for r in self.resources:
|
| 101 |
if r.id == "attacker-ip":
|
| 102 |
r.status = "blocked"
|
| 103 |
-
reward = 0.
|
| 104 |
msg = "IP Blocked. Now close the open port on web-server."
|
| 105 |
elif action.command == "close_port" and action.target_id == "web-server":
|
| 106 |
if ip_blocked:
|
| 107 |
for r in self.resources:
|
| 108 |
if r.id == "web-server":
|
| 109 |
r.status = "port_22_closed"
|
| 110 |
-
reward = 0.
|
| 111 |
done = True
|
| 112 |
msg = "SUCCESS: Attack stopped and port secured."
|
| 113 |
else:
|
| 114 |
-
reward =
|
| 115 |
msg = "Port close failed. Block the attacker IP first."
|
| 116 |
else:
|
| 117 |
-
reward =
|
| 118 |
-
msg = "Unknown action.
|
| 119 |
|
| 120 |
self.logs.append(f"Step {self._state.step_count}: {msg} (reward={reward})")
|
| 121 |
return self._generate_observation(msg), reward, done, {}
|
|
|
|
| 51 |
|
| 52 |
def step(self, action: SentinelAction) -> Tuple[SentinelObservation, float, bool, Dict[str, Any]]:
|
| 53 |
self._state.step_count += 1
|
| 54 |
+
reward = 0.05
|
| 55 |
done = False
|
| 56 |
msg = "No change detected."
|
| 57 |
task = self._state.current_task_id
|
|
|
|
| 60 |
if task == "easy-lockdown":
|
| 61 |
if action.command == "lockdown" and action.target_id == "s3-vault":
|
| 62 |
self.resources[0].status = "private"
|
| 63 |
+
reward = 0.95
|
| 64 |
done = True
|
| 65 |
msg = "SUCCESS: S3 Bucket is now private."
|
| 66 |
elif action.command == "scan" and action.target_id == "s3-vault":
|
|
|
|
| 67 |
reward = 0.1
|
| 68 |
msg = "Scan complete: S3 bucket is public. Apply lockdown."
|
| 69 |
else:
|
| 70 |
+
reward = 0.05
|
| 71 |
msg = "Wrong action. Try: command='lockdown', target_id='s3-vault'."
|
| 72 |
|
| 73 |
# ββ TASK 2: MEDIUM β Least Privilege ββ
|
| 74 |
elif task == "medium-access":
|
| 75 |
if action.command == "revoke_admin" and action.target_id == "user-dev-01":
|
| 76 |
self.resources[0].status = "read_only"
|
| 77 |
+
reward = 0.95
|
| 78 |
done = True
|
| 79 |
msg = "SUCCESS: Admin rights revoked. User set to Read-Only."
|
| 80 |
elif action.command == "audit" and action.target_id == "user-dev-01":
|
|
|
|
| 81 |
reward = 0.1
|
| 82 |
msg = "Audit complete: user-dev-01 has admin_access. Revoke it."
|
| 83 |
else:
|
| 84 |
+
reward = 0.05
|
| 85 |
msg = "Wrong action. Try: command='revoke_admin', target_id='user-dev-01'."
|
| 86 |
|
| 87 |
# ββ TASK 3: HARD β Incident Response ββ
|
|
|
|
| 91 |
for r in self.resources
|
| 92 |
)
|
| 93 |
if action.command == "investigate" and action.target_id == "attacker-ip":
|
|
|
|
| 94 |
reward = 0.1
|
| 95 |
+
msg = "Investigation complete: brute force from attacker-ip. Block it."
|
| 96 |
elif action.command == "block_ip" and action.target_id == "attacker-ip":
|
| 97 |
for r in self.resources:
|
| 98 |
if r.id == "attacker-ip":
|
| 99 |
r.status = "blocked"
|
| 100 |
+
reward = 0.45
|
| 101 |
msg = "IP Blocked. Now close the open port on web-server."
|
| 102 |
elif action.command == "close_port" and action.target_id == "web-server":
|
| 103 |
if ip_blocked:
|
| 104 |
for r in self.resources:
|
| 105 |
if r.id == "web-server":
|
| 106 |
r.status = "port_22_closed"
|
| 107 |
+
reward = 0.50
|
| 108 |
done = True
|
| 109 |
msg = "SUCCESS: Attack stopped and port secured."
|
| 110 |
else:
|
| 111 |
+
reward = 0.05
|
| 112 |
msg = "Port close failed. Block the attacker IP first."
|
| 113 |
else:
|
| 114 |
+
reward = 0.05
|
| 115 |
+
msg = "Unknown action."
|
| 116 |
|
| 117 |
self.logs.append(f"Step {self._state.step_count}: {msg} (reward={reward})")
|
| 118 |
return self._generate_observation(msg), reward, done, {}
|