Connect and authenticate to Metasploit RPC.
(self)
| 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.""" |
no test coverage detected