Client for Metasploit Framework RPC API.
| 22 | |
| 23 | class MetasploitClient: |
| 24 | """Client for Metasploit Framework RPC API.""" |
| 25 | |
| 26 | def __init__( |
| 27 | self, |
| 28 | host: str = "127.0.0.1", |
| 29 | port: int = 55553, |
| 30 | username: str = "msf", |
| 31 | password: str = "", |
| 32 | ssl: bool = True, |
| 33 | ): |
| 34 | self.host = host |
| 35 | self.port = port |
| 36 | self.username = username |
| 37 | self.password = password |
| 38 | self.ssl = ssl |
| 39 | self.token: str | None = None |
| 40 | self.logger = get_logger("exploit.metasploit") |
| 41 | |
| 42 | @property |
| 43 | def base_url(self) -> str: |
| 44 | protocol = "https" if self.ssl else "http" |
| 45 | return f"{protocol}://{self.host}:{self.port}/api" |
| 46 | |
| 47 | async def connect(self) -> bool: |
| 48 | """Connect and authenticate to Metasploit RPC.""" |
| 49 | try: |
| 50 | result = await self._call("auth.login", [self.username, self.password]) |
| 51 | # msgpack may return byte or string keys depending on version — handle both |
| 52 | res_val = result.get("result") or result.get(b"result") |
| 53 | token = result.get("token") or result.get(b"token") |
| 54 | if res_val in ("success", b"success") and token: |
| 55 | self.token = token.decode() if isinstance(token, bytes) else token |
| 56 | self.logger.info("Connected to Metasploit RPC") |
| 57 | return True |
| 58 | else: |
| 59 | self.logger.error(f"Authentication failed: {result}") |
| 60 | return False |
| 61 | except Exception as e: |
| 62 | self.logger.error(f"Failed to connect to Metasploit: {e}") |
| 63 | return False |
| 64 | |
| 65 | @staticmethod |
| 66 | def _decode_result(raw: dict) -> dict: |
| 67 | """Normalise msgpack response — convert byte keys/values to strings.""" |
| 68 | out = {} |
| 69 | for k, v in raw.items(): |
| 70 | key = k.decode() if isinstance(k, bytes) else k |
| 71 | if isinstance(v, bytes): |
| 72 | val = v.decode(errors="replace") |
| 73 | elif isinstance(v, dict): |
| 74 | val = MetasploitClient._decode_result(v) |
| 75 | elif isinstance(v, list): |
| 76 | val = [ |
| 77 | MetasploitClient._decode_result(i) if isinstance(i, dict) |
| 78 | else (i.decode(errors="replace") if isinstance(i, bytes) else i) |
| 79 | for i in v |
| 80 | ] |
| 81 | else: |