(self, handler)
| 110 | handler.wfile.write(json.dumps({"status": "error", "message": str(e)}).encode('utf-8')) |
| 111 | |
| 112 | def execute_manual_attack(self, handler): |
| 113 | try: |
| 114 | content_length = int(handler.headers['Content-Length']) |
| 115 | post_data = handler.rfile.read(content_length).decode('utf-8') |
| 116 | params = json.loads(post_data) |
| 117 | ip = params['ip'] |
| 118 | port = params['port'] |
| 119 | action_class = params['action'] |
| 120 | |
| 121 | self.logger.info(f"Received request to execute {action_class} on {ip}:{port}") |
| 122 | |
| 123 | # Charger les actions si ce n'est pas déjà fait |
| 124 | self.load_actions() |
| 125 | |
| 126 | action_instance = next((action for action in self.actions if action.action_name == action_class), None) |
| 127 | if action_instance is None: |
| 128 | raise Exception(f"Action class {action_class} not found") |
| 129 | |
| 130 | # Charger les données actuelles |
| 131 | current_data = self.shared_data.read_data() |
| 132 | row = next((r for r in current_data if r["IPs"] == ip), None) |
| 133 | |
| 134 | if row is None: |
| 135 | raise Exception(f"No data found for IP: {ip}") |
| 136 | |
| 137 | action_key = action_instance.action_name |
| 138 | self.logger.info(f"Executing {action_key} on {ip}:{port}") |
| 139 | result = action_instance.execute(ip, port, row, action_key) |
| 140 | |
| 141 | timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| 142 | if result == 'success': |
| 143 | row[action_key] = f'success_{timestamp}' |
| 144 | self.logger.info(f"Action {action_key} executed successfully on {ip}:{port}") |
| 145 | else: |
| 146 | row[action_key] = f'failed_{timestamp}' |
| 147 | self.logger.error(f"Action {action_key} failed on {ip}:{port}") |
| 148 | self.shared_data.write_data(current_data) |
| 149 | |
| 150 | handler.send_response(200) |
| 151 | handler.send_header("Content-type", "application/json") |
| 152 | handler.end_headers() |
| 153 | handler.wfile.write(json.dumps({"status": "success", "message": "Manual attack executed"}).encode('utf-8')) |
| 154 | except Exception as e: |
| 155 | self.logger.error(f"Error executing manual attack: {e}") |
| 156 | handler.send_response(500) |
| 157 | handler.send_header("Content-type", "application/json") |
| 158 | handler.end_headers() |
| 159 | handler.wfile.write(json.dumps({"status": "error", "message": str(e)}).encode('utf-8')) |
| 160 | |
| 161 | |
| 162 | def serve_logs(self, handler): |
nothing calls this directly
no test coverage detected