Helper to execute a module synchronously via console.
(
module_type: str,
module_name: str, # Can be full path or base name
module_options: Dict[str, Any],
command: str, # Typically 'exploit', 'run', or 'check'
payload_spec: Optional[Union[str, Dict[str, Any]]] = None,
timeout: int = LONG_CONSOLE_READ_TIMEOUT
)
| 601 | return {"status": "error", "message": f"Unexpected server error running {full_module_path}: {e}"} |
| 602 | |
| 603 | async def _execute_module_console( |
| 604 | module_type: str, |
| 605 | module_name: str, # Can be full path or base name |
| 606 | module_options: Dict[str, Any], |
| 607 | command: str, # Typically 'exploit', 'run', or 'check' |
| 608 | payload_spec: Optional[Union[str, Dict[str, Any]]] = None, |
| 609 | timeout: int = LONG_CONSOLE_READ_TIMEOUT |
| 610 | ) -> Dict[str, Any]: |
| 611 | """Helper to execute a module synchronously via console.""" |
| 612 | # Determine full path needed for 'use' command |
| 613 | if '/' not in module_name: |
| 614 | full_module_path = f"{module_type}/{module_name}" |
| 615 | else: |
| 616 | # Assume full path or relative path was given; ensure type prefix |
| 617 | if not module_name.startswith(module_type + '/'): |
| 618 | # e.g., got 'windows/x', type 'exploit' -> 'exploit/windows/x' |
| 619 | # e.g., got 'exploit/windows/x', type 'exploit' -> 'exploit/windows/x' (no change) |
| 620 | if not any(module_name.startswith(pfx + '/') for pfx in ['exploit', 'payload', 'post', 'auxiliary', 'encoder', 'nop']): |
| 621 | full_module_path = f"{module_type}/{module_name}" |
| 622 | else: # Already has a type prefix, use it as is |
| 623 | full_module_path = module_name |
| 624 | else: # Starts with correct type prefix |
| 625 | full_module_path = module_name |
| 626 | |
| 627 | logger.info(f"Executing {full_module_path} synchronously via console (command: {command})...") |
| 628 | |
| 629 | payload_name_for_log = None |
| 630 | payload_options_for_log = None |
| 631 | |
| 632 | async with get_msf_console() as console: |
| 633 | try: |
| 634 | setup_commands = [f"use {full_module_path}"] |
| 635 | |
| 636 | # Add module options |
| 637 | for key, value in module_options.items(): |
| 638 | val_str = str(value) |
| 639 | if isinstance(value, str) and any(c in val_str for c in [' ', '"', "'", '\\']): |
| 640 | val_str = shlex.quote(val_str) |
| 641 | elif isinstance(value, bool): |
| 642 | val_str = str(value).lower() # MSF console expects lowercase bools |
| 643 | setup_commands.append(f"set {key} {val_str}") |
| 644 | |
| 645 | # Add payload and payload options (if applicable) |
| 646 | if payload_spec: |
| 647 | payload_name = None |
| 648 | payload_options = {} |
| 649 | if isinstance(payload_spec, str): |
| 650 | payload_name = payload_spec |
| 651 | elif isinstance(payload_spec, dict) and 'name' in payload_spec: |
| 652 | payload_name = payload_spec['name'] |
| 653 | payload_options = payload_spec.get('options', {}) |
| 654 | |
| 655 | if payload_name: |
| 656 | payload_name_for_log = payload_name |
| 657 | payload_options_for_log = payload_options |
| 658 | # Need base name for 'set PAYLOAD' |
| 659 | if '/' in payload_name: |
| 660 | parts = payload_name.split('/') |
no test coverage detected