| import logging |
| from typing import List |
| from src.adware_manager import Payload |
|
|
| class PayloadManager: |
| def __init__(self, logger: logging.Logger): |
| """ |
| Initializes the PayloadManager with a logger. |
| |
| Args: |
| logger (Logging.Logger): The logger instance to use. |
| """ |
| self.logger = logger |
|
|
| def add_payload(self, name: str, description: str, file_path: str) -> Payload: |
| """ |
| Adds a new payload to the database. |
| |
| Args: |
| name (str): The name of the payload. |
| description (str): A description of the payload. |
| file_path (str): The path to the payload file. |
| |
| Returns: |
| Payload: The created payload object. |
| """ |
| payload = Payload(name=name, description=description, file_path=file_path) |
| payload.save() |
| self.logger.info(f"Payload '{name}' added successfully.") |
| return payload |
|
|
| def get_payload(self, payload_id: int) -> Payload: |
| """ |
| Retrieves a payload by its ID. |
| |
| Args: |
| payload_id (int): The ID of the payload to retrieve. |
| |
| Returns: |
| Payload: The payload object, or None if not found. |
| """ |
| payload = Payload.get_or_none(Payload.id == payload_id) |
| if not payload: |
| self.logger.warning(f"Payload with ID {payload_id} not found.") |
| return payload |
|
|
| def update_payload(self, payload_id: int, name: str = None, description: str = None, file_path: str = None) -> Payload: |
| """ |
| Updates an existing payload. |
| |
| Args: |
| payload_id (int): The ID of the payload to update. |
| name (str, optional): The new name of the payload. |
| description (str, optional): The new description of the payload. |
| file_path (str, optional): The new path to the payload file. |
| |
| Returns: |
| Payload: The updated payload object, or None if not found. |
| """ |
| payload = self.get_payload(payload_id) |
| if not payload: |
| return None |
|
|
| if name: |
| payload.name = name |
| if description: |
| payload.description = description |
| if file_path: |
| payload.file_path = file_path |
|
|
| payload.save() |
| self.logger.info(f"Payload '{payload.name}' updated successfully.") |
| return payload |
|
|
| def delete_payload(self, payload_id: int) -> bool: |
| """ |
| Deletes a payload by its ID. |
| |
| Args: |
| payload_id (int): The ID of the payload to delete. |
| |
| Returns: |
| bool: True if the payload was deleted, False otherwise. |
| """ |
| payload = self.get_payload(payload_id) |
| if not payload: |
| return False |
|
|
| payload.delete_instance() |
| self.logger.info(f"Payload '{payload.name}' deleted successfully.") |
| return True |
|
|
| def list_payloads(self) -> List[Payload]: |
| """ |
| Lists all available payloads. |
| |
| Returns: |
| List[Payload]: A list of all payload objects. |
| """ |
| payload_list = list(Payload.select()) |
| return payload_list |
|
|