Spaces:
Runtime error
Runtime error
File size: 3,396 Bytes
4431575 d9194e2 4431575 3ff9f99 4431575 7cdbef9 4431575 5b93dcc 4431575 5b93dcc e33b52e 4431575 93b1bab 4431575 5b93dcc e33b52e 5b93dcc 7cdbef9 5b93dcc d9194e2 7cdbef9 5b93dcc 4431575 5b93dcc 7cdbef9 e33b52e 7cdbef9 e33b52e 7cdbef9 d9194e2 e33b52e 7cdbef9 d9194e2 e33b52e 7cdbef9 86252d9 d9194e2 e33b52e 86252d9 d9194e2 e33b52e 86252d9 e33b52e 93b1bab e33b52e 5b93dcc 93b1bab 7cdbef9 d9194e2 e33b52e 7cdbef9 e33b52e 5b93dcc 93b1bab 7cdbef9 d9194e2 e33b52e 7cdbef9 e33b52e 5b93dcc e33b52e e9b9ee2 e33b52e 7cdbef9 e33b52e e9b9ee2 7cdbef9 e33b52e 93b1bab e33b52e 5b93dcc d9194e2 5b93dcc 4431575 d9194e2 4431575 7cdbef9 d9194e2 4431575 7cdbef9 d9194e2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | import requests
from env.deception import deploy_honeypot, fake_database, block_attacker
from models import Observation, Reward
SERVER = "http://127.0.0.1:7860"
class DeceptionEnv:
def __init__(self):
self._state = {}
self.done = False
self.max_steps = 5
self.current_step = 0
self.detected = False
self.deployed = False
def reset(self):
requests.post(f"{SERVER}/reset")
self.done = False
self.current_step = 0
self.detected = False
self.deployed = False
logs = requests.get(f"{SERVER}/logs").json()
self._state = logs
return Observation(**self._state)
def step(self, action):
reward = 0.0
self.current_step += 1
logs = requests.get(f"{SERVER}/logs").json()
failed_logins = logs.get("failed_logins", 0)
requests_log = logs.get("requests", [])
# ---------------- Detect Attack ----------------
if action == "detect_attack":
detected_any = False
if failed_logins > 3:
reward += 0.15
detected_any = True
for r in requests_log:
if isinstance(r, dict) and r.get("type") == "port_scan":
reward += 0.15
detected_any = True
break
for r in requests_log:
if isinstance(r, dict) and r.get("type") == "sql_injection":
reward += 0.15
detected_any = True
break
for r in requests_log:
if isinstance(r, dict) and r.get("type") == "directory_traversal":
reward += 0.15
detected_any = True
break
if detected_any:
self.detected = True
else:
reward -= 0.10
# ---------------- Deploy Honeypot ----------------
elif action == "deploy_honeypot":
deploy_honeypot()
reward += 0.30
self.deployed = True
# ---------------- Fake Database ----------------
elif action == "fake_database":
fake_database()
reward += 0.20
self.deployed = True
# ---------------- Block Attacker ----------------
elif action == "block_ip":
if (
logs.get("suspicious_ips")
and self.detected
and self.deployed
and self.current_step >= 3
):
ip = logs["suspicious_ips"][-1]
block_attacker(ip)
reward += 0.70
self.done = True
else:
reward -= 0.10
# ---------------- Episode Boundary ----------------
if self.current_step >= self.max_steps:
self.done = True
reward = min(max(reward, 0.0), 1.0)
self._state = logs
observation = Observation(**self._state)
reward_obj = Reward(reward=reward, done=self.done)
return observation, reward_obj.reward, reward_obj.done, {}
def state(self):
return Observation(**self._state)
def action_space(self):
return [
"detect_attack",
"deploy_honeypot",
"fake_database",
"block_ip"
] |