Make RPC call to Metasploit.
(self, method: str, params: list = None)
| 85 | |
| 86 | async def _call(self, method: str, params: list = None) -> dict: |
| 87 | """Make RPC call to Metasploit.""" |
| 88 | params = params or [] |
| 89 | |
| 90 | # Add token to authenticated calls |
| 91 | if self.token and not method.startswith("auth."): |
| 92 | params = [self.token] + params |
| 93 | |
| 94 | request_data = msgpack.packb([method] + params, use_bin_type=True) |
| 95 | |
| 96 | async with HTTPClient() as client: |
| 97 | response = await client.post( |
| 98 | self.base_url, |
| 99 | data=request_data, |
| 100 | headers={"Content-Type": "binary/message-pack"}, |
| 101 | ) |
| 102 | |
| 103 | if response.status_code == 200: |
| 104 | raw = msgpack.unpackb(response.content, raw=True) |
| 105 | return self._decode_result(raw) |
| 106 | else: |
| 107 | raise Exception(f"RPC error: {response.status_code}") |
| 108 | |
| 109 | async def get_version(self) -> dict: |
| 110 | """Get Metasploit version info.""" |
| 111 | return await self._call("core.version") |
no test coverage detected